diff --git a/go/api/database/client.go b/go/api/database/client.go index b5756f4877..60a231d69b 100644 --- a/go/api/database/client.go +++ b/go/api/database/client.go @@ -26,6 +26,18 @@ type QueryOptions struct { After time.Time OrderAsc bool // When true, order results by created_at ASC (chronological). Default is DESC (newest first). } + +// ListTasksForUserParams filters and paginates a user's tasks. SessionID empty +// lists across every session the user owns. Status empty (TaskStateUnspecified) +// disables the status filter; StatusTimestampAfter nil disables the timestamp +// filter. Results are ordered by task id. +type ListTasksForUserParams struct { + SessionID string + Status a2a.TaskState + StatusTimestampAfter *time.Time + Limit int + Offset int +} type LangGraphCheckpointTuple struct { Checkpoint *LangGraphCheckpoint Writes []*LangGraphCheckpointWrite @@ -62,6 +74,7 @@ type Client interface { ListTools(ctx context.Context) ([]Tool, error) ListFeedback(ctx context.Context, userID string) ([]Feedback, error) ListTasksForSession(ctx context.Context, sessionID string, userID string) ([]*a2a.Task, error) + ListTasksForUser(ctx context.Context, userID string, params ListTasksForUserParams) (tasks []*a2a.Task, total int, err error) ListSessions(ctx context.Context, userID string) ([]Session, error) ListSessionsForAgent(ctx context.Context, agentID string, userID string) ([]SessionWithShareToken, error) ListSessionsForAgentAllUsers(ctx context.Context, agentID string) ([]Session, error) diff --git a/go/core/internal/a2a/task_query_store.go b/go/core/internal/a2a/task_query_store.go index a121af1353..3f73e6a20f 100644 --- a/go/core/internal/a2a/task_query_store.go +++ b/go/core/internal/a2a/task_query_store.go @@ -1,11 +1,9 @@ package a2a import ( - "cmp" "context" "encoding/base64" "fmt" - "slices" "strconv" a2atype "github.com/a2aproject/a2a-go/v2/a2a" @@ -21,11 +19,11 @@ const ( // TaskStore is the subset of the persistent store ListTasks reads from. // *database.Client satisfies it. GetSession errors (including a missing or -// other-user session) surface to the caller. +// other-user session) surface to the caller. ListTasksForUser filters, orders, +// and paginates server-side and returns the full filtered count. type TaskStore interface { GetSession(ctx context.Context, sessionID, userID string) (*dbpkg.Session, error) - ListSessions(ctx context.Context, userID string) ([]dbpkg.Session, error) - ListTasksForSession(ctx context.Context, sessionID, userID string) ([]*a2atype.Task, error) + ListTasksForUser(ctx context.Context, userID string, params dbpkg.ListTasksForUserParams) (tasks []*a2atype.Task, total int, err error) } // storeTaskQueryHandler answers ListTasks from kagent's task store, which is @@ -75,34 +73,38 @@ func (h *storeTaskQueryHandler) ListTasks(ctx context.Context, req *a2atype.List return &a2atype.ListTasksResponse{Tasks: []*a2atype.Task{}, PageSize: pageSize}, nil } - tasks, err := h.collectUserTasks(ctx, userID, req.ContextID) + offset, err := decodePageToken(req.PageToken) if err != nil { - return nil, err + return nil, a2atype.NewError(a2atype.ErrInvalidParams, "invalid pageToken") } - filtered := filterTasks(tasks, req) - // Order by task id so the page-token offset is stable across calls: task - // ids are immutable, unlike session updated_at (which reorders on writes). - slices.SortFunc(filtered, func(a, b *a2atype.Task) int { return cmp.Compare(a.ID, b.ID) }) + // A single-session query fails closed: a missing or other-user session is an + // error, not an empty page. The join in ListTasksForUser also scopes by user, so + // this only distinguishes the error case and keeps share-context validation. + if req.ContextID != "" { + if _, err := h.store.GetSession(ctx, req.ContextID, userID); err != nil { + return nil, fmt.Errorf("get session %s: %w", req.ContextID, err) + } + } - offset, err := decodePageToken(req.PageToken) + tasks, total, err := h.store.ListTasksForUser(ctx, userID, dbpkg.ListTasksForUserParams{ + SessionID: req.ContextID, + Status: req.Status, + StatusTimestampAfter: req.StatusTimestampAfter, + Limit: pageSize, + Offset: offset, + }) if err != nil { - return nil, a2atype.NewError(a2atype.ErrInvalidParams, "invalid pageToken") - } - total := len(filtered) - if offset > total { - offset = total + return nil, err } - end := min(offset+pageSize, total) - page := filtered[offset:end] - shaped := make([]*a2atype.Task, 0, len(page)) - for _, t := range page { + shaped := make([]*a2atype.Task, 0, len(tasks)) + for _, t := range tasks { shaped = append(shaped, shapeTask(t, req.HistoryLength, req.IncludeArtifacts)) } nextToken := "" - if end < total { + if end := offset + len(tasks); end < total { nextToken = encodePageToken(end) } @@ -114,52 +116,6 @@ func (h *storeTaskQueryHandler) ListTasks(ctx context.Context, req *a2atype.List }, nil } -// collectUserTasks returns the caller's tasks, either for a single session -// (contextId) or across every session the user owns. Both paths are strictly -// scoped to userID. -func (h *storeTaskQueryHandler) collectUserTasks(ctx context.Context, userID, contextID string) ([]*a2atype.Task, error) { - if contextID != "" { - if _, err := h.store.GetSession(ctx, contextID, userID); err != nil { - return nil, fmt.Errorf("get session %s: %w", contextID, err) - } - return h.store.ListTasksForSession(ctx, contextID, userID) - } - - 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, userID) - if err != nil { - return nil, fmt.Errorf("list tasks for session %s: %w", s.ID, err) - } - all = append(all, tasks...) - } - return all, nil -} - -func filterTasks(tasks []*a2atype.Task, req *a2atype.ListTasksRequest) []*a2atype.Task { - filtered := make([]*a2atype.Task, 0, len(tasks)) - for _, t := range tasks { - if t == nil { - continue - } - if req.Status != a2atype.TaskStateUnspecified && t.Status.State != req.Status { - continue - } - if req.StatusTimestampAfter != nil { - ts := t.Status.Timestamp - if ts == nil || !ts.After(*req.StatusTimestampAfter) { - continue - } - } - filtered = append(filtered, t) - } - return filtered -} - // shapeTask returns a copy of task with history capped and artifacts included // only when requested. includeArtifacts defaults to false, in which case // artifacts are omitted entirely (nil slice + omitempty). diff --git a/go/core/internal/a2a/task_query_store_test.go b/go/core/internal/a2a/task_query_store_test.go index bab407d9a4..4dbb634144 100644 --- a/go/core/internal/a2a/task_query_store_test.go +++ b/go/core/internal/a2a/task_query_store_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "sort" "strings" "testing" "time" @@ -47,21 +48,44 @@ func (f *fakeTaskStore) GetSession(_ context.Context, sessionID, userID string) return &s, nil } -func (f *fakeTaskStore) ListSessions(_ context.Context, userID string) ([]dbpkg.Session, error) { - var out []dbpkg.Session - for _, s := range f.sessions { - if s.UserID == userID { - out = append(out, s) +// ListTasksForUser mirrors the SQL query's semantics in memory: scope to the +// user's own sessions (optionally one), filter status/timestamp, order by task +// id, and paginate, returning the full filtered count as total. +func (f *fakeTaskStore) ListTasksForUser(_ context.Context, userID string, params dbpkg.ListTasksForUserParams) ([]*a2atype.Task, int, error) { + var filtered []*a2atype.Task + for sessionID, tasks := range f.tasks { + s, ok := f.sessions[sessionID] + if !ok || s.UserID != userID { + continue + } + if params.SessionID != "" && sessionID != params.SessionID { + continue + } + for _, t := range tasks { + if t == nil { + continue + } + if params.Status != a2atype.TaskStateUnspecified && t.Status.State != params.Status { + continue + } + if params.StatusTimestampAfter != nil { + ts := t.Status.Timestamp + if ts == nil || !ts.After(*params.StatusTimestampAfter) { + continue + } + } + filtered = append(filtered, t) } } - return out, nil -} + sort.Slice(filtered, func(i, j int) bool { return filtered[i].ID < filtered[j].ID }) -func (f *fakeTaskStore) ListTasksForSession(_ context.Context, sessionID, userID string) ([]*a2atype.Task, error) { - if s, ok := f.sessions[sessionID]; !ok || s.UserID != userID { - return nil, nil + total := len(filtered) + offset := min(params.Offset, total) + end := total + if params.Limit > 0 && offset+params.Limit < end { + end = offset + params.Limit } - return f.tasks[sessionID], nil + return filtered[offset:end], total, nil } // fakeSession injects a user principal into the request context. @@ -269,12 +293,8 @@ func (f failingTaskStore) GetSession(context.Context, string, string) (*dbpkg.Se return nil, f.err } -func (f failingTaskStore) ListSessions(context.Context, string) ([]dbpkg.Session, error) { - return nil, f.err -} - -func (f failingTaskStore) ListTasksForSession(context.Context, string, string) ([]*a2atype.Task, error) { - return nil, f.err +func (f failingTaskStore) ListTasksForUser(context.Context, string, dbpkg.ListTasksForUserParams) ([]*a2atype.Task, int, error) { + return nil, 0, f.err } func TestListTasks_BackendFailurePropagates(t *testing.T) { diff --git a/go/core/internal/database/client_postgres.go b/go/core/internal/database/client_postgres.go index bcf9b39b96..b7e4ac9341 100644 --- a/go/core/internal/database/client_postgres.go +++ b/go/core/internal/database/client_postgres.go @@ -350,6 +350,57 @@ func (c *postgresClient) ListTasksForSession(ctx context.Context, sessionID, use return tasks, nil } +func (c *postgresClient) ListTasksForUser(ctx context.Context, userID string, params dbpkg.ListTasksForUserParams) ([]*a2a.Task, int, error) { + arg := dbgen.ListTasksForUserParams{ + UserID: userID, + StatusAfter: params.StatusTimestampAfter, + PageOffset: int32(params.Offset), + PageLimit: int32(params.Limit), + } + if params.SessionID != "" { + arg.SessionID = ¶ms.SessionID + } + if params.Status != a2a.TaskStateUnspecified { + status := string(params.Status) + arg.Status = &status + } + + rows, err := c.q.ListTasksForUser(ctx, arg) + if err != nil { + return nil, 0, fmt.Errorf("failed to list user tasks: %w", err) + } + + tasks := make([]*a2a.Task, 0, len(rows)) + for _, r := range rows { + task, err := parseVersionedTask(r.Data, r.ProtocolVersion) + if err != nil { + return nil, 0, fmt.Errorf("failed to parse task %s: %w", r.ID, err) + } + tasks = append(tasks, task) + } + total := 0 + if len(rows) > 0 { + total = int(rows[0].Total) + } + + // COUNT(*) OVER() rides on the returned rows, so a page requested past the + // end of the set comes back empty and carries no total. Recover it directly + // so the caller still sees the true filtered count. + if len(rows) == 0 && params.Offset > 0 { + count, err := c.q.CountTasksForUser(ctx, dbgen.CountTasksForUserParams{ + UserID: arg.UserID, + SessionID: arg.SessionID, + Status: arg.Status, + StatusAfter: arg.StatusAfter, + }) + if err != nil { + return nil, 0, fmt.Errorf("failed to count user tasks: %w", err) + } + total = int(count) + } + return tasks, total, nil +} + func (c *postgresClient) DeleteTask(ctx context.Context, taskID, userID string) error { if err := c.q.SoftDeleteTask(ctx, dbgen.SoftDeleteTaskParams{ID: taskID, UserID: &userID}); err != nil { return fmt.Errorf("failed to delete task %s: %w", taskID, err) diff --git a/go/core/internal/database/client_test.go b/go/core/internal/database/client_test.go index bf1959ca6b..78e6f41527 100644 --- a/go/core/internal/database/client_test.go +++ b/go/core/internal/database/client_test.go @@ -1121,6 +1121,114 @@ func TestSearchAgentMemoryConcurrentAccessCount(t *testing.T) { } } +// TestListTasksForUser exercises the ListTasksForUser SQL query against a real +// database: cross-user scoping via the session join, the optional single-session +// predicate, status filtering, the timestamp filter, and LIMIT/OFFSET pagination +// with a stable COUNT(*) OVER() total. +func TestListTasksForUser(t *testing.T) { + db := setupTestDB(t) + client := NewClient(db) + ctx := context.Background() + + for _, s := range []struct{ id, user string }{ + {"s1", "alice"}, {"s2", "alice"}, {"s3", "bob"}, + } { + require.NoError(t, client.StoreSession(ctx, &dbpkg.Session{ID: s.id, UserID: s.user})) + } + + early := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + late := time.Date(2026, 7, 8, 0, 0, 0, 0, time.UTC) + mkTask := func(id, contextID string, state a2a.TaskState, ts time.Time) *a2a.Task { + return &a2a.Task{ + ID: a2a.TaskID(id), + ContextID: contextID, + Status: a2a.TaskStatus{State: state, Timestamp: &ts}, + } + } + + require.NoError(t, client.StoreTask(ctx, mkTask("t1", "s1", a2a.TaskStateWorking, early), "alice")) + require.NoError(t, client.StoreTask(ctx, mkTask("t2", "s1", a2a.TaskStateCompleted, late), "alice")) + require.NoError(t, client.StoreTask(ctx, mkTask("t3", "s2", a2a.TaskStateWorking, late), "alice")) + require.NoError(t, client.StoreTask(ctx, mkTask("t4", "s3", a2a.TaskStateWorking, late), "bob")) + require.NoError(t, client.StoreTask(ctx, mkTask("t5", "s1", a2a.TaskStateInputRequired, late), "alice")) + + // t6 sits in alice's session s1 but is owned by mallory (a fresh insert has + // no ownership check against the session, unlike an UpsertTask conflict on + // an existing row): the session join alone must not be enough to hand it + // to alice. + require.NoError(t, client.StoreTask(ctx, mkTask("t6", "s1", a2a.TaskStateWorking, late), "mallory")) + + ids := func(tasks []*a2a.Task) []string { + out := make([]string, len(tasks)) + for i, tk := range tasks { + out[i] = string(tk.ID) + } + return out + } + + t.Run("all sessions ordered by id", func(t *testing.T) { + tasks, total, err := client.ListTasksForUser(ctx, "alice", dbpkg.ListTasksForUserParams{Limit: 50}) + require.NoError(t, err) + require.Equal(t, 4, total) + require.Equal(t, []string{"t1", "t2", "t3", "t5"}, ids(tasks)) + }) + + t.Run("cross-user isolation", func(t *testing.T) { + tasks, total, err := client.ListTasksForUser(ctx, "bob", dbpkg.ListTasksForUserParams{Limit: 50}) + require.NoError(t, err) + require.Equal(t, 1, total) + require.Equal(t, []string{"t4"}, ids(tasks)) + }) + + t.Run("session ownership alone does not grant a foreign-owned task", func(t *testing.T) { + tasks, total, err := client.ListTasksForUser(ctx, "alice", dbpkg.ListTasksForUserParams{SessionID: "s1", Limit: 50}) + require.NoError(t, err) + require.Equal(t, 3, total, "t6 belongs to mallory even though it lives in alice's session s1") + require.NotContains(t, ids(tasks), "t6") + }) + + t.Run("single session predicate", func(t *testing.T) { + tasks, total, err := client.ListTasksForUser(ctx, "alice", dbpkg.ListTasksForUserParams{SessionID: "s1", Limit: 50}) + require.NoError(t, err) + require.Equal(t, 3, total) + require.Equal(t, []string{"t1", "t2", "t5"}, ids(tasks)) + }) + + t.Run("status filter", func(t *testing.T) { + tasks, total, err := client.ListTasksForUser(ctx, "alice", dbpkg.ListTasksForUserParams{Status: a2a.TaskStateWorking, Limit: 50}) + require.NoError(t, err) + require.Equal(t, 2, total) + require.Equal(t, []string{"t1", "t3"}, ids(tasks)) + }) + + t.Run("status timestamp after", func(t *testing.T) { + cutoff := time.Date(2026, 7, 5, 0, 0, 0, 0, time.UTC) + tasks, total, err := client.ListTasksForUser(ctx, "alice", dbpkg.ListTasksForUserParams{StatusTimestampAfter: &cutoff, Limit: 50}) + require.NoError(t, err) + require.Equal(t, 3, total) // t1 is early and excluded + require.Equal(t, []string{"t2", "t3", "t5"}, ids(tasks)) + }) + + t.Run("pagination with stable total", func(t *testing.T) { + p1, total, err := client.ListTasksForUser(ctx, "alice", dbpkg.ListTasksForUserParams{Limit: 2, Offset: 0}) + require.NoError(t, err) + require.Equal(t, 4, total) + require.Equal(t, []string{"t1", "t2"}, ids(p1)) + + p2, total, err := client.ListTasksForUser(ctx, "alice", dbpkg.ListTasksForUserParams{Limit: 2, Offset: 2}) + require.NoError(t, err) + require.Equal(t, 4, total) + require.Equal(t, []string{"t3", "t5"}, ids(p2)) + }) + + t.Run("offset past end keeps the true total", func(t *testing.T) { + tasks, total, err := client.ListTasksForUser(ctx, "alice", dbpkg.ListTasksForUserParams{Limit: 2, Offset: 10}) + require.NoError(t, err) + require.Empty(t, tasks) + require.Equal(t, 4, total, "an empty over-range page must still report the full count") + }) +} + // TestSingleRowReadsMapMissingToErrNotFound verifies that every single-row // read maps the driver's no-rows error to dbpkg.ErrNotFound, so callers can // match with errors.Is without importing pgx. diff --git a/go/core/internal/database/gen/querier.go b/go/core/internal/database/gen/querier.go index 7355ec8a4b..3eb73ce75b 100644 --- a/go/core/internal/database/gen/querier.go +++ b/go/core/internal/database/gen/querier.go @@ -10,6 +10,11 @@ import ( type Querier interface { CountAgentInstanceTasks(ctx context.Context, arg CountAgentInstanceTasksParams) (int64, error) + // The full filtered count for ListTasksForUser, independent of LIMIT/OFFSET. Used + // to recover total when a requested page lands past the end of the set (an empty + // page carries no COUNT(*) OVER()). The WHERE clause is identical to + // ListTasksForUser and must stay in sync with it. + CountTasksForUser(ctx context.Context, arg CountTasksForUserParams) (int64, error) CreateAgentInstanceShare(ctx context.Context, arg CreateAgentInstanceShareParams) (AgentInstanceShare, error) CreateSessionShare(ctx context.Context, arg CreateSessionShareParams) (SessionShare, error) DeleteAgentInstance(ctx context.Context, id string) error @@ -82,6 +87,25 @@ type Querier interface { ListSessionsForAgent(ctx context.Context, arg ListSessionsForAgentParams) ([]ListSessionsForAgentRow, error) ListSessionsForAgentAllUsers(ctx context.Context, agentID *string) ([]Session, error) ListTasksForSession(ctx context.Context, arg ListTasksForSessionParams) ([]Task, error) + // Lists a user's tasks across every session they own (or a single session when + // session_id is set), filtering, ordering, and paginating server-side. total is + // the COUNT(*) OVER() of the full filtered set, before LIMIT/OFFSET. + // + // The session join alone isn't sufficient ownership proof: a task's own + // user_id (with the NULL-owner fallback documented above, for rows predating + // the owner column) is checked too, matching GetTask/ListTasksForSession, so a + // foreign task parked in the caller's session is never handed to the caller. + // + // status matches task.data's persisted state string (e.g. 'TASK_STATE_WORKING'). + // Post-v1.0 cutover, task.data is always v1-shaped, so a single spelling is + // enough. data is always a JSON object (json.Marshal output), so ::jsonb never + // errors; the timestamp cast is guarded by a CASE (ordered evaluation) against + // a present-but-malformed value. + // + // COUNT(*) OVER() rides on the returned rows, so a page past the end of the set + // carries no total; callers recover it with CountTasksForUser, whose WHERE clause + // must stay identical to this one. + ListTasksForUser(ctx context.Context, arg ListTasksForUserParams) ([]ListTasksForUserRow, error) ListToolServers(ctx context.Context) ([]Toolserver, error) ListTools(ctx context.Context) ([]Tool, error) ListToolsForServer(ctx context.Context, arg ListToolsForServerParams) ([]Tool, error) diff --git a/go/core/internal/database/gen/tasks.sql.go b/go/core/internal/database/gen/tasks.sql.go index bde0409573..81f3f5968b 100644 --- a/go/core/internal/database/gen/tasks.sql.go +++ b/go/core/internal/database/gen/tasks.sql.go @@ -7,8 +7,59 @@ package dbgen import ( "context" + "time" ) +const countTasksForUser = `-- name: CountTasksForUser :one +SELECT COUNT(*) +FROM task +JOIN session ON session.id = task.session_id +WHERE session.user_id = $1 + AND (task.user_id = $1 OR (task.user_id IS NULL AND $1 = ( + SELECT MIN(s.user_id) FROM session s + WHERE s.id = task.session_id AND s.created_at <= task.created_at + HAVING COUNT(DISTINCT s.user_id) = 1))) + AND task.deleted_at IS NULL + AND session.deleted_at IS NULL + AND ($2::text IS NULL OR task.session_id = $2) + AND ( + $3::text IS NULL + OR (task.data::jsonb -> 'status' ->> 'state') = $3 + ) + AND ( + $4::timestamptz IS NULL + OR ( + CASE + WHEN (task.data::jsonb -> 'status' ->> 'timestamp') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]+)?(Z|[+-][0-9]{2}:?[0-9]{2})$' + THEN (task.data::jsonb -> 'status' ->> 'timestamp')::timestamptz + END + ) > $4 + ) +` + +type CountTasksForUserParams struct { + UserID string + SessionID *string + Status *string + StatusAfter *time.Time +} + +// The full filtered count for ListTasksForUser, independent of LIMIT/OFFSET. Used +// to recover total when a requested page lands past the end of the set (an empty +// page carries no COUNT(*) OVER()). The WHERE clause is identical to +// ListTasksForUser and must stay in sync with it. +func (q *Queries) CountTasksForUser(ctx context.Context, arg CountTasksForUserParams) (int64, error) { + row := q.db.QueryRow(ctx, countTasksForUser, + arg.UserID, + arg.SessionID, + arg.Status, + arg.StatusAfter, + ) + var count int64 + err := row.Scan(&count) + return count, err +} + const getTask = `-- name: GetTask :one SELECT id, created_at, updated_at, deleted_at, data, session_id, protocol_version, user_id FROM task @@ -109,6 +160,111 @@ func (q *Queries) ListTasksForSession(ctx context.Context, arg ListTasksForSessi return items, nil } +const listTasksForUser = `-- name: ListTasksForUser :many +SELECT task.id, task.created_at, task.updated_at, task.deleted_at, task.data, task.session_id, task.protocol_version, task.user_id, COUNT(*) OVER() AS total +FROM task +JOIN session ON session.id = task.session_id +WHERE session.user_id = $1 + AND (task.user_id = $1 OR (task.user_id IS NULL AND $1 = ( + SELECT MIN(s.user_id) FROM session s + WHERE s.id = task.session_id AND s.created_at <= task.created_at + HAVING COUNT(DISTINCT s.user_id) = 1))) + AND task.deleted_at IS NULL + AND session.deleted_at IS NULL + AND ($2::text IS NULL OR task.session_id = $2) + AND ( + $3::text IS NULL + OR (task.data::jsonb -> 'status' ->> 'state') = $3 + ) + AND ( + $4::timestamptz IS NULL + OR ( + CASE + WHEN (task.data::jsonb -> 'status' ->> 'timestamp') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]+)?(Z|[+-][0-9]{2}:?[0-9]{2})$' + THEN (task.data::jsonb -> 'status' ->> 'timestamp')::timestamptz + END + ) > $4 + ) +ORDER BY task.id +LIMIT $6::int OFFSET $5::int +` + +type ListTasksForUserParams struct { + UserID string + SessionID *string + Status *string + StatusAfter *time.Time + PageOffset int32 + PageLimit int32 +} + +type ListTasksForUserRow struct { + ID string + CreatedAt *time.Time + UpdatedAt *time.Time + DeletedAt *time.Time + Data string + SessionID *string + ProtocolVersion *string + UserID *string + Total int64 +} + +// Lists a user's tasks across every session they own (or a single session when +// session_id is set), filtering, ordering, and paginating server-side. total is +// the COUNT(*) OVER() of the full filtered set, before LIMIT/OFFSET. +// +// The session join alone isn't sufficient ownership proof: a task's own +// user_id (with the NULL-owner fallback documented above, for rows predating +// the owner column) is checked too, matching GetTask/ListTasksForSession, so a +// foreign task parked in the caller's session is never handed to the caller. +// +// status matches task.data's persisted state string (e.g. 'TASK_STATE_WORKING'). +// Post-v1.0 cutover, task.data is always v1-shaped, so a single spelling is +// enough. data is always a JSON object (json.Marshal output), so ::jsonb never +// errors; the timestamp cast is guarded by a CASE (ordered evaluation) against +// a present-but-malformed value. +// +// COUNT(*) OVER() rides on the returned rows, so a page past the end of the set +// carries no total; callers recover it with CountTasksForUser, whose WHERE clause +// must stay identical to this one. +func (q *Queries) ListTasksForUser(ctx context.Context, arg ListTasksForUserParams) ([]ListTasksForUserRow, error) { + rows, err := q.db.Query(ctx, listTasksForUser, + arg.UserID, + arg.SessionID, + arg.Status, + arg.StatusAfter, + arg.PageOffset, + arg.PageLimit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListTasksForUserRow + for rows.Next() { + var i ListTasksForUserRow + if err := rows.Scan( + &i.ID, + &i.CreatedAt, + &i.UpdatedAt, + &i.DeletedAt, + &i.Data, + &i.SessionID, + &i.ProtocolVersion, + &i.UserID, + &i.Total, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const softDeleteTask = `-- name: SoftDeleteTask :exec UPDATE task SET deleted_at = NOW() WHERE task.id = $1 AND task.deleted_at IS NULL diff --git a/go/core/internal/database/queries/tasks.sql b/go/core/internal/database/queries/tasks.sql index 14e20793f8..f6cdf504bf 100644 --- a/go/core/internal/database/queries/tasks.sql +++ b/go/core/internal/database/queries/tasks.sql @@ -38,6 +38,82 @@ WHERE task.session_id = $1 AND task.deleted_at IS NULL HAVING COUNT(DISTINCT s.user_id) = 1))) ORDER BY created_at ASC; +-- name: ListTasksForUser :many +-- Lists a user's tasks across every session they own (or a single session when +-- session_id is set), filtering, ordering, and paginating server-side. total is +-- the COUNT(*) OVER() of the full filtered set, before LIMIT/OFFSET. +-- +-- The session join alone isn't sufficient ownership proof: a task's own +-- user_id (with the NULL-owner fallback documented above, for rows predating +-- the owner column) is checked too, matching GetTask/ListTasksForSession, so a +-- foreign task parked in the caller's session is never handed to the caller. +-- +-- status matches task.data's persisted state string (e.g. 'TASK_STATE_WORKING'). +-- Post-v1.0 cutover, task.data is always v1-shaped, so a single spelling is +-- enough. data is always a JSON object (json.Marshal output), so ::jsonb never +-- errors; the timestamp cast is guarded by a CASE (ordered evaluation) against +-- a present-but-malformed value. +-- +-- COUNT(*) OVER() rides on the returned rows, so a page past the end of the set +-- carries no total; callers recover it with CountTasksForUser, whose WHERE clause +-- must stay identical to this one. +SELECT task.*, COUNT(*) OVER() AS total +FROM task +JOIN session ON session.id = task.session_id +WHERE session.user_id = @user_id + AND (task.user_id = @user_id OR (task.user_id IS NULL AND @user_id = ( + SELECT MIN(s.user_id) FROM session s + WHERE s.id = task.session_id AND s.created_at <= task.created_at + HAVING COUNT(DISTINCT s.user_id) = 1))) + AND task.deleted_at IS NULL + AND session.deleted_at IS NULL + AND (sqlc.narg('session_id')::text IS NULL OR task.session_id = sqlc.narg('session_id')) + AND ( + sqlc.narg('status')::text IS NULL + OR (task.data::jsonb -> 'status' ->> 'state') = sqlc.narg('status') + ) + AND ( + sqlc.narg('status_after')::timestamptz IS NULL + OR ( + CASE + WHEN (task.data::jsonb -> 'status' ->> 'timestamp') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]+)?(Z|[+-][0-9]{2}:?[0-9]{2})$' + THEN (task.data::jsonb -> 'status' ->> 'timestamp')::timestamptz + END + ) > sqlc.narg('status_after') + ) +ORDER BY task.id +LIMIT @page_limit::int OFFSET @page_offset::int; + +-- name: CountTasksForUser :one +-- The full filtered count for ListTasksForUser, independent of LIMIT/OFFSET. Used +-- to recover total when a requested page lands past the end of the set (an empty +-- page carries no COUNT(*) OVER()). The WHERE clause is identical to +-- ListTasksForUser and must stay in sync with it. +SELECT COUNT(*) +FROM task +JOIN session ON session.id = task.session_id +WHERE session.user_id = @user_id + AND (task.user_id = @user_id OR (task.user_id IS NULL AND @user_id = ( + SELECT MIN(s.user_id) FROM session s + WHERE s.id = task.session_id AND s.created_at <= task.created_at + HAVING COUNT(DISTINCT s.user_id) = 1))) + AND task.deleted_at IS NULL + AND session.deleted_at IS NULL + AND (sqlc.narg('session_id')::text IS NULL OR task.session_id = sqlc.narg('session_id')) + AND ( + sqlc.narg('status')::text IS NULL + OR (task.data::jsonb -> 'status' ->> 'state') = sqlc.narg('status') + ) + AND ( + sqlc.narg('status_after')::timestamptz IS NULL + OR ( + CASE + WHEN (task.data::jsonb -> 'status' ->> 'timestamp') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]+)?(Z|[+-][0-9]{2}:?[0-9]{2})$' + THEN (task.data::jsonb -> 'status' ->> 'timestamp')::timestamptz + END + ) > sqlc.narg('status_after') + ); + -- UpsertTask returns the upserted id, or no rows when the write was rejected: -- the id belongs to another user, or it belongs to a soft-deleted task (a -- deleted id is never updated or resurrected, it stays burned). Callers map