From 4d7c1659d3d835bfe6347efc73a3ebec90d2abbf Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 11 Oct 2026 15:15:13 -0400 Subject: [PATCH] feat(store): status board tables, patch merge and board reads (RIG-5170) Adds agent_boards and agent_board_tasks to 0001_init.sql, with tenant RLS and updated_at triggers. ApplyBoardPatch merges a presence-aware patch in one transaction: it enforces the caps, defaults a new task to ACTIVE, clears focus when the focused task is removed, merges subagents by id (dropping the oldest settled one past the cap), and leaves freshness alone for automatic patches. Board reads, task search, viewer-clipped asks by author, open authored refs, and coordinate lookups for pull requests and issues serve the later board derivation. Ask gains a stored Ref. Ledger-impact: none Co-authored-by: Matt Wilkinson --- go/internal/store/agent_boards.go | 865 ++++++++++++++++++ go/internal/store/agent_boards_pgtest_test.go | 835 +++++++++++++++++ go/internal/store/blocks.go | 5 +- go/internal/store/db/agent_boards.sql.go | 668 ++++++++++++++ go/internal/store/db/models.go | 30 + go/internal/store/db/querier.go | 16 + go/internal/store/migrations/0001_init.sql | 38 + go/internal/store/queries/agent_boards.sql | 232 +++++ go/internal/store/rls_pgtest_test.go | 1 + go/internal/store/store.go | 2 + go/internal/store/types.go | 2 + 11 files changed, 2692 insertions(+), 2 deletions(-) create mode 100644 go/internal/store/agent_boards.go create mode 100644 go/internal/store/agent_boards_pgtest_test.go create mode 100644 go/internal/store/db/agent_boards.sql.go create mode 100644 go/internal/store/queries/agent_boards.sql diff --git a/go/internal/store/agent_boards.go b/go/internal/store/agent_boards.go new file mode 100644 index 000000000..2c262028e --- /dev/null +++ b/go/internal/store/agent_boards.go @@ -0,0 +1,865 @@ +package store + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "math" + "strconv" + "strings" + "time" + "unicode/utf8" + + "github.com/jackc/pgx/v5" + + "github.com/RigelBuild/compass/go/internal/store/db" +) + +const ( + maxBoardTasks = 30 + maxBoardRefs = 10 + maxBoardWaiting = 5 + maxBoardSubagents = 20 + maxBoardAsks = 20 + maxBoardSearchAgents = 100 + maxBoardTitleChars = 120 + maxBoardNoteChars = 500 + maxBoardHaltChars = 200 + maxBoardWhatChars = 200 +) + +// BoardTaskPatch updates only fields that are present in the patch. +type BoardTaskPatch struct { + Key string + Title *string + PrimaryRef *string + Note *string + State *int16 + Refs *[]string + Waiting *[]byte + UpsertSubagents []byte +} + +// BoardPatch is a presence-aware update to one agent's board. +type BoardPatch struct { + Upsert []BoardTaskPatch + Remove []string + Focus *string + Halt *BoardHalt + ClearHalt bool + Turn uint64 + Automatic bool + NowUnixMs int64 +} + +// BoardHalt describes why the agent has stopped or is waiting. +type BoardHalt struct { + Kind int16 + Ref string + Reason string +} + +// BoardRow is the stored board metadata and its ordered tasks. +type BoardRow struct { + Agent AccountID + Focus string + Halt BoardHalt + WrittenTurn uint64 + WrittenAtUnixMs int64 + Tasks []BoardTaskRow +} + +// BoardTaskRow is one stored task and its agent-authored annotations. +type BoardTaskRow struct { + Key string + Title string + PrimaryRef string + Note string + Refs []string + State int16 + Waiting []byte + Subagents []byte + WrittenAtUnixMs int64 +} + +// BoardTaskFilter selects tasks across an optional agent set. +type BoardTaskFilter struct { + Agents []AccountID + Query string + Ref string + WaitingOn int16 + State int16 + Halt int16 + Limit int +} + +// AskRow is a visible ask message authored by one agent. +type AskRow struct { + AskID string + MessageID string + ChannelID string + Ref string + Header string + Answered bool + PostedAtUnixMs int64 +} + +// AuthoredRef is an open forge coordinate owned by an agent. +type AuthoredRef struct { + Coord ForgeCoord + Kind ForgeArtifactKind +} + +// ApplyBoardPatch validates and applies one complete patch atomically. +func (s *Store) ApplyBoardPatch(ctx context.Context, agent AccountID, p BoardPatch) (result BoardRow, retErr error) { + if agent == "" { + return BoardRow{}, fmt.Errorf("%w: agent account id is required", ErrInvalidArgument) + } + if !p.Automatic && p.Turn > math.MaxInt64 { + return BoardRow{}, fmt.Errorf("%w: board turn %d exceeds signed BIGINT range", ErrInvalidArgument, p.Turn) + } + + tx, err := s.beginTenantTx(ctx) + if err != nil { + return BoardRow{}, fmt.Errorf("store: begin board patch: %w", err) + } + defer func() { + if rollbackErr := tx.Rollback(ctx); rollbackErr != nil && !errors.Is(rollbackErr, pgx.ErrTxClosed) { + retErr = errors.Join(retErr, fmt.Errorf("store: roll back board patch: %w", rollbackErr)) + } + }() + qtx := s.q.WithTx(tx) + if err := lockBoardAgent(ctx, qtx, agent); err != nil { + return BoardRow{}, err + } + current, positions, err := boardForAgent(ctx, qtx, agent) + if err != nil { + return BoardRow{}, err + } + if p.Automatic && emptyBoardPatch(p) { + return current, nil + } + + next, positions, err := mergeBoardPatch(current, positions, p) + if err != nil { + return BoardRow{}, err + } + if len(next.Tasks) > maxBoardTasks { + return BoardRow{}, fmt.Errorf("%w: board has %d tasks; maximum is %d", ErrInvalidArgument, len(next.Tasks), maxBoardTasks) + } + if next.Focus != "" && !boardHasTask(next.Tasks, next.Focus) { + return BoardRow{}, fmt.Errorf("%w: focus task %q is not on the board", ErrInvalidArgument, next.Focus) + } + if err := upsertBoardMetadata(ctx, qtx, agent, current.Focus, next, p); err != nil { + return BoardRow{}, err + } + if err := persistBoardTasks(ctx, qtx, agent, next, positions, p); err != nil { + return BoardRow{}, err + } + if err := tx.Commit(ctx); err != nil { + return BoardRow{}, fmt.Errorf("store: commit board patch: %w", err) + } + return next, nil +} + +func emptyBoardPatch(p BoardPatch) bool { + return len(p.Upsert) == 0 && len(p.Remove) == 0 && p.Focus == nil && p.Halt == nil && !p.ClearHalt +} + +func lockBoardAgent(ctx context.Context, q *db.Queries, agent AccountID) error { + _, err := q.LockBoardAgent(ctx, string(agent)) + if noRows(err) { + return fmt.Errorf("%w: agent account %q does not exist", ErrInvalidArgument, agent) + } + if err != nil { + return fmt.Errorf("store: lock board agent: %w", err) + } + return nil +} + +func boardForAgent(ctx context.Context, q *db.Queries, agent AccountID) (BoardRow, map[string]int32, error) { + out := BoardRow{Agent: agent, Tasks: make([]BoardTaskRow, 0)} + positions := make(map[string]int32) + boards, err := q.GetAgentBoards(ctx, []string{string(agent)}) + if err != nil { + return BoardRow{}, nil, fmt.Errorf("store: read agent board: %w", err) + } + if len(boards) > 0 { + r := boards[0] + out.Focus = r.FocusTaskKey + out.Halt = BoardHalt{Kind: r.HaltKind, Ref: r.HaltRef, Reason: r.HaltReason} + out.WrittenTurn = uint64(r.WrittenTurn) //nolint:gosec // schema writes nonnegative turns within signed BIGINT + out.WrittenAtUnixMs = r.WrittenAtUnixMs + } + tasks, err := q.GetAgentBoardTasks(ctx, []string{string(agent)}) + if err != nil { + return BoardRow{}, nil, fmt.Errorf("store: read agent board tasks: %w", err) + } + for _, row := range tasks { + refs := row.Refs + if len(refs) == 0 { + refs = nil + } + positions[row.TaskKey] = row.Position + out.Tasks = append(out.Tasks, BoardTaskRow{ + Key: row.TaskKey, Title: row.Title, PrimaryRef: row.PrimaryRef, Note: row.Note, + Refs: refs, State: row.State, Waiting: row.Waiting, Subagents: row.Subagents, + WrittenAtUnixMs: row.WrittenAtUnixMs, + }) + } + return out, positions, nil +} + +func mergeBoardPatch(current BoardRow, positions map[string]int32, p BoardPatch) (BoardRow, map[string]int32, error) { + if err := validateHalt(current.Halt); err != nil { + return BoardRow{}, nil, err + } + next := current + next.Tasks = append([]BoardTaskRow(nil), current.Tasks...) + for _, row := range current.Tasks { + if err := validateTask(row); err != nil { + return BoardRow{}, nil, fmt.Errorf("store: stored board task %q is invalid: %w", row.Key, err) + } + } + if err := removeBoardTasks(&next, positions, p.Remove); err != nil { + return BoardRow{}, nil, err + } + if err := applyBoardTaskUpserts(&next, positions, p); err != nil { + return BoardRow{}, nil, err + } + if p.Focus != nil { + next.Focus = *p.Focus + } + if err := applyBoardHalt(&next, p); err != nil { + return BoardRow{}, nil, err + } + if !p.Automatic { + next.WrittenTurn = p.Turn + next.WrittenAtUnixMs = p.NowUnixMs + } + return next, positions, nil +} + +func applyBoardTaskUpserts(next *BoardRow, positions map[string]int32, p BoardPatch) error { + for _, patch := range p.Upsert { + if patch.Key == "" { + return fmt.Errorf("%w: task key is required", ErrInvalidArgument) + } + index := boardTaskIndex(next.Tasks, patch.Key) + if index < 0 { + if patch.Title == nil { + return fmt.Errorf("%w: new task %q requires a title", ErrInvalidArgument, patch.Key) + } + if len(next.Tasks) == maxBoardTasks { + return fmt.Errorf("%w: board exceeds %d tasks", ErrInvalidArgument, maxBoardTasks) + } + position, err := nextBoardPosition(positions) + if err != nil { + return err + } + positions[patch.Key] = position + next.Tasks = append(next.Tasks, BoardTaskRow{ + Key: patch.Key, State: 1, Refs: []string{}, Waiting: []byte("[]"), Subagents: []byte("[]"), + }) + index = len(next.Tasks) - 1 + } + updated, err := applyTaskPatch(next.Tasks[index], patch) + if err != nil { + return err + } + if !p.Automatic { + updated.WrittenAtUnixMs = p.NowUnixMs + } + next.Tasks[index] = updated + } + return nil +} + +func boardTaskIndex(tasks []BoardTaskRow, key string) int { + for i := range tasks { + if tasks[i].Key == key { + return i + } + } + return -1 +} + +func applyBoardHalt(next *BoardRow, p BoardPatch) error { + if p.Halt != nil { + if err := validateHalt(*p.Halt); err != nil { + return err + } + next.Halt = *p.Halt + } + if p.ClearHalt { + next.Halt = BoardHalt{} + } + return nil +} + +func removeBoardTasks(next *BoardRow, positions map[string]int32, keys []string) error { + remove := make(map[string]struct{}, len(keys)) + for _, key := range keys { + if key == "" { + return fmt.Errorf("%w: removed task key is required", ErrInvalidArgument) + } + remove[key] = struct{}{} + delete(positions, key) + if next.Focus == key { + next.Focus = "" + } + } + kept := next.Tasks[:0] + for _, task := range next.Tasks { + if _, removed := remove[task.Key]; !removed { + kept = append(kept, task) + } + } + next.Tasks = kept + return nil +} + +func nextBoardPosition(positions map[string]int32) (int32, error) { + var highest int32 = -1 + for _, position := range positions { + if position > highest { + highest = position + } + } + if highest == math.MaxInt32 { + return 0, fmt.Errorf("%w: task position exceeds signed INT range", ErrInvalidArgument) + } + return highest + 1, nil +} + +func applyTaskPatch(row BoardTaskRow, patch BoardTaskPatch) (BoardTaskRow, error) { + if patch.Title != nil { + row.Title = *patch.Title + } + if patch.PrimaryRef != nil { + row.PrimaryRef = *patch.PrimaryRef + } + if patch.Note != nil { + row.Note = *patch.Note + } + if patch.State != nil { + row.State = *patch.State + } + if patch.Refs != nil { + row.Refs = append([]string(nil), (*patch.Refs)...) + } + if patch.Waiting != nil { + waiting, err := validateWaiting(*patch.Waiting) + if err != nil { + return BoardTaskRow{}, err + } + row.Waiting = waiting + } + if len(patch.UpsertSubagents) != 0 { + subagents, err := mergeSubagents(row.Subagents, patch.UpsertSubagents) + if err != nil { + return BoardTaskRow{}, err + } + row.Subagents = subagents + } + if err := validateTask(row); err != nil { + return BoardTaskRow{}, err + } + return row, nil +} + +func validateTask(row BoardTaskRow) error { + if row.Title == "" { + return fmt.Errorf("%w: task %q title is required", ErrInvalidArgument, row.Key) + } + if n := utf8.RuneCountInString(row.Title); n > maxBoardTitleChars { + return fmt.Errorf("%w: task %q title has %d characters; maximum is %d", ErrInvalidArgument, row.Key, n, maxBoardTitleChars) + } + if n := utf8.RuneCountInString(row.Note); n > maxBoardNoteChars { + return fmt.Errorf("%w: task %q note has %d characters; maximum is %d", ErrInvalidArgument, row.Key, n, maxBoardNoteChars) + } + if len(row.Refs) > maxBoardRefs { + return fmt.Errorf("%w: task %q has %d refs; maximum is %d", ErrInvalidArgument, row.Key, len(row.Refs), maxBoardRefs) + } + if row.State < 1 || row.State > 5 { + return fmt.Errorf("%w: task %q has invalid state %d", ErrInvalidArgument, row.Key, row.State) + } + if _, err := validateWaiting(row.Waiting); err != nil { + return err + } + subagents, err := validateSubagents(row.Subagents, false) + if err != nil { + return err + } + if len(subagents) > maxBoardSubagents { + return fmt.Errorf("%w: task %q has %d subagents; maximum is %d", ErrInvalidArgument, row.Key, len(subagents), maxBoardSubagents) + } + return nil +} + +func validateHalt(halt BoardHalt) error { + if halt.Kind < 0 || halt.Kind > 3 { + return fmt.Errorf("%w: invalid halt kind %d", ErrInvalidArgument, halt.Kind) + } + if n := utf8.RuneCountInString(halt.Reason); n > maxBoardHaltChars { + return fmt.Errorf("%w: halt reason has %d characters; maximum is %d", ErrInvalidArgument, n, maxBoardHaltChars) + } + return nil +} + +func validateWaiting(raw []byte) ([]byte, error) { + var entries []json.RawMessage + if err := json.Unmarshal(raw, &entries); err != nil || entries == nil { + return nil, fmt.Errorf("%w: waiting must be a JSON array", ErrInvalidArgument) + } + if len(entries) > maxBoardWaiting { + return nil, fmt.Errorf("%w: task has %d waiting entries; maximum is %d", ErrInvalidArgument, len(entries), maxBoardWaiting) + } + for _, entry := range entries { + var value map[string]json.RawMessage + if err := json.Unmarshal(entry, &value); err != nil || value == nil { + return nil, fmt.Errorf("%w: waiting entry must be an object", ErrInvalidArgument) + } + rawOn, found := value["on"] + var on string + if !found || json.Unmarshal(rawOn, &on) != nil { + return nil, fmt.Errorf("%w: waiting on is required", ErrInvalidArgument) + } + if on != "BOARD_WAITING_ON_OPERATOR" && on != "BOARD_WAITING_ON_AGENT" && on != "BOARD_WAITING_ON_EXTERNAL" { + return nil, fmt.Errorf("%w: waiting on must be a recognized BOARD_WAITING_ON value", ErrInvalidArgument) + } + if rawWhat, found := value["what"]; found { + var what string + if err := json.Unmarshal(rawWhat, &what); err != nil { + return nil, fmt.Errorf("%w: waiting what must be a string", ErrInvalidArgument) + } + if n := utf8.RuneCountInString(what); n > maxBoardWhatChars { + return nil, fmt.Errorf("%w: waiting what has %d characters; maximum is %d", ErrInvalidArgument, n, maxBoardWhatChars) + } + } + } + if len(entries) == 0 { + return []byte("[]"), nil + } + return append([]byte(nil), raw...), nil +} + +func mergeSubagents(existing, updates []byte) ([]byte, error) { + current, err := validateSubagents(existing, false) + if err != nil { + return nil, fmt.Errorf("store: decode stored subagents: %w", err) + } + incoming, err := validateSubagents(updates, true) + if err != nil { + return nil, err + } + positions := make(map[string]int, len(current)+len(incoming)) + for i, item := range current { + id, err := subagentID(item) + if err != nil { + return nil, fmt.Errorf("store: decode stored subagent id: %w", err) + } + positions[id] = i + } + for _, item := range incoming { + id, err := subagentID(item) + if err != nil { + return nil, fmt.Errorf("%w: subagent id is required", ErrInvalidArgument) + } + if index, found := positions[id]; found { + current[index] = item + continue + } + positions[id] = len(current) + current = append(current, item) + } + for len(current) > maxBoardSubagents { + oldest := -1 + oldestAt := int64(math.MaxInt64) + for i, item := range current { + settled, startedAt := subagentSettled(item) + if settled && (oldest < 0 || startedAt < oldestAt) { + oldest = i + oldestAt = startedAt + } + } + if oldest < 0 { + return nil, fmt.Errorf("%w: task has more than %d unsettled subagents", ErrInvalidArgument, maxBoardSubagents) + } + current = append(current[:oldest], current[oldest+1:]...) + } + if current == nil { + return []byte("[]"), nil + } + out, err := json.Marshal(current) + if err != nil { + return nil, fmt.Errorf("store: encode subagents: %w", err) + } + return out, nil +} + +func validateSubagents(raw []byte, callerInput bool) ([]json.RawMessage, error) { + var entries []json.RawMessage + if err := json.Unmarshal(raw, &entries); err != nil || entries == nil { + if callerInput { + return nil, fmt.Errorf("%w: subagents must be a JSON array", ErrInvalidArgument) + } + return nil, errors.New("subagents must be a JSON array") + } + seen := make(map[string]struct{}, len(entries)) + for _, entry := range entries { + var value map[string]json.RawMessage + if err := json.Unmarshal(entry, &value); err != nil || value == nil { + if callerInput { + return nil, fmt.Errorf("%w: subagent entry must be an object", ErrInvalidArgument) + } + return nil, errors.New("subagent entry must be an object") + } + id, err := subagentID(entry) + if err != nil || id == "" { + if callerInput { + return nil, fmt.Errorf("%w: subagent id is required", ErrInvalidArgument) + } + return nil, errors.New("subagent id is required") + } + if callerInput { + if _, duplicate := seen[id]; duplicate { + return nil, fmt.Errorf("%w: duplicate subagent id %q", ErrInvalidArgument, id) + } + seen[id] = struct{}{} + } + } + return entries, nil +} + +func subagentID(raw json.RawMessage) (string, error) { + var value struct { + ID string `json:"id"` + } + if err := json.Unmarshal(raw, &value); err != nil { + return "", err + } + return value.ID, nil +} + +func subagentSettled(raw json.RawMessage) (bool, int64) { + var value struct { + Status string `json:"status"` + EndedAtUnixMs json.RawMessage `json:"endedAtUnixMs"` + EndedAtUnixMsSnake json.RawMessage `json:"ended_at_unix_ms"` + StartedAtUnixMs json.RawMessage `json:"startedAtUnixMs"` + StartedAtUnixMsSnake json.RawMessage `json:"started_at_unix_ms"` + } + if err := json.Unmarshal(raw, &value); err != nil { + return false, 0 + } + started, _ := jsonInt64(value.StartedAtUnixMs) + if started == 0 { + started, _ = jsonInt64(value.StartedAtUnixMsSnake) + } + ended, _ := jsonInt64(value.EndedAtUnixMs) + if ended == 0 { + ended, _ = jsonInt64(value.EndedAtUnixMsSnake) + } + switch value.Status { + case "completed", "failed", "aborted": + return true, started + default: + return ended != 0, started + } +} + +func jsonInt64(raw json.RawMessage) (int64, error) { + var value int64 + if err := json.Unmarshal(raw, &value); err == nil { + return value, nil + } + var text string + if err := json.Unmarshal(raw, &text); err != nil { + return 0, err + } + return strconv.ParseInt(text, 10, 64) +} + +func boardHasTask(tasks []BoardTaskRow, key string) bool { + for _, task := range tasks { + if task.Key == key { + return true + } + } + return false +} + +func upsertBoardMetadata(ctx context.Context, q *db.Queries, agent AccountID, currentFocus string, next BoardRow, p BoardPatch) error { + if err := q.UpsertAgentBoard(ctx, db.UpsertAgentBoardParams{ + AgentAccountID: string(agent), + Column2: p.Focus != nil || next.Focus != currentFocus, + Column3: next.Focus, + Column4: p.Halt != nil, + Column5: next.Halt.Kind, + Column6: next.Halt.Ref, + Column7: next.Halt.Reason, + Column8: p.ClearHalt, + Column9: p.Automatic, + Column10: int64(p.Turn), //nolint:gosec // validated against MaxInt64 before tx + Column11: p.NowUnixMs, + }); err != nil { + return fmt.Errorf("store: upsert board metadata: %w", err) + } + return nil +} + +func persistBoardTasks(ctx context.Context, q *db.Queries, agent AccountID, next BoardRow, positions map[string]int32, p BoardPatch) error { + if len(p.Remove) > 0 { + if err := q.DeleteAgentBoardTasks(ctx, db.DeleteAgentBoardTasksParams{AgentAccountID: string(agent), Column2: p.Remove}); err != nil { + return fmt.Errorf("store: delete board tasks: %w", err) + } + } + if len(p.Upsert) == 0 { + return nil + } + rows := make(map[string]BoardTaskRow, len(next.Tasks)) + for _, task := range next.Tasks { + rows[task.Key] = task + } + for _, patch := range p.Upsert { + task, ok := rows[patch.Key] + if !ok { + continue + } + if task.Refs == nil { + task.Refs = []string{} + } + if task.Waiting == nil { + task.Waiting = []byte("[]") + } + if task.Subagents == nil { + task.Subagents = []byte("[]") + } + if err := q.UpsertAgentBoardTask(ctx, db.UpsertAgentBoardTaskParams{ + AgentAccountID: string(agent), TaskKey: task.Key, Title: task.Title, + PrimaryRef: task.PrimaryRef, Refs: task.Refs, State: task.State, + Note: task.Note, Waiting: task.Waiting, Subagents: task.Subagents, + Position: positions[task.Key], WrittenAtUnixMs: task.WrittenAtUnixMs, + }); err != nil { + return fmt.Errorf("store: upsert board task %q: %w", task.Key, err) + } + } + return nil +} + +// AgentBoards returns stored board rows for the requested agents. +func (s *Store) AgentBoards(ctx context.Context, agents []AccountID) (map[AccountID]BoardRow, error) { + ids := uniqueAccountStrings(agents) + out := make(map[AccountID]BoardRow, len(ids)) + if len(ids) == 0 { + return out, nil + } + tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly}) + if err != nil { + return nil, fmt.Errorf("store: begin board snapshot: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + if err := armTx(ctx, s, tx); err != nil { + return nil, err + } + qtx := s.q.WithTx(tx) + boards, err := qtx.GetAgentBoards(ctx, ids) + if err != nil { + return nil, fmt.Errorf("store: read agent boards: %w", err) + } + if s.afterBoardMetaRead != nil { + s.afterBoardMetaRead() + } + for _, row := range boards { + id := AccountID(row.AgentAccountID) + if row.WrittenTurn < 0 { + return nil, fmt.Errorf("store: board %q has negative written turn %d", id, row.WrittenTurn) + } + out[id] = BoardRow{ + Agent: id, Focus: row.FocusTaskKey, + Halt: BoardHalt{Kind: row.HaltKind, Ref: row.HaltRef, Reason: row.HaltReason}, + WrittenTurn: uint64(row.WrittenTurn), WrittenAtUnixMs: row.WrittenAtUnixMs, + Tasks: make([]BoardTaskRow, 0), + } + } + tasks, err := qtx.GetAgentBoardTasks(ctx, ids) + if err != nil { + return nil, fmt.Errorf("store: read agent board tasks: %w", err) + } + for _, row := range tasks { + id := AccountID(row.AgentAccountID) + board, found := out[id] + if !found { + continue + } + refs := row.Refs + if len(refs) == 0 { + refs = nil + } + board.Tasks = append(board.Tasks, BoardTaskRow{ + Key: row.TaskKey, Title: row.Title, PrimaryRef: row.PrimaryRef, + Refs: refs, State: row.State, Note: row.Note, Waiting: row.Waiting, + Subagents: row.Subagents, WrittenAtUnixMs: row.WrittenAtUnixMs, + }) + out[id] = board + } + if err := tx.Commit(ctx); err != nil { + return nil, fmt.Errorf("store: commit board snapshot: %w", err) + } + return out, nil +} + +// SearchAgentBoards returns task keys matched by all supplied filters. +func (s *Store) SearchAgentBoards(ctx context.Context, f BoardTaskFilter) (map[AccountID][]string, error) { + limit := f.Limit + if limit <= 0 { + limit = 20 + } else if limit > maxBoardSearchAgents { + limit = maxBoardSearchAgents + } + rows, err := s.q.SearchAgentBoardTasks(ctx, db.SearchAgentBoardTasksParams{ + AgentAccountIds: uniqueAccountStrings(f.Agents), Query: strings.TrimSpace(f.Query), AgentLimit: int32(limit), + Ref: f.Ref, WaitingOn: f.WaitingOn, State: f.State, Halt: f.Halt, + }) + if err != nil { + return nil, fmt.Errorf("store: search agent boards: %w", err) + } + out := make(map[AccountID][]string) + for _, row := range rows { + id := AccountID(row.AgentAccountID) + out[id] = append(out[id], row.TaskKey) + } + return out, nil +} + +func uniqueAccountStrings(agents []AccountID) []string { + out := make([]string, 0, len(agents)) + seen := make(map[AccountID]struct{}, len(agents)) + for _, agent := range agents { + if agent == "" { + continue + } + if _, found := seen[agent]; found { + continue + } + seen[agent] = struct{}{} + out = append(out, string(agent)) + } + return out +} + +// AsksByAuthor returns recent answered asks and every pending ask visible to viewer. +func (s *Store) AsksByAuthor(ctx context.Context, viewer, author AccountID, answeredSince time.Time, limit int) ([]AskRow, error) { + if viewer == "" || author == "" { + return nil, fmt.Errorf("%w: viewer and author account ids are required", ErrInvalidArgument) + } + if limit <= 0 || limit > maxBoardAsks { + limit = maxBoardAsks + } + rows, err := s.q.ListAsksByAuthor(ctx, db.ListAsksByAuthorParams{ + AccountID: string(viewer), AuthorAccountID: string(author), + AnsweredSince: answeredSince.UnixMilli(), RowLimit: int32(limit), + }) + if err != nil { + return nil, fmt.Errorf("store: list asks by author: %w", err) + } + out := make([]AskRow, 0, len(rows)) + for _, row := range rows { + out = append(out, AskRow{ + AskID: row.AskID, MessageID: row.MessageID, ChannelID: row.ChannelID, + Ref: row.Ref, Header: row.Header, Answered: row.Answered, PostedAtUnixMs: row.AtUnixMs, + }) + } + return out, nil +} + +// OpenAuthoredRefs returns the agent's authored issues and PRs that remain open. +func (s *Store) OpenAuthoredRefs(ctx context.Context, agent AccountID) ([]AuthoredRef, error) { + if agent == "" { + return nil, fmt.Errorf("%w: agent account id is required", ErrInvalidArgument) + } + rows, err := s.q.ListOpenAuthoredRefs(ctx, string(agent)) + if err != nil { + return nil, fmt.Errorf("store: list open authored refs: %w", err) + } + out := make([]AuthoredRef, 0, len(rows)) + for _, row := range rows { + out = append(out, AuthoredRef{ + Coord: coordFromDB(row.ForgeProvider, row.ForgeHost, row.Repo, row.Number), + Kind: ForgeArtifactKind(row.Kind), + }) + } + return out, nil +} + +// PullRequestsByCoords reads stored pull request rows for normalized coordinates. +func (s *Store) PullRequestsByCoords(ctx context.Context, coords []ForgeCoord) ([]PullRequestRow, error) { + want, err := normalizeCoords(coords) + if err != nil { + return nil, err + } + if len(want) == 0 { + return nil, nil + } + arg := db.PullRequestsByCoordsParams{ + ForgeProviders: make([]int16, 0, len(want)), ForgeHosts: make([]string, 0, len(want)), + ForgeRepos: make([]string, 0, len(want)), ForgeNumbers: make([]int64, 0, len(want)), + } + for _, coord := range want { + value := coord.db() + arg.ForgeProviders = append(arg.ForgeProviders, value.provider) + arg.ForgeHosts = append(arg.ForgeHosts, value.host) + arg.ForgeRepos = append(arg.ForgeRepos, value.repo) + arg.ForgeNumbers = append(arg.ForgeNumbers, value.number) + } + rows, err := s.q.PullRequestsByCoords(ctx, arg) + if err != nil { + return nil, fmt.Errorf("store: pull requests by coordinates: %w", err) + } + out := make([]PullRequestRow, 0, len(rows)) + for _, row := range rows { + out = append(out, PullRequestRow{ + Coord: coordFromDB(row.ForgeProvider, row.ForgeHost, row.Repo, row.Number), + State: row.ForgeState, CreatedAt: row.ForgeCreatedAt.Time, + UpdatedAt: row.ForgeUpdatedAt.Time, PR: row.Pr, + }) + } + return out, nil +} + +// IssuesByCoords reads stored issue rows for normalized coordinates. +func (s *Store) IssuesByCoords(ctx context.Context, coords []ForgeCoord) ([]Issue, error) { + want, err := normalizeCoords(coords) + if err != nil { + return nil, err + } + if len(want) == 0 { + return nil, nil + } + arg := db.IssuesByCoordsParams{ + ForgeProviders: make([]int16, 0, len(want)), ForgeHosts: make([]string, 0, len(want)), + ForgeRepos: make([]string, 0, len(want)), ForgeNumbers: make([]int64, 0, len(want)), + } + for _, coord := range want { + value := coord.db() + arg.ForgeProviders = append(arg.ForgeProviders, value.provider) + arg.ForgeHosts = append(arg.ForgeHosts, value.host) + arg.ForgeRepos = append(arg.ForgeRepos, value.repo) + arg.ForgeNumbers = append(arg.ForgeNumbers, value.number) + } + rows, err := s.q.IssuesByCoords(ctx, arg) + if err != nil { + return nil, fmt.Errorf("store: issues by coordinates: %w", err) + } + out := make([]Issue, 0, len(rows)) + for _, row := range rows { + out = append(out, issueFromColumns( + row.ID, row.ForgeProvider, row.ForgeHost, row.Repo, row.Number, + row.Title, row.Body, row.ForgeState, row.Url, row.ForgeAccount, row.Labels, row.AgentHandle, + row.State, row.Priority, row.Assignee, row.Summary, row.Branch, + )) + } + return out, nil +} diff --git a/go/internal/store/agent_boards_pgtest_test.go b/go/internal/store/agent_boards_pgtest_test.go new file mode 100644 index 000000000..f365ca7ad --- /dev/null +++ b/go/internal/store/agent_boards_pgtest_test.go @@ -0,0 +1,835 @@ +//go:build pgtest + +package store + +import ( + "encoding/json" + "fmt" + "reflect" + "slices" + "strconv" + "strings" + "testing" + "time" +) + +func boardTask(key, title string) BoardTaskPatch { + return BoardTaskPatch{Key: key, Title: &title} +} + +func boardJSON(t *testing.T, value any) []byte { + t.Helper() + data, err := json.Marshal(value) + if err != nil { + t.Fatalf("marshal board json: %v", err) + } + return data +} + +func boardWaitingEntries(t *testing.T, count, whatLength int) []byte { + t.Helper() + entries := make([]map[string]any, count) + for i := range entries { + entries[i] = map[string]any{"on": "BOARD_WAITING_ON_OPERATOR", "what": strings.Repeat("w", whatLength)} + } + return boardJSON(t, entries) +} + +func boardSubagentEntries(t *testing.T, count, settled int) []byte { + t.Helper() + entries := make([]map[string]any, count) + for i := range entries { + settledAt := int64(0) + status := "running" + if i < settled { + settledAt = int64(i + 1) + status = "completed" + } + entries[i] = map[string]any{ + "id": fmt.Sprintf("sub-%02d", i), "agent": "implement", "description": fmt.Sprintf("task %d", i), + "status": status, "started_at_unix_ms": int64(i + 1), "ended_at_unix_ms": settledAt, + } + } + return boardJSON(t, entries) +} + +func boardRowCount(t *testing.T, s *Store, agent AccountID) (int, int) { + t.Helper() + var boards, tasks int + if err := s.pool.QueryRow(t.Context(), "SELECT count(*) FROM agent_boards WHERE agent_account_id = $1", string(agent)).Scan(&boards); err != nil { + t.Fatalf("count board rows: %v", err) + } + if err := s.pool.QueryRow(t.Context(), "SELECT count(*) FROM agent_board_tasks WHERE agent_account_id = $1", string(agent)).Scan(&tasks); err != nil { + t.Fatalf("count task rows: %v", err) + } + return boards, tasks +} + +func TestApplyBoardPatchMergesFieldsAndDefaultsState(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-merge-owner") + agent := mustAgent(t, s, owner.ID, "board-merge-agent") + title, primary, note := "Status board", "RIG-12", "first note" + state := int16(2) + refs := []string{"RIG-12", "owner/repo#4"} + waiting := boardWaitingEntries(t, 1, 12) + subagents := boardSubagentEntries(t, 1, 0) + got, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{{Key: "build", Title: &title, PrimaryRef: &primary, Note: ¬e, + State: &state, Refs: &refs, Waiting: &waiting, UpsertSubagents: subagents}}, + Focus: new("build"), Halt: &BoardHalt{Kind: 2, Ref: "RIG-9", Reason: "blocked"}, Turn: 7, NowUnixMs: 100, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch(initial): %v", err) + } + if got.WrittenTurn != 7 || got.WrittenAtUnixMs != 100 || got.Focus != "build" || got.Halt.Kind != 2 { + t.Fatalf("initial board metadata = %+v", got) + } + if len(got.Tasks) != 1 || got.Tasks[0].State != 2 || !reflect.DeepEqual(got.Tasks[0].Refs, refs) { + t.Fatalf("initial task = %+v", got.Tasks) + } + + newPrimary, newNote := "", "updated note" + emptyRefs := []string{} + merged, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{{Key: "build", PrimaryRef: &newPrimary, Note: &newNote, Refs: &emptyRefs, + UpsertSubagents: boardJSON(t, []map[string]any{{"id": "sub-00", "agent": "review", "description": "reviewing", "status": "running"}})}}, + Turn: 8, NowUnixMs: 200, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch(merge): %v", err) + } + if len(merged.Tasks) != 1 { + t.Fatalf("merged task count = %d, want 1", len(merged.Tasks)) + } + task := merged.Tasks[0] + if task.Title != title || task.PrimaryRef != "" || task.Note != newNote || task.State != state || len(task.Refs) != 0 { + t.Fatalf("merged task fields = %+v", task) + } + var gotWaiting, gotSubagents []map[string]any + if err := json.Unmarshal(task.Waiting, &gotWaiting); err != nil { + t.Fatalf("decode waiting: %v", err) + } + if err := json.Unmarshal(task.Subagents, &gotSubagents); err != nil { + t.Fatalf("decode subagents: %v", err) + } + if len(gotWaiting) != 1 || gotWaiting[0]["what"] != strings.Repeat("w", 12) { + t.Fatalf("waiting list was not kept: %s", task.Waiting) + } + if len(gotSubagents) != 1 || gotSubagents[0]["agent"] != "review" { + t.Fatalf("subagent id upsert did not replace existing entry: %s", task.Subagents) + } + if merged.WrittenTurn != 8 || merged.WrittenAtUnixMs != 200 || merged.Halt.Reason != "blocked" { + t.Fatalf("merged board metadata = %+v", merged) + } + + defaultStateTitle := "default state" + withDefault, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("default", defaultStateTitle)}, Turn: 9, NowUnixMs: 300, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch(default state): %v", err) + } + if len(withDefault.Tasks) != 2 || withDefault.Tasks[1].Key != "default" || withDefault.Tasks[1].State != 1 { + t.Fatalf("new task without state = %+v, want ACTIVE(1)", withDefault.Tasks) + } +} + +func TestApplyBoardPatchRemovingFocusedTaskClearsFocus(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-remove-owner") + agent := mustAgent(t, s, owner.ID, "board-remove-agent") + title := "focused task" + got, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("focus", title)}, Focus: new("focus"), NowUnixMs: 10, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch: %v", err) + } + if got.Focus != "focus" { + t.Fatalf("focus = %q, want focus", got.Focus) + } + got, err = s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Remove: []string{"focus"}, NowUnixMs: 20}) + if err != nil { + t.Fatalf("ApplyBoardPatch(remove): %v", err) + } + if got.Focus != "" || len(got.Tasks) != 0 { + t.Fatalf("board after focused task removal = %+v", got) + } + boards, err := s.AgentBoards(t.Context(), []AccountID{agent.ID}) + if err != nil { + t.Fatalf("AgentBoards after removal: %v", err) + } + if boards[agent.ID].Focus != "" { + t.Fatalf("persisted focus = %q after removing focused task", boards[agent.ID].Focus) + } + title = "later task" + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("later", title)}, NowUnixMs: 30, + }); err != nil { + t.Fatalf("ApplyBoardPatch after removal: %v", err) + } +} + +func TestApplyBoardPatchExplicitEmptyFocusClearsFocus(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-focus-clear-owner") + agent := mustAgent(t, s, owner.ID, "board-focus-clear-agent") + title := "focused" + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("focus", title)}, Focus: new("focus"), NowUnixMs: 1, + }); err != nil { + t.Fatalf("ApplyBoardPatch(initial): %v", err) + } + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Focus: new(""), NowUnixMs: 2}); err != nil { + t.Fatalf("ApplyBoardPatch(clear focus): %v", err) + } + boards, err := s.AgentBoards(t.Context(), []AccountID{agent.ID}) + if err != nil { + t.Fatalf("AgentBoards: %v", err) + } + if boards[agent.ID].Focus != "" { + t.Fatalf("focus = %q, want empty", boards[agent.ID].Focus) + } +} + +func TestApplyBoardPatchRejectsUnknownFocusAndClearsWaiting(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-focus-validation-owner") + agent := mustAgent(t, s, owner.ID, "board-focus-validation-agent") + title := "waiting task" + waiting := boardWaitingEntries(t, 1, 8) + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{{Key: "task", Title: &title, Waiting: &waiting}}, NowUnixMs: 1, + }); err != nil { + t.Fatalf("ApplyBoardPatch(initial): %v", err) + } + _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Focus: new("missing"), NowUnixMs: 2}) + sentinelIs(t, err, ErrInvalidArgument, "focus task not on board") + emptyWaiting := []byte("[]") + got, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{{Key: "task", Waiting: &emptyWaiting}}, NowUnixMs: 3, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch(clear waiting): %v", err) + } + var entries []json.RawMessage + if err := json.Unmarshal(got.Tasks[0].Waiting, &entries); err != nil || len(entries) != 0 { + t.Fatalf("waiting after clear = %s, err=%v", got.Tasks[0].Waiting, err) + } +} + +func TestApplyBoardPatchReplaceTaskAtCapacity(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-replace-full-owner") + agent := mustAgent(t, s, owner.ID, "board-replace-full-agent") + upserts := make([]BoardTaskPatch, maxBoardTasks) + for i := range upserts { + upserts[i] = boardTask(fmt.Sprintf("task-%02d", i), "task") + } + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Upsert: upserts, NowUnixMs: 1}); err != nil { + t.Fatalf("ApplyBoardPatch(fill board): %v", err) + } + title := "replacement" + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Remove: []string{"task-00"}, Upsert: []BoardTaskPatch{boardTask("replacement", title)}, NowUnixMs: 2, + }); err != nil { + t.Fatalf("ApplyBoardPatch(replace at capacity): %v", err) + } + boards, err := s.AgentBoards(t.Context(), []AccountID{agent.ID}) + if err != nil { + t.Fatalf("AgentBoards after replacement: %v", err) + } + got := boards[agent.ID].Tasks + if len(got) != maxBoardTasks || got[0].Key != "task-01" || got[len(got)-1].Key != "replacement" { + t.Fatalf("replacement board has unexpected tasks: count=%d first=%q last=%q", len(got), got[0].Key, got[len(got)-1].Key) + } + for i, task := range got { + want := fmt.Sprintf("task-%02d", i+1) + if i == len(got)-1 { + want = "replacement" + } + if task.Key != want { + t.Fatalf("stored task %d = %q, want %q", i, task.Key, want) + } + } +} + +func TestApplyBoardPatchCapBreachPreservesExistingBoard(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-existing-cap-owner") + agent := mustAgent(t, s, owner.ID, "board-existing-cap-agent") + title := "existing" + initial, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("existing", title)}, Focus: new("existing"), NowUnixMs: 1, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch(initial): %v", err) + } + tooMany := make([]BoardTaskPatch, maxBoardTasks) + for i := range tooMany { + tooMany[i] = boardTask(fmt.Sprintf("extra-%02d", i), "extra") + } + _, err = s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Upsert: tooMany, NowUnixMs: 2}) + sentinelIs(t, err, ErrInvalidArgument, "existing board task cap") + boards, err := s.AgentBoards(t.Context(), []AccountID{agent.ID}) + if err != nil { + t.Fatalf("AgentBoards after rejected patch: %v", err) + } + got := boards[agent.ID] + if len(got.Tasks) != 1 || got.Focus != initial.Focus || got.WrittenTurn != initial.WrittenTurn || got.WrittenAtUnixMs != initial.WrittenAtUnixMs || + got.Tasks[0].Key != "existing" || got.Tasks[0].Title != "existing" || got.Tasks[0].WrittenAtUnixMs != initial.Tasks[0].WrittenAtUnixMs { + t.Fatalf("board changed after rejected patch: got %+v, want %+v", got, initial) + } +} + +func TestApplyBoardPatchValidationWritesNothing(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-invalid-owner") + tooLong := strings.Repeat("x", 121) + tooLongNote := strings.Repeat("x", 501) + tooLongReason := strings.Repeat("x", 201) + tooLongWhat := boardWaitingEntries(t, 1, 201) + tooManyRefs := make([]string, 11) + tooManyWaits := boardWaitingEntries(t, 6, 1) + tooManyActive := boardSubagentEntries(t, 21, 0) + tooManyTasks := make([]BoardTaskPatch, 31) + for i := range tooManyTasks { + tooManyTasks[i] = boardTask(fmt.Sprintf("task-%02d", i), "task") + } + missingWaitingOn := []byte(`[{"what":"review"}]`) + numericWaitingOn := []byte(`[{"on":1,"what":"review"}]`) + unknownWaitingOn := []byte(`[{"on":"BOARD_WAITING_ON_UNKNOWN","what":"review"}]`) + cases := []struct { + name string + patch BoardPatch + }{ + {name: "waiting on missing", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "waiting-missing-on", Title: new("task"), Waiting: &missingWaitingOn}}}}, + {name: "waiting on numeric", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "waiting-numeric-on", Title: new("task"), Waiting: &numericWaitingOn}}}}, + {name: "waiting on unknown", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "waiting-unknown-on", Title: new("task"), Waiting: &unknownWaitingOn}}}}, + {name: "new key requires title", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "missing-title"}}}}, + {name: "title cap", patch: BoardPatch{Upsert: []BoardTaskPatch{boardTask("title", tooLong)}}}, + {name: "note cap", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "note", Title: new("task"), Note: &tooLongNote}}}}, + {name: "refs cap", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "refs", Title: new("task"), Refs: &tooManyRefs}}}}, + {name: "waiting cap", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "waiting", Title: new("task"), Waiting: &tooManyWaits}}}}, + {name: "waiting what cap", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "what", Title: new("task"), Waiting: &tooLongWhat}}}}, + {name: "active subagent cap", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "subagents", Title: new("task"), UpsertSubagents: tooManyActive}}}}, + {name: "task cap", patch: BoardPatch{Upsert: tooManyTasks}}, + {name: "halt reason cap", patch: BoardPatch{Halt: &BoardHalt{Kind: 2, Reason: tooLongReason}}}, + {name: "invalid state", patch: BoardPatch{Upsert: []BoardTaskPatch{{Key: "state", Title: new("task"), State: new(int16(0))}}}}, + } + for i, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + caseAgent := mustAgent(t, s, owner.ID, fmt.Sprintf("board-invalid-agent-%02d", i)) + _, err := s.ApplyBoardPatch(t.Context(), caseAgent.ID, tc.patch) + sentinelIs(t, err, ErrInvalidArgument, tc.name) + boards, tasks := boardRowCount(t, s, caseAgent.ID) + if boards != 0 || tasks != 0 { + t.Fatalf("invalid patch wrote %d board rows and %d task rows", boards, tasks) + } + }) + } +} + +func TestApplyBoardPatchDropsOldestSettledSubagent(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-subagent-owner") + agent := mustAgent(t, s, owner.ID, "board-subagent-agent") + title := "subagents" + entries := make([]map[string]any, 21) + for i := range entries { + entries[i] = map[string]any{ + "id": fmt.Sprintf("sub-%02d", i), "agent": "implement", "description": fmt.Sprintf("task %d", i), + "status": "running", "startedAtUnixMs": strconv.Itoa(1000 - i), + } + } + entries[0]["status"] = "completed" + entries[0]["startedAtUnixMs"] = "200" + entries[0]["endedAtUnixMs"] = "300" + entries[10]["status"] = "aborted" + entries[10]["startedAtUnixMs"] = "100" + entries[20]["status"] = "started" + entries[20]["startedAtUnixMs"] = "1" + got, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{{Key: "workers", Title: &title, UpsertSubagents: boardJSON(t, entries)}}, NowUnixMs: 10, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch: %v", err) + } + var subagents []map[string]any + if err := json.Unmarshal(got.Tasks[0].Subagents, &subagents); err != nil { + t.Fatalf("decode subagents: %v", err) + } + if len(subagents) != 20 { + t.Fatalf("retained %d subagents, want 20", len(subagents)) + } + for _, entry := range subagents { + if entry["id"] == "sub-10" { + t.Fatalf("oldest settled subagent remained: %+v", entry) + } + } + if subagents[len(subagents)-1]["id"] != "sub-20" { + t.Fatalf("oldest live subagent was not retained: %+v", subagents[len(subagents)-1]) + } + for _, status := range []string{"", "running", "pending", "started", "queued", "unknown", "cancelled", "error"} { + settled, _ := subagentSettled(json.RawMessage(fmt.Sprintf(`{"status":%q}`, status))) + if settled { + t.Errorf("subagent status %q was treated as settled", status) + } + } + for _, status := range []string{"completed", "failed", "aborted"} { + settled, _ := subagentSettled(json.RawMessage(fmt.Sprintf(`{"status":%q}`, status))) + if !settled { + t.Errorf("subagent status %q was not treated as settled", status) + } + } +} + +func TestAutomaticBoardPatchPreservesFreshnessAndEmptyPatchWritesNothing(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-auto-owner") + agent := mustAgent(t, s, owner.ID, "board-auto-agent") + title := "automatic task" + initial, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("automatic", title)}, Turn: 4, NowUnixMs: 123, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch(initial): %v", err) + } + updated, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{{Key: "automatic", UpsertSubagents: boardSubagentEntries(t, 1, 0)}}, + Automatic: true, Turn: 99, NowUnixMs: 999, + }) + if err != nil { + t.Fatalf("ApplyBoardPatch(automatic): %v", err) + } + if updated.WrittenTurn != 4 || updated.WrittenAtUnixMs != 123 || updated.Tasks[0].WrittenAtUnixMs != initial.Tasks[0].WrittenAtUnixMs { + t.Fatalf("automatic patch changed freshness: %+v", updated) + } + noOp, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Automatic: true, Turn: 100, NowUnixMs: 1000}) + if err != nil { + t.Fatalf("ApplyBoardPatch(empty automatic): %v", err) + } + if noOp.Focus != updated.Focus || len(noOp.Tasks) != 1 || noOp.Tasks[0].Key != updated.Tasks[0].Key { + t.Fatalf("empty automatic patch changed board content: %+v", noOp) + } + other := mustAgent(t, s, owner.ID, "board-empty-auto-agent") + empty, err := s.ApplyBoardPatch(t.Context(), other.ID, BoardPatch{Automatic: true, Turn: 77, NowUnixMs: 77}) + if err != nil { + t.Fatalf("ApplyBoardPatch(empty automatic without row): %v", err) + } + boards, tasks := boardRowCount(t, s, other.ID) + if boards != 0 || tasks != 0 || empty.Agent != other.ID { + t.Fatalf("empty automatic patch created rows or lost agent: board=%+v rows=%d/%d", empty, boards, tasks) + } +} + +func TestSearchAgentBoardsCombinesFilters(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-search-owner") + agent := mustAgent(t, s, owner.ID, "board-search-agent") + excluded := mustAgent(t, s, owner.ID, "board-search-excluded-agent") + other := mustAgent(t, s, owner.ID, "board-search-other") + title, note := "Status board storage", "migration details" + state := int16(2) + refs := []string{"RIG-12"} + wrongRefs := []string{"RIG-99"} + waiting := boardJSON(t, []map[string]any{{"on": "BOARD_WAITING_ON_OPERATOR", "what": "review"}}) + wrongWaiting := boardJSON(t, []map[string]any{{"on": "BOARD_WAITING_ON_AGENT", "what": "review"}}) + wrongState := int16(3) + setup := []struct { + agent AccountID + key string + note string + refs []string + waiting []byte + state int16 + halt int16 + }{ + {agent: agent.ID, key: "match", note: note, refs: refs, waiting: waiting, state: state, halt: 2}, + {agent: agent.ID, key: "wrong-query", note: "release notes", refs: refs, waiting: waiting, state: state, halt: 2}, + {agent: agent.ID, key: "wrong-ref", note: note, refs: wrongRefs, waiting: waiting, state: state, halt: 2}, + {agent: agent.ID, key: "wrong-waiting", note: note, refs: refs, waiting: wrongWaiting, state: state, halt: 2}, + {agent: agent.ID, key: "wrong-state", note: note, refs: refs, waiting: waiting, state: wrongState, halt: 2}, + {agent: excluded.ID, key: "wrong-agent", note: note, refs: refs, waiting: waiting, state: state, halt: 2}, + {agent: other.ID, key: "wrong-halt", note: note, refs: refs, waiting: waiting, state: state, halt: 1}, + } + for i := range setup { + item := &setup[i] + if _, err := s.ApplyBoardPatch(t.Context(), item.agent, BoardPatch{ + Upsert: []BoardTaskPatch{{Key: item.key, Title: &title, Note: &item.note, State: &item.state, Refs: &item.refs, Waiting: &item.waiting}}, + Halt: &BoardHalt{Kind: item.halt}, NowUnixMs: int64(i + 1), + }); err != nil { + t.Fatalf("ApplyBoardPatch(%s): %v", item.key, err) + } + } + got, err := s.SearchAgentBoards(t.Context(), BoardTaskFilter{ + Agents: []AccountID{agent.ID, other.ID}, Query: "migration", Ref: "RIG-12", WaitingOn: 1, State: 2, Halt: 2, + }) + if err != nil { + t.Fatalf("SearchAgentBoards: %v", err) + } + if !reflect.DeepEqual(got, map[AccountID][]string{agent.ID: {"match"}}) { + t.Fatalf("SearchAgentBoards = %v, want only the task matching every filter", got) + } +} + +func TestSearchAgentBoardsTitleAndPrimaryRefOnly(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-search-title-ref-owner") + agent := mustAgent(t, s, owner.ID, "board-search-title-ref-agent") + titleOnly := "Quartz milestone" + quietNote := "ordinary notes" + noRefs := []string{} + primaryOnly := "No matching words" + primaryRef := "RIG-884" + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Upsert: []BoardTaskPatch{ + {Key: "title-only", Title: &titleOnly, Note: &quietNote, Refs: &noRefs}, + {Key: "primary-only", Title: &primaryOnly, PrimaryRef: &primaryRef, Refs: &noRefs}, + }, NowUnixMs: 1}); err != nil { + t.Fatalf("ApplyBoardPatch: %v", err) + } + t.Run("title contains query", func(t *testing.T) { + got, err := s.SearchAgentBoards(t.Context(), BoardTaskFilter{Agents: []AccountID{agent.ID}, Query: "quartz"}) + if err != nil { + t.Fatalf("SearchAgentBoards(title query): %v", err) + } + if !reflect.DeepEqual(got[agent.ID], []string{"title-only"}) { + t.Fatalf("title query results = %v, want title-only", got) + } + }) + t.Run("primary ref matches without refs entry", func(t *testing.T) { + got, err := s.SearchAgentBoards(t.Context(), BoardTaskFilter{Agents: []AccountID{agent.ID}, Ref: primaryRef}) + if err != nil { + t.Fatalf("SearchAgentBoards(primary ref): %v", err) + } + if !reflect.DeepEqual(got[agent.ID], []string{"primary-only"}) { + t.Fatalf("primary ref results = %v, want primary-only", got) + } + }) +} + +func TestSearchAgentBoardsClampsLimit(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-search-limit-owner") + agents := make([]AccountID, maxBoardSearchAgents+1) + taskKeys := []string{"task-a", "task-b", "task-c"} + title := "searchable" + for i := range agents { + agent := mustAgent(t, s, owner.ID, fmt.Sprintf("board-search-limit-%03d", i)) + agents[i] = agent.ID + patches := make([]BoardTaskPatch, len(taskKeys)) + for j, key := range taskKeys { + patches[j] = boardTask(key, title) + } + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Upsert: patches, NowUnixMs: 1}); err != nil { + t.Fatalf("ApplyBoardPatch(%d): %v", i, err) + } + } + got, err := s.SearchAgentBoards(t.Context(), BoardTaskFilter{Agents: agents, Limit: maxBoardSearchAgents + 1}) + if err != nil { + t.Fatalf("SearchAgentBoards: %v", err) + } + if len(got) != maxBoardSearchAgents { + t.Fatalf("SearchAgentBoards returned %d agents, want cap %d", len(got), maxBoardSearchAgents) + } + wantStoreIDs := append([]AccountID(nil), agents...) + slices.Sort(wantStoreIDs) + wantStoreIDs = wantStoreIDs[:maxBoardSearchAgents] + gotStoreIDs := make([]AccountID, 0, maxBoardSearchAgents) + for agentID, keys := range got { + gotStoreIDs = append(gotStoreIDs, agentID) + if !reflect.DeepEqual(keys, taskKeys) { + t.Fatalf("agent %q returned task keys %v, want %v", agentID, keys, taskKeys) + } + } + slices.Sort(gotStoreIDs) + if !reflect.DeepEqual(gotStoreIDs, wantStoreIDs) { + t.Fatalf("SearchAgentBoards returned unexpected capped agents: got %v, want %v", gotStoreIDs, wantStoreIDs) + } +} + +func TestSearchAgentBoardsZeroLimitDefaultsToTwenty(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-search-zero-limit-owner") + agents := make([]AccountID, 25) + taskKeys := []string{"task-a", "task-b"} + title := "searchable" + for i := range agents { + agent := mustAgent(t, s, owner.ID, fmt.Sprintf("board-search-zero-limit-%02d", i)) + agents[i] = agent.ID + patches := make([]BoardTaskPatch, len(taskKeys)) + for j, key := range taskKeys { + patches[j] = boardTask(key, title) + } + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{Upsert: patches, NowUnixMs: 1}); err != nil { + t.Fatalf("ApplyBoardPatch(%d): %v", i, err) + } + } + got, err := s.SearchAgentBoards(t.Context(), BoardTaskFilter{Limit: 0}) + if err != nil { + t.Fatalf("SearchAgentBoards(zero limit): %v", err) + } + if len(got) != 20 { + t.Fatalf("SearchAgentBoards(zero limit) returned %d agents, want 20", len(got)) + } + wantAgents := append([]AccountID(nil), agents...) + slices.Sort(wantAgents) + wantAgents = wantAgents[:20] + gotAgents := make([]AccountID, 0, len(got)) + for agent, keys := range got { + gotAgents = append(gotAgents, agent) + if !reflect.DeepEqual(keys, taskKeys) { + t.Fatalf("agent %q returned keys %v, want %v", agent, keys, taskKeys) + } + } + slices.Sort(gotAgents) + if !reflect.DeepEqual(gotAgents, wantAgents) { + t.Fatalf("SearchAgentBoards(zero limit) agents = %v, want %v", gotAgents, wantAgents) + } +} + +func TestAgentBoardsUsesOneSnapshotAcrossConcurrentPatch(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-snapshot-owner") + agent := mustAgent(t, s, owner.ID, "board-snapshot-agent") + oldTitle, newTitle := "before patch", "after patch" + if _, err := s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("old", oldTitle)}, Focus: new("old"), NowUnixMs: 1, + }); err != nil { + t.Fatalf("ApplyBoardPatch(initial): %v", err) + } + var patchErr error + s.afterBoardMetaRead = func() { + s.afterBoardMetaRead = nil + _, patchErr = s.ApplyBoardPatch(t.Context(), agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("new", newTitle)}, Focus: new("new"), NowUnixMs: 2, + }) + } + got, err := s.AgentBoards(t.Context(), []AccountID{agent.ID}) + if err != nil { + t.Fatalf("AgentBoards: %v", err) + } + if patchErr != nil { + t.Fatalf("ApplyBoardPatch(concurrent): %v", patchErr) + } + board := got[agent.ID] + if board.Focus != "old" || len(board.Tasks) != 1 || board.Tasks[0].Key != "old" || board.Tasks[0].Title != oldTitle { + t.Fatalf("AgentBoards mixed snapshots: %+v", board) + } + after, err := s.AgentBoards(t.Context(), []AccountID{agent.ID}) + if err != nil { + t.Fatalf("AgentBoards(after patch): %v", err) + } + board = after[agent.ID] + if board.Focus != "new" || len(board.Tasks) != 2 || board.Tasks[1].Key != "new" || board.Tasks[1].Title != newTitle { + t.Fatalf("committed patch missing from next snapshot: %+v", board) + } +} + +func TestBoardCoordinateReadsAndOpenAuthoredRefs(t *testing.T) { + s := newTestStore(t) + agent, owner := seedAgent(t, s, "board-coords") + openIssue, closedIssue := ghCoord("Owner/Repo", 41), ghCoord("Owner/Repo", 42) + for _, tc := range []struct { + coord ForgeCoord + state string + }{{openIssue, "open"}, {closedIssue, "closed"}} { + if _, err := s.UpsertIssueForgeFields(t.Context(), IssueForgeFields{ + ForgeProvider: tc.coord.Provider, ForgeHost: tc.coord.Host, Repo: tc.coord.Repo, + Number: uint32(tc.coord.Number), Title: fmt.Sprintf("issue %d", tc.coord.Number), ForgeState: tc.state, + }); err != nil { + t.Fatalf("UpsertIssueForgeFields(%d): %v", tc.coord.Number, err) + } + if err := s.RecordAuthoredArtifact(t.Context(), AuthoredArtifact{ + Provider: tc.coord.Provider, Host: tc.coord.Host, Repo: tc.coord.Repo, + Kind: ForgeArtifactKindIssue, Number: tc.coord.Number, AgentAccountID: agent, + OwnerUserID: owner, CreatedAtUnixMS: int64(tc.coord.Number), + }); err != nil { + t.Fatalf("RecordAuthoredArtifact(issue %d): %v", tc.coord.Number, err) + } + } + openPR, closedPR, mergedPR := ghCoord("OWNER/REPO", 51), ghCoord("OWNER/REPO", 52), ghCoord("OWNER/REPO", 53) + for _, tc := range []struct { + coord ForgeCoord + state string + }{{openPR, "open"}, {closedPR, "closed"}, {mergedPR, "merged"}} { + pr := prRow(tc.coord.Repo, tc.coord.Number, tc.state, prBase, prBase.Add(time.Hour)) + if _, err := s.UpsertPullRequest(t.Context(), pr, nil); err != nil { + t.Fatalf("UpsertPullRequest(%d): %v", tc.coord.Number, err) + } + if err := s.RecordAuthoredArtifact(t.Context(), authoredPR(agent, owner, pr)); err != nil { + t.Fatalf("RecordAuthoredArtifact(PR %d): %v", tc.coord.Number, err) + } + } + open, err := s.OpenAuthoredRefs(t.Context(), agent) + if err != nil { + t.Fatalf("OpenAuthoredRefs: %v", err) + } + if len(open) != 2 || open[0].Coord.Number != openIssue.Number || open[0].Kind != ForgeArtifactKindIssue || + open[1].Coord.Number != openPR.Number || open[1].Kind != ForgeArtifactKindPullRequest { + t.Fatalf("open authored refs = %+v, want open issue and PR only", open) + } + mixedCasePRs, err := s.PullRequestsByCoords(t.Context(), []ForgeCoord{ghCoord("OWNER/REPO", openPR.Number)}) + if err != nil { + t.Fatalf("PullRequestsByCoords(mixed-case repo): %v", err) + } + if len(mixedCasePRs) != 1 || mixedCasePRs[0].Coord.Number != openPR.Number { + t.Fatalf("PullRequestsByCoords(mixed-case repo) = %+v, want PR %d", mixedCasePRs, openPR.Number) + } + prs, err := s.PullRequestsByCoords(t.Context(), []ForgeCoord{openPR, closedPR, openPR}) + if err != nil { + t.Fatalf("PullRequestsByCoords: %v", err) + } + if len(prs) != 2 || prs[0].Coord.Number != openPR.Number || prs[1].Coord.Number != closedPR.Number { + t.Fatalf("PullRequestsByCoords = %+v", prs) + } + issues, err := s.IssuesByCoords(t.Context(), []ForgeCoord{openIssue, closedIssue}) + if err != nil { + t.Fatalf("IssuesByCoords: %v", err) + } + if len(issues) != 2 || issues[0].Number != uint32(openIssue.Number) || issues[1].Number != uint32(closedIssue.Number) { + t.Fatalf("IssuesByCoords = %+v", issues) + } +} + +func TestAsksByAuthorClipsInvisibleChannels(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-asks-owner") + viewer := mustUser(t, s, "board-asks-viewer") + author := mustAgent(t, s, owner.ID, "board-asks-author") + private, err := s.CreateChannel(t.Context(), owner.ID, NewChannel{ + Name: "board-private", MemberAccountIDs: []AccountID{owner.ID, author.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel(private): %v", err) + } + group, err := s.CreateChannelGroup(t.Context(), owner.ID, NewChannelGroup{Name: "board-shared", Visibility: VisibilityShared}) + if err != nil { + t.Fatalf("CreateChannelGroup(shared): %v", err) + } + shared, err := s.CreateChannel(t.Context(), owner.ID, NewChannel{ + Name: "board-shared-room", Kind: ChannelKindChannel, GroupID: group.ID, MemberAccountIDs: []AccountID{author.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel(shared): %v", err) + } + for _, tc := range []struct { + channel ChannelID + ref string + }{{private.ID, "RIG-13"}, {shared.ID, "RIG-12"}} { + _, _, err := s.AppendMessage(t.Context(), Message{ + AuthorAccountID: author.ID, + Blocks: []MessageBlock{{Ask: &Ask{Ref: tc.ref, Questions: []AskQuestion{{QuestionID: "q", Question: "Need a decision?"}}}}}, + }, string(tc.channel), TopicRef{Name: "general", Create: true}, "") + if err != nil { + t.Fatalf("AppendMessage(ask %s): %v", tc.ref, err) + } + } + tasks, err := s.AsksByAuthor(t.Context(), viewer.ID, author.ID, time.Unix(1, 0), 20) + if err != nil { + t.Fatalf("AsksByAuthor: %v", err) + } + if len(tasks) != 1 || tasks[0].Ref != "RIG-12" || tasks[0].ChannelID != string(shared.ID) || tasks[0].Header != "Need a decision?" { + t.Fatalf("AsksByAuthor = %+v, want only shared-channel ask", tasks) + } +} + +func TestAsksByAuthorUsesAnswerMessageTimestamp(t *testing.T) { + s := newTestStore(t) + owner := mustUser(t, s, "board-ask-time-owner") + author := mustAgent(t, s, owner.ID, "board-ask-time-author") + channel, err := s.CreateChannel(t.Context(), owner.ID, NewChannel{ + Name: "board-ask-time", MemberAccountIDs: []AccountID{owner.ID, author.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + answerIDs := make([]string, maxBoardAsks) + askIDs := make([]string, maxBoardAsks) + askTimes := make([]int64, maxBoardAsks) + for i := range answerIDs { + answerAt := int64(2_000 + i) + askTimes[i] = int64(1_000 + i) + ask, _, err := s.AppendMessage(t.Context(), Message{ + AuthorAccountID: author.ID, + Blocks: []MessageBlock{{Ask: &Ask{Questions: []AskQuestion{{QuestionID: "q", Question: "Question"}}}}}, + }, string(channel.ID), TopicRef{Name: "general", Create: true}, "") + if err != nil { + t.Fatalf("AppendMessage(%d): %v", i, err) + } + askIDs[i] = ask.Blocks[0].Ask.AskID + _, answer, err := s.AnswerAsk(t.Context(), owner.ID, askIDs[i], []AskAnswer{{QuestionID: "q"}}) + if err != nil { + t.Fatalf("AnswerAsk(%d): %v", i, err) + } + answerIDs[i] = string(answer.ID) + if _, err := s.pool.Exec(t.Context(), "UPDATE messages SET at_unix_ms = $1 WHERE id = $2", askTimes[i], string(ask.ID)); err != nil { + t.Fatalf("age ask %d: %v", i, err) + } + if _, err := s.pool.Exec(t.Context(), "UPDATE messages SET at_unix_ms = $1 WHERE id = $2", answerAt, answerIDs[i]); err != nil { + t.Fatalf("set answer time %d: %v", i, err) + } + } + oldPending, _, err := s.AppendMessage(t.Context(), Message{ + AuthorAccountID: author.ID, + Blocks: []MessageBlock{{Ask: &Ask{Questions: []AskQuestion{{QuestionID: "q", Question: "Old pending ask"}}}}}, + }, string(channel.ID), TopicRef{Name: "general", Create: true}, "") + if err != nil { + t.Fatalf("AppendMessage(old pending ask): %v", err) + } + if _, err := s.pool.Exec(t.Context(), "UPDATE messages SET at_unix_ms = $1 WHERE id = $2", int64(100), string(oldPending.ID)); err != nil { + t.Fatalf("age pending ask: %v", err) + } + tasks, err := s.AsksByAuthor(t.Context(), owner.ID, author.ID, time.UnixMilli(1_200), 20) + if err != nil { + t.Fatalf("AsksByAuthor: %v", err) + } + if len(tasks) != maxBoardAsks { + t.Fatalf("AsksByAuthor returned %d asks, want pending ask and %d answered asks", len(tasks), maxBoardAsks-1) + } + if tasks[0].AskID != oldPending.Blocks[0].Ask.AskID || tasks[0].Answered || tasks[0].PostedAtUnixMs != 100 { + t.Fatalf("first ask = %+v, want old pending ask %q", tasks[0], oldPending.Blocks[0].Ask.AskID) + } + for i := 1; i < len(tasks); i++ { + answerIndex := maxBoardAsks - i + if tasks[i].AskID != askIDs[answerIndex] || tasks[i].PostedAtUnixMs != askTimes[answerIndex] || !tasks[i].Answered { + t.Fatalf("answer %d = %+v, want ask %q posted at %d", i, tasks[i], askIDs[answerIndex], askTimes[answerIndex]) + } + } +} + +func TestAgentBoardsTenantIsolation(t *testing.T) { + s := newTestStore(t) + tenantA := s.EffectiveTenant(t.Context()) + tenantB := seedTenant(t, s, "board-tenant-b") + ctxA := WithTenant(t.Context(), tenantA) + owner, err := s.CreateUser(ctxA, NewUser{Handle: "board-tenant-owner", DisplayName: "Owner"}) + if err != nil { + t.Fatalf("CreateUser(tenant A): %v", err) + } + agent, err := s.CreateAgent(ctxA, owner.ID, NewAgent{Handle: "board-tenant-agent", DisplayName: "Agent"}) + if err != nil { + t.Fatalf("CreateAgent(tenant A): %v", err) + } + title := "tenant board" + if _, err := s.ApplyBoardPatch(ctxA, agent.ID, BoardPatch{ + Upsert: []BoardTaskPatch{boardTask("private", title)}, NowUnixMs: 10, + }); err != nil { + t.Fatalf("ApplyBoardPatch(tenant A): %v", err) + } + boards, err := s.AgentBoards(WithTenant(t.Context(), tenantB), []AccountID{agent.ID}) + if err != nil { + t.Fatalf("AgentBoards(tenant B): %v", err) + } + if len(boards) != 0 { + t.Fatalf("tenant B saw tenant A board: %+v", boards) + } + matches, err := s.SearchAgentBoards(WithTenant(t.Context(), tenantB), BoardTaskFilter{Agents: []AccountID{agent.ID}}) + if err != nil { + t.Fatalf("SearchAgentBoards(tenant B): %v", err) + } + if len(matches) != 0 { + t.Fatalf("tenant B saw tenant A task matches: %+v", matches) + } +} diff --git a/go/internal/store/blocks.go b/go/internal/store/blocks.go index 2339bc0e8..9daa51c2c 100644 --- a/go/internal/store/blocks.go +++ b/go/internal/store/blocks.go @@ -41,6 +41,7 @@ type storedAskAnswer struct { // storedAsk is the JSONB shape of an Ask block. type storedAsk struct { AskID string `json:"ask_id"` + Ref string `json:"ref,omitempty"` Questions []storedAskQuestion `json:"questions,omitempty"` // Answered mirrors Ask.Answered: true once answered, so the pending/answered // distinction survives the JSONB round-trip even for a fully-skipped ask. @@ -317,7 +318,7 @@ func toStoredAsk(a *Ask) *storedAsk { TimedOut: q.TimedOut, }) } - return &storedAsk{AskID: a.AskID, Questions: qs, Answered: a.Answered} + return &storedAsk{AskID: a.AskID, Ref: a.Ref, Questions: qs, Answered: a.Answered} } // fromStoredAsk maps a stored Ask back to the domain type. @@ -340,7 +341,7 @@ func fromStoredAsk(s *storedAsk) *Ask { TimedOut: q.TimedOut, }) } - return &Ask{AskID: s.AskID, Questions: qs, Answered: s.Answered} + return &Ask{AskID: s.AskID, Ref: s.Ref, Questions: qs, Answered: s.Answered} } // toStoredAskAnswer maps the domain AskAnswerBlock onto its JSONB shape, diff --git a/go/internal/store/db/agent_boards.sql.go b/go/internal/store/db/agent_boards.sql.go new file mode 100644 index 000000000..a0354af76 --- /dev/null +++ b/go/internal/store/db/agent_boards.sql.go @@ -0,0 +1,668 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: agent_boards.sql + +package db + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const deleteAgentBoardTasks = `-- name: DeleteAgentBoardTasks :exec +DELETE FROM agent_board_tasks + WHERE agent_account_id = $1 AND task_key = ANY($2::text[]) +` + +type DeleteAgentBoardTasksParams struct { + AgentAccountID string + Column2 []string +} + +func (q *Queries) DeleteAgentBoardTasks(ctx context.Context, arg DeleteAgentBoardTasksParams) error { + _, err := q.db.Exec(ctx, deleteAgentBoardTasks, arg.AgentAccountID, arg.Column2) + return err +} + +const getAgentBoardTasks = `-- name: GetAgentBoardTasks :many +SELECT agent_account_id, task_key, title, primary_ref, refs, state, note, + waiting, subagents, position, written_at_unix_ms + FROM agent_board_tasks + WHERE agent_account_id = ANY($1::text[]) + ORDER BY position, task_key +` + +type GetAgentBoardTasksRow struct { + AgentAccountID string + TaskKey string + Title string + PrimaryRef string + Refs []string + State int16 + Note string + Waiting []byte + Subagents []byte + Position int32 + WrittenAtUnixMs int64 +} + +func (q *Queries) GetAgentBoardTasks(ctx context.Context, dollar_1 []string) ([]GetAgentBoardTasksRow, error) { + rows, err := q.db.Query(ctx, getAgentBoardTasks, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetAgentBoardTasksRow + for rows.Next() { + var i GetAgentBoardTasksRow + if err := rows.Scan( + &i.AgentAccountID, + &i.TaskKey, + &i.Title, + &i.PrimaryRef, + &i.Refs, + &i.State, + &i.Note, + &i.Waiting, + &i.Subagents, + &i.Position, + &i.WrittenAtUnixMs, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getAgentBoards = `-- name: GetAgentBoards :many +SELECT agent_account_id, focus_task_key, halt_kind, halt_ref, halt_reason, + written_turn, written_at_unix_ms + FROM agent_boards + WHERE agent_account_id = ANY($1::text[]) +` + +type GetAgentBoardsRow struct { + AgentAccountID string + FocusTaskKey string + HaltKind int16 + HaltRef string + HaltReason string + WrittenTurn int64 + WrittenAtUnixMs int64 +} + +func (q *Queries) GetAgentBoards(ctx context.Context, dollar_1 []string) ([]GetAgentBoardsRow, error) { + rows, err := q.db.Query(ctx, getAgentBoards, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetAgentBoardsRow + for rows.Next() { + var i GetAgentBoardsRow + if err := rows.Scan( + &i.AgentAccountID, + &i.FocusTaskKey, + &i.HaltKind, + &i.HaltRef, + &i.HaltReason, + &i.WrittenTurn, + &i.WrittenAtUnixMs, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const issuesByCoords = `-- name: IssuesByCoords :many +WITH want AS ( + SELECT unnest($1::smallint[]) AS provider, + unnest($2::text[]) AS host, + unnest($3::text[]) AS repo, + unnest($4::bigint[]) AS number +) +SELECT i.id, i.forge_provider, i.forge_host, i.repo, i.number, + i.title, i.body, i.forge_state, i.url, i.forge_account, i.labels, i.agent_handle, + i.state, i.priority, i.assignee, i.summary, i.branch + FROM issues i + JOIN want w ON i.forge_provider = w.provider AND i.forge_host = w.host + AND (CASE WHEN i.forge_provider = 1 THEN lower(i.repo) ELSE i.repo END) = w.repo + AND i.number = w.number + ORDER BY i.forge_provider, i.forge_host, i.repo, i.number +` + +type IssuesByCoordsParams struct { + ForgeProviders []int16 + ForgeHosts []string + ForgeRepos []string + ForgeNumbers []int64 +} + +type IssuesByCoordsRow struct { + ID string + ForgeProvider int16 + ForgeHost string + Repo string + Number int64 + Title string + Body string + ForgeState string + Url string + ForgeAccount string + Labels []string + AgentHandle string + State int16 + Priority string + Assignee string + Summary string + Branch string +} + +func (q *Queries) IssuesByCoords(ctx context.Context, arg IssuesByCoordsParams) ([]IssuesByCoordsRow, error) { + rows, err := q.db.Query(ctx, issuesByCoords, + arg.ForgeProviders, + arg.ForgeHosts, + arg.ForgeRepos, + arg.ForgeNumbers, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []IssuesByCoordsRow + for rows.Next() { + var i IssuesByCoordsRow + if err := rows.Scan( + &i.ID, + &i.ForgeProvider, + &i.ForgeHost, + &i.Repo, + &i.Number, + &i.Title, + &i.Body, + &i.ForgeState, + &i.Url, + &i.ForgeAccount, + &i.Labels, + &i.AgentHandle, + &i.State, + &i.Priority, + &i.Assignee, + &i.Summary, + &i.Branch, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listAsksByAuthor = `-- name: ListAsksByAuthor :many +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +), +viewer AS ( + SELECT owner_user_id AS uid FROM agent_accounts WHERE account_id = $1 + UNION ALL + SELECT $1 AS uid +), +asks AS ( + SELECT m.id, m.topic_id, m.seq, m.at_unix_ms, ask.value->'ask' AS a + FROM messages m + CROSS JOIN LATERAL jsonb_array_elements(m.blocks) AS ask(value) + WHERE m.author_account_id = $2 + AND ask.value->>'kind' = 'ask' +), +answered_topics AS ( + SELECT DISTINCT topic_id + FROM asks + WHERE COALESCE((a->>'answered')::boolean, FALSE) +), +candidate_messages AS MATERIALIZED ( + SELECT answer.topic_id, answer.seq, answer.at_unix_ms, answer.blocks + FROM messages answer + WHERE answer.topic_id IN (SELECT topic_id FROM answered_topics) + AND answer.at_unix_ms >= $3::bigint +), +answer_candidates AS ( + SELECT answer.topic_id, answer.seq, answer.at_unix_ms, + answer_block.value->'ask_answer'->'ask'->>'ask_id' AS ask_id + FROM candidate_messages answer + CROSS JOIN LATERAL jsonb_array_elements(answer.blocks) AS answer_block(value) + WHERE answer_block.value->>'kind' = 'ask_answer' +), +answered AS ( + SELECT k.id, k.a->>'ask_id' AS ask_id, min(candidate.at_unix_ms) AS at_unix_ms + FROM asks k + JOIN answer_candidates candidate + ON candidate.topic_id = k.topic_id AND candidate.seq > k.seq + AND candidate.ask_id = k.a->>'ask_id' + WHERE COALESCE((k.a->>'answered')::boolean, FALSE) + GROUP BY k.id, k.a->>'ask_id' +) +SELECT k.id AS message_id, t.channel_id, + k.at_unix_ms::bigint AS at_unix_ms, + (k.a->>'ask_id')::text AS ask_id, + COALESCE(k.a->>'ref', '')::text AS ref, + COALESCE(NULLIF(k.a->'questions'->0->>'header', ''), + left(COALESCE(k.a->'questions'->0->>'question', ''), 80))::text AS header, + COALESCE((k.a->>'answered')::boolean, FALSE)::boolean AS answered + FROM asks k + JOIN topics t ON t.id = k.topic_id + LEFT JOIN answered an ON an.id = k.id AND an.ask_id = k.a->>'ask_id' + WHERE (COALESCE((k.a->>'answered')::boolean, FALSE) = FALSE + OR an.at_unix_ms >= $3::bigint) + AND EXISTS ( + SELECT 1 + FROM channels c + WHERE c.id = t.channel_id AND ( + EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = c.id AND cm.account_id = $1 + ) + OR ( + c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 + ) + ) + OR ( + c.parent_agent_id IS NOT NULL + AND (SELECT aa.owner_user_id FROM agent_accounts aa WHERE aa.account_id = c.parent_agent_id) + IN (SELECT uid FROM viewer) + ) + ) + ) + ORDER BY CASE WHEN COALESCE((k.a->>'answered')::boolean, FALSE) THEN 1 ELSE 0 END, + COALESCE(an.at_unix_ms, k.at_unix_ms) DESC, k.seq DESC + LIMIT $4 +` + +type ListAsksByAuthorParams struct { + AccountID string + AuthorAccountID string + AnsweredSince int64 + RowLimit int32 +} + +type ListAsksByAuthorRow struct { + MessageID string + ChannelID string + AtUnixMs int64 + AskID string + Ref string + Header string + Answered bool +} + +// The EXISTS body is ChannelVisibleTo's predicate with the channel spelled +// t.channel_id, so it is diffable against the other copies. +// AnswerAsk posts the answer in the ask's topic, after the ask. +func (q *Queries) ListAsksByAuthor(ctx context.Context, arg ListAsksByAuthorParams) ([]ListAsksByAuthorRow, error) { + rows, err := q.db.Query(ctx, listAsksByAuthor, + arg.AccountID, + arg.AuthorAccountID, + arg.AnsweredSince, + arg.RowLimit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListAsksByAuthorRow + for rows.Next() { + var i ListAsksByAuthorRow + if err := rows.Scan( + &i.MessageID, + &i.ChannelID, + &i.AtUnixMs, + &i.AskID, + &i.Ref, + &i.Header, + &i.Answered, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listOpenAuthoredRefs = `-- name: ListOpenAuthoredRefs :many +SELECT a.forge_provider, a.forge_host, a.repo, a.kind, a.number + FROM forge_authored_artifacts a + WHERE a.agent_account_id = $1 + AND ((a.kind = 2 AND EXISTS ( + SELECT 1 FROM pull_requests p + WHERE p.forge_provider = a.forge_provider AND p.forge_host = a.forge_host + AND p.repo = CASE WHEN a.forge_provider = 1 THEN lower(a.repo) ELSE a.repo END + AND p.number = a.number AND p.forge_state = 'open')) + OR (a.kind = 1 AND EXISTS ( + SELECT 1 FROM issues i + WHERE i.forge_provider = a.forge_provider AND i.forge_host = a.forge_host + AND (CASE WHEN i.forge_provider = 1 THEN lower(i.repo) ELSE i.repo END) = + (CASE WHEN a.forge_provider = 1 THEN lower(a.repo) ELSE a.repo END) + AND i.number = a.number AND i.forge_state = 'open'))) + ORDER BY a.created_at_unix_ms, a.forge_provider, a.forge_host, a.repo, a.kind, a.number +` + +type ListOpenAuthoredRefsRow struct { + ForgeProvider int16 + ForgeHost string + Repo string + Kind int16 + Number int64 +} + +func (q *Queries) ListOpenAuthoredRefs(ctx context.Context, agentAccountID string) ([]ListOpenAuthoredRefsRow, error) { + rows, err := q.db.Query(ctx, listOpenAuthoredRefs, agentAccountID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListOpenAuthoredRefsRow + for rows.Next() { + var i ListOpenAuthoredRefsRow + if err := rows.Scan( + &i.ForgeProvider, + &i.ForgeHost, + &i.Repo, + &i.Kind, + &i.Number, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const lockBoardAgent = `-- name: LockBoardAgent :one + +SELECT account_id FROM agent_accounts WHERE account_id = $1 FOR NO KEY UPDATE +` + +// Agent status board reads and writes. +// NO KEY UPDATE serializes board writers without blocking FK inserts on the agent. +func (q *Queries) LockBoardAgent(ctx context.Context, accountID string) (string, error) { + row := q.db.QueryRow(ctx, lockBoardAgent, accountID) + var account_id string + err := row.Scan(&account_id) + return account_id, err +} + +const pullRequestsByCoords = `-- name: PullRequestsByCoords :many +WITH want AS ( + SELECT unnest($1::smallint[]) AS provider, + unnest($2::text[]) AS host, + unnest($3::text[]) AS repo, + unnest($4::bigint[]) AS number +) +SELECT p.forge_provider, p.forge_host, p.repo, p.number, p.forge_state, + p.forge_created_at, p.forge_updated_at, p.pr + FROM pull_requests p + JOIN want w ON p.forge_provider = w.provider AND p.forge_host = w.host + AND p.repo = w.repo + AND p.number = w.number + ORDER BY p.forge_provider, p.forge_host, p.repo, p.number +` + +type PullRequestsByCoordsParams struct { + ForgeProviders []int16 + ForgeHosts []string + ForgeRepos []string + ForgeNumbers []int64 +} + +type PullRequestsByCoordsRow struct { + ForgeProvider int16 + ForgeHost string + Repo string + Number int64 + ForgeState string + ForgeCreatedAt pgtype.Timestamptz + ForgeUpdatedAt pgtype.Timestamptz + Pr []byte +} + +func (q *Queries) PullRequestsByCoords(ctx context.Context, arg PullRequestsByCoordsParams) ([]PullRequestsByCoordsRow, error) { + rows, err := q.db.Query(ctx, pullRequestsByCoords, + arg.ForgeProviders, + arg.ForgeHosts, + arg.ForgeRepos, + arg.ForgeNumbers, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []PullRequestsByCoordsRow + for rows.Next() { + var i PullRequestsByCoordsRow + if err := rows.Scan( + &i.ForgeProvider, + &i.ForgeHost, + &i.Repo, + &i.Number, + &i.ForgeState, + &i.ForgeCreatedAt, + &i.ForgeUpdatedAt, + &i.Pr, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const searchAgentBoardTasks = `-- name: SearchAgentBoardTasks :many +WITH filtered_tasks AS ( + SELECT t.agent_account_id, t.task_key, t.position + FROM agent_board_tasks t + JOIN agent_boards b ON b.agent_account_id = t.agent_account_id + WHERE (cardinality($1::text[]) = 0 OR t.agent_account_id = ANY($1::text[])) + AND ($2::text = '' OR to_tsvector('english', t.title || ' ' || t.note) + @@ websearch_to_tsquery('english', $2::text)) + AND ($3::text = '' OR $3::text = ANY(t.refs) OR t.primary_ref = $3::text) + AND ($4::smallint = 0 OR EXISTS ( + SELECT 1 FROM jsonb_array_elements(t.waiting) AS w(value) + WHERE w.value->>'on' = CASE $4::smallint + WHEN 1 THEN 'BOARD_WAITING_ON_OPERATOR' + WHEN 2 THEN 'BOARD_WAITING_ON_AGENT' + WHEN 3 THEN 'BOARD_WAITING_ON_EXTERNAL' + ELSE '' END)) + AND ($5::smallint = 0 OR t.state = $5::smallint) + AND ($6::smallint = 0 OR b.halt_kind = $6::smallint) +), +matching_agents AS ( + SELECT DISTINCT agent_account_id + FROM filtered_tasks + ORDER BY agent_account_id + LIMIT $7 +) +SELECT f.agent_account_id, f.task_key + FROM filtered_tasks f + JOIN matching_agents a USING (agent_account_id) + ORDER BY f.agent_account_id, f.position, f.task_key +` + +type SearchAgentBoardTasksParams struct { + AgentAccountIds []string + Query string + Ref string + WaitingOn int16 + State int16 + Halt int16 + AgentLimit int32 +} + +type SearchAgentBoardTasksRow struct { + AgentAccountID string + TaskKey string +} + +func (q *Queries) SearchAgentBoardTasks(ctx context.Context, arg SearchAgentBoardTasksParams) ([]SearchAgentBoardTasksRow, error) { + rows, err := q.db.Query(ctx, searchAgentBoardTasks, + arg.AgentAccountIds, + arg.Query, + arg.Ref, + arg.WaitingOn, + arg.State, + arg.Halt, + arg.AgentLimit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []SearchAgentBoardTasksRow + for rows.Next() { + var i SearchAgentBoardTasksRow + if err := rows.Scan(&i.AgentAccountID, &i.TaskKey); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const upsertAgentBoard = `-- name: UpsertAgentBoard :exec +INSERT INTO agent_boards + (agent_account_id, focus_task_key, halt_kind, halt_ref, halt_reason, + written_turn, written_at_unix_ms) +VALUES ($1, CASE WHEN $2::boolean THEN $3::text ELSE '' END, + CASE WHEN $4::boolean THEN $5::smallint ELSE 0 END, + CASE WHEN $4::boolean THEN $6::text ELSE '' END, + CASE WHEN $4::boolean THEN $7::text ELSE '' END, + CASE WHEN $9::boolean THEN 0 ELSE $10::bigint END, + CASE WHEN $9::boolean THEN 0 ELSE $11::bigint END) +ON CONFLICT (agent_account_id) DO UPDATE + SET focus_task_key = CASE WHEN $2::boolean THEN $3::text ELSE agent_boards.focus_task_key END, + halt_kind = CASE WHEN $8::boolean THEN 0 + WHEN $4::boolean THEN $5::smallint + ELSE agent_boards.halt_kind END, + halt_ref = CASE WHEN $8::boolean THEN '' + WHEN $4::boolean THEN $6::text + ELSE agent_boards.halt_ref END, + halt_reason = CASE WHEN $8::boolean THEN '' + WHEN $4::boolean THEN $7::text + ELSE agent_boards.halt_reason END, + written_turn = CASE WHEN $9::boolean THEN agent_boards.written_turn ELSE $10::bigint END, + written_at_unix_ms = CASE WHEN $9::boolean THEN agent_boards.written_at_unix_ms ELSE $11::bigint END +` + +type UpsertAgentBoardParams struct { + AgentAccountID string + Column2 bool + Column3 string + Column4 bool + Column5 int16 + Column6 string + Column7 string + Column8 bool + Column9 bool + Column10 int64 + Column11 int64 +} + +func (q *Queries) UpsertAgentBoard(ctx context.Context, arg UpsertAgentBoardParams) error { + _, err := q.db.Exec(ctx, upsertAgentBoard, + arg.AgentAccountID, + arg.Column2, + arg.Column3, + arg.Column4, + arg.Column5, + arg.Column6, + arg.Column7, + arg.Column8, + arg.Column9, + arg.Column10, + arg.Column11, + ) + return err +} + +const upsertAgentBoardTask = `-- name: UpsertAgentBoardTask :exec +INSERT INTO agent_board_tasks + (agent_account_id, task_key, title, primary_ref, refs, state, note, + waiting, subagents, position, written_at_unix_ms) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) +ON CONFLICT (agent_account_id, task_key) DO UPDATE + SET title = EXCLUDED.title, + primary_ref = EXCLUDED.primary_ref, + refs = EXCLUDED.refs, + state = EXCLUDED.state, + note = EXCLUDED.note, + waiting = EXCLUDED.waiting, + subagents = EXCLUDED.subagents, + position = EXCLUDED.position, + written_at_unix_ms = EXCLUDED.written_at_unix_ms +` + +type UpsertAgentBoardTaskParams struct { + AgentAccountID string + TaskKey string + Title string + PrimaryRef string + Refs []string + State int16 + Note string + Waiting []byte + Subagents []byte + Position int32 + WrittenAtUnixMs int64 +} + +func (q *Queries) UpsertAgentBoardTask(ctx context.Context, arg UpsertAgentBoardTaskParams) error { + _, err := q.db.Exec(ctx, upsertAgentBoardTask, + arg.AgentAccountID, + arg.TaskKey, + arg.Title, + arg.PrimaryRef, + arg.Refs, + arg.State, + arg.Note, + arg.Waiting, + arg.Subagents, + arg.Position, + arg.WrittenAtUnixMs, + ) + return err +} diff --git a/go/internal/store/db/models.go b/go/internal/store/db/models.go index d95972b2f..0b71ade37 100644 --- a/go/internal/store/db/models.go +++ b/go/internal/store/db/models.go @@ -58,6 +58,36 @@ type AgentActivity struct { TenantID string } +type AgentBoard struct { + AgentAccountID string + FocusTaskKey string + HaltKind int16 + HaltRef string + HaltReason string + WrittenTurn int64 + WrittenAtUnixMs int64 + CreatedAt pgtype.Timestamptz + UpdatedAt pgtype.Timestamptz + TenantID string +} + +type AgentBoardTask struct { + AgentAccountID string + TaskKey string + Title string + PrimaryRef string + Refs []string + State int16 + Note string + Waiting []byte + Subagents []byte + Position int32 + WrittenAtUnixMs int64 + CreatedAt pgtype.Timestamptz + UpdatedAt pgtype.Timestamptz + TenantID string +} + type AgentConfigBundle struct { Singleton bool Version string diff --git a/go/internal/store/db/querier.go b/go/internal/store/db/querier.go index 4e323f12d..0a1a6f541 100644 --- a/go/internal/store/db/querier.go +++ b/go/internal/store/db/querier.go @@ -136,6 +136,7 @@ type Querier interface { // interface still consumes (value-free, all scopes). DeclaredSecrets(ctx context.Context) ([]DeclaredSecretsRow, error) DeclaredServerSecrets(ctx context.Context) ([]ServerSecret, error) + DeleteAgentBoardTasks(ctx context.Context, arg DeleteAgentBoardTasksParams) error DeleteAgentConfig(ctx context.Context) error // Scoped to the calling agent (id AND agent). RETURNING the coordinate drives the // one-tx GC of the artifact cursor when this was the last subscription. @@ -211,6 +212,8 @@ type Querier interface { GetAccountByGlobalHandle(ctx context.Context, handle string) (GetAccountByGlobalHandleRow, error) GetAccountByOwnerHandle(ctx context.Context, arg GetAccountByOwnerHandleParams) (GetAccountByOwnerHandleRow, error) GetAccountHandle(ctx context.Context, accountID string) (string, error) + GetAgentBoardTasks(ctx context.Context, dollar_1 []string) ([]GetAgentBoardTasksRow, error) + GetAgentBoards(ctx context.Context, dollar_1 []string) ([]GetAgentBoardsRow, error) GetAgentOwner(ctx context.Context, accountID string) (string, error) GetAgentParent(ctx context.Context, accountID string) (pgtype.Text, error) GetAgentWorkspaceID(ctx context.Context, agentAccountID string) (string, error) @@ -397,12 +400,17 @@ type Querier interface { // (scope_kind 1) referential check the UpsertSecret door runs in lieu of an FK // (A9). The agent-scope check reuses IsAgentAccount. IsUserAccount(ctx context.Context, accountID string) (bool, error) + IssuesByCoords(ctx context.Context, arg IssuesByCoordsParams) ([]IssuesByCoordsRow, error) LatestCheckpointSeq(ctx context.Context, sessionID string) (int64, error) LatestSessionForAccount(ctx context.Context, agentAccountID string) (string, error) LinearAgentSession(ctx context.Context, linearSessionID string) (LinearAgentSessionRow, error) // An account's own rows only; the owner's grants are not included. ListAgentForgeScopeRepos(ctx context.Context, arg ListAgentForgeScopeReposParams) ([]string, error) ListAgentPlacementsForRunner(ctx context.Context, runnerID string) ([]ListAgentPlacementsForRunnerRow, error) + // The EXISTS body is ChannelVisibleTo's predicate with the channel spelled + // t.channel_id, so it is diffable against the other copies. + // AnswerAsk posts the answer in the ask's topic, after the ask. + ListAsksByAuthor(ctx context.Context, arg ListAsksByAuthorParams) ([]ListAsksByAuthorRow, error) ListAuthoredArtifactsByAgent(ctx context.Context, agentAccountID string) ([]ForgeAuthoredArtifact, error) ListChannelGroups(ctx context.Context, accountID string) ([]ListChannelGroupsRow, error) ListChannels(ctx context.Context, accountID string) ([]ListChannelsRow, error) @@ -418,6 +426,7 @@ type Querier interface { ListGatewayCredentials(ctx context.Context, arg ListGatewayCredentialsParams) ([]ListGatewayCredentialsRow, error) ListIssues(ctx context.Context) ([]ListIssuesRow, error) ListMessages(ctx context.Context, arg ListMessagesParams) ([]ListMessagesRow, error) + ListOpenAuthoredRefs(ctx context.Context, agentAccountID string) ([]ListOpenAuthoredRefsRow, error) ListOpenPullRequests(ctx context.Context, arg ListOpenPullRequestsParams) ([]int64, error) ListPullRequestLinks(ctx context.Context, arg ListPullRequestLinksParams) ([]ListPullRequestLinksRow, error) // ListTenantIDs lists every tenant. tenants has no row-level security, so the @@ -448,6 +457,9 @@ type Querier interface { // back to the domain time.Time / ForgeRepoSubscription. LoadForgeRepoWatermark(ctx context.Context, arg LoadForgeRepoWatermarkParams) (LoadForgeRepoWatermarkRow, error) LoadPRsBackfilledAt(ctx context.Context, arg LoadPRsBackfilledAtParams) (pgtype.Timestamptz, error) + // Agent status board reads and writes. + // NO KEY UPDATE serializes board writers without blocking FK inserts on the agent. + LockBoardAgent(ctx context.Context, accountID string) (string, error) // Channel-pins (pinned board) queries (sqlc adoption T3, RIG-3034). These // replace the inline SQL literals in internal/store/channel_pins.go; the // hand-written Store methods and the in-tx FOR UPDATE lock / cap-check control @@ -549,6 +561,7 @@ type Querier interface { PruneTranscriptEntries(ctx context.Context, arg PruneTranscriptEntriesParams) (int64, error) PullRequestForgeUpdatedAt(ctx context.Context, arg PullRequestForgeUpdatedAtParams) (pgtype.Timestamptz, error) PullRequestLinks(ctx context.Context, arg PullRequestLinksParams) ([]PullRequestLinksRow, error) + PullRequestsByCoords(ctx context.Context, arg PullRequestsByCoordsParams) ([]PullRequestsByCoordsRow, error) // Explicit links attach directly; a closing reference attaches only when none // of the PR's explicit targets is an issue on the board. PullRequestsForIssues(ctx context.Context, arg PullRequestsForIssuesParams) ([]PullRequestsForIssuesRow, error) @@ -641,6 +654,7 @@ type Querier interface { // selects a genuinely existing column from a genuinely existing table // (tenants, 0001_init.sql), so sqlc compiles it against the real schema. ScaffoldGetTenant(ctx context.Context, id string) (Tenant, error) + SearchAgentBoardTasks(ctx context.Context, arg SearchAgentBoardTasksParams) ([]SearchAgentBoardTasksRow, error) SearchIssues(ctx context.Context, arg SearchIssuesParams) ([]SearchIssuesRow, error) SearchMessages(ctx context.Context, arg SearchMessagesParams) ([]SearchMessagesRow, error) // SecretRecordsForAgent collapses the A9 precedence in SQL: DISTINCT ON keeps the @@ -749,6 +763,8 @@ type Querier interface { // UpdateModelRegistry applies a compare-and-set only to a live row. UpdateModelRegistry(ctx context.Context, arg UpdateModelRegistryParams) (int64, error) UpdateTopicLastSeq(ctx context.Context, arg UpdateTopicLastSeqParams) error + UpsertAgentBoard(ctx context.Context, arg UpsertAgentBoardParams) error + UpsertAgentBoardTask(ctx context.Context, arg UpsertAgentBoardTaskParams) error UpsertChannelMember(ctx context.Context, arg UpsertChannelMemberParams) error // A TREE channel's per-account subscribe override; it has no member rows. UpsertChannelSubscription(ctx context.Context, arg UpsertChannelSubscriptionParams) error diff --git a/go/internal/store/migrations/0001_init.sql b/go/internal/store/migrations/0001_init.sql index 7f27eddfa..a30bf74bd 100644 --- a/go/internal/store/migrations/0001_init.sql +++ b/go/internal/store/migrations/0001_init.sql @@ -411,6 +411,10 @@ CREATE INDEX messages_mentions_unrouted_idx ON messages (seq) WHERE mentions_rou -- filters channel_id, keying on the same table-monotonic seq. CREATE INDEX messages_topic_seq_idx ON messages (topic_id, seq DESC); +-- Answer-window reads constrain one topic and timestamp lower bound before +-- expanding answer blocks. +CREATE INDEX messages_topic_at_idx ON messages (tenant_id, topic_id, at_unix_ms); + -- Full-text search index (design.md:1137-1139): GIN over the generated tsvector. CREATE INDEX messages_search_idx ON messages USING gin (search_tsv); @@ -893,6 +897,38 @@ CREATE TABLE agent_activity ( activity_at_unix_ms BIGINT NOT NULL, tenant_id TEXT NOT NULL DEFAULT current_setting('compass.tenant_id', TRUE) ); +-- The agent's durable status board is split into one metadata row and ordered +-- task rows so updates can merge individual tasks without replacing the board. +CREATE TABLE agent_boards ( + agent_account_id TEXT PRIMARY KEY REFERENCES agent_accounts (account_id) ON DELETE RESTRICT, + focus_task_key TEXT NOT NULL DEFAULT '', + halt_kind SMALLINT NOT NULL DEFAULT 0 CHECK (halt_kind BETWEEN 0 AND 3), + halt_ref TEXT NOT NULL DEFAULT '', + halt_reason TEXT NOT NULL DEFAULT '', + written_turn BIGINT NOT NULL DEFAULT 0, + written_at_unix_ms BIGINT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + tenant_id TEXT NOT NULL DEFAULT current_setting('compass.tenant_id', TRUE) +); + +CREATE TABLE agent_board_tasks ( + agent_account_id TEXT NOT NULL REFERENCES agent_boards (agent_account_id) ON DELETE CASCADE, + task_key TEXT NOT NULL CHECK (task_key <> ''), + title TEXT NOT NULL, + primary_ref TEXT NOT NULL DEFAULT '', + refs TEXT[] NOT NULL DEFAULT '{}', + state SMALLINT NOT NULL DEFAULT 1 CHECK (state BETWEEN 1 AND 5), + note TEXT NOT NULL DEFAULT '', + waiting JSONB NOT NULL DEFAULT '[]', + subagents JSONB NOT NULL DEFAULT '[]', + position INT NOT NULL, + written_at_unix_ms BIGINT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + tenant_id TEXT NOT NULL DEFAULT current_setting('compass.tenant_id', TRUE), + PRIMARY KEY (agent_account_id, task_key) +); -- ── Agent config bundle (fleet singleton) ───────────────────────────────────── -- The Server-side fleet CONFIG-BUNDLE store (RIG-1624 T1): the ONE fleet-wide @@ -1635,6 +1671,7 @@ DECLARE 'agent_session_blobs', 'agent_delivery_cursors', 'owed_mentions', 'agent_activity', 'agent_forge_subscriptions', 'forge_authored_artifacts', + 'agent_boards', 'agent_board_tasks', 'linear_agent_sessions', 'issues', 'forge_repo_subscriptions', 'forge_artifact_cursors', 'forge_state_transitions', 'account_forge_scopes', @@ -1723,6 +1760,7 @@ DECLARE 'model_registry', 'forge_repo_subscriptions', 'forge_state_transitions', + 'agent_boards', 'agent_board_tasks', 'pull_requests', 'server_secrets', 'server_key_state', diff --git a/go/internal/store/queries/agent_boards.sql b/go/internal/store/queries/agent_boards.sql new file mode 100644 index 000000000..a5aa442d0 --- /dev/null +++ b/go/internal/store/queries/agent_boards.sql @@ -0,0 +1,232 @@ +-- Agent status board reads and writes. + +-- name: LockBoardAgent :one +-- NO KEY UPDATE serializes board writers without blocking FK inserts on the agent. +SELECT account_id FROM agent_accounts WHERE account_id = $1 FOR NO KEY UPDATE; + +-- name: UpsertAgentBoard :exec +INSERT INTO agent_boards + (agent_account_id, focus_task_key, halt_kind, halt_ref, halt_reason, + written_turn, written_at_unix_ms) +VALUES ($1, CASE WHEN $2::boolean THEN $3::text ELSE '' END, + CASE WHEN $4::boolean THEN $5::smallint ELSE 0 END, + CASE WHEN $4::boolean THEN $6::text ELSE '' END, + CASE WHEN $4::boolean THEN $7::text ELSE '' END, + CASE WHEN $9::boolean THEN 0 ELSE $10::bigint END, + CASE WHEN $9::boolean THEN 0 ELSE $11::bigint END) +ON CONFLICT (agent_account_id) DO UPDATE + SET focus_task_key = CASE WHEN $2::boolean THEN $3::text ELSE agent_boards.focus_task_key END, + halt_kind = CASE WHEN $8::boolean THEN 0 + WHEN $4::boolean THEN $5::smallint + ELSE agent_boards.halt_kind END, + halt_ref = CASE WHEN $8::boolean THEN '' + WHEN $4::boolean THEN $6::text + ELSE agent_boards.halt_ref END, + halt_reason = CASE WHEN $8::boolean THEN '' + WHEN $4::boolean THEN $7::text + ELSE agent_boards.halt_reason END, + written_turn = CASE WHEN $9::boolean THEN agent_boards.written_turn ELSE $10::bigint END, + written_at_unix_ms = CASE WHEN $9::boolean THEN agent_boards.written_at_unix_ms ELSE $11::bigint END; + +-- name: UpsertAgentBoardTask :exec +INSERT INTO agent_board_tasks + (agent_account_id, task_key, title, primary_ref, refs, state, note, + waiting, subagents, position, written_at_unix_ms) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) +ON CONFLICT (agent_account_id, task_key) DO UPDATE + SET title = EXCLUDED.title, + primary_ref = EXCLUDED.primary_ref, + refs = EXCLUDED.refs, + state = EXCLUDED.state, + note = EXCLUDED.note, + waiting = EXCLUDED.waiting, + subagents = EXCLUDED.subagents, + position = EXCLUDED.position, + written_at_unix_ms = EXCLUDED.written_at_unix_ms; + +-- name: DeleteAgentBoardTasks :exec +DELETE FROM agent_board_tasks + WHERE agent_account_id = $1 AND task_key = ANY($2::text[]); + +-- name: GetAgentBoards :many +SELECT agent_account_id, focus_task_key, halt_kind, halt_ref, halt_reason, + written_turn, written_at_unix_ms + FROM agent_boards + WHERE agent_account_id = ANY($1::text[]); + +-- name: GetAgentBoardTasks :many +SELECT agent_account_id, task_key, title, primary_ref, refs, state, note, + waiting, subagents, position, written_at_unix_ms + FROM agent_board_tasks + WHERE agent_account_id = ANY($1::text[]) + ORDER BY position, task_key; + +-- name: SearchAgentBoardTasks :many +WITH filtered_tasks AS ( + SELECT t.agent_account_id, t.task_key, t.position + FROM agent_board_tasks t + JOIN agent_boards b ON b.agent_account_id = t.agent_account_id + WHERE (cardinality(@agent_account_ids::text[]) = 0 OR t.agent_account_id = ANY(@agent_account_ids::text[])) + AND (@query::text = '' OR to_tsvector('english', t.title || ' ' || t.note) + @@ websearch_to_tsquery('english', @query::text)) + AND (@ref::text = '' OR @ref::text = ANY(t.refs) OR t.primary_ref = @ref::text) + AND (@waiting_on::smallint = 0 OR EXISTS ( + SELECT 1 FROM jsonb_array_elements(t.waiting) AS w(value) + WHERE w.value->>'on' = CASE @waiting_on::smallint + WHEN 1 THEN 'BOARD_WAITING_ON_OPERATOR' + WHEN 2 THEN 'BOARD_WAITING_ON_AGENT' + WHEN 3 THEN 'BOARD_WAITING_ON_EXTERNAL' + ELSE '' END)) + AND (@state::smallint = 0 OR t.state = @state::smallint) + AND (@halt::smallint = 0 OR b.halt_kind = @halt::smallint) +), +matching_agents AS ( + SELECT DISTINCT agent_account_id + FROM filtered_tasks + ORDER BY agent_account_id + LIMIT sqlc.arg(agent_limit) +) +SELECT f.agent_account_id, f.task_key + FROM filtered_tasks f + JOIN matching_agents a USING (agent_account_id) + ORDER BY f.agent_account_id, f.position, f.task_key; + +-- name: ListAsksByAuthor :many +-- The EXISTS body is ChannelVisibleTo's predicate with the channel spelled +-- t.channel_id, so it is diffable against the other copies. +WITH RECURSIVE ancestry AS ( + SELECT id, parent_group_id, visibility AS min_vis + FROM channel_groups + UNION ALL + SELECT a.id, g.parent_group_id, LEAST(a.min_vis, g.visibility) + FROM ancestry a + JOIN channel_groups g ON g.id = a.parent_group_id +), +effective AS ( + SELECT id, MIN(min_vis) AS eff_vis + FROM ancestry + GROUP BY id +), +viewer AS ( + SELECT owner_user_id AS uid FROM agent_accounts WHERE account_id = $1 + UNION ALL + SELECT $1 AS uid +), +asks AS ( + SELECT m.id, m.topic_id, m.seq, m.at_unix_ms, ask.value->'ask' AS a + FROM messages m + CROSS JOIN LATERAL jsonb_array_elements(m.blocks) AS ask(value) + WHERE m.author_account_id = $2 + AND ask.value->>'kind' = 'ask' +), +answered_topics AS ( + SELECT DISTINCT topic_id + FROM asks + WHERE COALESCE((a->>'answered')::boolean, FALSE) +), +-- AnswerAsk posts the answer in the ask's topic, after the ask. +candidate_messages AS MATERIALIZED ( + SELECT answer.topic_id, answer.seq, answer.at_unix_ms, answer.blocks + FROM messages answer + WHERE answer.topic_id IN (SELECT topic_id FROM answered_topics) + AND answer.at_unix_ms >= sqlc.arg(answered_since)::bigint +), +answer_candidates AS ( + SELECT answer.topic_id, answer.seq, answer.at_unix_ms, + answer_block.value->'ask_answer'->'ask'->>'ask_id' AS ask_id + FROM candidate_messages answer + CROSS JOIN LATERAL jsonb_array_elements(answer.blocks) AS answer_block(value) + WHERE answer_block.value->>'kind' = 'ask_answer' +), +answered AS ( + SELECT k.id, k.a->>'ask_id' AS ask_id, min(candidate.at_unix_ms) AS at_unix_ms + FROM asks k + JOIN answer_candidates candidate + ON candidate.topic_id = k.topic_id AND candidate.seq > k.seq + AND candidate.ask_id = k.a->>'ask_id' + WHERE COALESCE((k.a->>'answered')::boolean, FALSE) + GROUP BY k.id, k.a->>'ask_id' +) +SELECT k.id AS message_id, t.channel_id, + k.at_unix_ms::bigint AS at_unix_ms, + (k.a->>'ask_id')::text AS ask_id, + COALESCE(k.a->>'ref', '')::text AS ref, + COALESCE(NULLIF(k.a->'questions'->0->>'header', ''), + left(COALESCE(k.a->'questions'->0->>'question', ''), 80))::text AS header, + COALESCE((k.a->>'answered')::boolean, FALSE)::boolean AS answered + FROM asks k + JOIN topics t ON t.id = k.topic_id + LEFT JOIN answered an ON an.id = k.id AND an.ask_id = k.a->>'ask_id' + WHERE (COALESCE((k.a->>'answered')::boolean, FALSE) = FALSE + OR an.at_unix_ms >= sqlc.arg(answered_since)::bigint) + AND EXISTS ( + SELECT 1 + FROM channels c + WHERE c.id = t.channel_id AND ( + EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.channel_id = c.id AND cm.account_id = $1 + ) + OR ( + c.kind = 0 AND c.group_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM effective e WHERE e.id = c.group_id AND e.eff_vis = 1 + ) + ) + OR ( + c.parent_agent_id IS NOT NULL + AND (SELECT aa.owner_user_id FROM agent_accounts aa WHERE aa.account_id = c.parent_agent_id) + IN (SELECT uid FROM viewer) + ) + ) + ) + ORDER BY CASE WHEN COALESCE((k.a->>'answered')::boolean, FALSE) THEN 1 ELSE 0 END, + COALESCE(an.at_unix_ms, k.at_unix_ms) DESC, k.seq DESC + LIMIT sqlc.arg(row_limit); + +-- name: ListOpenAuthoredRefs :many +SELECT a.forge_provider, a.forge_host, a.repo, a.kind, a.number + FROM forge_authored_artifacts a + WHERE a.agent_account_id = $1 + AND ((a.kind = 2 AND EXISTS ( + SELECT 1 FROM pull_requests p + WHERE p.forge_provider = a.forge_provider AND p.forge_host = a.forge_host + AND p.repo = CASE WHEN a.forge_provider = 1 THEN lower(a.repo) ELSE a.repo END + AND p.number = a.number AND p.forge_state = 'open')) + OR (a.kind = 1 AND EXISTS ( + SELECT 1 FROM issues i + WHERE i.forge_provider = a.forge_provider AND i.forge_host = a.forge_host + AND (CASE WHEN i.forge_provider = 1 THEN lower(i.repo) ELSE i.repo END) = + (CASE WHEN a.forge_provider = 1 THEN lower(a.repo) ELSE a.repo END) + AND i.number = a.number AND i.forge_state = 'open'))) + ORDER BY a.created_at_unix_ms, a.forge_provider, a.forge_host, a.repo, a.kind, a.number; + +-- name: PullRequestsByCoords :many +WITH want AS ( + SELECT unnest(@forge_providers::smallint[]) AS provider, + unnest(@forge_hosts::text[]) AS host, + unnest(@forge_repos::text[]) AS repo, + unnest(@forge_numbers::bigint[]) AS number +) +SELECT p.forge_provider, p.forge_host, p.repo, p.number, p.forge_state, + p.forge_created_at, p.forge_updated_at, p.pr + FROM pull_requests p + JOIN want w ON p.forge_provider = w.provider AND p.forge_host = w.host + AND p.repo = w.repo + AND p.number = w.number + ORDER BY p.forge_provider, p.forge_host, p.repo, p.number; + +-- name: IssuesByCoords :many +WITH want AS ( + SELECT unnest(@forge_providers::smallint[]) AS provider, + unnest(@forge_hosts::text[]) AS host, + unnest(@forge_repos::text[]) AS repo, + unnest(@forge_numbers::bigint[]) AS number +) +SELECT i.id, i.forge_provider, i.forge_host, i.repo, i.number, + i.title, i.body, i.forge_state, i.url, i.forge_account, i.labels, i.agent_handle, + i.state, i.priority, i.assignee, i.summary, i.branch + FROM issues i + JOIN want w ON i.forge_provider = w.provider AND i.forge_host = w.host + AND (CASE WHEN i.forge_provider = 1 THEN lower(i.repo) ELSE i.repo END) = w.repo + AND i.number = w.number + ORDER BY i.forge_provider, i.forge_host, i.repo, i.number; diff --git a/go/internal/store/rls_pgtest_test.go b/go/internal/store/rls_pgtest_test.go index a7d2e9c26..aae0261f4 100644 --- a/go/internal/store/rls_pgtest_test.go +++ b/go/internal/store/rls_pgtest_test.go @@ -642,6 +642,7 @@ func TestRLSCatalogEnabledAndForced(t *testing.T) { "agent_session_transcript_entries", "agent_session_archive_segments", "agent_delivery_cursors", "owed_mentions", "agent_activity", "agent_forge_subscriptions", "forge_authored_artifacts", "agent_session_blobs", + "agent_boards", "agent_board_tasks", "linear_agent_sessions", "issues", "forge_repo_subscriptions", "forge_artifact_cursors", "forge_state_transitions", diff --git a/go/internal/store/store.go b/go/internal/store/store.go index 121b02a82..e10dfcf00 100644 --- a/go/internal/store/store.go +++ b/go/internal/store/store.go @@ -51,6 +51,8 @@ type Store struct { // Set once in Open, immutable thereafter, so it is safe for concurrent use // exactly like the pool it wraps. q *db.Queries + // afterBoardMetaRead is a test seam for the snapshot test. + afterBoardMetaRead func() // objectStore is the archive-tier object-store seam (RIG-1667 T4), injected // via SetObjectStore. nil until slice B wires a real client (store tests // inject an in-memory fake); a flush against a nil store fails loudly. diff --git a/go/internal/store/types.go b/go/internal/store/types.go index 3f8c19f64..60577c4f6 100644 --- a/go/internal/store/types.go +++ b/go/internal/store/types.go @@ -455,6 +455,8 @@ type Ask struct { // primitive; internal callers may set one), so uniqueness rests on the edge // strip, not on the store rejecting a supplied id. AskID string + // Ref is the explicit issue or pull request this ask concerns. + Ref string // Questions are the questions the agent asked, in order. At least one; a // zero-question ask is rejected on both marshal and unmarshal. Questions []AskQuestion