ai/byre
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 9 Wiki Activity

Give each workflow run a parent task with no agent

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-21 09:02:09 -07:00
parent d062d7ccae
commit a130b9556d
23 changed files with 661 additions and 474 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/store/migrateruns.go
+70
View File
@@ -0,0 +1,70 @@
package store
import (
"database/sql"
"fmt"
"g.fc.run/ai/byre/internal/ids"
)
func migrateRunTasks(tx *sql.Tx, _ string) error {
rows, err := tx.Query(`SELECT workflow_run FROM tasks WHERE workflow_run != '' GROUP BY workflow_run ORDER BY min(rowid)`)
if err != nil {
return err
}
runs := []string{}
for rows.Next() {
run := ""
err = rows.Scan(&run)
if err != nil {
rows.Close()
return err
}
runs = append(runs, run)
}
rows.Close()
err = rows.Err()
if err != nil {
return err
}
for _, run := range runs {
id := ids.Derive(ids.Task, run)
_, err = tx.Exec(`
INSERT INTO tasks (id, project_id, title, branch, worktree_path, base_ref, status, created_by, created_at, closed_at, todo_item, color,
workflow_name, workflow_definition_id, workflow_brief, workflow_title, audit)
SELECT ?1, f.project_id, f.workflow_title, '', '', '',
CASE
WHEN EXISTS (SELECT 1 FROM tasks s WHERE s.workflow_run = ?2 AND s.status = 'active') THEN 'active'
WHEN EXISTS (SELECT 1 FROM tasks s WHERE s.workflow_run = ?2 AND s.status = 'merged') THEN 'merged'
ELSE 'discarded'
END,
f.created_by, f.created_at,
CASE WHEN EXISTS (SELECT 1 FROM tasks s WHERE s.workflow_run = ?2 AND s.status = 'active') THEN NULL
ELSE (SELECT max(s.closed_at) FROM tasks s WHERE s.workflow_run = ?2) END,
f.todo_item, f.color, f.workflow_name, f.workflow_definition_id, f.workflow_brief, f.workflow_title, f.audit
FROM tasks f WHERE f.workflow_run = ?2 ORDER BY f.created_at LIMIT 1`, id, run)
if err != nil {
return fmt.Errorf("run %s: %w", run, err)
}
_, err = tx.Exec(`
UPDATE events SET task_id = ?1, payload = json_set(payload, '$.id', ?1)
WHERE kind = 'task.workflow_finished' AND task_id IN (SELECT id FROM tasks WHERE workflow_run = ?2)`, id, run)
if err != nil {
return fmt.Errorf("run %s: %w", run, err)
}
_, err = tx.Exec(`
UPDATE tasks SET parent_task_id = ?1, workflow_name = '', workflow_definition_id = NULL, workflow_brief = '', workflow_title = '', audit = '', todo_item = ''
WHERE workflow_run = ?2`, id, run)
if err != nil {
return fmt.Errorf("run %s: %w", run, err)
}
_, err = tx.Exec(`UPDATE tasks SET state = `+taskStateExpr+` WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("run %s: %w", run, err)
}
}
_, err = tx.Exec(`ALTER TABLE tasks DROP COLUMN workflow_run`)
return err
}
internal/store/store.go
+64 -25
View File
@@ -35,6 +35,7 @@ var goMigrations = map[string]func(tx *sql.Tx, dataDir string) error{
"071-workflow-steps": migrateWorkflowSteps,
"078-normalize-workflow-definitions": migrateWorkflowDefinitions,
"085-prefix-ids": migratePrefixIDs,
"088-workflow-run-tasks": migrateRunTasks,
}
func loadMigrations() ([]migration, error) {
@@ -331,16 +332,15 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
var workflowDefinitionID *string
workflowBrief := ""
workflowTitle := ""
workflowRun := ""
workflowStage := ""
if frozen != nil {
workflowName = frozen.Name
workflowDefinitionID = &frozen.DefinitionID
workflowBrief = frozen.Brief
workflowTitle = frozen.Title
workflowRun = *t.WorkflowRun
}
workflowStage := ""
if t.WorkflowStage != nil {
workflowStage = *t.WorkflowStage
t.Workflow = &workflowName
}
todoItem := ""
if t.TodoItem != nil {
@@ -365,7 +365,6 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
audit := ""
if t.Audit != nil {
audit = *t.Audit
t.Background = true
}
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
if parentTaskID != "" {
@@ -379,16 +378,18 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
return ErrParentNotOpen
}
}
err := tx.QueryRowContext(ctx, `
SELECT min(candidate) FROM (
SELECT ? AS candidate
UNION ALL SELECT port_base + ? FROM tasks WHERE status = ?
) WHERE candidate NOT IN (SELECT port_base FROM tasks WHERE status = ?)
`, s.portFirst, PortBlock, string(api.TaskStatusActive), string(api.TaskStatusActive)).Scan(&t.PortBase)
if err != nil {
return err
if t.WorktreePath != "" {
err := tx.QueryRowContext(ctx, `
SELECT min(candidate) FROM (
SELECT ? AS candidate
UNION ALL SELECT port_base + ? FROM tasks WHERE status = ? AND port_base != 0
) WHERE candidate NOT IN (SELECT port_base FROM tasks WHERE status = ?)
`, s.portFirst, PortBlock, string(api.TaskStatusActive), string(api.TaskStatusActive)).Scan(&t.PortBase)
if err != nil {
return err
}
}
err = tx.QueryRowContext(ctx, `
err := tx.QueryRowContext(ctx, `
WITH RECURSIVE palette(c) AS (SELECT 0 UNION ALL SELECT c + 1 FROM palette WHERE c + 1 < ?)
SELECT c FROM palette
ORDER BY c IN (SELECT color FROM tasks WHERE status = ?), coalesce((SELECT max(rowid) FROM tasks WHERE color = c), -1), c
@@ -398,9 +399,13 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
return err
}
_, err = tx.ExecContext(ctx, `
INSERT INTO tasks (id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, todo_item, forked_from, forked_at, workflow_name, workflow_definition_id, workflow_brief, workflow_title, workflow_run, workflow_stage, parent_task_id, audit)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, t.Id, t.ProjectId, t.Title, mergeTitle, t.Branch, t.WorktreePath, t.BaseRef, t.PortBase, t.Color, string(t.Status), t.CreatedBy, formatTime(t.CreatedAt), todoItem, forkedFrom, forkedAt, workflowName, workflowDefinitionID, workflowBrief, workflowTitle, workflowRun, workflowStage, parentTaskID, audit)
INSERT INTO tasks (id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, todo_item, forked_from, forked_at, workflow_name, workflow_definition_id, workflow_brief, workflow_title, workflow_stage, parent_task_id, audit)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, t.Id, t.ProjectId, t.Title, mergeTitle, t.Branch, t.WorktreePath, t.BaseRef, t.PortBase, t.Color, string(t.Status), t.CreatedBy, formatTime(t.CreatedAt), todoItem, forkedFrom, forkedAt, workflowName, workflowDefinitionID, workflowBrief, workflowTitle, workflowStage, parentTaskID, audit)
if err != nil {
return err
}
t, err = scanTask(tx.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE id = ?`, t.Id))
if err != nil {
return err
}
@@ -954,7 +959,11 @@ func (s *Store) closeTask(ctx context.Context, id string, status api.TaskStatus,
if err != nil {
return err
}
return dismissTaskInbox(ctx, tx, emit, id, actor)
err = dismissTaskInbox(ctx, tx, emit, id, actor)
if err != nil {
return err
}
return closeRun(ctx, tx, emit, id, actor)
})
if err != nil {
return api.Task{}, err
@@ -962,6 +971,24 @@ func (s *Store) closeTask(ctx context.Context, id string, status api.TaskStatus,
return s.GetTask(ctx, id)
}
func closeRun(ctx context.Context, tx *sql.Tx, emit emitFunc, stageID, actor string) error {
runID := ""
err := tx.QueryRowContext(ctx, `
UPDATE tasks SET closed_at = ?,
status = CASE WHEN EXISTS (SELECT 1 FROM tasks c WHERE c.parent_task_id = tasks.id AND c.status = ?) THEN ? ELSE ? END
WHERE id = (SELECT parent_task_id FROM tasks WHERE id = ? AND workflow_stage != '') AND status = ?
AND NOT EXISTS (SELECT 1 FROM tasks c WHERE c.parent_task_id = tasks.id AND c.status = ?)
RETURNING id`,
now(), string(api.TaskStatusMerged), string(api.TaskStatusMerged), string(api.TaskStatusDiscarded), stageID, string(api.TaskStatusActive), string(api.TaskStatusActive)).Scan(&runID)
if errors.Is(err, sql.ErrNoRows) {
return nil
}
if err != nil {
return err
}
return settleTaskState(ctx, tx, emit, runID, actor)
}
func (s *Store) Participants(ctx context.Context, taskID string) ([]api.Participant, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT u.login, u.display_name, u.git_name, u.git_email
@@ -1583,7 +1610,8 @@ func settleTaskState(ctx context.Context, tx *sql.Tx, emit emitFunc, taskID, act
projectID := ""
before := ""
now := ""
err := tx.QueryRowContext(ctx, `SELECT project_id, state, `+taskStateExpr+` FROM tasks WHERE id = ?`, taskID).Scan(&projectID, &before, &now)
runID := ""
err := tx.QueryRowContext(ctx, `SELECT project_id, state, `+taskStateExpr+`, CASE WHEN workflow_stage != '' THEN parent_task_id ELSE '' END FROM tasks WHERE id = ?`, taskID).Scan(&projectID, &before, &now, &runID)
if errors.Is(err, sql.ErrNoRows) {
return nil
}
@@ -1597,17 +1625,26 @@ func settleTaskState(ctx context.Context, tx *sql.Tx, emit emitFunc, taskID, act
if err != nil {
return err
}
return emit(api.Event{
err = emit(api.Event{
Kind: "task.state",
ProjectId: &projectID,
TaskId: &taskID,
Actor: actor,
Payload: map[string]any{"from": before, "to": now},
})
if err != nil || runID == "" {
return err
}
return settleTaskState(ctx, tx, emit, runID, actor)
}
const taskStateExpr = `CASE
WHEN tasks.status != 'active' THEN tasks.status
WHEN tasks.workflow_name != '' THEN CASE
WHEN EXISTS (SELECT 1 FROM events e WHERE e.task_id = tasks.id AND e.kind = 'task.workflow_finished') THEN 'finished'
WHEN EXISTS (SELECT 1 FROM tasks c WHERE c.parent_task_id = tasks.id AND c.status = 'active' AND c.state = 'running') THEN 'running'
ELSE 'idle'
END
WHEN tasks.discard_state != '' THEN 'discarding'
WHEN EXISTS (SELECT 1 FROM inbox_items i WHERE i.task_id = tasks.id AND i.state = 'open') THEN 'asking'
WHEN tasks.merging_by != '' OR tasks.finalizing != '' OR tasks.parent_merge_by != '' THEN 'merging'
@@ -1683,9 +1720,11 @@ func (s *Store) ClearReviewRequests(ctx context.Context) error {
const projectColumns = `id, name, repo_path, default_branch, remote, created_by, created_at, review_workflow, audit_workflow`
const taskColumns = `id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, closed_at, merge_commit, todo_item, forked_from, forked_at, merge_queued_at, workflow_name, workflow_run, workflow_stage, parent_task_id, parent_merge_by, merge_after_review_by != '', state, merging_by != '' OR finalizing != '', discard_state != '' AND status = 'active', (SELECT coalesce(max(seq), 0) FROM events WHERE events.task_id = tasks.id AND kind = 'turn.ended' AND json_extract(payload, '$.summary') = 1), audit, ` + backgroundExpr
const taskColumns = `id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, closed_at, merge_commit, todo_item, forked_from, forked_at, merge_queued_at, workflow_name, workflow_stage, parent_task_id, parent_merge_by, merge_after_review_by != '', state, merging_by != '' OR finalizing != '', discard_state != '' AND status = 'active', (SELECT coalesce(max(seq), 0) FROM events WHERE events.task_id = tasks.id AND kind = 'turn.ended' AND json_extract(payload, '$.summary') = 1), audit, ` + backgroundExpr
const backgroundExpr = `audit != '' AND NOT EXISTS (SELECT 1 FROM events WHERE events.task_id = tasks.id AND kind IN ('inbox.created', 'task.ready', 'session.error', 'session.rate_limited'))`
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 c.id FROM tasks c WHERE c.id = tasks.id OR (c.parent_task_id = tasks.id AND tasks.workflow_name != ''))
AND kind IN ('inbox.created', 'task.ready', 'session.error', 'session.rate_limited'))`
const sessionColumns = `id, coalesce(task_id, ''), owner, vendor, model, vendor_session_id, state, role, member, created_by, created_at, forked_from, forked_at, subagents`
@@ -1715,13 +1754,12 @@ func scanTask(row scanner) (api.Task, error) {
forkedAt := 0
mergeQueuedAt := ""
workflowName := ""
workflowRun := ""
workflowStage := ""
parentTaskID := ""
parentMergeBy := ""
taskState := ""
audit := ""
err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.MergeTitle, &t.Branch, &t.WorktreePath, &t.BaseRef, &t.PortBase, &t.Color, &status, &t.CreatedBy, &createdAt, &closedAt, &mergeCommit, &todoItem, &forkedFrom, &forkedAt, &mergeQueuedAt, &workflowName, &workflowRun, &workflowStage, &parentTaskID, &parentMergeBy, &t.MergeAfterReview, &taskState, &t.Merging, &t.Discarding, &t.SummarySeq, &audit, &t.Background)
err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.MergeTitle, &t.Branch, &t.WorktreePath, &t.BaseRef, &t.PortBase, &t.Color, &status, &t.CreatedBy, &createdAt, &closedAt, &mergeCommit, &todoItem, &forkedFrom, &forkedAt, &mergeQueuedAt, &workflowName, &workflowStage, &parentTaskID, &parentMergeBy, &t.MergeAfterReview, &taskState, &t.Merging, &t.Discarding, &t.SummarySeq, &audit, &t.Background)
if err != nil {
return t, err
}
@@ -1735,7 +1773,8 @@ func scanTask(row scanner) (api.Task, error) {
t.MergingIntoParent = parentMergeBy != ""
if workflowName != "" {
t.Workflow = &workflowName
t.WorkflowRun = &workflowRun
}
if workflowStage != "" {
t.WorkflowStage = &workflowStage
}
if t.MergeTitle == "" {
internal/store/workflow.go
+11 -8
View File
@@ -160,7 +160,10 @@ func (s *Store) FreezeWorkflow(ctx context.Context, name string) (FrozenWorkflow
func (s *Store) TaskWorkflow(ctx context.Context, taskID string) (FrozenWorkflow, error) {
f := FrozenWorkflow{}
defID := sql.NullString{}
err := s.db.QueryRowContext(ctx, `SELECT workflow_name, workflow_definition_id, workflow_brief, workflow_title, audit FROM tasks WHERE id = ?`, taskID).Scan(&f.Name, &defID, &f.Brief, &f.Title, &f.Audit)
err := s.db.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)
if errors.Is(err, sql.ErrNoRows) {
return FrozenWorkflow{}, fmt.Errorf("%w: no task %s", ErrNotFound, taskID)
}
@@ -187,13 +190,13 @@ func (s *Store) LastAudit(ctx context.Context, projectID string) (string, bool,
}
func (s *Store) RunTasks(ctx context.Context, run string) ([]api.Task, error) {
return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE workflow_run = ? ORDER BY created_at`, run)
return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE parent_task_id = ? AND workflow_stage != '' ORDER BY created_at`, run)
}
func (s *Store) RunEvents(ctx context.Context, run string) ([]api.Event, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT `+eventColumns+` FROM events
WHERE task_id IN (SELECT id FROM tasks WHERE workflow_run = ?) AND kind IN ('task.workflow_stage', 'task.discarded', 'task.workflow_finished')
WHERE task_id IN (SELECT id FROM tasks WHERE parent_task_id = ?1 AND workflow_stage != '') AND kind IN ('task.workflow_stage', 'task.discarded')
ORDER BY seq
`, run)
if err != nil {
@@ -283,7 +286,7 @@ func clearStage(ctx context.Context, tx *sql.Tx, sessionID string) error {
return err
}
func (s *Store) FinishWorkflow(ctx context.Context, task api.Task, sessionID, workflowName, output string, kept []string) error {
func (s *Store) FinishWorkflow(ctx context.Context, run api.Task, sessionID, workflowName, output string, kept []string) error {
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
err := clearStage(ctx, tx, sessionID)
if err != nil {
@@ -291,16 +294,16 @@ func (s *Store) FinishWorkflow(ctx context.Context, task api.Task, sessionID, wo
}
return emit(api.Event{
Kind: "task.workflow_finished",
ProjectId: &task.ProjectId,
TaskId: &task.Id,
ProjectId: &run.ProjectId,
TaskId: &run.Id,
Actor: "byre",
Payload: map[string]any{"id": task.Id, "workflow": workflowName, "output": output, "kept": kept},
Payload: map[string]any{"id": run.Id, "workflow": workflowName, "output": output, "kept": kept},
})
})
}
func (s *Store) WorkflowFinished(ctx context.Context, run string) (bool, error) {
finished := false
err := s.db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM events e JOIN tasks t ON t.id = e.task_id WHERE t.workflow_run = ? AND e.kind = 'task.workflow_finished')`, run).Scan(&finished)
err := s.db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM events WHERE task_id = ? AND kind = 'task.workflow_finished')`, run).Scan(&finished)
return finished, err
}