byre/internal/agents/afterreview.go

package agents

import (
	"context"
	"errors"
	"fmt"
	"log/slog"

	"g.fc.run/ai/byre/internal/api"
	"g.fc.run/ai/byre/internal/store"
)

func (m *Manager) MergeAfterReview(ctx context.Context, task api.Task, user api.User, target api.MergeTaskJSONBodyTarget, title string) (api.Task, error) {
	if task.MergeQueuedAt != nil || task.Merging {
		return api.Task{}, fmt.Errorf("%w: task %s is already merging or waiting to merge", ErrConflict, task.Id)
	}
	if task.MergingIntoParent {
		return api.Task{}, fmt.Errorf("%w: task %s is merging into its parent", ErrConflict, task.Id)
	}
	err := m.checkNoSubtasks(ctx, task)
	if err != nil {
		return api.Task{}, err
	}
	if target == api.MergeTargetParent && task.ParentTaskId == nil {
		return api.Task{}, fmt.Errorf("%w: task %s is not a subtask", ErrInvalid, task.Id)
	}
	if target == api.MergeTargetMain {
		if user.GitName == "" || user.GitEmail == "" {
			return api.Task{}, ErrNoGitIdentity
		}
		title, err = cleanMergeTitle(title)
		if err != nil {
			return api.Task{}, err
		}
	}

	latest, err := m.store.LastReviewable(ctx, task.Id)
	if errors.Is(err, store.ErrNotFound) {
		return api.Task{}, fmt.Errorf("%w: task %s has no snapshot under review", ErrConflict, task.Id)
	}
	if err != nil {
		return api.Task{}, err
	}
	reviews, err := m.store.ListReviews(ctx, task.Id)
	if err != nil {
		return api.Task{}, err
	}
	reviewID := ""
	for _, r := range reviews {
		if r.SnapshotN == latest.N && r.State == api.ReviewStateRunning {
			reviewID = r.Id
		}
	}
	if reviewID == "" {
		return api.Task{}, fmt.Errorf("%w: the review of snapshot %d of task %s isn't running", ErrConflict, latest.N, task.Id)
	}

	if target == api.MergeTargetMain && title != task.MergeTitle {
		err = m.storeMergeTitle(ctx, task.Id, title, user.Login)
		if err != nil {
			return api.Task{}, err
		}
	}
	err = m.store.SetMergeAfterReview(ctx, task.Id, store.MergeAfterReview{By: user.Login, ReviewID: reviewID, Target: target})
	if errors.Is(err, store.ErrNotFound) {
		return api.Task{}, fmt.Errorf("%w: task %s is closed, being discarded or merging, or its review has finished", ErrConflict, task.Id)
	}
	if err != nil {
		return api.Task{}, err
	}
	return m.store.GetTask(ctx, task.Id)
}

func (m *Manager) clearMergeAfterReview(ctx context.Context, taskID string) {
	_, err := m.store.TakeMergeAfterReview(ctx, taskID, "")
	if err != nil {
		slog.Error("clear merge after review", "task", taskID, "err", err)
	}
}

func (m *Manager) recoverMergeAfterReviews(ctx context.Context) {
	ids, err := m.store.FinishedMergeAfterReviews(ctx)
	if err != nil {
		slog.Error("list merges after finished reviews", "err", err)
		return
	}
	for _, id := range ids {
		m.reviewFinished(id)
	}
}

func (m *Manager) reviewFinished(reviewID string) {
	ctx := context.Background()
	r, err := m.store.GetReview(ctx, reviewID)
	if err != nil {
		slog.Error("load finished review", "review", reviewID, "err", err)
		return
	}
	mar, err := m.store.TakeMergeAfterReview(ctx, r.TaskId, r.Id)
	if err != nil {
		slog.Error("take merge after review", "task", r.TaskId, "err", err)
		return
	}
	if mar.By == "" {
		return
	}
	reason := m.startMergeAfterReview(ctx, r, mar)
	if reason == "" {
		slog.Info("merge after review started", "task", r.TaskId, "review", r.Id, "by", mar.By)
		return
	}
	slog.Info("merge after review dropped", "task", r.TaskId, "review", r.Id, "reason", reason)
	task, err := m.store.GetTask(ctx, r.TaskId)
	if err != nil {
		slog.Error("load task for dropped merge after review", "task", r.TaskId, "err", err)
		return
	}
	err = m.store.AppendEvent(ctx, api.Event{Kind: "task.merge_after_review_dropped", ProjectId: &task.ProjectId, TaskId: &task.Id, Actor: "byre", Payload: map[string]any{"id": task.Id, "reason": reason}})
	if err != nil {
		slog.Error("record dropped merge after review", "task", task.Id, "err", err)
	}
}

func (m *Manager) startMergeAfterReview(ctx context.Context, r api.Review, mar store.MergeAfterReview) string {
	switch {
	case r.State == api.ReviewStateFailed:
		return "the review failed"
	case r.State != api.ReviewStateDone:
		return fmt.Sprintf("the review ended %s", r.State)
	}
	findings := []any{}
	if r.Result != nil {
		findings, _ = (*r.Result)["findings"].([]any)
	}
	switch len(findings) {
	case 0:
	case 1:
		return "the review found 1 finding"
	default:
		return fmt.Sprintf("the review found %d findings", len(findings))
	}
	task, project, err := m.activeTask(ctx, r.TaskId)
	if err != nil {
		return err.Error()
	}
	if task.Discarding {
		return "the task is being discarded"
	}
	latest, err := m.store.LastReviewable(ctx, task.Id)
	if err != nil {
		return err.Error()
	}
	if latest.N != r.SnapshotN {
		return "a newer snapshot was taken"
	}
	sess, found, err := m.liveSession(ctx, task.Id)
	if err != nil {
		return err.Error()
	}
	if found && sess.State == api.AgentSessionStateRunning {
		return "the agent is mid-turn"
	}
	user, err := m.store.GetUser(ctx, mar.By)
	if err != nil {
		return err.Error()
	}
	if mar.Target == api.MergeTargetParent {
		_, err = m.MergeIntoParent(ctx, task, user)
	} else {
		_, err = m.Merge(ctx, task, project, user, task.MergeTitle)
	}
	if err != nil {
		return err.Error()
	}
	return ""
}