diff --git a/docs/operations.md b/docs/operations.md index 2fd428a..939387c 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -86,7 +86,7 @@ Vendor setup and protocol quirks are in [Claude Code](claude-protocol.md), [Code ## Coordination -Merge and discard coordination must decide from the database's current state, never from which event woke it. This lets startup use the same decisions as ordinary operation rather than maintaining separate recovery logic. Slow work must run outside the decision transaction, with its progress recorded so another pass cannot start it twice. +Coordination of merges, discards, reviews and ready-for-merge prompts must decide from the database's current state, never from which event woke it. This lets startup use the same decisions as ordinary operation rather than maintaining separate recovery logic. Slow work must run outside the decision transaction, with its progress recorded so another pass cannot start it twice. ## git diff --git a/internal/agents/afterreview.go b/internal/agents/afterreview.go index d74f022..b0c1322 100644 --- a/internal/agents/afterreview.go +++ b/internal/agents/afterreview.go @@ -78,64 +78,16 @@ func (m *Manager) clearMergeAfterReview(ctx context.Context, taskID string) { } } -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, false) - } -} - -func (m *Manager) reviewFinished(reviewID string, deliver bool) { - ctx := context.Background() - r, err := m.store.GetReview(ctx, reviewID) - if err != nil { - slog.Error("load finished review", "review", reviewID, "err", err) - return - } - if m.advanceReviews(ctx, r) { - return - } - reviews, err := m.store.SnapshotReviews(ctx, r.TaskId, r.SnapshotN) - if err != nil { - slog.Error("list snapshot reviews", "task", r.TaskId, "err", err) - return - } - settled, err := m.reviewsSettled(ctx, reviews) - if err != nil { - slog.Error("check snapshot reviews settled", "task", r.TaskId, "err", err) - return - } - if !settled { - return - } - if deliver { - m.deliverFindings(ctx, r, reviews) - } - ids := []string{} - for _, sr := range reviews { - ids = append(ids, sr.Id) - } - mar, err := m.store.TakeMergeAfterReview(ctx, r.TaskId, ids) - if err != nil { - slog.Error("take merge after review", "task", r.TaskId, "err", err) - return - } - if mar.By == "" { - return - } - reason := m.startMergeAfterReview(ctx, r, reviews, mar) +func (m *Manager) mergeAfterReview(ctx context.Context, ref store.SnapshotRef, reviews []api.Review, mar store.MergeAfterReview) { + reason := m.startMergeAfterReview(ctx, ref, reviews, mar) if reason == "" { - slog.Info("merge after review started", "task", r.TaskId, "review", r.Id, "by", mar.By) + slog.Info("merge after review started", "task", ref.TaskID, "snapshot", ref.N, "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) + slog.Info("merge after review dropped", "task", ref.TaskID, "snapshot", ref.N, "reason", reason) + task, err := m.store.GetTask(ctx, ref.TaskID) if err != nil { - slog.Error("load task for dropped merge after review", "task", r.TaskId, "err", err) + slog.Error("load task for dropped merge after review", "task", ref.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}}) @@ -144,7 +96,7 @@ func (m *Manager) reviewFinished(reviewID string, deliver bool) { } } -func (m *Manager) startMergeAfterReview(ctx context.Context, r api.Review, reviews []api.Review, mar store.MergeAfterReview) string { +func (m *Manager) startMergeAfterReview(ctx context.Context, ref store.SnapshotRef, reviews []api.Review, mar store.MergeAfterReview) string { findings := []any{} for _, sr := range reviews { switch { @@ -165,7 +117,7 @@ func (m *Manager) startMergeAfterReview(ctx context.Context, r api.Review, revie default: return fmt.Sprintf("the review found %d findings", len(findings)) } - task, project, err := m.activeTask(ctx, r.TaskId) + task, project, err := m.activeTask(ctx, ref.TaskID) if err != nil { return err.Error() } @@ -176,7 +128,7 @@ func (m *Manager) startMergeAfterReview(ctx context.Context, r api.Review, revie if err != nil { return err.Error() } - if latest.N != r.SnapshotN { + if latest.N != ref.N { return "a newer snapshot was taken" } sess, found, err := m.liveSession(ctx, task.Id) diff --git a/internal/agents/coordinator.go b/internal/agents/coordinator.go index 5487c80..71212c4 100644 --- a/internal/agents/coordinator.go +++ b/internal/agents/coordinator.go @@ -2,6 +2,7 @@ package agents import ( "context" + "fmt" "log/slog" ) @@ -43,6 +44,9 @@ func (m *Manager) coordinatePass(ctx context.Context, started map[string]bool) { for _, id := range work.StartDiscards { m.startDiscard(ctx, id) } + for _, r := range work.EndReviewers { + go m.cleanUpReviewer(r) + } current := map[string]bool{} once := func(key string, run func()) { @@ -59,6 +63,18 @@ func (m *Manager) coordinatePass(ctx context.Context, started map[string]bool) { for _, id := range work.FinishDiscards { once("discard:"+id, func() { m.finishDiscard(id) }) } + for _, r := range work.StartReviews { + once(fmt.Sprintf("review:%s:%d", r.TaskID, r.N), func() { m.startReviews(r) }) + } + for _, s := range work.StartStages { + once(fmt.Sprintf("stage:%s:%d:%s", s.TaskID, s.SnapshotN, s.Stage), func() { m.startReviewStage(s) }) + } + for _, r := range work.SettleReviews { + once(fmt.Sprintf("settle:%s:%d", r.TaskID, r.N), func() { m.settleReviews(r) }) + } + for _, p := range work.ReadyPrompts { + once("ready:"+p.TaskID, func() { m.promptReady(p) }) + } for key := range started { if !current[key] { delete(started, key) diff --git a/internal/agents/lead.go b/internal/agents/lead.go index 3c79024..fcfcae0 100644 --- a/internal/agents/lead.go +++ b/internal/agents/lead.go @@ -366,9 +366,6 @@ func (m *Manager) Observe(events []api.Event) { if ev.Seq != 0 { m.poke() } - if to, _ := ev.Payload["to"].(string); ev.Kind == "task.state" && readyState(api.TaskState(to)) && ev.TaskId != nil { - go m.promptReady(*ev.TaskId, ev.Payload["from"]) - } m.observeFacts(ev) switch { case (ev.Kind == "task.merged" || ev.Kind == "task.merged_to_main" || ev.Kind == "task.discarded") && ev.ProjectId != nil: diff --git a/internal/agents/manager.go b/internal/agents/manager.go index bafad8b..f1990ee 100644 --- a/internal/agents/manager.go +++ b/internal/agents/manager.go @@ -461,7 +461,7 @@ func (m *Manager) Recover(ctx context.Context) error { m.logUndelivered(ctx, sess.Id, "idle at startup") } - err = m.recoverReviews(ctx) + err = m.store.ClearReviewRequests(ctx) if err != nil { return err } @@ -471,13 +471,18 @@ func (m *Manager) Recover(ctx context.Context) error { go m.coordinate() m.poke() m.MarkFacts(allTasks) - go m.recoverMergeAfterReviews(context.Background()) go m.auditAll(context.Background()) return nil } func (m *Manager) failResume(ctx context.Context, sess api.AgentSession, task api.Task, err error) error { slog.Error("resume session", "session", sess.Id, "err", err) + if sess.Role == api.Reviewer { + r, rerr := m.store.ReviewBySession(ctx, sess.Id) + if rerr == nil { + m.failReview(ctx, r.Id, "byre restarted and couldn't resume the reviewer: "+err.Error()) + } + } m.endMergeTurn(sess.Id, true) m.logUndelivered(ctx, sess.Id, "resume failed") derr := m.store.DropStageOutput(ctx, sess.Id) @@ -664,6 +669,10 @@ func (m *Manager) failStart(ctx context.Context, sessionID, actor string, err er cctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) defer cancel() + r, rerr := m.store.ReviewBySession(cctx, sessionID) + if rerr == nil { + m.failReview(cctx, r.Id, "the reviewer didn't start: "+err.Error()) + } serr := m.store.SetSessionState(cctx, sessionID, api.AgentSessionStateIdle, actor, map[string]any{"error": err.Error()}) if serr != nil { slog.Error("record failed start", "session", sessionID, "err", serr) @@ -981,6 +990,10 @@ func (m *Manager) pump(l *live) { } m.logUndelivered(ctx, l.id, "agent process exited mid-turn") + if l.reviewID != "" { + m.finishReviewer(l, "the reviewer's process exited mid-turn") + return + } if l.stage != "" { err = m.store.DropStageOutput(ctx, l.id) if err != nil { @@ -992,10 +1005,6 @@ func (m *Manager) pump(l *live) { if err != nil { slog.Error("record session exit", "session", l.id, "err", err) } - if l.reviewID != "" { - m.finishReviewer(l, "the reviewer's process exited mid-turn") - return - } if end.Discard != store.DiscardNone { return } diff --git a/internal/agents/mergein.go b/internal/agents/mergein.go index faef9d9..a6db30d 100644 --- a/internal/agents/mergein.go +++ b/internal/agents/mergein.go @@ -397,7 +397,6 @@ func (m *Manager) RemoveTask(ctx context.Context, task api.Task, project api.Pro if err != nil { return err } - m.cancelReviews(ctx, task, project) err = m.dropSnapshots(project.RepoPath, task.Id) if err != nil { return err diff --git a/internal/agents/ready.go b/internal/agents/ready.go index ef02c4c..bd9e543 100644 --- a/internal/agents/ready.go +++ b/internal/agents/ready.go @@ -8,7 +8,6 @@ import ( "log/slog" "strings" - "g.fc.run/ai/byre/internal/api" "g.fc.run/ai/byre/internal/mcp" "g.fc.run/ai/byre/internal/store" "g.fc.run/ai/byre/internal/workflow" @@ -142,121 +141,42 @@ func (m *Manager) cancelReadyTool(l *live) mcp.Tool { } } -func readyState(s api.TaskState) bool { - return s == api.TaskStateIdle || s == api.TaskStateFinished || s == api.TaskStateReadyToMerge || s == api.TaskStateDiscardSuggested -} - -func (m *Manager) promptReady(taskID string, from any) { +func (m *Manager) promptReady(p store.ReadyPrompt) { ctx := context.Background() - err := m.tryPromptReady(ctx, taskID, from == string(api.TaskStateReviewing)) + err := m.sendReadyPrompt(ctx, p) if err != nil { - slog.Error("prompt ready for merge", "task", taskID, "err", err) + slog.Info("ready for merge prompt not sent", "task", p.TaskID, "snapshot", p.Snapshot, "session", p.SessionID, "err", err) + return } + slog.Info("ready for merge prompted", "task", p.TaskID, "snapshot", p.Snapshot, "session", p.SessionID) } -func (m *Manager) tryPromptReady(ctx context.Context, taskID string, fromReview bool) error { - sess, found, err := m.liveSession(ctx, taskID) - if err != nil || !found { - return err - } - unlock := m.lockDelivery(sess.Id) - defer unlock() - - task, err := m.store.GetTask(ctx, taskID) - if err != nil { - return err - } - if task.Status != api.TaskStatusActive || !readyState(task.State) { - return nil - } - stagePrompt := "" - if task.WorkflowStage != nil { - frozen, err := m.store.TaskWorkflow(ctx, taskID) - if err != nil { - return err - } - output, _, err := m.store.StageResult(ctx, taskID) - if err != nil { - return err - } - if output != "" && len(workflow.Next(frozen.Definition, *task.WorkflowStage)) > 0 { - return nil - } - finished, err := m.store.WorkflowFinished(ctx, *task.ParentTaskId) - if err != nil { - return err - } - if output == "" && !finished { - stagePrompt = finishStagePrompt(frozen, *task.WorkflowStage) - } - } - sess, err = m.store.GetSession(ctx, sess.Id) - if err != nil || sess.State == api.AgentSessionStateRunning || m.interrupted(sess.Id) { - return err - } - latest, err := m.store.LastReviewable(ctx, taskID) - if errors.Is(err, store.ErrNotFound) { - return nil - } - if err != nil { - return err - } - reviews, err := m.store.SnapshotReviews(ctx, taskID, latest.N) - if err != nil { - return err - } - if len(reviews) == 0 { - project, err := m.store.GetProject(ctx, task.ProjectId) - if err != nil || project.ReviewWorkflow != "" { - return err - } - } - settled, err := m.reviewsSettled(ctx, reviews) - if err != nil || !settled { - return err - } - for _, r := range reviews { - if r.State == api.ReviewStateFailed && task.Background { - m.auditError(ctx, task, fmt.Sprintf("Review %q of snapshot %d failed: %s. The audit can't go on without you.", r.Stage, latest.N, r.Error)) - return nil - } - if r.State != api.ReviewStateDone && r.State != api.ReviewStateSkipped { - return nil - } - if fromReview && r.Result != nil { - findings, _ := (*r.Result)["findings"].([]any) - if len(findings) > 0 { - return nil - } - } - } - due, err := m.store.ReadyPromptDue(ctx, taskID) - if err != nil || !due { - return err - } - open, err := m.store.ListInbox(ctx, store.InboxFilter{TaskID: taskID, State: api.Open}) - if err != nil || len(open) > 0 { - return err - } +func (m *Manager) sendReadyPrompt(ctx context.Context, p store.ReadyPrompt) error { prompt := readyPrompt - if len(reviews) == 1 && reviews[0].State == api.ReviewStateSkipped { + switch p.Kind { + case store.ReadyPromptUnchanged: prompt = readyPromptUnchanged + case store.ReadyPromptFinishStage: + task, err := m.store.GetTask(ctx, p.TaskID) + if err != nil { + return err + } + frozen, err := m.store.TaskWorkflow(ctx, p.TaskID) + if err != nil { + return err + } + prompt = finishStagePrompt(frozen, *task.WorkflowStage) } - if stagePrompt != "" { - prompt = stagePrompt - } - _, err = m.Send(ctx, sess.Id, "byre", prompt, "", "", "") + + unlock := m.lockDelivery(p.SessionID) + defer unlock() + _, err := m.Send(ctx, p.SessionID, "byre", prompt, "", "", "") if err != nil { return err } - seq, err := m.store.LastTaskSeq(ctx, taskID) + seq, err := m.store.LastTaskSeq(ctx, p.TaskID) if err != nil { return err } - err = m.store.SetReadyPromptSeq(ctx, taskID, seq) - if err != nil { - return err - } - slog.Info("ready for merge prompted", "task", taskID, "snapshot", latest.N, "session", sess.Id) - return nil + return m.store.SetReadyPromptSeq(ctx, p.TaskID, seq) } diff --git a/internal/agents/review.go b/internal/agents/review.go index c4b3629..866ae75 100644 --- a/internal/agents/review.go +++ b/internal/agents/review.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "log/slog" - "math" "net/http" "os" "path/filepath" @@ -78,92 +77,6 @@ func reviewSchema() json.RawMessage { return b } -func (m *Manager) afterSnapshot(snap *api.Snapshot) { - if snap == nil { - return - } - ctx := context.Background() - m.reviewMu.Lock() - defer m.reviewMu.Unlock() - - task, err := m.store.GetTask(ctx, snap.TaskId) - if err != nil { - slog.Error("load task to review", "task", snap.TaskId, "err", err) - return - } - if task.Status != api.TaskStatusActive { - return - } - reviews, err := m.store.ListReviews(ctx, task.Id) - if err != nil { - slog.Error("list reviews to review", "task", task.Id, "err", err) - return - } - for _, r := range reviews { - if r.SnapshotN > snap.N { - slog.Info("review not started, a newer snapshot has one", "task", task.Id, "snapshot", snap.N, "newer", r.SnapshotN) - return - } - } - err = m.startReviews(ctx, task, *snap) - if err != nil { - slog.Error("start review", "task", task.Id, "snapshot", snap.N, "err", err) - } -} - -func (m *Manager) reviewsSettled(ctx context.Context, reviews []api.Review) (bool, error) { - byStage := map[string]api.Review{} - for _, r := range reviews { - if r.State == api.ReviewStateRunning { - return false, nil - } - byStage[r.Stage] = r - } - for _, r := range reviews { - if r.State != api.ReviewStateDone { - continue - } - _, def, err := m.store.ReviewDefinition(ctx, r.Id) - if err != nil { - return false, err - } - if len(def.Steps) == 0 { - continue - } - if len(pendingReviewStages(def, r.Stage, byStage)) > 0 { - return false, nil - } - } - return true, nil -} - -type pendingReviewStage struct { - name string - inputs []api.Review -} - -func pendingReviewStages(def api.WorkflowDefinition, stage string, byStage map[string]api.Review) []pendingReviewStage { - pending := []pendingReviewStage{} - for _, next := range workflow.Next(def, stage) { - if _, found := byStage[next]; found { - continue - } - inputs := []api.Review{} - after := workflow.After(def, next) - for _, a := range after { - in := byStage[a] - if in.State != api.ReviewStateDone { - break - } - inputs = append(inputs, in) - } - if len(inputs) == len(after) { - pending = append(pending, pendingReviewStage{name: next, inputs: inputs}) - } - } - return pending -} - func (m *Manager) reviewDir(projectID, reviewID string) string { return filepath.Join(m.worktreeRoot, projectID, "review-"+reviewID) } @@ -180,24 +93,39 @@ func playwrightMCP(outputDir string) mcp.Stdio { } } -func (m *Manager) startReviews(ctx context.Context, task api.Task, snap api.Snapshot) error { - project, err := m.store.GetProject(ctx, task.ProjectId) +func (m *Manager) startReviews(ref store.SnapshotRef) { + ctx := context.Background() + m.reviewMu.Lock() + defer m.reviewMu.Unlock() + err := m.beginReviews(ctx, ref) + if err != nil { + slog.Error("start review", "task", ref.TaskID, "snapshot", ref.N, "err", err) + } +} + +func (m *Manager) beginReviews(ctx context.Context, ref store.SnapshotRef) error { + task, project, err := m.activeTask(ctx, ref.TaskID) if err != nil { return err } - defer m.failRunningReviews(ctx, task.Id, project, snap.N, fmt.Sprintf("superseded by snapshot %d", snap.N)) - if project.ReviewWorkflow == "" { - slog.Info("review not started, the project has no review workflow", "task", task.Id, "snapshot", snap.N) - return nil - } - changes, err := git.Compare(project.RepoPath, project.DefaultBranch, snap.CommitSha, snap.TreeSha, "") + snap, err := m.store.GetSnapshot(ctx, task.Id, ref.N) if err != nil { return err } now := time.Now().UTC() + finished := func(state api.ReviewState, stage, problem string) error { + return m.store.CreateReview(ctx, api.Review{Id: ids.New(ids.Review), TaskId: task.Id, SnapshotN: snap.N, Stage: stage, State: state, Error: problem, CreatedAt: now, FinishedAt: &now}, "") + } + if project.ReviewWorkflow == "" { + return finished(api.ReviewStateFailed, "", fmt.Sprintf("project %s has no review workflow", project.Name)) + } + changes, err := git.Compare(project.RepoPath, project.DefaultBranch, snap.CommitSha, snap.TreeSha, "") + if err != nil { + return finished(api.ReviewStateFailed, project.ReviewWorkflow, fmt.Sprintf("byre couldn't compare snapshot %d with %s: %v", snap.N, project.DefaultBranch, err)) + } if len(changes.Files) == 0 { slog.Info("review skipped, snapshot has no diff", "task", task.Id, "snapshot", snap.N) - return m.store.CreateReview(ctx, api.Review{Id: ids.New(ids.Review), TaskId: task.Id, SnapshotN: snap.N, State: api.ReviewStateSkipped, CreatedAt: now, FinishedAt: &now}, "") + return finished(api.ReviewStateSkipped, "", "") } frozen, err := m.store.FreezeWorkflow(ctx, project.ReviewWorkflow) @@ -205,167 +133,102 @@ func (m *Manager) startReviews(ctx context.Context, task api.Task, snap api.Snap err = reviewProblem(frozen.Definition) } if err != nil { - return m.store.CreateReview(ctx, api.Review{ - Id: ids.New(ids.Review), - TaskId: task.Id, - SnapshotN: snap.N, - Stage: project.ReviewWorkflow, - State: api.ReviewStateFailed, - Error: fmt.Sprintf("project %s reviews with workflow %q, which can't run: %v", project.Name, project.ReviewWorkflow, err), - CreatedAt: now, - FinishedAt: &now, - }, "") + return finished(api.ReviewStateFailed, project.ReviewWorkflow, fmt.Sprintf("project %s reviews with workflow %q, which can't run: %v", project.Name, project.ReviewWorkflow, err)) } + stages := []store.StageStart{} for _, stage := range workflow.Starts(frozen.Definition) { - err = m.startReviewer(ctx, task, project, snap, frozen.DefinitionID, frozen.Definition, stage, nil) - if err != nil { - slog.Error("start review stage", "task", task.Id, "snapshot", snap.N, "stage", stage, "err", err) - } + stages = append(stages, store.StageStart{TaskID: task.Id, SnapshotN: snap.N, DefinitionID: frozen.DefinitionID, Definition: frozen.Definition, Stage: stage}) } - return nil + return m.startReviewers(ctx, task, project, snap, stages) } -func (m *Manager) advanceReviews(ctx context.Context, r api.Review) bool { - if r.State != api.ReviewStateDone { - return false - } +func (m *Manager) startReviewStage(s store.StageStart) { + ctx := context.Background() m.reviewMu.Lock() defer m.reviewMu.Unlock() - - defID, def, err := m.store.ReviewDefinition(ctx, r.Id) + err := m.beginReviewStage(ctx, s) if err != nil { - slog.Error("load review definition", "review", r.Id, "err", err) - return false + slog.Error("start review stage", "task", s.TaskID, "snapshot", s.SnapshotN, "stage", s.Stage, "err", err) } - if len(def.Steps) == 0 { - return false - } - task, project, err := m.activeTask(ctx, r.TaskId) - if err != nil { - return false - } - all, err := m.store.ListReviews(ctx, task.Id) - if err != nil { - slog.Error("list reviews to advance", "task", task.Id, "err", err) - return false - } - byStage := map[string]api.Review{} - for _, sr := range all { - if sr.SnapshotN > r.SnapshotN { - return false - } - if sr.SnapshotN == r.SnapshotN { - byStage[sr.Stage] = sr - } - } - snap, err := m.store.GetSnapshot(ctx, task.Id, r.SnapshotN) - if err != nil { - slog.Error("load snapshot to advance review", "task", task.Id, "snapshot", r.SnapshotN, "err", err) - return false - } - - started := false - for _, next := range pendingReviewStages(def, r.Stage, byStage) { - err = m.startReviewer(ctx, task, project, snap, defID, def, next.name, next.inputs) - if err != nil { - slog.Error("start review stage", "task", task.Id, "snapshot", r.SnapshotN, "stage", next.name, "err", err) - continue - } - started = true - } - return started } -func (m *Manager) startReviewer(ctx context.Context, task api.Task, project api.Project, snap api.Snapshot, defID string, def api.WorkflowDefinition, stage string, inputs []api.Review) (err error) { - model := workflow.Get(def, stage).Model - vendor := vendorOf(model) - reviewID := ids.New(ids.Review) - recorded := false - defer func() { - if err == nil || recorded { - return +func (m *Manager) beginReviewStage(ctx context.Context, s store.StageStart) error { + task, project, err := m.activeTask(ctx, s.TaskID) + if err != nil { + return err + } + snap, err := m.store.GetSnapshot(ctx, task.Id, s.SnapshotN) + if err != nil { + return err + } + return m.startReviewers(ctx, task, project, snap, []store.StageStart{s}) +} + +func (m *Manager) startReviewers(ctx context.Context, task api.Task, project api.Project, snap api.Snapshot, stages []store.StageStart) error { + if len(stages) == 0 { + return nil + } + now := time.Now().UTC() + reviewers := []store.Reviewer{} + failed := []api.Review{} + prompts := map[string]string{} + for _, s := range stages { + model := workflow.Get(s.Definition, s.Stage).Model + vendor := vendorOf(model) + r := api.Review{Id: ids.New(ids.Review), TaskId: task.Id, SnapshotN: snap.N, Stage: s.Stage, Vendor: string(vendor), Model: model, CreatedAt: now} + prompt, err := m.reviewPrompt(ctx, task, project, snap, s.Definition, s.Stage, s.Inputs) + if err != nil { + r.State = api.ReviewStateFailed + r.Error = "the reviewer didn't start: " + err.Error() + r.FinishedAt = &now + failed = append(failed, r) + continue } - now := time.Now().UTC() - rerr := m.store.CreateReview(ctx, api.Review{ - Id: reviewID, - TaskId: task.Id, - SnapshotN: snap.N, - Stage: stage, - Vendor: string(vendor), - Model: model, - State: api.ReviewStateFailed, - Error: "the reviewer didn't start: " + err.Error(), - CreatedAt: now, - FinishedAt: &now, - }, defID) - if rerr != nil { - slog.Error("record review that didn't start", "task", task.Id, "stage", stage, "err", rerr) - return - } - go m.reviewFinished(reviewID, true) - }() - - prompt, err := m.reviewPrompt(ctx, task, project, snap, def, stage, inputs) + prompts[r.Id] = prompt + reviewers = append(reviewers, store.Reviewer{ + Session: api.AgentSession{ + Id: ids.New(ids.Session), + TaskId: task.Id, + Vendor: vendor, + Model: model, + VendorSessionId: initialVendorSessionID(vendor), + State: api.AgentSessionStateRunning, + Role: api.Reviewer, + CreatedBy: task.CreatedBy, + CreatedAt: now, + }, + Review: r, + }) + } + err := m.store.StartReviewers(ctx, project.Id, stages[0].DefinitionID, reviewers, failed) if err != nil { return err } - - dir := m.reviewDir(project.Id, reviewID) - err = git.AddDetachedWorktree(project.RepoPath, dir, snap.CommitSha) - if err != nil { - return err + for _, rv := range reviewers { + m.launchReviewer(ctx, task, project, snap, rv, prompts[rv.Review.Id]) } - err = os.MkdirAll(m.reviewMediaDir(project.Id, reviewID), 0o755) - if err != nil { - m.removeReviewDir(project, reviewID) - return err - } - - id := ids.New(ids.Session) - sess := api.AgentSession{ - Id: id, - TaskId: task.Id, - Vendor: vendor, - Model: model, - VendorSessionId: initialVendorSessionID(vendor), - State: api.AgentSessionStateIdle, - Role: api.Reviewer, - CreatedBy: task.CreatedBy, - CreatedAt: time.Now().UTC(), - } - err = m.store.CreateSession(ctx, sess, project.Id) - if err != nil { - m.removeReviewDir(project, reviewID) - return err - } - err = m.store.CreateReview(ctx, api.Review{ - Id: reviewID, - TaskId: task.Id, - SnapshotN: snap.N, - Stage: stage, - AgentSessionId: sess.Id, - Vendor: string(vendor), - Model: model, - State: api.ReviewStateRunning, - CreatedAt: time.Now().UTC(), - }, defID) - if err != nil { - m.endReviewer(ctx, sess.Id, project, reviewID) - return err - } - recorded = true - - _, err = m.turn(ctx, sess, task, "byre", prompt, "", "", nil) - if err != nil { - m.endReviewer(ctx, sess.Id, project, reviewID) - m.failReview(ctx, reviewID, "the reviewer didn't start: "+err.Error()) - return err - } - slog.Info("review started", "review", reviewID, "task", task.Id, "snapshot", snap.N, "stage", stage, "vendor", vendor, "model", model) return nil } +func (m *Manager) launchReviewer(ctx context.Context, task api.Task, project api.Project, snap api.Snapshot, rv store.Reviewer, prompt string) { + r := rv.Review + err := git.AddDetachedWorktree(project.RepoPath, m.reviewDir(project.Id, r.Id), snap.CommitSha) + if err == nil { + err = os.MkdirAll(m.reviewMediaDir(project.Id, r.Id), 0o755) + } + if err == nil { + _, err = m.launch(ctx, rv.Session, task, "byre", prompt, "", "", nil) + } + if err != nil { + slog.Error("start reviewer", "review", r.Id, "task", task.Id, "stage", r.Stage, "err", err) + m.failReview(ctx, r.Id, "the reviewer didn't start: "+err.Error()) + m.endReviewer(ctx, rv.Session.Id, project, r.Id) + return + } + slog.Info("review started", "review", r.Id, "task", task.Id, "snapshot", snap.N, "stage", r.Stage, "vendor", r.Vendor, "model", r.Model) +} + func (m *Manager) reviewPrompt(ctx context.Context, task api.Task, project api.Project, snap api.Snapshot, def api.WorkflowDefinition, stage string, inputs []api.Review) (string, error) { changes, err := git.Compare(project.RepoPath, project.DefaultBranch, snap.CommitSha, snap.TreeSha, "") if err != nil { @@ -465,43 +328,65 @@ func (m *Manager) submitReview(ctx context.Context, l *live, raw json.RawMessage if err != nil { return "", err } - go m.reviewFinished(l.reviewID, true) return "Review recorded. End your turn now with one line.", nil } -func (m *Manager) deliverFindings(ctx context.Context, r api.Review, reviews []api.Review) { - last := r - for _, sr := range reviews { - if sr.FinishedAt != nil && last.FinishedAt != nil && (sr.FinishedAt.After(*last.FinishedAt) || sr.FinishedAt.Equal(*last.FinishedAt) && sr.Id > last.Id) { - last = sr - } - } - if last.Id != r.Id { +func (m *Manager) settleReviews(ref store.SnapshotRef) { + ctx := context.Background() + m.settle(ctx, ref) + err := m.store.FinishSettling(ctx, ref.TaskID, ref.N) + if err != nil { + slog.Error("finish settling reviews", "task", ref.TaskID, "snapshot", ref.N, "err", err) return } - latest, err := m.store.LastReviewable(ctx, r.TaskId) - if err != nil || latest.N != r.SnapshotN { + m.poke() +} + +func (m *Manager) settle(ctx context.Context, ref store.SnapshotRef) { + reviews, err := m.store.SnapshotReviews(ctx, ref.TaskID, ref.N) + if err != nil { + slog.Error("list reviews to settle", "task", ref.TaskID, "err", err) return } - text := findingsMessage(r.SnapshotN, reviews) + reviewIDs := []string{} + for _, r := range reviews { + reviewIDs = append(reviewIDs, r.Id) + } + m.deliverFindings(ctx, ref, reviews) + mar, err := m.store.TakeMergeAfterReview(ctx, ref.TaskID, reviewIDs) + if err != nil { + slog.Error("take merge after review", "task", ref.TaskID, "err", err) + return + } + if mar.By != "" { + m.mergeAfterReview(ctx, ref, reviews, mar) + } +} + +func (m *Manager) deliverFindings(ctx context.Context, ref store.SnapshotRef, reviews []api.Review) { + text := findingsMessage(ref.N, reviews) if text == "" { return } - s, found, err := m.liveSession(ctx, r.TaskId) + latest, err := m.store.LastReviewable(ctx, ref.TaskID) + if err != nil || latest.N != ref.N { + return + } + s, found, err := m.liveSession(ctx, ref.TaskID) if err != nil { - slog.Error("list sessions to deliver findings", "task", r.TaskId, "err", err) + slog.Error("list sessions to deliver findings", "task", ref.TaskID, "err", err) return } if !found { - slog.Info("review findings not delivered, task has no live agent", "task", r.TaskId, "snapshot", r.SnapshotN) + slog.Info("review findings not delivered, task has no live agent", "task", ref.TaskID, "snapshot", ref.N) return } _, _, err = m.Message(ctx, s.Id, "byre", text, "", "", "") if err != nil { - slog.Error("deliver review findings", "task", r.TaskId, "snapshot", r.SnapshotN, "session", s.Id, "err", err) + slog.Error("deliver review findings", "task", ref.TaskID, "snapshot", ref.N, "session", s.Id, "err", err) return } - slog.Info("review findings delivered", "task", r.TaskId, "snapshot", r.SnapshotN, "session", s.Id) + slog.Info("review findings delivered", "task", ref.TaskID, "snapshot", ref.N, "session", s.Id) } const findingsPreamble = "The review of snapshot #" @@ -691,8 +576,8 @@ func (m *Manager) finishReviewer(l *live, reason string) { slog.Error("load project to end reviewer", "review", l.reviewID, "err", err) return } - m.endReviewer(ctx, l.id, project, l.reviewID) m.failReview(ctx, l.reviewID, reason) + m.endReviewer(ctx, l.id, project, l.reviewID) } func (m *Manager) endReviewer(ctx context.Context, sessionID string, project api.Project, reviewID string) { @@ -709,6 +594,11 @@ func (m *Manager) endReviewer(ctx context.Context, sessionID string, project api m.removeReviewDir(project, reviewID) } +func (m *Manager) cleanUpReviewer(r store.EndedReviewer) { + m.stopLive(r.SessionID) + m.removeReviewDir(api.Project{Id: r.ProjectID, RepoPath: r.RepoPath}, r.ReviewID) +} + func (m *Manager) removeReviewDir(project api.Project, reviewID string) { err := git.RemoveDetachedWorktree(project.RepoPath, m.reviewDir(project.Id, reviewID)) if err != nil { @@ -730,28 +620,6 @@ func (m *Manager) failReview(ctx context.Context, reviewID, reason string) { return } slog.Info("review failed", "review", reviewID, "reason", reason) - go m.reviewFinished(reviewID, true) -} - -func (m *Manager) cancelReviews(ctx context.Context, task api.Task, project api.Project) { - m.reviewMu.Lock() - defer m.reviewMu.Unlock() - m.failRunningReviews(ctx, task.Id, project, math.MaxInt, "the task was closed") -} - -func (m *Manager) failRunningReviews(ctx context.Context, taskID string, project api.Project, before int, reason string) { - reviews, err := m.store.ListReviews(ctx, taskID) - if err != nil { - slog.Error("list reviews to cancel", "task", taskID, "err", err) - return - } - for _, r := range reviews { - if r.State != api.ReviewStateRunning || r.SnapshotN >= before { - continue - } - m.endReviewer(ctx, r.AgentSessionId, project, r.Id) - m.failReview(ctx, r.Id, reason) - } } func (m *Manager) StopReview(ctx context.Context, reviewID, reason string) (api.Review, error) { @@ -772,8 +640,8 @@ func (m *Manager) StopReview(ctx context.Context, reviewID, reason string) (api. if err != nil { return api.Review{}, err } - m.endReviewer(ctx, r.AgentSessionId, project, r.Id) m.failReview(ctx, r.Id, reason) + m.endReviewer(ctx, r.AgentSessionId, project, r.Id) return m.store.GetReview(ctx, r.Id) } @@ -799,44 +667,6 @@ func (m *Manager) stopStuckReviewers(ctx context.Context, h Health) { } } -func (m *Manager) recoverReviews(ctx context.Context) error { - err := m.store.ClearReviewRequests(ctx) - if err != nil { - return err - } - running, err := m.store.RunningReviews(ctx) - if err != nil { - return err - } - for _, r := range running { - sess, err := m.store.GetSession(ctx, r.AgentSessionId) - if err != nil { - return err - } - if sess.State == api.AgentSessionStateRunning { - continue - } - task, err := m.store.GetTask(ctx, r.TaskId) - if err != nil { - return err - } - project, err := m.store.GetProject(ctx, task.ProjectId) - if err != nil { - return err - } - m.endReviewer(ctx, r.AgentSessionId, project, r.Id) - m.failReview(ctx, r.Id, "byre restarted before the reviewer finished") - } - done, err := m.store.LatestDoneStages(ctx) - if err != nil { - return err - } - for _, id := range done { - go m.reviewFinished(id, false) - } - return nil -} - func reviewName(r api.Review) string { if r.Stage == "" { return fmt.Sprintf("review of snapshot %d", r.SnapshotN) diff --git a/internal/agents/snapshots.go b/internal/agents/snapshots.go index 08087ac..0f5e880 100644 --- a/internal/agents/snapshots.go +++ b/internal/agents/snapshots.go @@ -86,15 +86,12 @@ func (m *Manager) runSnapshots(st *snapState, req *snapRequest) { closed := st.closed m.snapMu.Unlock() if !closed { - snap, err := m.takeSnapshot(context.Background(), t, req.trigger, req.seq, req.force) + snap, err := m.takeSnapshot(context.Background(), t, req.trigger, req.seq, req.force, req.review) if err != nil { slog.Error("take snapshot", "task", t.taskID, "trigger", req.trigger, "err", err) } - if req.review { - go func() { - m.afterSnapshot(snap) - m.endReviewRequest(t.taskID) - }() + if snap == nil { + m.dropRequest(t.taskID, req) } } else { m.dropRequest(t.taskID, req) @@ -128,7 +125,7 @@ func (m *Manager) endReviewRequest(taskID string) { } } -func (m *Manager) takeSnapshot(ctx context.Context, t snapTarget, trigger api.SnapshotTrigger, seq int64, force bool) (*api.Snapshot, error) { +func (m *Manager) takeSnapshot(ctx context.Context, t snapTarget, trigger api.SnapshotTrigger, seq int64, force, review bool) (*api.Snapshot, error) { if seq == 0 { var err error seq, err = m.store.LastTaskSeq(ctx, t.taskID) @@ -171,7 +168,7 @@ func (m *Manager) takeSnapshot(ctx context.Context, t snapTarget, trigger api.Sn Trigger: trigger, CreatedAt: time.Now().UTC(), } - err = m.store.CreateSnapshot(ctx, snap, t.projectID) + err = m.store.CreateSnapshot(ctx, snap, t.projectID, review) if err != nil { return nil, err } @@ -253,7 +250,7 @@ func (m *Manager) cutSnapshot(ctx context.Context, t snapTarget, held bool, trig if closed { return nil, nil } - return m.takeSnapshot(ctx, t, trigger, 0, true) + return m.takeSnapshot(ctx, t, trigger, 0, true, false) } func (m *Manager) takeSnapshotTool(l *live) mcp.Tool { diff --git a/internal/store/coordinate.go b/internal/store/coordinate.go index ecdb361..0b4e9d2 100644 --- a/internal/store/coordinate.go +++ b/internal/store/coordinate.go @@ -11,6 +11,11 @@ type Work struct { FinalizeMerges []FinalizingMerge StartDiscards []string FinishDiscards []string + EndReviewers []EndedReviewer + StartReviews []SnapshotRef + StartStages []StageStart + SettleReviews []SnapshotRef + ReadyPrompts []ReadyPrompt } func (s *Store) Coordinate(ctx context.Context) (Work, error) { @@ -44,6 +49,26 @@ func (s *Store) Coordinate(ctx context.Context) (Work, error) { return err } work.FinishDiscards, err = queryIDs(ctx, tx, `SELECT id FROM tasks WHERE status = 'active' AND discard_state = 'closing'`) + if err != nil { + return err + } + work.EndReviewers, err = endReviews(ctx, tx, emit) + if err != nil { + return err + } + work.StartReviews, err = startableReviews(ctx, tx) + if err != nil { + return err + } + work.StartStages, err = pendingStages(ctx, tx) + if err != nil { + return err + } + work.SettleReviews, err = settleReviews(ctx, tx, work.StartStages) + if err != nil { + return err + } + work.ReadyPrompts, err = readyPrompts(ctx, tx, emit) return err }) return work, err diff --git a/internal/store/migrations/094-add-snapshot-review-progress.sql b/internal/store/migrations/094-add-snapshot-review-progress.sql new file mode 100644 index 0000000..ff83fef --- /dev/null +++ b/internal/store/migrations/094-add-snapshot-review-progress.sql @@ -0,0 +1,14 @@ +ALTER TABLE snapshots ADD COLUMN review INTEGER NOT NULL DEFAULT 0; +ALTER TABLE snapshots ADD COLUMN settle_state TEXT NOT NULL DEFAULT ''; + +UPDATE snapshots SET review = 1 +WHERE EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = snapshots.task_id AND r.snapshot_n = snapshots.n); + +UPDATE snapshots SET settle_state = 'done' +WHERE NOT EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = snapshots.task_id AND r.snapshot_n = snapshots.n AND r.state = 'running') + AND NOT EXISTS ( + SELECT 1 FROM tasks t JOIN reviews r ON r.id = t.merge_after_review_id + WHERE t.id = snapshots.task_id AND t.merge_after_review_by != '' AND r.snapshot_n = snapshots.n + ); + +CREATE INDEX reviews_running ON reviews(state) WHERE state = 'running'; diff --git a/internal/store/readyprompt.go b/internal/store/readyprompt.go new file mode 100644 index 0000000..4cd9e98 --- /dev/null +++ b/internal/store/readyprompt.go @@ -0,0 +1,144 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "g.fc.run/ai/byre/internal/api" + "g.fc.run/ai/byre/internal/workflow" +) + +type ReadyPromptKind string + +const ( + ReadyPromptReady ReadyPromptKind = "ready" + ReadyPromptUnchanged ReadyPromptKind = "unchanged" + ReadyPromptFinishStage ReadyPromptKind = "finish_stage" +) + +type ReadyPrompt struct { + TaskID string + SessionID string + Kind ReadyPromptKind + Snapshot int +} + +func (s *Store) SetReadyPromptSeq(ctx context.Context, id string, seq int64) error { + return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { + _, err := tx.ExecContext(ctx, `UPDATE tasks SET ready_prompt_seq = ? WHERE id = ?`, seq, id) + return err + }) +} + +func readyPrompts(ctx context.Context, tx *sql.Tx, emit emitFunc) ([]ReadyPrompt, error) { + type candidate struct { + ReadyPrompt + projectID string + stage string + run string + output string + background bool + workflow string + } + rows, err := tx.QueryContext(ctx, ` + SELECT tasks.id, tasks.project_id, s.id, tasks.workflow_stage, coalesce(tasks.parent_task_id, ''), tasks.stage_output, `+backgroundExpr+`, + (SELECT p.review_workflow FROM projects p WHERE p.id = tasks.project_id) + FROM tasks JOIN agent_sessions s ON s.id = ( + SELECT l.id FROM agent_sessions l + WHERE l.task_id = tasks.id AND l.role != 'reviewer' AND l.state != 'completed' + AND l.member = (SELECT f.member FROM agent_sessions f WHERE f.task_id = tasks.id AND f.role != 'reviewer' ORDER BY f.created_at LIMIT 1) + ORDER BY l.created_at DESC LIMIT 1) + WHERE tasks.status = 'active' AND tasks.state IN ('idle', 'finished', 'ready_to_merge', 'discard_suggested') AND s.state = 'idle' + AND NOT EXISTS (SELECT 1 FROM inbox_items i WHERE i.task_id = tasks.id AND i.state = 'open') + AND NOT EXISTS (SELECT 1 FROM task_inputs q WHERE q.task_id = tasks.id) + AND EXISTS (SELECT 1 FROM events e WHERE e.task_id = tasks.id AND e.seq > tasks.ready_prompt_seq + AND ((e.kind = 'snapshot.created' AND json_extract(e.payload, '$.trigger') NOT IN `+cutPointTriggers+`) + OR (e.kind = 'user.message' AND e.actor != ? AND e.actor NOT LIKE '% (lead)' AND json_extract(e.payload, '$.bounce') IS NOT 1)))`, agentActor) + if err != nil { + return nil, err + } + found := []candidate{} + for rows.Next() { + c := candidate{} + err = rows.Scan(&c.TaskID, &c.projectID, &c.SessionID, &c.stage, &c.run, &c.output, &c.background, &c.workflow) + if err != nil { + rows.Close() + return nil, err + } + found = append(found, c) + } + err = errors.Join(rows.Err(), rows.Close()) + if err != nil { + return nil, err + } + + prompts := []ReadyPrompt{} + for _, c := range found { + stopped, err := sessionStopped(ctx, tx, c.TaskID, c.SessionID) + if err != nil { + return nil, err + } + if stopped { + continue + } + c.Kind = ReadyPromptReady + if c.stage != "" { + frozen, err := taskWorkflow(ctx, tx, c.TaskID) + if err != nil { + return nil, err + } + if c.output != "" && len(workflow.Next(frozen.Definition, c.stage)) > 0 { + continue + } + finished := false + err = tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM events WHERE task_id = ? AND kind = 'task.workflow_finished')`, c.run).Scan(&finished) + if err != nil { + return nil, err + } + if c.output == "" && !finished { + c.Kind = ReadyPromptFinishStage + } + } + + settle := "" + err = tx.QueryRowContext(ctx, `SELECT n, settle_state FROM snapshots WHERE task_id = ? AND trigger NOT IN `+cutPointTriggers+` ORDER BY n DESC LIMIT 1`, c.TaskID).Scan(&c.Snapshot, &settle) + if errors.Is(err, sql.ErrNoRows) { + continue + } + if err != nil { + return nil, err + } + reviews, err := snapshotReviews(ctx, tx, c.TaskID, c.Snapshot) + if err != nil { + return nil, err + } + if len(reviews) == 0 && c.workflow != "" || len(reviews) > 0 && settle != "done" { + continue + } + blocked := false + for _, r := range reviews { + if r.State == api.ReviewStateFailed && c.background { + err = emit(api.Event{Kind: "session.error", ProjectId: &c.projectID, TaskId: &c.TaskID, Actor: "byre", Payload: map[string]any{ + "error": fmt.Sprintf("Review %q of snapshot %d failed: %s. The audit can't go on without you.", r.Stage, c.Snapshot, r.Error), + }}) + if err != nil { + return nil, err + } + } + if r.State != api.ReviewStateDone && r.State != api.ReviewStateSkipped { + blocked = true + break + } + } + if blocked { + continue + } + if c.Kind == ReadyPromptReady && len(reviews) == 1 && reviews[0].State == api.ReviewStateSkipped { + c.Kind = ReadyPromptUnchanged + } + prompts = append(prompts, c.ReadyPrompt) + } + return prompts, nil +} diff --git a/internal/store/review.go b/internal/store/review.go index b148d13..f897f1d 100644 --- a/internal/store/review.go +++ b/internal/store/review.go @@ -5,9 +5,11 @@ import ( "database/sql" "encoding/json" "errors" + "fmt" "time" "g.fc.run/ai/byre/internal/api" + "g.fc.run/ai/byre/internal/workflow" ) const reviewColumns = `id, task_id, snapshot_n, stage, coalesce(agent_session_id, ''), vendor, model, state, result, error, created_at, finished_at, finding_states, finding_responses` @@ -18,20 +20,280 @@ func (s *Store) CreateReview(ctx context.Context, r api.Review, definitionID str 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, stage, definition_id, agent_session_id, vendor, model, state, error, created_at, finished_at) - VALUES (?, ?, ?, ?, nullif(?, ''), nullif(?, ''), ?, ?, ?, ?, ?, ?) - `, r.Id, r.TaskId, r.SnapshotN, r.Stage, definitionID, r.AgentSessionId, r.Vendor, r.Model, string(r.State), r.Error, formatTime(r.CreatedAt), finishedAt) - if err != nil { - return err - } - if r.State == api.ReviewStateRunning { - return emitReview(ctx, tx, emit, "review.started", r.Id, "byre") - } - return emitReview(ctx, tx, emit, "review.finished", r.Id, "byre") + return insertReview(ctx, tx, emit, r, definitionID, finishedAt) }) } +func insertReview(ctx context.Context, tx *sql.Tx, emit emitFunc, r api.Review, definitionID, finishedAt string) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO reviews (id, task_id, snapshot_n, stage, definition_id, agent_session_id, vendor, model, state, error, created_at, finished_at) + VALUES (?, ?, ?, ?, nullif(?, ''), nullif(?, ''), ?, ?, ?, ?, ?, ?) + `, r.Id, r.TaskId, r.SnapshotN, r.Stage, definitionID, r.AgentSessionId, r.Vendor, r.Model, string(r.State), r.Error, formatTime(r.CreatedAt), finishedAt) + if err != nil { + return err + } + if r.State == api.ReviewStateRunning { + return emitReview(ctx, tx, emit, "review.started", r.Id, "byre") + } + return emitReview(ctx, tx, emit, "review.finished", r.Id, "byre") +} + +type Reviewer struct { + Session api.AgentSession + Review api.Review +} + +func (s *Store) StartReviewers(ctx context.Context, projectID, definitionID string, reviewers []Reviewer, failed []api.Review) error { + return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { + for _, r := range failed { + err := insertReview(ctx, tx, emit, r, definitionID, formatTime(*r.FinishedAt)) + if err != nil { + return err + } + } + for _, r := range reviewers { + r.Session.State = api.AgentSessionStateRunning + err := insertSession(ctx, tx, emit, r.Session, projectID) + if err != nil { + return err + } + r.Review.State = api.ReviewStateRunning + r.Review.AgentSessionId = r.Session.Id + err = insertReview(ctx, tx, emit, r.Review, definitionID, "") + if err != nil { + return err + } + } + return nil + }) +} + +func (s *Store) FinishSettling(ctx context.Context, taskID string, n int) error { + return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { + _, err := tx.ExecContext(ctx, `UPDATE snapshots SET settle_state = 'done' WHERE task_id = ? AND n = ?`, taskID, n) + return err + }) +} + +type SnapshotRef struct { + TaskID string + N int +} + +type StageStart struct { + TaskID string + SnapshotN int + DefinitionID string + Definition api.WorkflowDefinition + Stage string + Inputs []api.Review +} + +type EndedReviewer struct { + ReviewID string + SessionID string + ProjectID string + RepoPath string +} + +const currentReview = `(SELECT max(sn.n) FROM snapshots sn WHERE sn.task_id = tasks.id AND sn.review = 1)` + +func endReviews(ctx context.Context, tx *sql.Tx, emit emitFunc) ([]EndedReviewer, error) { + type running struct { + EndedReviewer + n int + current int + active bool + state string + } + rows, err := tx.QueryContext(ctx, ` + SELECT reviews.id, coalesce(reviews.agent_session_id, ''), tasks.project_id, (SELECT p.repo_path FROM projects p WHERE p.id = tasks.project_id), + reviews.snapshot_n, coalesce(`+currentReview+`, 0), tasks.status = 'active', + coalesce((SELECT s.state FROM agent_sessions s WHERE s.id = reviews.agent_session_id), '') + FROM reviews JOIN tasks ON tasks.id = reviews.task_id + WHERE reviews.state = 'running'`) + if err != nil { + return nil, err + } + found := []running{} + for rows.Next() { + r := running{} + err = rows.Scan(&r.ReviewID, &r.SessionID, &r.ProjectID, &r.RepoPath, &r.n, &r.current, &r.active, &r.state) + if err != nil { + rows.Close() + return nil, err + } + found = append(found, r) + } + err = errors.Join(rows.Err(), rows.Close()) + if err != nil { + return nil, err + } + + ended := []EndedReviewer{} + for _, r := range found { + reason := "" + switch { + case !r.active: + reason = "the task was closed" + case r.n < r.current: + reason = fmt.Sprintf("superseded by snapshot %d", r.current) + case r.state != string(api.AgentSessionStateRunning): + reason = "the reviewer stopped before submitting a review" + default: + continue + } + err = finishReview(ctx, tx, emit, r.ReviewID, api.ReviewStateFailed, "", reason) + if err != nil { + return nil, err + } + if r.state != "" && r.state != string(api.AgentSessionStateCompleted) { + err = transitionSessionTx(ctx, tx, emit, r.SessionID, nil, api.AgentSessionStateCompleted, "byre", nil) + if err != nil { + return nil, err + } + } + ended = append(ended, r.EndedReviewer) + } + return ended, nil +} + +func startableReviews(ctx context.Context, tx *sql.Tx) ([]SnapshotRef, error) { + rows, err := tx.QueryContext(ctx, ` + SELECT c.id, c.n FROM (SELECT tasks.id, `+currentReview+` AS n FROM tasks WHERE status = 'active') c + WHERE c.n IS NOT NULL AND NOT EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = c.id AND r.snapshot_n >= c.n)`) + if err != nil { + return nil, err + } + return scanSnapshotRefs(rows) +} + +func scanSnapshotRefs(rows *sql.Rows) ([]SnapshotRef, error) { + defer rows.Close() + + refs := []SnapshotRef{} + for rows.Next() { + r := SnapshotRef{} + err := rows.Scan(&r.TaskID, &r.N) + if err != nil { + return nil, err + } + refs = append(refs, r) + } + return refs, rows.Err() +} + +func pendingStages(ctx context.Context, tx *sql.Tx) ([]StageStart, error) { + rows, err := tx.QueryContext(ctx, `SELECT `+reviewColumns+`, coalesce(definition_id, '') FROM reviews + WHERE (task_id, snapshot_n) IN (SELECT tasks.id, `+currentReview+` FROM tasks WHERE status = 'active')`) + if err != nil { + return nil, err + } + type defined struct { + review api.Review + defID string + } + found := []defined{} + for rows.Next() { + d := defined{} + d.review, err = scanReview(trailingScanner{rows, []any{&d.defID}}) + if err != nil { + rows.Close() + return nil, err + } + found = append(found, d) + } + err = errors.Join(rows.Err(), rows.Close()) + if err != nil { + return nil, err + } + + byTask := map[string]map[string]api.Review{} + for _, d := range found { + if byTask[d.review.TaskId] == nil { + byTask[d.review.TaskId] = map[string]api.Review{} + } + byTask[d.review.TaskId][d.review.Stage] = d.review + } + defs := map[string]api.WorkflowDefinition{} + starts := []StageStart{} + seen := map[string]bool{} + for _, d := range found { + if d.review.State != api.ReviewStateDone || d.defID == "" { + continue + } + def, loaded := defs[d.defID] + if !loaded { + def, err = loadDefinition(ctx, tx, d.defID) + if err != nil { + return nil, err + } + defs[d.defID] = def + } + byStage := byTask[d.review.TaskId] + for _, next := range workflow.Next(def, d.review.Stage) { + key := d.review.TaskId + "\x00" + next + if _, started := byStage[next]; started || seen[key] { + continue + } + inputs := []api.Review{} + after := workflow.After(def, next) + for _, a := range after { + in := byStage[a] + if in.State != api.ReviewStateDone { + break + } + inputs = append(inputs, in) + } + if len(inputs) != len(after) { + continue + } + seen[key] = true + starts = append(starts, StageStart{TaskID: d.review.TaskId, SnapshotN: d.review.SnapshotN, DefinitionID: d.defID, Definition: def, Stage: next, Inputs: inputs}) + } + } + return starts, nil +} + +func settleReviews(ctx context.Context, tx *sql.Tx, pending []StageStart) ([]SnapshotRef, error) { + waiting := map[SnapshotRef]bool{} + for _, p := range pending { + waiting[SnapshotRef{TaskID: p.TaskID, N: p.SnapshotN}] = true + _, err := tx.ExecContext(ctx, `UPDATE snapshots SET settle_state = '' WHERE task_id = ? AND n = ?`, p.TaskID, p.SnapshotN) + if err != nil { + return nil, err + } + } + rows, err := tx.QueryContext(ctx, ` + SELECT sn.task_id, sn.n FROM tasks JOIN snapshots sn ON sn.task_id = tasks.id + WHERE tasks.status = 'active' AND sn.settle_state = '' + AND EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = sn.task_id AND r.snapshot_n = sn.n) + AND NOT EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = sn.task_id AND r.snapshot_n = sn.n AND r.state = 'running')`) + if err != nil { + return nil, err + } + settled, err := scanSnapshotRefs(rows) + if err != nil { + return nil, err + } + for _, ref := range settled { + if waiting[ref] { + continue + } + _, err = tx.ExecContext(ctx, `UPDATE snapshots SET settle_state = 'running' WHERE task_id = ? AND n = ?`, ref.TaskID, ref.N) + if err != nil { + return nil, err + } + } + rows, err = tx.QueryContext(ctx, ` + SELECT sn.task_id, sn.n FROM tasks JOIN snapshots sn ON sn.task_id = tasks.id + WHERE tasks.status = 'active' AND sn.settle_state = 'running'`) + if err != nil { + return nil, err + } + return scanSnapshotRefs(rows) +} + func (s *Store) ReviewDefinition(ctx context.Context, id string) (string, api.WorkflowDefinition, error) { defID := sql.NullString{} err := s.db.QueryRowContext(ctx, `SELECT definition_id FROM reviews WHERE id = ?`, id).Scan(&defID) @@ -70,29 +332,17 @@ func (s *Store) ListReviews(ctx context.Context, taskID string) ([]api.Review, e } func (s *Store) SnapshotReviews(ctx context.Context, taskID string, n int) ([]api.Review, error) { - rows, err := s.db.QueryContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE task_id = ? AND snapshot_n = ? ORDER BY created_at`, taskID, n) + return snapshotReviews(ctx, s.db, taskID, n) +} + +func snapshotReviews(ctx context.Context, q querier, taskID string, n int) ([]api.Review, error) { + rows, err := q.QueryContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE task_id = ? AND snapshot_n = ? ORDER BY created_at`, taskID, n) 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) LatestDoneStages(ctx context.Context) ([]string, error) { - return queryIDs(ctx, s.db, ` - SELECT r.id FROM reviews r JOIN tasks t ON t.id = r.task_id - WHERE t.status = ? AND r.state = ? AND r.definition_id IS NOT NULL - AND r.snapshot_n = (SELECT max(snapshot_n) FROM reviews WHERE task_id = r.task_id) - ORDER BY r.created_at`, string(api.TaskStatusActive), string(api.ReviewStateDone)) -} - func (s *Store) FinishReview(ctx context.Context, id string, state api.ReviewState, result map[string]any, errText string) error { text := "" if result != nil { @@ -103,22 +353,26 @@ func (s *Store) FinishReview(ctx context.Context, id string, state api.ReviewSta 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, "byre") + return finishReview(ctx, tx, emit, id, state, text, errText) }) } +func finishReview(ctx context.Context, tx *sql.Tx, emit emitFunc, id string, state api.ReviewState, result, errText string) error { + res, err := tx.ExecContext(ctx, `UPDATE reviews SET state = ?, result = ?, error = ?, finished_at = ? WHERE id = ? AND state = ?`, + string(state), result, 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, "byre") +} + func (s *Store) SetFindingState(ctx context.Context, id string, index int, state api.FindingState, actor string) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { res, err := tx.ExecContext(ctx, ` diff --git a/internal/store/snapshot.go b/internal/store/snapshot.go index 945d612..549300e 100644 --- a/internal/store/snapshot.go +++ b/internal/store/snapshot.go @@ -12,13 +12,20 @@ const cutPointTriggers = `('answer', 'fork', 'ask_human')` const snapshotColumns = `id, task_id, agent_session_id, n, commit_sha, tree_sha, event_seq, trigger, created_at` -func (s *Store) CreateSnapshot(ctx context.Context, snap api.Snapshot, projectID string) error { +func (s *Store) CreateSnapshot(ctx context.Context, snap api.Snapshot, projectID string, review bool) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - _, err := tx.ExecContext(ctx, `INSERT INTO snapshots (`+snapshotColumns+`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, - snap.Id, snap.TaskId, snap.AgentSessionId, snap.N, snap.CommitSha, snap.TreeSha, snap.EventSeq, string(snap.Trigger), formatTime(snap.CreatedAt)) + _, err := tx.ExecContext(ctx, `INSERT INTO snapshots (`+snapshotColumns+`, review) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, + ? AND EXISTS (SELECT 1 FROM projects WHERE id = ? AND review_workflow != ''))`, + snap.Id, snap.TaskId, snap.AgentSessionId, snap.N, snap.CommitSha, snap.TreeSha, snap.EventSeq, string(snap.Trigger), formatTime(snap.CreatedAt), review, projectID) if err != nil { return err } + if review { + _, err = tx.ExecContext(ctx, `UPDATE tasks SET review_requests = max(review_requests - 1, 0) WHERE id = ?`, snap.TaskId) + if err != nil { + return err + } + } _, err = tx.ExecContext(ctx, `UPDATE tasks SET ready = '' WHERE id = ? AND ? NOT IN `+cutPointTriggers, snap.TaskId, string(snap.Trigger)) if err != nil { return err diff --git a/internal/store/store.go b/internal/store/store.go index 17e8291..6d229ad 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -705,12 +705,6 @@ func (s *Store) CancelMergeAfterReview(ctx context.Context, id, actor string) er }) } -func (s *Store) FinishedMergeAfterReviews(ctx context.Context) ([]string, error) { - return queryIDs(ctx, s.db, ` - SELECT tasks.merge_after_review_id FROM tasks JOIN reviews ON reviews.id = tasks.merge_after_review_id - WHERE tasks.merge_after_review_by != '' AND reviews.state != ?`, string(api.ReviewStateRunning)) -} - func (s *Store) TakeMergeAfterReview(ctx context.Context, id string, reviewIDs []string) (MergeAfterReview, error) { mar := MergeAfterReview{} err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { @@ -752,22 +746,6 @@ func (s *Store) SetTaskSummary(ctx context.Context, id, summary, actor string) e }) } -func (s *Store) ReadyPromptDue(ctx context.Context, id string) (bool, error) { - due := false - err := s.db.QueryRowContext(ctx, `SELECT EXISTS ( - SELECT 1 FROM events e, tasks t - WHERE t.id = ? AND e.task_id = t.id AND e.seq > t.ready_prompt_seq - AND ((e.kind = 'snapshot.created' AND json_extract(e.payload, '$.trigger') NOT IN `+cutPointTriggers+`) - OR (e.kind = 'user.message' AND e.actor != ? AND e.actor NOT LIKE '% (lead)' AND json_extract(e.payload, '$.bounce') IS NOT 1)) - )`, id, agentActor).Scan(&due) - return due, err -} - -func (s *Store) SetReadyPromptSeq(ctx context.Context, id string, seq int64) error { - _, err := s.db.ExecContext(ctx, `UPDATE tasks SET ready_prompt_seq = ? WHERE id = ?`, seq, id) - return err -} - func (s *Store) MarkReady(ctx context.Context, id, overview, merge, actor string) error { return s.setReady(ctx, id, actor, merge, "task.ready", map[string]any{"id": id, "overview": overview, "merge": merge}) } @@ -1092,8 +1070,12 @@ func (s *Store) TaskHandoffs(ctx context.Context, taskID string) ([]Handoff, err } func (s *Store) SessionStopped(ctx context.Context, taskID, sessionID string) (bool, error) { + return sessionStopped(ctx, s.db, taskID, sessionID) +} + +func sessionStopped(ctx context.Context, q querier, taskID, sessionID string) (bool, error) { stopped := false - err := s.db.QueryRowContext(ctx, ` + err := q.QueryRowContext(ctx, ` SELECT coalesce(json_extract(e.payload, '$.stopped'), 0) = 1 AND NOT EXISTS ( SELECT 1 FROM events a WHERE a.task_id = e.task_id AND a.agent_session_id = e.agent_session_id AND a.seq > e.seq @@ -1630,7 +1612,8 @@ const taskStateExpr = `CASE WHEN s.state != 'idle' THEN s.state WHEN s.subagents > 0 THEN 'running' WHEN EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = tasks.id AND r.state = 'running') THEN 'reviewing' - WHEN tasks.review_requests > 0 THEN 'review_pending' + WHEN tasks.review_requests > 0 OR EXISTS (SELECT 1 FROM snapshots sn WHERE sn.task_id = tasks.id AND sn.review = 1 + AND NOT EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = sn.task_id AND r.snapshot_n >= sn.n)) THEN 'review_pending' WHEN tasks.ready IN ('complete', 'continue') THEN 'ready_to_merge' WHEN tasks.ready = 'discard' THEN 'discard_suggested' WHEN tasks.stage_output != '' THEN 'finished' diff --git a/internal/store/workflow.go b/internal/store/workflow.go index 79a6145..42e2b8c 100644 --- a/internal/store/workflow.go +++ b/internal/store/workflow.go @@ -158,9 +158,13 @@ func (s *Store) FreezeWorkflow(ctx context.Context, name string) (FrozenWorkflow } func (s *Store) TaskWorkflow(ctx context.Context, taskID string) (FrozenWorkflow, error) { + return taskWorkflow(ctx, s.db, taskID) +} + +func taskWorkflow(ctx context.Context, q querier, taskID string) (FrozenWorkflow, error) { f := FrozenWorkflow{} defID := sql.NullString{} - err := s.db.QueryRowContext(ctx, ` + err := q.QueryRowContext(ctx, ` SELECT r.workflow_name, r.workflow_definition_id, r.workflow_brief, r.workflow_title, r.audit FROM tasks t JOIN tasks r ON r.id = CASE WHEN t.workflow_stage != '' THEN t.parent_task_id ELSE t.id END WHERE t.id = ?`, taskID).Scan(&f.Name, &defID, &f.Brief, &f.Title, &f.Audit) @@ -174,7 +178,7 @@ func (s *Store) TaskWorkflow(ctx context.Context, taskID string) (FrozenWorkflow return FrozenWorkflow{}, fmt.Errorf("%w: task %s has no workflow", ErrNotFound, taskID) } f.DefinitionID = defID.String - f.Definition, err = loadDefinition(ctx, s.db, f.DefinitionID) + f.Definition, err = loadDefinition(ctx, q, f.DefinitionID) return f, err }