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

Fold task merge progress columns into one merge state

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-23 11:29:46 -07:00
parent 395f76e59b
commit 27542c9e9f
9 changed files with 95 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/agents/manager.go
+4 -4
View File
@@ -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 {
internal/agents/mergein.go
+4 -10
View File
@@ -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})
internal/agents/snapshots.go
+2 -2
View File
@@ -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)
internal/store/coordinate.go
+1 -1
View File
@@ -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`)
internal/store/discard.go
+6 -7
View File
@@ -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
}
internal/store/landing.go
+1 -1
View File
@@ -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)`)
internal/store/merges.go
+19 -19
View File
@@ -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
}
internal/store/migrations/108-fold-task-merge-progress.sql
+20
View File
@@ -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;
internal/store/store.go
+38 -56
View File
@@ -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, &current, &role, &discard, &finalizing, &landing)
`, id).Scan(&taskID, &projectID, &owner, &current, &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 != ''))