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