Review every new snapshot with an agent from the other vendor

2026-10-15 16:33:17 -07:00
parent a539ddab43
commit e9039cceb4
39 changed files with 1737 additions and 218 deletions
@@ -11,7 +11,7 @@ import (
 	"g.fc.run/misc/byre/internal/api"
 )

-var priorities = []api.InboxItemPriority{api.Blocked, api.Gate, api.Review, api.Fyi}
+var priorities = []api.InboxItemPriority{api.Blocked, api.Gate, api.ReviewPriority, api.Fyi}

 func priorityRank(p api.InboxItemPriority) int {
 	for i, q := range priorities {
@@ -357,6 +357,10 @@ func getItem(ctx context.Context, q querier, id string) (api.InboxItem, error) {
 	}

 	item.Responses, err = getResponses(ctx, q, id)
+	if err != nil {
+		return item, err
+	}
+	item.Reviews, err = questionReviews(ctx, q, item)
 	return item, err
 }

@@ -0,0 +1,190 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"g.fc.run/misc/byre/internal/api"
+)
+
+const reviewColumns = `id, task_id, snapshot_n, agent_session_id, vendor, model, state, result, error, created_at, finished_at`
+
+func (s *Store) CreateReview(ctx context.Context, r api.Review) error {
+	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)
+			VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+		`, r.Id, r.TaskId, r.SnapshotN, r.AgentSessionId, r.Vendor, r.Model, string(r.State), formatTime(r.CreatedAt))
+		if err != nil {
+			return err
+		}
+		return emitReview(ctx, tx, emit, "review.started", 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
+}
@@ -0,0 +1,102 @@
+package store_test
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"g.fc.run/misc/byre/internal/api"
+	"g.fc.run/misc/byre/internal/store"
+)
+
+func TestReviewLifecycle(t *testing.T) {
+	ctx := context.Background()
+	st, err := store.Open(filepath.Join(t.TempDir(), "byre.db"), store.DefaultPortFirst, func(api.Event) {})
+	if err != nil {
+		t.Fatalf("open store: %v", err)
+	}
+	t.Cleanup(func() { st.Close() })
+
+	now := time.Now().UTC()
+	_, err = st.EnsureUser(ctx, "alice", "Alice")
+	if err != nil {
+		t.Fatalf("ensure user: %v", err)
+	}
+	err = st.CreateProject(ctx, api.Project{Id: "p1", Name: "demo", RepoPath: "/nonexistent", DefaultBranch: "main", Remote: "none", CreatedBy: "alice", CreatedAt: now})
+	if err != nil {
+		t.Fatalf("create project: %v", err)
+	}
+	err = st.CreateTask(ctx, api.Task{Id: "t1", ProjectId: "p1", Title: "t1", Branch: "t1", WorktreePath: "/nonexistent", BaseRef: "x", Status: api.TaskStatusActive, CreatedBy: "alice", CreatedAt: now}, "alice")
+	if err != nil {
+		t.Fatalf("create task: %v", err)
+	}
+	for _, s := range []struct {
+		id   string
+		role api.AgentSessionRole
+	}{{"impl", api.Implementer}, {"rev1", api.Reviewer}, {"rev2", api.Reviewer}} {
+		err = st.CreateSession(ctx, api.AgentSession{Id: s.id, TaskId: "t1", Vendor: api.Claude, VendorSessionId: s.id, State: api.AgentSessionStateIdle, Role: s.role, CreatedBy: "alice", CreatedAt: now}, "p1")
+		if err != nil {
+			t.Fatalf("create session: %v", err)
+		}
+	}
+	sessions, err := st.ListSessions(ctx, "t1")
+	if err != nil || len(sessions) != 1 || sessions[0].Id != "impl" {
+		t.Fatalf("task sessions %+v %v, want only the implementer", sessions, err)
+	}
+
+	snap := func(n int, at time.Time) {
+		t.Helper()
+		err := st.CreateSnapshot(ctx, api.Snapshot{Id: "s" + string(rune('0'+n)), TaskId: "t1", AgentSessionId: "impl", N: n, CommitSha: "c", TreeSha: "tree" + string(rune('0'+n)), EventSeq: 1, Trigger: api.SnapshotTriggerTurnEnd, CreatedAt: at}, "p1")
+		if err != nil {
+			t.Fatalf("create snapshot %d: %v", n, err)
+		}
+	}
+	snap(1, now)
+
+	asked := now.Add(time.Second)
+	err = st.CreateInboxItem(ctx, api.InboxItem{Id: "q1", ProjectId: "p1", TaskId: "t1", AgentSessionId: "impl", ThreadId: "q1",
+		Kind: api.Question, Priority: api.Blocked, Blocking: true, Payload: map[string]any{}, State: api.Open, CreatedAt: asked})
+	if err != nil {
+		t.Fatalf("create question: %v", err)
+	}
+	snap(2, asked.Add(time.Second))
+
+	for _, r := range []api.Review{
+		{Id: "r1", TaskId: "t1", SnapshotN: 1, AgentSessionId: "rev1", Vendor: "codex", Model: "m", State: api.ReviewStateRunning, CreatedAt: now},
+		{Id: "r2", TaskId: "t1", SnapshotN: 2, AgentSessionId: "rev2", Vendor: "codex", Model: "m", State: api.ReviewStateRunning, CreatedAt: now},
+	} {
+		err = st.CreateReview(ctx, r)
+		if err != nil {
+			t.Fatalf("create review: %v", err)
+		}
+	}
+
+	byRev, err := st.ReviewBySession(ctx, "rev2")
+	if err != nil || byRev.Id != "r2" {
+		t.Fatalf("review by session %+v %v", byRev, err)
+	}
+
+	err = st.FinishReview(ctx, "r2", api.ReviewStateDone, map[string]any{"summary": "fine", "findings": []any{}}, "")
+	if err != nil {
+		t.Fatalf("finish: %v", err)
+	}
+	err = st.FinishReview(ctx, "r2", api.ReviewStateFailed, nil, "late")
+	if !errors.Is(err, store.ErrNotFound) {
+		t.Fatalf("finishing a finished review: %v, want not found", err)
+	}
+
+	item, err := st.GetInboxItem(ctx, "q1")
+	if err != nil {
+		t.Fatalf("get question: %v", err)
+	}
+	if len(item.Reviews) != 1 || item.Reviews[0].Id != "r2" || item.Reviews[0].Result == nil || (*item.Reviews[0].Result)["summary"] != "fine" {
+		t.Fatalf("question's reviews %+v, want the review of the snapshot after it", item.Reviews)
+	}
+
+	running, err := st.RunningReviews(ctx)
+	if err != nil || len(running) != 1 || running[0].Id != "r1" {
+		t.Fatalf("running reviews %+v %v", running, err)
+	}
+}
@@ -289,6 +289,22 @@ UPDATE tasks SET color = (
 `}, {"drop-turn-queue", `
 UPDATE agent_sessions SET state = 'idle' WHERE state = 'queued';
 DROP TABLE turn_queue;
+`}, {"add-snapshot-reviews", `
+CREATE TABLE reviews (
+	id TEXT PRIMARY KEY,
+	task_id TEXT NOT NULL REFERENCES tasks(id),
+	snapshot_n INTEGER NOT NULL,
+	agent_session_id TEXT NOT NULL REFERENCES agent_sessions(id),
+	vendor TEXT NOT NULL,
+	model TEXT NOT NULL,
+	state TEXT NOT NULL,
+	result TEXT NOT NULL DEFAULT '',
+	error TEXT NOT NULL DEFAULT '',
+	created_at TEXT NOT NULL,
+	finished_at TEXT NOT NULL DEFAULT '',
+	UNIQUE (task_id, snapshot_n)
+);
+CREATE UNIQUE INDEX reviews_session ON reviews(agent_session_id);
 `}}

 const (
@@ -952,7 +968,7 @@ func (s *Store) GetSession(ctx context.Context, id string) (api.AgentSession, er
 }

 func (s *Store) ListSessions(ctx context.Context, taskID string) ([]api.AgentSession, error) {
-	return s.querySessions(ctx, `SELECT `+sessionColumns+` FROM agent_sessions WHERE task_id = ? ORDER BY created_at`, taskID)
+	return s.querySessions(ctx, `SELECT `+sessionColumns+` FROM agent_sessions WHERE task_id = ? AND role != ? ORDER BY created_at`, taskID, string(api.Reviewer))
 }

 func (s *Store) SessionsInState(ctx context.Context, state api.AgentSessionState) ([]api.AgentSession, error) {