Run discards on the state-driven coordinator

docs/operations.md

@@ -86,7 +86,7 @@ Vendor setup and protocol quirks are in [Claude Code](claude-protocol.md), [Code
 
 ## Coordination
 
-Merge coordination must decide from the database's current state, never from which event woke it. This lets startup use the same decisions as ordinary operation rather than maintaining separate recovery logic. Slow work must run outside the decision transaction, with its progress recorded so another pass cannot start it twice.
+Merge and discard coordination must decide from the database's current state, never from which event woke it. This lets startup use the same decisions as ordinary operation rather than maintaining separate recovery logic. Slow work must run outside the decision transaction, with its progress recorded so another pass cannot start it twice.
 
 ## git
 

internal/agents/compact.go

@@ -110,12 +110,7 @@ func (m *Manager) handoff(ctx context.Context, sessionID, handoff string) error
 	next.State = api.AgentSessionStateRunning
 	err = m.store.ReplaceSession(ctx, sess.Id, next, task.ProjectId, "byre")
 	if errors.Is(err, store.ErrStateChanged) {
-		err = transition(ctx, m.store, sess.Id, api.AgentSessionStateRunning, api.AgentSessionStateIdle, "byre")
-		if err != nil {
-			return err
-		}
-		m.startDiscard(task.Id)
-		return nil
+		return transition(ctx, m.store, sess.Id, api.AgentSessionStateRunning, api.AgentSessionStateIdle, "byre")
 	}
 	if err != nil {
 		return err

internal/agents/coordinator.go

@@ -15,43 +15,53 @@ func (m *Manager) poke() {
 }
 
 func (m *Manager) coordinate() {
-	finalizing := map[string]bool{}
+	started := map[string]bool{}
 	for {
 		select {
 		case <-m.watchStop:
 			return
 		case <-m.pokes:
-			m.coordinateMerges(context.Background(), finalizing)
+			m.coordinatePass(context.Background(), started)
 		}
 	}
 }
 
-func (m *Manager) coordinateMerges(ctx context.Context, finalizing map[string]bool) {
-	work, err := m.store.CoordinateMerges(ctx)
+func (m *Manager) coordinatePass(ctx context.Context, started map[string]bool) {
+	work, err := m.store.Coordinate(ctx)
 	if err != nil {
-		slog.Error("coordinate merges", "err", err)
+		slog.Error("coordinate", "err", err)
 		return
 	}
-	for _, id := range work.Abandoned {
+	for _, id := range work.AbandonedMerges {
 		slog.Info("merge turn ended unfinished", "task", id)
 		m.fileError(ctx, id, abandonedMergeText)
 		m.deliverIdle(ctx, id)
 	}
-	for _, c := range work.Start {
+	for _, c := range work.StartMerges {
 		m.startClaimedMerge(ctx, c)
 	}
+	for _, id := range work.StartDiscards {
+		m.startDiscard(ctx, id)
+	}
+
 	current := map[string]bool{}
-	for _, f := range work.Finalize {
-		current[f.Task.Id] = true
-		if finalizing[f.Task.Id] {
-			continue
+	once := func(key string, run func()) {
+		current[key] = true
+		if started[key] {
+			return
 		}
-		finalizing[f.Task.Id] = true
-		go m.finalizeMerge(f)
+		started[key] = true
+		go run()
 	}
-	for id := range finalizing {
-		if !current[id] {
-			delete(finalizing, id)
+	for _, f := range work.FinalizeMerges {
+		once("merge:"+f.Task.Id, func() { m.finalizeMerge(f) })
+	}
+	for _, id := range work.FinishDiscards {
+		once("discard:"+id, func() { m.finishDiscard(id) })
+	}
+	for key := range started {
+		if !current[key] {
+			delete(started, key)
 		}
 	}
 }

internal/agents/manager.go

@@ -61,7 +61,6 @@ type Manager struct {
 	moveRefused   map[string]bool
 	startedTodo   map[string]time.Time
 	startRefused  map[string]bool
-	finishing     map[string]bool
 	handoffModel  map[string]string
 	mergeTurns    map[string]mergeTurn
 	landing       map[string]landing
@@ -167,7 +166,6 @@ func New(st *store.Store, worktreeRoot, leadRoot, handsRoot, codexHome, antigrav
 		moveRefused:      map[string]bool{},
 		startedTodo:      map[string]time.Time{},
 		startRefused:     map[string]bool{},
-		finishing:        map[string]bool{},
 		handoffModel:     map[string]string{},
 		mergeTurns:       map[string]mergeTurn{},
 		landing:          map[string]landing{},
@@ -468,11 +466,6 @@ func (m *Manager) Recover(ctx context.Context) error {
 		return err
 	}
 
-	err = m.recoverDiscards(ctx)
-	if err != nil {
-		return err
-	}
-
 	go m.deliverAllLeads()
 	go m.deliverAllQueued(context.Background())
 	go m.coordinate()
@@ -491,17 +484,8 @@ func (m *Manager) failResume(ctx context.Context, sess api.AgentSession, task ap
 	if derr != nil {
 		return derr
 	}
-	end, serr := m.store.EndTurn(ctx, sess.Id, "byre", map[string]any{"error": err.Error()}, nil, false, false, false)
-	if serr != nil {
-		return serr
-	}
-	switch end.Discard {
-	case store.DiscardClosing:
-		go m.finishDiscard(task.Id)
-	case store.DiscardHeld:
-		go m.startDiscard(task.Id)
-	}
-	return nil
+	_, err = m.store.EndTurn(ctx, sess.Id, "byre", map[string]any{"error": err.Error()}, nil, false, false, false)
+	return err
 }
 
 func (m *Manager) logUndelivered(ctx context.Context, sessionID, reason string) {
@@ -1012,12 +996,7 @@ func (m *Manager) pump(l *live) {
 		m.finishReviewer(l, "the reviewer's process exited mid-turn")
 		return
 	}
-	switch end.Discard {
-	case store.DiscardClosing:
-		m.finishDiscard(l.taskID)
-		return
-	case store.DiscardHeld:
-		m.startDiscard(l.taskID)
+	if end.Discard != store.DiscardNone {
 		return
 	}
 	if l.owner == "" {
@@ -1317,17 +1296,11 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
 		m.snapshotSoon(liveTarget(l), api.SnapshotTriggerTurnEnd, 0, review, false)
 	}
 
-	switch end.Discard {
-	case store.DiscardClosing:
+	if end.Discard != store.DiscardNone {
 		m.logUndelivered(ctx, l.id, "turn ended in discard")
-		go m.finishDiscard(l.taskID)
-		return nil
-	case store.DiscardHeld:
-		if stageFinished {
+		if end.Discard == store.DiscardHeld && stageFinished {
 			go m.advanceStage(l)
-			return nil
 		}
-		go m.startDiscard(l.taskID)
 		return nil
 	}
 	if handoff {
@@ -1348,22 +1321,13 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
 }
 
 func (m *Manager) afterTurn(ctx context.Context, l *live) {
-	discard, err := m.store.GetDiscard(ctx, l.taskID)
-	if err != nil {
-		slog.Error("read discard state", "task", l.taskID, "err", err)
-		return
-	}
-	if discard.State == store.DiscardHeld {
-		m.startDiscard(l.taskID)
-		return
-	}
 	task, err := m.store.GetTask(ctx, l.taskID)
 	if err != nil {
 		slog.Error("read task after turn", "task", l.taskID, "err", err)
 		return
 	}
-	if task.Merging {
-		m.logUndelivered(ctx, l.id, "task is merging")
+	if task.Merging || task.Discarding {
+		m.logUndelivered(ctx, l.id, "task is merging or being discarded")
 		return
 	}
 	m.deliverTeam(ctx, l)

internal/agents/summary.go

@@ -134,7 +134,7 @@ func (m *Manager) searchClosedText(ctx context.Context, projectID, query string,
 }
 
 func (m *Manager) Discard(ctx context.Context, task api.Task, actor string) (api.Task, error) {
-	state, err := m.store.RequestDiscard(ctx, task.Id, actor)
+	err := m.store.RequestDiscard(ctx, task.Id, actor)
 	switch {
 	case errors.Is(err, store.ErrNotFound):
 		return api.Task{}, fmt.Errorf("%w: task %s is closed", ErrConflict, task.Id)
@@ -149,94 +149,74 @@ func (m *Manager) Discard(ctx context.Context, task api.Task, actor string) (api
 	case err != nil:
 		return api.Task{}, err
 	}
-	if state == store.DiscardClosing {
-		go m.finishDiscard(task.Id)
-	} else {
-		go m.startDiscard(task.Id)
-	}
 	return m.store.GetTask(ctx, task.Id)
 }
 
-func (m *Manager) startDiscard(taskID string) {
-	ctx := context.Background()
-	started, err := m.launchDiscard(ctx, taskID)
+func (m *Manager) startDiscard(ctx context.Context, taskID string) {
+	err := m.beginDiscard(ctx, taskID)
 	if err == nil {
-		if started {
-			slog.Info("discard turn started", "task", taskID)
-		}
 		return
 	}
 	slog.Error("start discard turn", "task", taskID, "err", err)
-	err = m.store.CloseDiscard(ctx, taskID)
-	if err != nil {
-		slog.Error("close discard", "task", taskID, "err", err)
-		return
-	}
-	m.finishDiscard(taskID)
+	m.closeDiscard(ctx, taskID)
 }
 
-func (m *Manager) launchDiscard(ctx context.Context, taskID string) (bool, error) {
+func (m *Manager) closeDiscard(ctx context.Context, taskID string) {
+	err := m.store.CloseDiscard(ctx, taskID)
+	if err != nil {
+		slog.Error("close discard", "task", taskID, "err", err)
+	}
+	m.poke()
+}
+
+func (m *Manager) beginDiscard(ctx context.Context, taskID string) error {
 	task, err := m.store.GetTask(ctx, taskID)
 	if err != nil {
-		return false, err
+		return err
 	}
 	sess, found, err := m.liveSession(ctx, taskID)
 	if err != nil {
-		return false, err
+		return err
 	}
 	if !found {
 		creator, err := m.store.GetUser(ctx, task.CreatedBy)
 		if err != nil {
-			return false, err
+			return err
 		}
 		sess, err = m.newSession(ctx, task, creator, sess.Member, "")
 		if err != nil {
-			return false, err
+			return err
 		}
 	}
 
 	err = m.store.StartDiscardTurn(ctx, taskID, sess.Id, "byre")
 	if errors.Is(err, store.ErrStateChanged) {
-		return false, nil
+		return nil
 	}
 	if err != nil {
-		return false, err
+		return err
 	}
-	_, err = m.launch(ctx, sess, task, "byre", discardPrompt, "", "", nil)
-	return err == nil, err
+	slog.Info("discard turn started", "task", taskID)
+	go func() {
+		ctx := context.Background()
+		_, err := m.launch(ctx, sess, task, "byre", discardPrompt, "", "", nil)
+		if err == nil {
+			return
+		}
+		slog.Error("launch discard turn", "task", taskID, "err", err)
+		m.closeDiscard(ctx, taskID)
+	}()
+	return nil
 }
 
 func (m *Manager) finishDiscard(taskID string) {
-	m.mu.Lock()
-	if m.finishing[taskID] {
-		m.mu.Unlock()
-		return
-	}
-	m.finishing[taskID] = true
-	m.mu.Unlock()
-	defer func() {
-		m.mu.Lock()
-		delete(m.finishing, taskID)
-		m.mu.Unlock()
-	}()
-
 	ctx := context.Background()
 	err := m.discardNow(ctx, taskID)
 	if err == nil {
 		return
 	}
 	slog.Error("finish discard", "task", taskID, "err", err)
-	sessions, serr := m.store.ListSessions(ctx, taskID)
-	if serr != nil {
-		slog.Error("list sessions for discard error", "task", taskID, "err", serr)
-		return
-	}
-	task, terr := m.store.GetTask(ctx, taskID)
-	if terr != nil {
-		slog.Error("load task for discard error", "task", taskID, "err", terr)
-		return
-	}
-	m.fileTaskError(ctx, task, sessions, "Discarding the task failed: "+err.Error()+". Discard again to retry; byre also retries when it next starts.")
+	m.fileError(ctx, taskID, "Discarding the task failed: "+err.Error()+". byre retries when it next starts.")
 }
 
 func (m *Manager) discardNow(ctx context.Context, taskID string) error {
@@ -264,36 +244,3 @@ func (m *Manager) discardNow(ctx context.Context, taskID string) error {
 	_, err = m.store.DiscardTask(ctx, task.Id, discard.By)
 	return err
 }
-
-func (m *Manager) recoverDiscards(ctx context.Context) error {
-	tasks, err := m.store.PendingDiscards(ctx)
-	if err != nil {
-		return err
-	}
-	for _, t := range tasks {
-		discard, err := m.store.GetDiscard(ctx, t.Id)
-		if err != nil {
-			return err
-		}
-		switch discard.State {
-		case store.DiscardHeld:
-			go m.startDiscard(t.Id)
-		case store.DiscardClosing:
-			go m.finishDiscard(t.Id)
-		case store.DiscardRunning:
-			sess, err := m.store.GetSession(ctx, discard.Session)
-			if err != nil {
-				return err
-			}
-			if sess.State == api.AgentSessionStateRunning {
-				continue
-			}
-			err = m.store.CloseDiscard(ctx, t.Id)
-			if err != nil {
-				return err
-			}
-			go m.finishDiscard(t.Id)
-		}
-	}
-	return nil
-}

internal/store/coordinate.go

@@ -0,0 +1,69 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+)
+
+type Work struct {
+	AbandonedMerges []string
+	StartMerges     []ClaimedMerge
+	FinalizeMerges  []FinalizingMerge
+	StartDiscards   []string
+	FinishDiscards  []string
+}
+
+func (s *Store) Coordinate(ctx context.Context) (Work, error) {
+	work := Work{}
+	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		var err error
+		work.AbandonedMerges, err = abandonMerges(ctx, tx, emit)
+		if err != nil {
+			return err
+		}
+		err = claimHeldMerges(ctx, tx, emit)
+		if err != nil {
+			return err
+		}
+		work.StartMerges, err = startableMerges(ctx, tx)
+		if err != nil {
+			return err
+		}
+		work.FinalizeMerges, err = finalizingMerges(ctx, tx)
+		if err != nil {
+			return err
+		}
+		_, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = 'closing'
+			WHERE status = 'active' AND discard_state = 'running'
+				AND NOT EXISTS (SELECT 1 FROM agent_sessions s WHERE s.id = tasks.discard_session AND s.state = 'running')`)
+		if err != nil {
+			return err
+		}
+		work.StartDiscards, err = queryIDs(ctx, tx, `SELECT id FROM tasks WHERE status = 'active' AND discard_state = 'held' AND NOT `+implementerRunning)
+		if err != nil {
+			return err
+		}
+		work.FinishDiscards, err = queryIDs(ctx, tx, `SELECT id FROM tasks WHERE status = 'active' AND discard_state = 'closing'`)
+		return err
+	})
+	return work, err
+}
+
+func queryIDs(ctx context.Context, tx *sql.Tx, query string) ([]string, error) {
+	rows, err := tx.QueryContext(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	ids := []string{}
+	for rows.Next() {
+		id := ""
+		err = rows.Scan(&id)
+		if err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	return ids, rows.Err()
+}

internal/store/discard.go

@@ -42,9 +42,9 @@ func (s *Store) GetDiscard(ctx context.Context, taskID string) (Discard, error)
 	return d, err
 }
 
-func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) (DiscardState, error) {
+func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		state := DiscardNone
-	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		projectID := ""
 		status := ""
 		mergingBy := ""
@@ -75,7 +75,6 @@ func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) (Disca
 		case subtasks > 0:
 			return ErrHasSubtasks
 		}
-		state = DiscardHeld
 		_, 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 = ?`,
 			string(DiscardHeld), actor, taskID)
 		if err != nil {
@@ -83,7 +82,6 @@ func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) (Disca
 		}
 		return emit(api.Event{Kind: "task.discard_started", ProjectId: &projectID, TaskId: &taskID, Actor: actor, Payload: map[string]any{"id": taskID}})
 	})
-	return state, err
 }
 
 func (s *Store) StartDiscardTurn(ctx context.Context, taskID, sessionID, actor string) error {
@@ -176,8 +174,7 @@ func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload ma
 		case state == DiscardHeld:
 			end.Discard = DiscardHeld
 		case state == DiscardRunning && discardSession == sessionID:
-			end.Discard = DiscardClosing
-			_, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = ? WHERE id = ?`, string(DiscardClosing), taskID)
+			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)
 			if errors.Is(err, sql.ErrNoRows) {
@@ -195,7 +192,3 @@ func (s *Store) CloseDiscard(ctx context.Context, taskID string) error {
 		return err
 	})
 }
-
-func (s *Store) PendingDiscards(ctx context.Context) ([]api.Task, error) {
-	return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE status = ? AND discard_state != ''`, string(api.TaskStatusActive))
-}

internal/store/merges.go

@@ -62,34 +62,6 @@ type FinalizingMerge struct {
 	Unchanged bool
 }
 
-type MergeWork struct {
-	Abandoned []string
-	Start     []ClaimedMerge
-	Finalize  []FinalizingMerge
-}
-
-func (s *Store) CoordinateMerges(ctx context.Context) (MergeWork, error) {
-	work := MergeWork{}
-	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
-		var err error
-		work.Abandoned, err = abandonMerges(ctx, tx, emit)
-		if err != nil {
-			return err
-		}
-		err = claimHeldMerges(ctx, tx, emit)
-		if err != nil {
-			return err
-		}
-		work.Start, err = startableMerges(ctx, tx)
-		if err != nil {
-			return err
-		}
-		work.Finalize, err = finalizingMerges(ctx, tx)
-		return err
-	})
-	return work, err
-}
-
 type idRow struct {
 	id      string
 	project string