From 27542c9e9fbe0518f74411bb9da4d46f17f95676 Mon Sep 17 00:00:00 2001 From: Ian Gulliver Date: Fri, 23 Oct 2026 11:29:46 -0700 Subject: [PATCH] Fold task merge progress columns into one merge state --- internal/agents/manager.go | 8 +- internal/agents/mergein.go | 14 +-- internal/agents/snapshots.go | 4 +- internal/store/coordinate.go | 2 +- internal/store/discard.go | 13 ++- internal/store/landing.go | 2 +- internal/store/merges.go | 38 ++++---- .../108-fold-task-merge-progress.sql | 20 ++++ internal/store/store.go | 94 ++++++++----------- 9 files changed, 95 insertions(+), 100 deletions(-) create mode 100644 internal/store/migrations/108-fold-task-merge-progress.sql diff --git a/internal/agents/manager.go b/internal/agents/manager.go index fc028a0..f065d29 100644 --- a/internal/agents/manager.go +++ b/internal/agents/manager.go @@ -1270,15 +1270,15 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error { if err != nil { return err } - mergingBy := "" + merging := false if finished && !handoff { - mergingBy, err = m.store.MergingBy(ctx, l.taskID) + merging, err = m.store.MergeTurnPending(ctx, l.taskID) if err != nil { return err } } snapshot := discard.State == store.DiscardNone - review := snapshot && mergingBy == "" + review := snapshot && !merging switch { case handoff: payload["handoff"] = true @@ -1308,7 +1308,7 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error { stageFinished := false if l.stage != "" { - if finished && !handoff && mergingBy == "" { + if finished && !handoff && !merging { stageFinished, err = m.store.FinishStage(ctx, l.id) } if err == nil && !stageFinished { diff --git a/internal/agents/mergein.go b/internal/agents/mergein.go index 83c7930..4b5df62 100644 --- a/internal/agents/mergein.go +++ b/internal/agents/mergein.go @@ -120,7 +120,7 @@ func (m *Manager) startClaimedMerge(ctx context.Context, c store.ClaimedMerge) { return } slog.Error("start claimed merge", "task", c.TaskID, "err", err) - _, terr := m.store.TakeMergingBy(ctx, c.TaskID) + _, terr := m.store.DropMerge(ctx, c.TaskID) if terr != nil { slog.Error("drop claimed merge", "task", c.TaskID, "err", terr) } @@ -505,18 +505,12 @@ func (m *Manager) stopMerge(ctx context.Context, l *live, raw json.RawMessage) ( if !m.leadsTask(ctx, l.taskID, l.id) { return "", fmt.Errorf("%w: only the team's first agent runs the merge, so only it can stop it", ErrConflict) } - mergingBy, err := m.store.TakeMergingBy(ctx, l.taskID) + dropped, err := m.store.DropMerge(ctx, l.taskID) if err != nil { return "", err } - if mergingBy == "" { - queuedBy, err := m.store.TakeQueuedMerge(ctx, l.taskID) - if err != nil { - return "", err - } - if queuedBy == "" { - return "No merge is pending for this task; just end your turn.", nil - } + if !dropped { + return "No merge is pending for this task; just end your turn.", nil } err = m.emit(ctx, l, agentActor, "task.merge_stopped", map[string]any{"id": l.taskID, "reason": reason}) diff --git a/internal/agents/snapshots.go b/internal/agents/snapshots.go index 6a9e520..8001be5 100644 --- a/internal/agents/snapshots.go +++ b/internal/agents/snapshots.go @@ -187,11 +187,11 @@ func (m *Manager) RequestSnapshot(ctx context.Context, taskID string) error { if discard.State != store.DiscardNone { return fmt.Errorf("%w: task %s is being discarded", ErrConflict, task.Id) } - mergingBy, err := m.store.MergingBy(ctx, task.Id) + merging, err := m.store.MergeTurnPending(ctx, task.Id) if err != nil { return err } - if mergingBy != "" { + if merging { return fmt.Errorf("%w: task %s is merging", ErrConflict, task.Id) } sessions, err := m.store.ListSessions(ctx, task.Id) diff --git a/internal/store/coordinate.go b/internal/store/coordinate.go index 0ce67c6..779ca16 100644 --- a/internal/store/coordinate.go +++ b/internal/store/coordinate.go @@ -118,7 +118,7 @@ func selfCloses(ctx context.Context, tx *sql.Tx) ([]SelfClose, error) { 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.merge_state = '' 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`) diff --git a/internal/store/discard.go b/internal/store/discard.go index fb44fd8..e3f3869 100644 --- a/internal/store/discard.go +++ b/internal/store/discard.go @@ -47,14 +47,13 @@ func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) error state := DiscardNone projectID := "" status := "" - mergingBy := "" - finalizing := "" + mergeState := "" parentMergeBy := "" subtasks := 0 err := tx.QueryRowContext(ctx, ` - SELECT project_id, status, merging_by, finalizing, parent_merge_by, discard_state, + SELECT project_id, status, merge_state, 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, &finalizing, &parentMergeBy, &state, &subtasks) + FROM tasks WHERE id = ?`, string(api.TaskStatusActive), taskID).Scan(&projectID, &status, &mergeState, &parentMergeBy, &state, &subtasks) if errors.Is(err, sql.ErrNoRows) { return ErrNotFound } @@ -68,14 +67,14 @@ func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) error return nil case state != DiscardNone: return ErrDiscardPending - case mergingBy != "" || finalizing != "": + case mergeState != "" && mergeState != "held": return ErrMerging case parentMergeBy != "": return ErrMergingIntoParent case subtasks > 0: return ErrHasSubtasks } - _, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = ?, discarding_by = ?, discard_session = '', merge_queued_by = '', merge_queued_at = '', merge_after_review_by = '', merge_after_review_id = '', merge_after_review_target = '' WHERE id = ?`, + _, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = ?, discarding_by = ?, discard_session = '', merge_state = '', merge_by = '', merge_queued_at = '', merge_after_review_by = '', merge_after_review_id = '', merge_after_review_target = '' WHERE id = ?`, string(DiscardHeld), actor, taskID) if err != nil { return err @@ -176,7 +175,7 @@ func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload ma case state == DiscardRunning && discardSession == sessionID: end.Discard = DiscardRunning case state == DiscardNone && finalizeMerge && lead: - err = tx.QueryRowContext(ctx, `UPDATE tasks SET finalizing = merging_by, merging_by = '' WHERE id = ? AND merging_by != '' AND merge_after_turn = 0 RETURNING finalizing`, taskID).Scan(&end.FinalizeBy) + err = tx.QueryRowContext(ctx, `UPDATE tasks SET merge_state = 'finalizing' WHERE id = ? AND merge_state = 'turn' RETURNING merge_by`, taskID).Scan(&end.FinalizeBy) if errors.Is(err, sql.ErrNoRows) { err = nil } diff --git a/internal/store/landing.go b/internal/store/landing.go index 4aed096..d66e9d7 100644 --- a/internal/store/landing.go +++ b/internal/store/landing.go @@ -74,7 +74,7 @@ func queuedDeliveries(ctx context.Context, tx *sql.Tx) ([]QueuedDelivery, error) WHERE i.agent_session_id = s.id AND r.delivered_at IS NULL ORDER BY r.created_at LIMIT 1), '') FROM agent_sessions s JOIN tasks t ON t.id = s.task_id WHERE s.role = 'implementer' AND s.state = 'idle' - AND t.status = 'active' AND t.discard_state = '' AND t.merging_by = '' AND t.finalizing = '' AND t.landing_session = '' + AND t.status = 'active' AND t.discard_state = '' AND t.merge_state NOT IN ('claimed', 'turn', 'finalizing') AND t.landing_session = '' AND (EXISTS (SELECT 1 FROM task_inputs q WHERE q.task_id = s.task_id AND q.member = s.member) OR EXISTS (SELECT 1 FROM inbox_responses r JOIN inbox_items i ON i.id = r.item_id WHERE i.agent_session_id = s.id AND r.delivered_at IS NULL)) AND s.id = (SELECT l.id FROM agent_sessions l WHERE l.task_id = s.task_id AND l.member = s.member AND l.role = 'implementer' AND l.state != 'completed' ORDER BY l.created_at DESC LIMIT 1)`) diff --git a/internal/store/merges.go b/internal/store/merges.go index 94f9c00..b94186f 100644 --- a/internal/store/merges.go +++ b/internal/store/merges.go @@ -8,19 +8,19 @@ import ( "g.fc.run/ai/byre/internal/api" ) -const mergeSlotHolder = `o.project_id = tasks.project_id AND o.status = 'active' AND (o.merging_by != '' OR o.finalizing != '' OR o.pushing_to_main = 1)` +const mergeSlotHolder = `o.project_id = tasks.project_id AND o.status = 'active' AND (o.merge_state IN ('claimed', 'turn', 'finalizing') OR o.pushing_to_main = 1)` const implementerRunning = `EXISTS (SELECT 1 FROM agent_sessions s WHERE s.task_id = tasks.id AND s.role = 'implementer' AND s.state = 'running')` func (s *Store) RequestMerge(ctx context.Context, id, login string) (bool, error) { held := false err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - projectID, status, discardState, queuedBy, mergingBy, finalizing := "", "", "", "", "", "" + projectID, status, discardState, mergeState := "", "", "", "" err := tx.QueryRowContext(ctx, ` - SELECT project_id, status, discard_state, merge_queued_by, merging_by, finalizing, + SELECT project_id, status, discard_state, merge_state, EXISTS (SELECT 1 FROM tasks o WHERE `+mergeSlotHolder+`) - OR EXISTS (SELECT 1 FROM tasks o WHERE o.project_id = tasks.project_id AND o.status = 'active' AND o.merge_queued_by != '') - FROM tasks WHERE id = ?`, id).Scan(&projectID, &status, &discardState, &queuedBy, &mergingBy, &finalizing, &held) + OR EXISTS (SELECT 1 FROM tasks o WHERE o.project_id = tasks.project_id AND o.status = 'active' AND o.merge_state = 'held') + FROM tasks WHERE id = ?`, id).Scan(&projectID, &status, &discardState, &mergeState, &held) if errors.Is(err, sql.ErrNoRows) { return ErrNotFound } @@ -28,19 +28,19 @@ func (s *Store) RequestMerge(ctx context.Context, id, login string) (bool, error return err } switch { - case status != string(api.TaskStatusActive) || discardState != "" || queuedBy != "": + case status != string(api.TaskStatusActive) || discardState != "" || mergeState == "held": return ErrNotFound - case mergingBy != "" || finalizing != "": + case mergeState != "": return ErrMerging } if held { - _, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_queued_by = ?, merge_queued_at = ? WHERE id = ?`, login, now(), id) + _, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_state = 'held', merge_by = ?, merge_queued_at = ? WHERE id = ?`, login, now(), id) if err != nil { return err } return emit(api.Event{Kind: "task.merge_queued", ProjectId: &projectID, TaskId: &id, Actor: login, Payload: map[string]any{"id": id}}) } - _, err = tx.ExecContext(ctx, `UPDATE tasks SET merging_by = ?, merge_after_turn = 1 WHERE id = ?`, login, id) + _, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_state = 'claimed', merge_by = ? WHERE id = ?`, login, id) if err != nil { return err } @@ -88,14 +88,14 @@ func queryIDRows(ctx context.Context, tx *sql.Tx, query string) ([]idRow, error) } func abandonMerges(ctx context.Context, tx *sql.Tx, emit emitFunc) ([]string, error) { - found, err := queryIDRows(ctx, tx, `SELECT id, project_id, merging_by FROM tasks - WHERE status = 'active' AND merging_by != '' AND merge_after_turn = 0 AND NOT `+implementerRunning) + found, err := queryIDRows(ctx, tx, `SELECT id, project_id, merge_by FROM tasks + WHERE status = 'active' AND merge_state = 'turn' AND NOT `+implementerRunning) if err != nil { return nil, err } ids := []string{} for _, r := range found { - _, err = tx.ExecContext(ctx, `UPDATE tasks SET merging_by = '' WHERE id = ?`, r.id) + _, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_state = '', merge_by = '' WHERE id = ?`, r.id) if err != nil { return nil, err } @@ -109,16 +109,16 @@ func abandonMerges(ctx context.Context, tx *sql.Tx, emit emitFunc) ([]string, er } func claimHeldMerges(ctx context.Context, tx *sql.Tx, emit emitFunc) error { - found, err := queryIDRows(ctx, tx, `SELECT id, project_id, merge_queued_by FROM tasks - WHERE status = 'active' AND merge_queued_by != '' AND discard_state = '' + found, err := queryIDRows(ctx, tx, `SELECT id, project_id, merge_by FROM tasks + WHERE status = 'active' AND merge_state = 'held' AND discard_state = '' AND NOT EXISTS (SELECT 1 FROM tasks o WHERE `+mergeSlotHolder+`) - AND NOT EXISTS (SELECT 1 FROM tasks o WHERE o.project_id = tasks.project_id AND o.status = 'active' AND o.merge_queued_by != '' + AND NOT EXISTS (SELECT 1 FROM tasks o WHERE o.project_id = tasks.project_id AND o.status = 'active' AND o.merge_state = 'held' AND (o.merge_queued_at < tasks.merge_queued_at OR (o.merge_queued_at = tasks.merge_queued_at AND o.id < tasks.id)))`) if err != nil { return err } for _, r := range found { - _, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_queued_by = '', merge_queued_at = '', merging_by = ?, merge_after_turn = 1 WHERE id = ?`, r.by, r.id) + _, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_state = 'claimed', merge_queued_at = '' WHERE id = ?`, r.id) if err != nil { return err } @@ -135,8 +135,8 @@ func claimHeldMerges(ctx context.Context, tx *sql.Tx, emit emitFunc) error { } func startableMerges(ctx context.Context, tx *sql.Tx) ([]ClaimedMerge, error) { - found, err := queryIDRows(ctx, tx, `SELECT id, project_id, merging_by FROM tasks - WHERE status = 'active' AND merge_after_turn = 1 AND NOT `+implementerRunning) + found, err := queryIDRows(ctx, tx, `SELECT id, project_id, merge_by FROM tasks + WHERE status = 'active' AND merge_state = 'claimed' AND NOT `+implementerRunning) if err != nil { return nil, err } @@ -148,7 +148,7 @@ func startableMerges(ctx context.Context, tx *sql.Tx) ([]ClaimedMerge, error) { } func finalizingMerges(ctx context.Context, tx *sql.Tx) ([]FinalizingMerge, error) { - rows, err := tx.QueryContext(ctx, `SELECT `+taskColumns+`, finalizing, finalizing_commit, finalizing_notes, finalizing_unchanged FROM tasks WHERE status = 'active' AND finalizing != ''`) + rows, err := tx.QueryContext(ctx, `SELECT `+taskColumns+`, merge_by, finalizing_commit, finalizing_notes, finalizing_unchanged FROM tasks WHERE status = 'active' AND merge_state = 'finalizing'`) if err != nil { return nil, err } diff --git a/internal/store/migrations/108-fold-task-merge-progress.sql b/internal/store/migrations/108-fold-task-merge-progress.sql new file mode 100644 index 0000000..4ffec82 --- /dev/null +++ b/internal/store/migrations/108-fold-task-merge-progress.sql @@ -0,0 +1,20 @@ +ALTER TABLE tasks ADD COLUMN merge_state TEXT NOT NULL DEFAULT ''; +ALTER TABLE tasks ADD COLUMN merge_by TEXT NOT NULL DEFAULT ''; +UPDATE tasks SET + merge_state = CASE + WHEN finalizing != '' THEN 'finalizing' + WHEN merging_by != '' AND merge_after_turn = 1 THEN 'claimed' + WHEN merging_by != '' THEN 'turn' + WHEN merge_queued_by != '' THEN 'held' + ELSE '' + END, + merge_by = CASE + WHEN finalizing != '' THEN finalizing + WHEN merging_by != '' THEN merging_by + ELSE merge_queued_by + END; +UPDATE tasks SET merge_queued_at = '' WHERE merge_state != 'held'; +ALTER TABLE tasks DROP COLUMN merge_queued_by; +ALTER TABLE tasks DROP COLUMN merging_by; +ALTER TABLE tasks DROP COLUMN merge_after_turn; +ALTER TABLE tasks DROP COLUMN finalizing; diff --git a/internal/store/store.go b/internal/store/store.go index f494358..cdeb319 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -379,7 +379,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 = '' AND finalizing = '')`, + err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM tasks WHERE id = ? AND status = ? AND discard_state = '' AND merge_state != 'finalizing')`, parentTaskID, string(api.TaskStatusActive)).Scan(&open) if err != nil { return err @@ -430,7 +430,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 := updatedProject(ctx, tx, `UPDATE tasks SET merge_title = ? WHERE id = ? AND status = ? AND merging_by = '' AND finalizing = '' AND merge_queued_by = '' RETURNING project_id`, + projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET merge_title = ? WHERE id = ? AND status = ? AND merge_state = '' RETURNING project_id`, title, id, string(api.TaskStatusActive)) if err != nil { return err @@ -441,7 +441,7 @@ func (s *Store) SetMergeTitle(ctx context.Context, id, title, actor string) erro func (s *Store) SetTitles(ctx context.Context, id, title, mergeTitle, actor string) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET title = ?, merge_title = ? WHERE id = ? AND status = ? AND merging_by = '' AND finalizing = '' AND merge_queued_by = '' RETURNING project_id`, + projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET title = ?, merge_title = ? WHERE id = ? AND status = ? AND merge_state = '' RETURNING project_id`, title, mergeTitle, id, string(api.TaskStatusActive)) if err != nil { return err @@ -454,41 +454,46 @@ func (s *Store) SetTitles(ctx context.Context, id, title, mergeTitle, actor stri }) } -func (s *Store) MergingBy(ctx context.Context, id string) (string, error) { - login := "" - err := s.db.QueryRowContext(ctx, `SELECT merging_by FROM tasks WHERE id = ?`, id).Scan(&login) +func (s *Store) MergeTurnPending(ctx context.Context, id string) (bool, error) { + pending := false + err := s.db.QueryRowContext(ctx, `SELECT merge_state IN ('claimed', 'turn') FROM tasks WHERE id = ?`, id).Scan(&pending) if errors.Is(err, sql.ErrNoRows) { - return "", ErrNotFound + return false, ErrNotFound } - return login, err + return pending, err } -func (s *Store) TakeMergingBy(ctx context.Context, id string) (string, error) { - login := "" +func (s *Store) DropMerge(ctx context.Context, id string) (bool, error) { + dropped := false err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - projectID := "" - err := tx.QueryRowContext(ctx, `SELECT project_id, merging_by FROM tasks WHERE id = ?`, id).Scan(&projectID, &login) + projectID, state := "", "" + err := tx.QueryRowContext(ctx, `SELECT project_id, merge_state FROM tasks WHERE id = ?`, id).Scan(&projectID, &state) if errors.Is(err, sql.ErrNoRows) { return ErrNotFound } if err != nil { return err } - if login == "" { + if state == "" || state == "finalizing" { return nil } - _, err = tx.ExecContext(ctx, `UPDATE tasks SET merging_by = '', merge_after_turn = 0 WHERE id = ?`, id) + kind := "task.merge_ended" + if state == "held" { + kind = "task.merge_unqueued" + } + _, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_state = '', merge_by = '', merge_queued_at = '' WHERE id = ?`, id) 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}}) + dropped = true + return emit(api.Event{Kind: kind, ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}}) }) - return login, err + return dropped, err } func (s *Store) ClearFinalizing(ctx context.Context, id string) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET finalizing = '', finalizing_commit = '', finalizing_notes = '', finalizing_unchanged = 0 WHERE id = ? AND finalizing != '' RETURNING project_id`, id) + projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET merge_state = '', merge_by = '', finalizing_commit = '', finalizing_notes = '', finalizing_unchanged = 0 WHERE id = ? AND merge_state = 'finalizing' RETURNING project_id`, id) if errors.Is(err, ErrNotFound) { return nil } @@ -509,7 +514,7 @@ func (s *Store) SetFinalizingUnchanged(ctx context.Context, id string) error { func (s *Store) setFinalizing(ctx context.Context, id, commit, notes, pushed string, unchanged bool) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET finalizing_commit = ?, finalizing_notes = ?, finalizing_unchanged = ? WHERE id = ? AND status = ? AND finalizing != '' RETURNING project_id`, + projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET finalizing_commit = ?, finalizing_notes = ?, finalizing_unchanged = ? WHERE id = ? AND status = ? AND merge_state = 'finalizing' RETURNING project_id`, commit, notes, unchanged, id, string(api.TaskStatusActive)) if err != nil || commit == "" { return err @@ -559,9 +564,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, 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) + projectID, parentID, status, parentMergeBy, discardState, mergeState := "", "", "", "", "", "" + err := tx.QueryRowContext(ctx, `SELECT project_id, parent_task_id, status, parent_merge_by, discard_state, merge_state FROM tasks WHERE id = ?`, id). + Scan(&projectID, &parentID, &status, &parentMergeBy, &discardState, &mergeState) if errors.Is(err, sql.ErrNoRows) { return ErrNotFound } @@ -573,7 +578,7 @@ func (s *Store) SetParentMerge(ctx context.Context, id, login string) error { return ErrNotFound case discardState != "": return ErrDiscardPending - case finalizing != "": + case mergeState == "finalizing": return ErrMerging case parentMergeBy != "": return ErrParentMergeRequested @@ -608,33 +613,10 @@ func (s *Store) ParentMergeBy(ctx context.Context, id string) (string, error) { return login, err } -func (s *Store) TakeQueuedMerge(ctx context.Context, id string) (string, error) { - login := "" - err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - projectID := "" - err := tx.QueryRowContext(ctx, `SELECT project_id, merge_queued_by FROM tasks WHERE id = ?`, id).Scan(&projectID, &login) - if errors.Is(err, sql.ErrNoRows) { - return ErrNotFound - } - if err != nil { - return err - } - if login == "" { - return nil - } - _, err = tx.ExecContext(ctx, `UPDATE tasks SET merge_queued_by = '', merge_queued_at = '' WHERE id = ?`, id) - if err != nil { - return err - } - return emit(api.Event{Kind: "task.merge_unqueued", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}}) - }) - return login, err -} - func (s *Store) ReserveMergeTurn(ctx context.Context, id, sessionID string) (bool, error) { reserved := false err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - res, err := tx.ExecContext(ctx, `UPDATE tasks SET merge_after_turn = 0 WHERE id = ? AND merge_after_turn = 1 + res, err := tx.ExecContext(ctx, `UPDATE tasks SET merge_state = 'turn' WHERE id = ? AND merge_state = 'claimed' AND NOT EXISTS (SELECT 1 FROM agent_sessions s WHERE s.task_id = tasks.id AND s.role = ? AND s.state = ?)`, id, string(api.Implementer), string(api.AgentSessionStateRunning)) if err != nil { @@ -667,7 +649,7 @@ func (s *Store) SetMergeAfterReview(ctx context.Context, id string, mar MergeAft snapshotN := 0 err := tx.QueryRowContext(ctx, ` UPDATE tasks SET merge_after_review_by = ?, merge_after_review_id = ?, merge_after_review_target = ? - WHERE id = ? AND status = ? AND discard_state = '' AND merging_by = '' AND finalizing = '' AND merge_queued_by = '' AND parent_merge_by = '' + WHERE id = ? AND status = ? AND discard_state = '' AND merge_state = '' AND parent_merge_by = '' AND EXISTS (SELECT 1 FROM reviews WHERE reviews.id = ? AND reviews.task_id = tasks.id AND reviews.state = ?) RETURNING project_id, (SELECT snapshot_n FROM reviews WHERE reviews.id = ?)`, mar.By, mar.ReviewID, string(mar.Target), id, string(api.TaskStatusActive), mar.ReviewID, string(api.ReviewStateRunning), mar.ReviewID).Scan(&projectID, &snapshotN) @@ -843,7 +825,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 = ?, finalizing = '', finalizing_commit = '', finalizing_notes = '', finalizing_unchanged = 0, merge_after_review_by = '', merge_after_review_id = '', merge_after_review_target = '' WHERE id = ? AND status = ?`, + res, err := tx.ExecContext(ctx, `UPDATE tasks SET status = ?, closed_at = ?, merge_commit = ?, merge_state = '', merge_by = '', merge_queued_at = '', finalizing_commit = '', finalizing_notes = '', finalizing_unchanged = 0, merge_after_review_by = '', merge_after_review_id = '', merge_after_review_target = '' WHERE id = ? AND status = ?`, string(status), now(), commit, id, string(api.TaskStatusActive)) if err != nil { return err @@ -1324,12 +1306,12 @@ func transitionSessionTx(ctx context.Context, tx *sql.Tx, emit emitFunc, id stri current := "" role := "" discard := "" - finalizing := "" + mergeState := "" landing := "" err := tx.QueryRowContext(ctx, ` - SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state, s.role, coalesce(t.discard_state, ''), coalesce(t.finalizing, ''), coalesce(t.landing_session, '') + SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state, s.role, coalesce(t.discard_state, ''), coalesce(t.merge_state, ''), coalesce(t.landing_session, '') FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ? - `, id).Scan(&taskID, &projectID, &owner, ¤t, &role, &discard, &finalizing, &landing) + `, id).Scan(&taskID, &projectID, &owner, ¤t, &role, &discard, &mergeState, &landing) if errors.Is(err, sql.ErrNoRows) { return ErrNotFound } @@ -1343,7 +1325,7 @@ func transitionSessionTx(ctx context.Context, tx *sql.Tx, emit emitFunc, id stri 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 != "" { + if state == api.AgentSessionStateRunning && role == string(api.Implementer) && mergeState == "finalizing" { return fmt.Errorf("%w: task %s is finishing its merge", ErrStateChanged, taskID) } if state == api.AgentSessionStateRunning && role == string(api.Implementer) && landing != "" && landing != id { @@ -1647,9 +1629,9 @@ const taskStateExpr = `CASE 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.finalizing != '' AND tasks.finalizing_commit != '' THEN 'pushing' - WHEN tasks.merging_by != '' OR tasks.finalizing != '' OR tasks.parent_merge_by != '' THEN 'merging' - WHEN tasks.merge_queued_by != '' THEN 'merge_held' + WHEN tasks.merge_state = 'finalizing' AND tasks.finalizing_commit != '' THEN 'pushing' + WHEN tasks.merge_state IN ('claimed', 'turn', 'finalizing') OR tasks.parent_merge_by != '' THEN 'merging' + WHEN tasks.merge_state = 'held' THEN 'merge_held' WHEN tasks.merge_after_review_by != '' THEN 'merge_after_review' ELSE coalesce(( SELECT CASE @@ -1708,7 +1690,7 @@ 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, 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, ready, ` + backgroundExpr + `, ` + mergeFactsColumns +const taskColumns = `id, project_id, title, merge_title, branch, worktree_path, base_ref, 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, merge_state IN ('claimed', 'turn', '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, ready, ` + backgroundExpr + `, ` + mergeFactsColumns // 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 != ''))