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

Refuse turns, discard and subtasks while a merge is finalizing

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-16 12:46:28 -07:00
parent 0e52253d07
commit ecfae97d97
11 changed files with 287 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/store/discard.go
+13 -6
View File
@@ -23,7 +23,7 @@ var (
ErrMerging = errors.New("the task is merging")
ErrMergingIntoParent = errors.New("the task is merging into its parent task")
ErrHasSubtasks = errors.New("the task has active subtasks")
ErrParentNotOpen = errors.New("the parent task is closed or being discarded")
ErrParentNotOpen = errors.New("the parent task is closed, being discarded or finishing its merge")
ErrParentMergeRequested = errors.New("the subtask is already merging into its parent")
)
@@ -48,12 +48,13 @@ func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) (Disca
projectID := ""
status := ""
mergingBy := ""
finalizing := ""
parentMergeBy := ""
subtasks := 0
err := tx.QueryRowContext(ctx, `
SELECT project_id, status, merging_by, parent_merge_by, discard_state,
SELECT project_id, status, merging_by, finalizing, parent_merge_by, discard_state,
(SELECT count(*) FROM tasks AS sub WHERE sub.parent_task_id = tasks.id AND sub.status = ?)
FROM tasks WHERE id = ?`, string(api.TaskStatusActive), taskID).Scan(&projectID, &status, &mergingBy, &parentMergeBy, &state, &subtasks)
FROM tasks WHERE id = ?`, string(api.TaskStatusActive), taskID).Scan(&projectID, &status, &mergingBy, &finalizing, &parentMergeBy, &state, &subtasks)
if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound
}
@@ -67,7 +68,7 @@ func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) (Disca
return nil
case state != DiscardNone:
return ErrDiscardPending
case mergingBy != "":
case mergingBy != "" || finalizing != "":
return ErrMerging
case parentMergeBy != "":
return ErrMergingIntoParent
@@ -114,8 +115,9 @@ func (s *Store) StartDiscardTurn(ctx context.Context, taskID, sessionID, actor s
})
}
func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload map[string]any, turn *api.Event, reviewRequest bool) (DiscardState, error) {
func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload map[string]any, turn *api.Event, reviewRequest, finalizeMerge bool) (DiscardState, string, error) {
next := DiscardNone
finalizing := ""
err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
taskID := ""
projectID := ""
@@ -167,10 +169,15 @@ func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload ma
case state == DiscardRunning && discardSession == sessionID:
next = DiscardClosing
_, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = ? WHERE id = ?`, string(DiscardClosing), taskID)
case state == DiscardNone && finalizeMerge:
err = tx.QueryRowContext(ctx, `UPDATE tasks SET finalizing = merging_by, merging_by = '' WHERE id = ? AND merging_by != '' RETURNING finalizing`, taskID).Scan(&finalizing)
if errors.Is(err, sql.ErrNoRows) {
err = nil
}
}
return err
})
return next, err
return next, finalizing, err
}
func (s *Store) CloseDiscard(ctx context.Context, taskID string) error {
internal/store/store.go
+37 -11
View File
@@ -446,6 +446,8 @@ CREATE TABLE summary_reads (
seq INTEGER NOT NULL,
PRIMARY KEY (login, task_id)
);
`}, {"add-task-finalizing", `
ALTER TABLE tasks ADD COLUMN finalizing TEXT NOT NULL DEFAULT '';
`}, {"add-task-state", `
ALTER TABLE tasks ADD COLUMN state TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN review_requests INTEGER NOT NULL DEFAULT 0;
@@ -758,7 +760,7 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
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 = '')`,
err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM tasks WHERE id = ? AND status = ? AND discard_state = '' AND finalizing = '')`,
parentTaskID, string(api.TaskStatusActive)).Scan(&open)
if err != nil {
return err
@@ -811,7 +813,7 @@ func (s *Store) GetTask(ctx context.Context, id string) (api.Task, error) {
func (s *Store) SetMergeTitle(ctx context.Context, id, title, actor string) error {
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
projectID := ""
err := tx.QueryRowContext(ctx, `UPDATE tasks SET merge_title = ? WHERE id = ? AND status = ? AND merging_by = '' AND merge_queued_by = '' RETURNING project_id`,
err := tx.QueryRowContext(ctx, `UPDATE tasks SET merge_title = ? WHERE id = ? AND status = ? AND merging_by = '' AND finalizing = '' AND merge_queued_by = '' RETURNING project_id`,
title, id, string(api.TaskStatusActive)).Scan(&projectID)
if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound
@@ -826,7 +828,7 @@ func (s *Store) SetMergeTitle(ctx context.Context, id, title, actor string) erro
func (s *Store) SetMergingBy(ctx context.Context, id, login string) error {
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
projectID := ""
err := tx.QueryRowContext(ctx, `UPDATE tasks SET merging_by = ? WHERE id = ? AND discard_state = '' RETURNING project_id`, login, id).Scan(&projectID)
err := tx.QueryRowContext(ctx, `UPDATE tasks SET merging_by = ? WHERE id = ? AND discard_state = '' AND finalizing = '' RETURNING project_id`, login, id).Scan(&projectID)
if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound
}
@@ -869,6 +871,24 @@ func (s *Store) TakeMergingBy(ctx context.Context, id string) (string, error) {
return login, err
}
func (s *Store) ClearFinalizing(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 finalizing = '' WHERE id = ? AND finalizing != '' 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.merge_ended", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}})
})
}
func (s *Store) FinalizingTasks(ctx context.Context) ([]api.Task, error) {
return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE status = ? AND finalizing != ''`, string(api.TaskStatusActive))
}
func (s *Store) MergingTasks(ctx context.Context) ([]api.Task, error) {
return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE status = ? AND merging_by != ''`, string(api.TaskStatusActive))
}
@@ -921,9 +941,9 @@ func (s *Store) ActiveSubtasks(ctx context.Context, parentID string) ([]api.Task
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)
projectID, parentID, status, parentMergeBy, discardState, finalizing := "", "", "", "", "", ""
err := tx.QueryRowContext(ctx, `SELECT project_id, parent_task_id, status, parent_merge_by, discard_state, finalizing FROM tasks WHERE id = ?`, id).
Scan(&projectID, &parentID, &status, &parentMergeBy, &discardState, &finalizing)
if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound
}
@@ -935,6 +955,8 @@ func (s *Store) SetParentMerge(ctx context.Context, id, login string) error {
return ErrNotFound
case discardState != "":
return ErrDiscardPending
case finalizing != "":
return ErrMerging
case parentMergeBy != "":
return ErrParentMergeRequested
}
@@ -1085,7 +1107,7 @@ func (s *Store) DiscardTask(ctx context.Context, id, actor string) (api.Task, er
func (s *Store) closeTask(ctx context.Context, id string, status api.TaskStatus, commit, actor, kind string, payload map[string]any) (api.Task, error) {
err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
res, err := tx.ExecContext(ctx, `UPDATE tasks SET status = ?, closed_at = ?, merge_commit = ? WHERE id = ? AND status = ?`,
res, err := tx.ExecContext(ctx, `UPDATE tasks SET status = ?, closed_at = ?, merge_commit = ?, finalizing = '' WHERE id = ? AND status = ?`,
string(status), now(), commit, id, string(api.TaskStatusActive))
if err != nil {
return err
@@ -1373,9 +1395,10 @@ func (s *Store) transitionSession(ctx context.Context, id string, from *api.Agen
current := ""
role := ""
discard := ""
finalizing := ""
err := tx.QueryRowContext(ctx, `
SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state, s.role, coalesce(t.discard_state, '') FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
`, id).Scan(&taskID, &projectID, &owner, &current, &role, &discard)
SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state, s.role, coalesce(t.discard_state, ''), coalesce(t.finalizing, '') FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
`, id).Scan(&taskID, &projectID, &owner, &current, &role, &discard, &finalizing)
if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound
}
@@ -1389,6 +1412,9 @@ func (s *Store) transitionSession(ctx context.Context, id string, from *api.Agen
if state == api.AgentSessionStateRunning && role == string(api.Implementer) && discard != "" {
return fmt.Errorf("%w: task %s is being discarded", ErrStateChanged, taskID)
}
if state == api.AgentSessionStateRunning && role == string(api.Implementer) && finalizing != "" {
return fmt.Errorf("%w: task %s is finishing its merge", ErrStateChanged, taskID)
}
_, err = tx.ExecContext(ctx, `UPDATE agent_sessions SET state = ? WHERE id = ?`, string(state), id)
if err != nil {
@@ -1586,7 +1612,7 @@ func settleTaskState(ctx context.Context, tx *sql.Tx, emit emitFunc, taskID, act
const taskStateExpr = `CASE
WHEN tasks.status != 'active' THEN tasks.status
WHEN tasks.discard_state != '' THEN 'discarding'
WHEN tasks.merging_by != '' THEN 'merging'
WHEN tasks.merging_by != '' OR tasks.finalizing != '' THEN 'merging'
ELSE coalesce((
SELECT CASE
WHEN s.state != 'idle' THEN s.state
@@ -1643,7 +1669,7 @@ func (s *Store) ClearReviewRequests(ctx context.Context) error {
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, parent_task_id, parent_merge_by, state, merging_by != '', 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)`
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, 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)`
const sessionColumns = `id, coalesce(task_id, ''), owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at`