Let task agents create and manage subtasks
This commit is contained in:
+88
-5
@@ -421,6 +421,10 @@ ALTER TABLE tasks ADD COLUMN discarding_by TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE tasks ADD COLUMN discard_state TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE tasks ADD COLUMN discard_session TEXT NOT NULL DEFAULT '';
|
||||
UPDATE tasks SET discard_state = 'held' WHERE discarding_by != '' AND status = 'active';
|
||||
`}, {"add-subtasks", `
|
||||
ALTER TABLE tasks ADD COLUMN parent_task_id TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE tasks ADD COLUMN parent_merge_by TEXT NOT NULL DEFAULT '';
|
||||
CREATE INDEX tasks_parent ON tasks(parent_task_id) WHERE parent_task_id != '';
|
||||
`}}
|
||||
|
||||
const (
|
||||
@@ -722,7 +726,22 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
|
||||
if t.ForkedAt != nil {
|
||||
forkedAt = *t.ForkedAt
|
||||
}
|
||||
parentTaskID := ""
|
||||
if t.ParentTaskId != nil {
|
||||
parentTaskID = *t.ParentTaskId
|
||||
}
|
||||
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
if parentTaskID != "" {
|
||||
open := false
|
||||
err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM tasks WHERE id = ? AND status = ? AND discard_state = '')`,
|
||||
parentTaskID, string(api.TaskStatusActive)).Scan(&open)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !open {
|
||||
return ErrParentNotOpen
|
||||
}
|
||||
}
|
||||
err := tx.QueryRowContext(ctx, `
|
||||
SELECT min(candidate) FROM (
|
||||
SELECT ? AS candidate
|
||||
@@ -742,9 +761,9 @@ 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, workflow_brief, workflow_title, workflow_run, workflow_stage)
|
||||
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, workflowDefinition, workflowBrief, workflowTitle, workflowRun, workflowStage)
|
||||
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, workflow_brief, workflow_title, workflow_run, workflow_stage, parent_task_id)
|
||||
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, workflowDefinition, workflowBrief, workflowTitle, workflowRun, workflowStage, parentTaskID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -871,6 +890,60 @@ func (s *Store) queryTasks(ctx context.Context, query string, args ...any) ([]ap
|
||||
return tasks, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Store) ActiveSubtasks(ctx context.Context, parentID string) ([]api.Task, error) {
|
||||
return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE parent_task_id = ? AND status = ? ORDER BY created_at`, parentID, string(api.TaskStatusActive))
|
||||
}
|
||||
|
||||
func (s *Store) SetParentMerge(ctx context.Context, id, login string) error {
|
||||
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
projectID, parentID, status, parentMergeBy, discardState := "", "", "", "", ""
|
||||
err := tx.QueryRowContext(ctx, `SELECT project_id, parent_task_id, status, parent_merge_by, discard_state FROM tasks WHERE id = ?`, id).
|
||||
Scan(&projectID, &parentID, &status, &parentMergeBy, &discardState)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
switch {
|
||||
case status != string(api.TaskStatusActive) || parentID == "":
|
||||
return ErrNotFound
|
||||
case discardState != "":
|
||||
return ErrDiscardPending
|
||||
case parentMergeBy != "":
|
||||
return ErrParentMergeRequested
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `UPDATE tasks SET parent_merge_by = ? WHERE id = ?`, login, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return emit(api.Event{Kind: "task.parent_merge_started", ProjectId: &projectID, TaskId: &id, Actor: login, Payload: map[string]any{"id": id, "parent_task_id": parentID}})
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Store) ClearParentMerge(ctx context.Context, id string) error {
|
||||
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
projectID := ""
|
||||
err := tx.QueryRowContext(ctx, `UPDATE tasks SET parent_merge_by = '' WHERE id = ? AND parent_merge_by != '' RETURNING project_id`, id).Scan(&projectID)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return emit(api.Event{Kind: "task.parent_merge_stopped", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}})
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Store) ParentMergeBy(ctx context.Context, id string) (string, error) {
|
||||
login := ""
|
||||
err := s.db.QueryRowContext(ctx, `SELECT parent_merge_by FROM tasks WHERE id = ?`, id).Scan(&login)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return "", ErrNotFound
|
||||
}
|
||||
return login, err
|
||||
}
|
||||
|
||||
func (s *Store) QueueMerge(ctx context.Context, id, login string) error {
|
||||
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
projectID := ""
|
||||
@@ -977,6 +1050,10 @@ func (s *Store) MergeTask(ctx context.Context, id, commit, pushed, actor string)
|
||||
return s.closeTask(ctx, id, api.TaskStatusMerged, commit, actor, "task.merged", map[string]any{"id": id, "commit": commit, "pushed": pushed})
|
||||
}
|
||||
|
||||
func (s *Store) MergeTaskIntoParent(ctx context.Context, id, parentID, commit, actor string) (api.Task, error) {
|
||||
return s.closeTask(ctx, id, api.TaskStatusMerged, commit, actor, "task.merged", map[string]any{"id": id, "commit": commit, "parent_task_id": parentID})
|
||||
}
|
||||
|
||||
func (s *Store) DiscardTask(ctx context.Context, id, actor string) (api.Task, error) {
|
||||
return s.closeTask(ctx, id, api.TaskStatusDiscarded, "", actor, "task.discarded", map[string]any{"id": id})
|
||||
}
|
||||
@@ -1436,7 +1513,7 @@ func (s *Store) write(ctx context.Context, fn func(tx *sql.Tx, emit emitFunc) er
|
||||
|
||||
const projectColumns = `id, name, repo_path, default_branch, remote, created_by, created_at`
|
||||
|
||||
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, EXISTS (SELECT 1 FROM reviews WHERE reviews.task_id = tasks.id AND reviews.state = 'running'), merging_by != '', discard_state != '' AND status = 'active'`
|
||||
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, EXISTS (SELECT 1 FROM reviews WHERE reviews.task_id = tasks.id AND reviews.state = 'running'), merging_by != '', discard_state != '' AND status = 'active'`
|
||||
|
||||
const sessionColumns = `id, coalesce(task_id, ''), owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at`
|
||||
|
||||
@@ -1468,10 +1545,16 @@ func scanTask(row scanner) (api.Task, error) {
|
||||
workflowName := ""
|
||||
workflowRun := ""
|
||||
workflowStage := ""
|
||||
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, &t.Reviewing, &t.Merging, &t.Discarding)
|
||||
parentTaskID := ""
|
||||
parentMergeBy := ""
|
||||
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.Reviewing, &t.Merging, &t.Discarding)
|
||||
if err != nil {
|
||||
return t, err
|
||||
}
|
||||
if parentTaskID != "" {
|
||||
t.ParentTaskId = &parentTaskID
|
||||
}
|
||||
t.MergingIntoParent = parentMergeBy != ""
|
||||
if workflowName != "" {
|
||||
t.Workflow = &workflowName
|
||||
t.WorkflowRun = &workflowRun
|
||||
|
||||
Reference in New Issue
Block a user