package store import ( "context" "database/sql" "encoding/json" "errors" "time" "g.fc.run/misc/byre/internal/api" ) const reviewColumns = `id, task_id, snapshot_n, coalesce(agent_session_id, ''), vendor, model, state, result, error, created_at, finished_at` func (s *Store) CreateReview(ctx context.Context, r api.Review) error { finishedAt := "" if r.FinishedAt != nil { finishedAt = formatTime(*r.FinishedAt) } return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { _, err := tx.ExecContext(ctx, ` INSERT INTO reviews (id, task_id, snapshot_n, agent_session_id, vendor, model, state, created_at, finished_at) VALUES (?, ?, ?, nullif(?, ''), ?, ?, ?, ?, ?) `, r.Id, r.TaskId, r.SnapshotN, r.AgentSessionId, r.Vendor, r.Model, string(r.State), formatTime(r.CreatedAt), finishedAt) if err != nil { return err } if r.State == api.ReviewStateRunning { return emitReview(ctx, tx, emit, "review.started", r.Id) } return emitReview(ctx, tx, emit, "review.finished", r.Id) }) } func (s *Store) GetReview(ctx context.Context, id string) (api.Review, error) { r, err := scanReview(s.db.QueryRowContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE id = ?`, id)) if errors.Is(err, sql.ErrNoRows) { return r, ErrNotFound } return r, err } func (s *Store) ReviewBySession(ctx context.Context, sessionID string) (api.Review, error) { r, err := scanReview(s.db.QueryRowContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE agent_session_id = ?`, sessionID)) if errors.Is(err, sql.ErrNoRows) { return r, ErrNotFound } return r, err } func (s *Store) ListReviews(ctx context.Context, taskID string) ([]api.Review, error) { rows, err := s.db.QueryContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE task_id = ? ORDER BY snapshot_n`, taskID) if err != nil { return nil, err } return scanReviews(rows) } func (s *Store) RunningReviews(ctx context.Context) ([]api.Review, error) { rows, err := s.db.QueryContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE state = ? ORDER BY created_at`, string(api.ReviewStateRunning)) if err != nil { return nil, err } return scanReviews(rows) } func (s *Store) FinishReview(ctx context.Context, id string, state api.ReviewState, result map[string]any, errText string) error { text := "" if result != nil { b, err := json.Marshal(result) if err != nil { return err } text = string(b) } return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { res, err := tx.ExecContext(ctx, `UPDATE reviews SET state = ?, result = ?, error = ?, finished_at = ? WHERE id = ? AND state = ?`, string(state), text, errText, now(), id, string(api.ReviewStateRunning)) if err != nil { return err } n, err := res.RowsAffected() if err != nil { return err } if n == 0 { return ErrNotFound } return emitReview(ctx, tx, emit, "review.finished", id) }) } func (s *Store) ReviewerSessionIDs(ctx context.Context, taskID string) (map[string]bool, error) { rows, err := s.db.QueryContext(ctx, `SELECT id FROM agent_sessions WHERE task_id = ? AND role = ?`, taskID, string(api.Reviewer)) if err != nil { return nil, err } defer rows.Close() ids := map[string]bool{} for rows.Next() { id := "" err = rows.Scan(&id) if err != nil { return nil, err } ids[id] = true } return ids, rows.Err() } func (s *Store) FirstTaskMessage(ctx context.Context, taskID string) (string, error) { text := "" err := s.db.QueryRowContext(ctx, ` SELECT coalesce(json_extract(payload, '$.text'), '') FROM events WHERE task_id = ? AND kind = 'user.message' AND agent_session_id IN (SELECT id FROM agent_sessions WHERE task_id = ? AND role = ?) ORDER BY seq LIMIT 1 `, taskID, taskID, string(api.Implementer)).Scan(&text) if errors.Is(err, sql.ErrNoRows) { return "", nil } return text, err } func emitReview(ctx context.Context, tx *sql.Tx, emit emitFunc, kind, id string) error { r, err := scanReview(tx.QueryRowContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE id = ?`, id)) if err != nil { return err } projectID := "" err = tx.QueryRowContext(ctx, `SELECT project_id FROM tasks WHERE id = ?`, r.TaskId).Scan(&projectID) if err != nil { return err } return emit(api.Event{Kind: kind, ProjectId: &projectID, TaskId: &r.TaskId, Actor: "byre", Payload: ToPayload(r)}) } func questionReviews(ctx context.Context, q querier, item api.InboxItem) ([]api.Review, error) { if item.Kind != api.Question { return []api.Review{}, nil } rows, err := q.QueryContext(ctx, ` SELECT `+reviewColumns+` FROM reviews WHERE task_id = ? AND snapshot_n = ( SELECT min(n) FROM snapshots WHERE task_id = ? AND created_at >= ? ) `, item.TaskId, item.TaskId, formatTime(item.CreatedAt)) if err != nil { return nil, err } return scanReviews(rows) } func scanReviews(rows *sql.Rows) ([]api.Review, error) { defer rows.Close() reviews := []api.Review{} for rows.Next() { r, err := scanReview(rows) if err != nil { return nil, err } reviews = append(reviews, r) } return reviews, rows.Err() } func scanReview(row scanner) (api.Review, error) { r := api.Review{} state := "" result := "" createdAt := "" finishedAt := "" err := row.Scan(&r.Id, &r.TaskId, &r.SnapshotN, &r.AgentSessionId, &r.Vendor, &r.Model, &state, &result, &r.Error, &createdAt, &finishedAt) if err != nil { return r, err } r.State = api.ReviewState(state) if result != "" { res := map[string]any{} err = json.Unmarshal([]byte(result), &res) if err != nil { return r, err } r.Result = &res } r.CreatedAt, err = parseTime(createdAt) if err != nil { return r, err } if finishedAt != "" { var t time.Time t, err = parseTime(finishedAt) r.FinishedAt = &t } return r, err }