Add background audits of the default branch's newest commit

internal/agents/audit.go

@@ -0,0 +1,146 @@
+package agents
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strings"
+
+	"g.fc.run/ai/byre/internal/api"
+	"g.fc.run/ai/byre/internal/git"
+)
+
+func (m *Manager) SetAuditWorkflow(ctx context.Context, projectID, name, actor string) (api.Project, error) {
+	name = strings.TrimSpace(name)
+	if name != "" {
+		_, err := m.store.GetWorkflow(ctx, name)
+		if err != nil {
+			return api.Project{}, err
+		}
+		project, err := m.store.GetProject(ctx, projectID)
+		if err != nil {
+			return api.Project{}, err
+		}
+		creator, err := m.store.GetUser(ctx, project.CreatedBy)
+		if err != nil {
+			return api.Project{}, err
+		}
+		if creator.GitName == "" || creator.GitEmail == "" {
+			return api.Project{}, fmt.Errorf("%w: audits run as the project's creator %s, who has no git identity", ErrInvalid, creator.Login)
+		}
+	}
+	p, err := m.store.SetAuditWorkflow(ctx, projectID, name, actor)
+	if err != nil {
+		return api.Project{}, err
+	}
+	go m.audit(p.Id)
+	return p, nil
+}
+
+func (m *Manager) auditAll(ctx context.Context) {
+	projects, err := m.store.ListProjects(ctx)
+	if err != nil {
+		slog.Error("start audits", "err", err)
+		return
+	}
+	for _, p := range projects {
+		tasks, err := m.store.ListTasks(ctx, p.Id)
+		if err != nil {
+			slog.Error("start audits", "project", p.Id, "err", err)
+			continue
+		}
+		for _, t := range tasks {
+			if t.Status == api.TaskStatusActive && t.Background && t.State == api.TaskStateDiscardSuggested {
+				m.dismissAudit(t.Id)
+			}
+		}
+		m.audit(p.Id)
+	}
+}
+
+func (m *Manager) audit(projectID string) {
+	err := m.startAudit(context.Background(), projectID)
+	if err != nil {
+		slog.Error("start audit", "project", projectID, "err", err)
+	}
+}
+
+func (m *Manager) startAudit(ctx context.Context, projectID string) error {
+	m.auditMu.Lock()
+	defer m.auditMu.Unlock()
+
+	project, err := m.store.GetProject(ctx, projectID)
+	if err != nil || project.AuditWorkflow == "" {
+		return err
+	}
+	last, active, err := m.store.LastAudit(ctx, projectID)
+	if err != nil || active {
+		return err
+	}
+	head, err := git.BranchHead(project.RepoPath, project.DefaultBranch)
+	if err != nil {
+		return err
+	}
+	if head == last {
+		return nil
+	}
+	creator, err := m.store.GetUser(ctx, project.CreatedBy)
+	if err != nil {
+		return err
+	}
+	tasks, err := m.StartWorkflow(ctx, project, creator, "Audit of "+short(head), head, "byre", project.AuditWorkflow, auditBrief(project.DefaultBranch, head), nil, head)
+	if err == nil {
+		slog.Info("audit started", "project", projectID, "commit", head, "workflow", project.AuditWorkflow)
+		return nil
+	}
+	for _, t := range tasks {
+		m.auditError(ctx, t, fmt.Sprintf("The audit of %s didn't start: %v. Discard this task to let audits continue; the next commit on %s starts one.", short(head), err, project.DefaultBranch))
+	}
+	return err
+}
+
+func (m *Manager) auditError(ctx context.Context, task api.Task, msg string) {
+	err := m.store.AppendEvent(ctx, api.Event{
+		Kind:      "session.error",
+		ProjectId: &task.ProjectId,
+		TaskId:    &task.Id,
+		Actor:     "byre",
+		Payload:   map[string]any{"error": msg},
+	})
+	if err != nil {
+		slog.Error("record audit error", "task", task.Id, "err", err)
+	}
+}
+
+func auditBrief(branch, commit string) string {
+	return fmt.Sprintf("Audit of %s at %s. This isn't a review of a change: check the whole codebase as it stands against the project's standards and expectations, and fix what falls short. "+
+		"byre runs this task in the background: the human doesn't see it until you ask a question, call ready_for_merge or hit an error. "+
+		"Finish your stage with finish_stage as usual, even when you found nothing. Once the workflow has finished, byre asks whether the task is ready to merge; "+
+		"if nothing needed changing, answer with suggest_discard, and byre discards the task without bothering the human.", branch, commit)
+}
+
+func (m *Manager) dismissAudit(taskID string) {
+	ctx := context.Background()
+	task, err := m.store.GetTask(ctx, taskID)
+	if err != nil {
+		slog.Error("dismiss audit", "task", taskID, "err", err)
+		return
+	}
+	if !task.Background || task.Status != api.TaskStatusActive {
+		return
+	}
+	finished, err := m.store.WorkflowFinished(ctx, *task.WorkflowRun)
+	if err != nil {
+		slog.Error("dismiss audit", "task", taskID, "err", err)
+		return
+	}
+	if !finished {
+		return
+	}
+	_, err = m.Discard(ctx, task, "byre")
+	if err != nil {
+		slog.Error("dismiss audit", "task", taskID, "err", err)
+		return
+	}
+	slog.Info("audit dismissed itself", "task", taskID)
+}

internal/agents/lead.go

@@ -368,6 +368,12 @@ func (m *Manager) Observe(events []api.Event) {
 				go m.promptReady(id, nil)
 			}
 		}
+		switch {
+		case (ev.Kind == "task.merged" || ev.Kind == "task.merged_to_main" || ev.Kind == "task.discarded") && ev.ProjectId != nil:
+			go m.audit(*ev.ProjectId)
+		case ev.Kind == "task.discard_suggested" && ev.TaskId != nil:
+			go m.dismissAudit(*ev.TaskId)
+		}
 		switch ev.Kind {
 		case "inbox.created", "inbox.updated", "task.created", "task.merge_queued", "task.merge_after_review", "task.merge_after_review_cancelled", "task.merge_after_review_dropped", "task.merge_started", "task.merged", "task.discarded", "turn.ended", "task.workflow_stage", "task.workflow_finished", "review.started", "review.finished", "task.state", "task.ready", "task.discard_suggested":
 		case "task.ready_cleared":
@@ -416,6 +422,10 @@ func (m *Manager) signal(ctx context.Context, batch []api.Event) error {
 	if err != nil {
 		return err
 	}
+	batch, err = m.dropBackground(ctx, batch)
+	if err != nil || len(batch) == 0 {
+		return err
+	}
 
 	texts := []string{}
 	for _, ev := range batch {
@@ -464,6 +474,23 @@ func (m *Manager) signal(ctx context.Context, batch []api.Event) error {
 	return nil
 }
 
+func (m *Manager) dropBackground(ctx context.Context, batch []api.Event) ([]api.Event, error) {
+	kept := []api.Event{}
+	for _, ev := range batch {
+		if ev.TaskId != nil {
+			task, err := m.store.GetTask(ctx, *ev.TaskId)
+			if err != nil {
+				return nil, err
+			}
+			if task.Background {
+				continue
+			}
+		}
+		kept = append(kept, ev)
+	}
+	return kept, nil
+}
+
 func quietSignal(ev api.Event) bool {
 	switch ev.Kind {
 	case "task.state", "review.started":

internal/agents/leadtools.go

@@ -284,7 +284,7 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
 					}
 				}
 				if args.Workflow != "" {
-					tasks, err := m.StartWorkflow(ctx, project, user, args.Title, base, leadActor(user.Login), args.Workflow, args.Prompt, nil)
+					tasks, err := m.StartWorkflow(ctx, project, user, args.Title, base, leadActor(user.Login), args.Workflow, args.Prompt, nil, "")
 					if err != nil {
 						return "", err
 					}
@@ -897,7 +897,7 @@ func (m *Manager) listTasksText(ctx context.Context) (string, error) {
 		listed := false
 		started := map[string]bool{}
 		for _, t := range tasks {
-			if t.Status != api.TaskStatusActive {
+			if t.Status != api.TaskStatusActive || t.Background {
 				continue
 			}
 			listed = true

internal/agents/manager.go

@@ -84,6 +84,8 @@ type Manager struct {
 
 	workflowMu sync.Mutex
 
+	auditMu sync.Mutex
+
 	previewMu        sync.Mutex
 	previewListeners map[int]*previewListener
 
@@ -475,6 +477,7 @@ func (m *Manager) Recover(ctx context.Context) error {
 	go m.deliverAllQueued(context.Background())
 	go m.startHeldMerges(context.Background())
 	go m.recoverMergeAfterReviews(context.Background())
+	go m.auditAll(context.Background())
 	return nil
 }
 

internal/agents/ready.go

@@ -189,6 +189,10 @@ func (m *Manager) tryPromptReady(ctx context.Context, taskID string, fromReview
 		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
 		}

internal/agents/tasks.go

@@ -75,7 +75,7 @@ func (m *Manager) StartTodo(ctx context.Context, project api.Project, creator ap
 
 	started := []api.Task{}
 	if workflowName != "" {
-		started, err = m.StartWorkflow(ctx, project, creator, item.Title, project.DefaultBranch, actor, workflowName, todoPrompt(item, discuss), &item.Title)
+		started, err = m.StartWorkflow(ctx, project, creator, item.Title, project.DefaultBranch, actor, workflowName, todoPrompt(item, discuss), &item.Title, "")
 		if err != nil {
 			return nil, err
 		}

internal/agents/workflows.go

@@ -231,7 +231,7 @@ type parentStage struct {
 	commit string
 }
 
-func (m *Manager) StartWorkflow(ctx context.Context, project api.Project, creator api.User, title, base, actor, name, brief string, todoItem *string) ([]api.Task, error) {
+func (m *Manager) StartWorkflow(ctx context.Context, project api.Project, creator api.User, title, base, actor, name, brief string, todoItem *string, audit string) ([]api.Task, error) {
 	if creator.GitName == "" || creator.GitEmail == "" {
 		return nil, ErrNoGitIdentity
 	}
@@ -241,6 +241,7 @@ func (m *Manager) StartWorkflow(ctx context.Context, project api.Project, creato
 	}
 	frozen.Brief = brief
 	frozen.Title = title
+	frozen.Audit = audit
 
 	run := store.NewID()
 	tasks := []api.Task{}
@@ -250,23 +251,29 @@ func (m *Manager) StartWorkflow(ctx context.Context, project api.Project, creato
 			from = project.DefaultBranch
 		}
 		task, err := m.startStage(ctx, project, creator, frozen, run, stage, from, todoItem, nil, actor)
+		if task.Id != "" {
+			tasks = append(tasks, task)
+		}
 		if err != nil {
 			return tasks, fmt.Errorf("workflow %q: stage %q did not start: %w", name, stage, err)
 		}
-		tasks = append(tasks, task)
 	}
 	return tasks, nil
 }
 
 func (m *Manager) startStage(ctx context.Context, project api.Project, creator api.User, frozen store.FrozenWorkflow, run, stage, base string, todoItem *string, parents []parentStage, actor string) (api.Task, error) {
-	task, err := m.createTask(ctx, project, api.Task{
+	t := api.Task{
 		Title:         fmt.Sprintf("%s: %s", frozen.Title, stage),
 		MergeTitle:    frozen.Title,
 		CreatedBy:     creator.Login,
 		TodoItem:      todoItem,
 		WorkflowRun:   &run,
 		WorkflowStage: &stage,
-	}, &frozen, base, actor)
+	}
+	if frozen.Audit != "" {
+		t.Audit = &frozen.Audit
+	}
+	task, err := m.createTask(ctx, project, t, &frozen, base, actor)
 	if err != nil {
 		return api.Task{}, err
 	}
@@ -489,7 +496,11 @@ func (m *Manager) advanceTo(ctx context.Context, from api.Task, frozen store.Fro
 	}
 
 	base := parents[0].commit
-	if workflow.FromMain(frozen.Definition, stage) {
+	switch {
+	case !workflow.FromMain(frozen.Definition, stage):
+	case frozen.Audit != "":
+		base = frozen.Audit
+	default:
 		base = project.DefaultBranch
 	}
 	task, err := m.startStage(ctx, project, creator, frozen, *from.WorkflowRun, stage, base, from.TodoItem, parents, "byre")
@@ -811,6 +822,32 @@ func (m *Manager) workflowTools(tool func(name, description, schema string, run
 				return fmt.Sprintf("Project %s reviews with workflow %q from its next snapshot on.", p.Name, p.ReviewWorkflow), nil
 			}),
 
+		tool("set_audit_workflow",
+			"Set the audit workflow (any saved workflow) that byre runs in the background on a project's default branch: one run at a time, started whenever the branch's newest commit hasn't had one, to check the whole codebase against its standards rather than one change. "+
+				"An audit's tasks stay out of list_tasks and your signals until one asks a question, calls ready_for_merge or hits an error; one that suggests discarding first is discarded by byre. An empty workflow turns audits off. Only when the human asks.",
+			`{"type": "object", "properties": {
+				"project_id": {"type": "string"},
+				"workflow": {"type": "string"}
+			}, "required": ["project_id", "workflow"]}`,
+			func(ctx context.Context, user api.User, raw json.RawMessage) (string, error) {
+				args := struct {
+					ProjectID string `json:"project_id"`
+					Workflow  string `json:"workflow"`
+				}{}
+				err := json.Unmarshal(raw, &args)
+				if err != nil {
+					return "", err
+				}
+				p, err := m.SetAuditWorkflow(ctx, args.ProjectID, args.Workflow, leadActor(user.Login))
+				if err != nil {
+					return "", err
+				}
+				if p.AuditWorkflow == "" {
+					return fmt.Sprintf("Project %s has no audit workflow now: it isn't audited.", p.Name), nil
+				}
+				return fmt.Sprintf("Project %s is audited with workflow %q.", p.Name, p.AuditWorkflow), nil
+			}),
+
 		tool("delete_workflow",
 			"Delete a workflow from the server. Tasks that froze it keep their copy. Only when the human asks.",
 			`{"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}`,