feat: serve A2A ListTasks from the task store#2187
Conversation
0f862ef to
834dad4
Compare
QuentinBisson
changed the title
834dad4 to
bda1598
Compare
github-actions
Bot
added
enhancement
There was a problem hiding this comment.
Pull request overview
Adds a store-backed implementation of A2A ListTasks at the gateway layer so task listing no longer depends on downstream runtime transports (including legacy v0.3), while keeping all other A2A methods delegated to the existing passthrough handler.
Changes:
- Introduces
storeTaskQueryHandlerthat servesListTasksfrom the persistent task store with filtering, shaping, and pagination. - Adds a v0 (
tasks/list) JSON-RPC interceptor that bridges to the same store-backed implementation with legacy enum casing. - Wires the task store into the A2A handler mux via
NewA2AHttpMux(..., taskStore)and passes the DB client from app startup.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| go/core/pkg/app/app.go | Passes the DB client into the A2A mux so handlers can read from the persistent task store. |
| go/core/internal/a2a/task_query_store.go | Implements store-backed ListTasks with filtering, pagination tokening, and response shaping. |
| go/core/internal/a2a/task_query_store_test.go | Adds conformance-focused tests for pagination/filtering/shaping and wire casing differences. |
| go/core/internal/a2a/a2av0_tasks_list.go | Adds v0 tasks/list interceptor to serve store-backed listing on legacy wire. |
| go/core/internal/a2a/a2a_handler_mux.go | Wraps request handling to use the store-backed ListTasks and installs the v0 interceptor. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| legacyJSONRPCHandler := newV0TasksListInterceptor(a2av0.NewJSONRPCHandler(taskHandler), taskHandler) | ||
| v1JSONRPCHandler := a2asrv.NewJSONRPCHandler(taskHandler) |
There was a problem hiding this comment.
Fixed in 7274937: the interceptor is only installed when the store-backed handler is in use, so a nil store keeps the legacy wire's native method-not-found for tasks/list. Covered by TestWire_V0TasksListWithoutStoreIsMethodNotFound.
| if _, err := h.store.GetSession(ctx, contextID, userID); err != nil { | ||
| // Session does not exist or is not the caller's: no tasks to return. | ||
| return nil, nil | ||
| } |
There was a problem hiding this comment.
Fixed in 7274937: GetSession now wraps a database.ErrNotFound sentinel for missing rows, and collectUserTasks only maps that to an empty list; any other store error propagates. Covered by TestListTasks_BackendFailurePropagates.
| sessions, err := h.store.ListSessions(ctx, userID) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("list sessions: %w", err) | ||
| } | ||
| var all []*a2atype.Task | ||
| for _, s := range sessions { | ||
| tasks, err := h.store.ListTasksForSession(ctx, s.ID) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("list tasks for session %s: %w", s.ID, err) |
There was a problem hiding this comment.
Valid, but the proper fix wants a schema decision (task.data is TEXT, so SQL-side status filtering needs jsonb casts or generated columns). Split out as #2197 rather than growing this PR.
EItanya
left a comment
EItanya
left a comment
There was a problem hiding this comment.
I I just have the one scope comment, but I would also like @supreme-gg-gg to take a look as he is the SME on this code
| // ErrNotFound reports that the requested record does not exist (or is not | ||
| // visible to the given user). Match with errors.Is; implementations wrap it | ||
| // with call-site context. | ||
| var ErrNotFound = errors.New("record not found") |
There was a problem hiding this comment.
I think this makes sense overall since it's probably better to hide the pgx errors, however this would only be one instance of that with still many pgx errors around. Can you leave this change for a follow-up where we clean-up all pgx leakage?
|
I dropped the The read path now fails closed: |
kagent advertises an A2A server but is not conformant for ListTasks: the gateway's PassthroughRequestHandler proxies task queries to the agent runtime, whose legacy 0.3 transport returns ErrUnsupportedOperation on both wires. ListTasks has the wrong target: kagent persists tasks and is their source of truth, so it should answer this method itself. Add a store-backed request handler that answers ListTasks from the persistent task store, honoring the A2A v1.0.0 request/response semantics: contextId/status/statusTimestampAfter filtering, historyLength and includeArtifacts shaping, page-size default and cap, and a page token that is always present (empty on the final page). Every other method delegates to the passthrough proxy unchanged. Reads are scoped to the caller's user id so a caller cannot list another user's tasks. A share token grants one session, so when the request carries a ShareContext for exactly the requested contextId the owner's id is used instead (matching getEffectiveUserIDForSession in the session handlers); it never widens an all-sessions query. GetTask is intentionally left on the passthrough: it already works on both wires and resolves to the same persisted rows (the runtime's task store is a REST client of kagent's own /api/tasks), so store-backing it would change a working path for no functional gain. ListTasks is exposed on both wires the gateway serves. The v1 (a2asrv) wire is spec-native with uppercase TaskState. The legacy 0.3 (a2av0) wire predates the task-query methods, so tasks/list is bridged as a kagent compatibility extension answered by the same store-backed impl with lowercase TaskState; every other v0 method delegates to the upstream a2av0 handler unchanged, including the well-known agent-card route. Signed-off-by: QuentinBisson <quentin@giantswarm.io>
Without a task store the v0 interceptor is no longer installed, so the legacy wire keeps its native method-not-found for tasks/list instead of hitting the passthrough. GetSession now wraps database.ErrNotFound for missing rows; ListTasks only maps that sentinel to an empty result and propagates any other store error instead of silently returning no tasks. Signed-off-by: QuentinBisson <quentin@giantswarm.io>
…tinel Keep GetSession byte-identical to main so the pgx-error wrapping cleanup lands in its own follow-up. collectUserTasks now surfaces any GetSession error; a context the caller cannot see returns an error rather than an empty list. The unauthenticated path is unchanged (no user id, so the lookup is never attempted). Signed-off-by: QuentinBisson <quentin@giantswarm.io>
b675a30 to
988d358
Compare
pendingTasks was in-memory only. A gateway restart dropped the paused task's A2A id, so the next reply started a fresh turn instead of resolving the tool call, leaving the kagent session with a tool_use that had no tool_result. That corrupts the model history and every later message in the thread returns 400 until the session is reset. Persist the pending-task map to a JSON file next to the OBO link store and reload it on startup, so a paused approval is resumed after a restart. kagent owns the task; this only persists the client handle to resume it over A2A, and is removable once kagent serves A2A task discovery by contextId (kagent-dev/kagent#2187, tracked in #140).
|
Opened #2302 (draft) as the follow-up for #2197 — pushes the all-sessions |
## What Read-only share tokens can now read tasks over A2A (`ListTasks`, `GetTask`). They were blocked from every A2A method because `shareTokenMiddleware` rejects all non-GET requests on the A2A path, and A2A is JSON-RPC over POST. Read-only is the default share mode (`session_shares.go` defaults `read_only` to `true`), so the share-scoped task read path added in #2187 was unreachable for the common case. ## How Read-only enforcement for A2A moves from the transport verb to the request handler: - Reads pass through: `ListTasks`, `GetTask`, push-config get/list, `SubscribeToTask`. - Mutating methods are rejected under a read-only share via `requireWritableShare`: `SendMessage`, `SendStreamingMessage`, `CancelTask`, `CreateTaskPushConfig`, `DeleteTaskPushConfig`. - The session REST path keeps its existing verb-based read-only block. The middleware no longer blanket-blocks non-GET A2A requests for read-only shares, so the per-method guards are what stop a read-only share from mutating a session. Guarding at the handler covers both wires (v0 and v1 route the same methods to it). ## Behavior change A read-only share attempting a mutating A2A method now returns a JSON-RPC error result instead of an HTTP 403. Follows #2187 (store-backed `ListTasks`); branched from `main`, no code dependency on it. Signed-off-by: QuentinBisson <quentin@giantswarm.io> Co-authored-by: Eitan Yarmush <eitan.yarmush@solo.io>
|
Thanks for your help and reviews @EItanya :) |
The conformance gap
kagent advertises an A2A server but is not conformant for
ListTasks. The gateway'sPassthroughRequestHandler.ListTasks(go/core/internal/a2a/passthrough_handler.go) delegates to a downstream*a2aclient.Clientthat the registrar builds filtered to the agent pod's legacy 0.3 interface. The v0 compat transport'sListTasksreturnsErrUnsupportedOperation, so a client asking the gateway for a task list gets an error — on both wires (the registrar pins the 0.3 client regardless of the wire the client negotiates).Proxying the query to the runtime is the wrong target: kagent persists tasks and is their source of truth. In fact the runtime doesn't own task state at all — its task store (
KAgentTaskStore) is a REST client of kagent's own/api/tasks, so a proxied query would round-trip out to the pod and straight back to kagent's database.The fix
A store-backed request handler answers
ListTasksfrom the persistent task store, wrapping the passthrough for every other method. It honors the A2A v1.0.0 request/response semantics on both wires:contextId,status,statusTimestampAfterfilteringhistoryLengthtruncation (most-recent messages) andincludeArtifactsshaping (default false omits artifacts entirely)nextPageTokenalways present, empty string on the final page;totalSize/pageSizesetTasks are ordered by task id so the page-token offset stays stable across calls (session
updated_atreorders on writes; task ids do not). Status is filtered in Go from the stored Task JSON rather than with a JSONB query.User scoping and session sharing
Reads are scoped to the caller's
user_id— a caller cannot list another user's tasks. Session sharing is honored the same way the session handlers do it (getEffectiveUserIDForSession): a share token grants one session, so when the request carries aShareContextfor exactly the requestedcontextId, the owner's id is used instead. A share token never widens an all-sessions query (nocontextId) to the owner's whole account.Scope: ListTasks only, GetTask left on the passthrough
GetTaskalready works on both wires today and resolves to the same persisted rows the store-backed handler would read (runtime KAgentTaskStore.get()→GET /api/tasks/{id}→ the same Postgres row). Store-backing it would change a currently-working path for no functional gain, so this PR leavesGetTaskon the passthrough. (Separately, the existingGetTaskread path is not user-scoped; tightening that is a deliberate security change, tracked on its own, not bundled here.)Wire split: v1 conformant, v0 bridged
a2asrv) is the spec-native, fully A2A-1.0-conformant surface. MethodListTasks, uppercaseTaskState(TASK_STATE_INPUT_REQUIRED, ...).a2av0, legacy 0.3) has no nativetasks/list(0.3 predates the task-query methods). It is bridged as a deliberate kagent compatibility extension:tasks/listis intercepted and served by the same store-backed impl with lowercaseTaskState(input-required, ...) via thea2acompat/a2av0conversion helpers. Every other v0 method (and the well-known agent-card route) delegates to the upstream a2av0 handler unchanged.Both wires filter and paginate identically; only the enum casing differs. Serving
tasks/liston v0 is purely additive (the method errored before), so no existing v0 client behavior changes.Tests
Conformance-first, covering both wires with an in-memory store: multi-page pagination,
nextPageTokenalways present/empty on last page,includeArtifactson/off,status(incl.input-required) andcontextIdfilters,statusTimestampAfter,historyLengthtruncation, cross-user isolation, share-token access to an owner's session (and that it doesn't widen an all-sessions query), all-sessions aggregation, and end-to-end JSON-RPC drives asserting uppercaseTaskStateon v1 vs lowercase on v0 with identical filtering.go test ./core/internal/a2a/...green; gofmt/vet/golangci-lint clean.