Let workflows merge or discard their own tasks

This commit is contained in:
Ian Gulliver
2026-10-22 16:46:57 -07:00
parent b2cb15dd2a
commit efa1513568
13 changed files with 489 additions and 398 deletions
+36
View File
@@ -19,6 +19,7 @@ type Work struct {
StartStages []StageStart
SettleReviews []SnapshotRef
ReadyPrompts []ReadyPrompt
SelfCloses []SelfClose
AdvanceStages []string
LeadDeliveries []LeadDelivery
QueuedDelivery []QueuedDelivery
@@ -80,6 +81,10 @@ func (s *Store) Coordinate(ctx context.Context) (Work, error) {
if err != nil {
return err
}
work.SelfCloses, err = selfCloses(ctx, tx)
if err != nil {
return err
}
work.AdvanceStages, err = finishedStages(ctx, tx)
if err != nil {
return err
@@ -101,3 +106,34 @@ func (s *Store) Coordinate(ctx context.Context) (Work, error) {
})
return work, err
}
type SelfClose struct {
TaskID string
Ready string
By string
}
func selfCloses(ctx context.Context, tx *sql.Tx) ([]SelfClose, error) {
rows, err := tx.QueryContext(ctx, `
SELECT tasks.id, tasks.ready, r.created_by
FROM tasks JOIN tasks r ON r.id = tasks.parent_task_id JOIN workflow_definitions d ON d.id = r.workflow_definition_id
WHERE tasks.status = 'active' AND tasks.workflow_stage != '' AND tasks.discard_state = ''
AND tasks.merging_by = '' AND tasks.finalizing = '' AND tasks.merge_queued_by = '' AND tasks.parent_merge_by = '' AND tasks.landing_session = ''
AND ((tasks.ready IN ('complete', 'continue') AND d.auto_merge) OR (tasks.ready = 'discard' AND d.auto_discard))
AND NOT `+implementerRunning+`
ORDER BY tasks.created_at`)
if err != nil {
return nil, err
}
defer rows.Close()
closes := []SelfClose{}
for rows.Next() {
c := SelfClose{}
err = rows.Scan(&c.TaskID, &c.Ready, &c.By)
if err != nil {
return nil, err
}
closes = append(closes, c)
}
return closes, rows.Err()
}
+2 -2
View File
@@ -12,7 +12,7 @@ import (
func insertDefinition(ctx context.Context, tx *sql.Tx, def api.WorkflowDefinition) (string, error) {
id := ids.New(ids.WorkflowDefinition)
_, err := tx.ExecContext(ctx, `INSERT INTO workflow_definitions (id, description) VALUES (?, ?)`, id, def.Description)
_, err := tx.ExecContext(ctx, `INSERT INTO workflow_definitions (id, description, auto_merge, auto_discard) VALUES (?, ?, ?, ?)`, id, def.Description, def.AutoMerge, def.AutoDiscard)
if err != nil {
return "", err
}
@@ -63,7 +63,7 @@ type stepRow struct {
func loadDefinition(ctx context.Context, q querier, id string) (api.WorkflowDefinition, error) {
def := api.WorkflowDefinition{}
err := q.QueryRowContext(ctx, `SELECT description FROM workflow_definitions WHERE id = ?`, id).Scan(&def.Description)
err := q.QueryRowContext(ctx, `SELECT description, auto_merge, auto_discard FROM workflow_definitions WHERE id = ?`, id).Scan(&def.Description, &def.AutoMerge, &def.AutoDiscard)
if errors.Is(err, sql.ErrNoRows) {
return def, fmt.Errorf("%w: no workflow definition %s", ErrNotFound, id)
}
@@ -0,0 +1,5 @@
ALTER TABLE workflow_definitions ADD COLUMN auto_merge INTEGER NOT NULL DEFAULT 0;
ALTER TABLE workflow_definitions ADD COLUMN auto_discard INTEGER NOT NULL DEFAULT 0;
UPDATE workflow_definitions SET auto_discard = 1
WHERE id IN (SELECT definition_id FROM workflows WHERE name IN (SELECT audit_workflow FROM projects WHERE audit_workflow != ''))
OR id IN (SELECT workflow_definition_id FROM tasks WHERE audit != '' AND status = 'active' AND workflow_definition_id IS NOT NULL);
+23 -1
View File
@@ -762,6 +762,27 @@ func (s *Store) ClearReady(ctx context.Context, id, actor, reason string) error
})
}
func (s *Store) TakeReady(ctx context.Context, id, ready string) (bool, error) {
taken := false
err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
projectID := ""
err := tx.QueryRowContext(ctx, `UPDATE tasks SET ready = '' WHERE id = ? AND status = ? AND ready = ? RETURNING project_id`, id, string(api.TaskStatusActive), ready).Scan(&projectID)
if errors.Is(err, sql.ErrNoRows) {
return nil
}
if err != nil {
return err
}
taken = true
reason := "merge"
if ready == "discard" {
reason = "discard"
}
return emit(api.Event{Kind: "task.ready_cleared", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id, "reason": reason}})
})
return taken, err
}
func (s *Store) setReady(ctx context.Context, id, actor, ready, kind string, payload map[string]any) error {
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
projectID := ""
@@ -1667,7 +1688,8 @@ const taskColumns = `id, project_id, title, merge_title, branch, worktree_path,
// Keep the child lookup separate from the task itself so SQLite can use tasks_parent.
const backgroundExpr = `(audit != '' OR EXISTS (SELECT 1 FROM tasks r WHERE r.id = tasks.parent_task_id AND tasks.workflow_stage != '' AND r.audit != ''))
AND NOT EXISTS (SELECT 1 FROM events WHERE events.task_id IN (SELECT tasks.id UNION ALL SELECT c.id FROM tasks c WHERE c.parent_task_id = tasks.id AND c.parent_task_id != '' AND tasks.workflow_name != '')
AND kind IN ('inbox.created', 'task.ready', 'session.error', 'session.rate_limited'))`
AND (kind IN ('inbox.created', 'task.ready', 'session.error', 'session.rate_limited')
OR (kind = 'task.discard_suggested' AND NOT EXISTS (SELECT 1 FROM tasks r JOIN workflow_definitions d ON d.id = r.workflow_definition_id WHERE r.id IN (tasks.id, tasks.parent_task_id) AND d.auto_discard))))`
const sessionColumns = `id, coalesce(task_id, ''), owner, vendor, model, vendor_session_id, state, role, member, created_by, created_at, forked_from, forked_at, subagents`