From 1b5f42a01a38c02f1d32118ab20bc50d750d959a Mon Sep 17 00:00:00 2001 From: Ian Gulliver Date: Wed, 21 Oct 2026 11:53:37 -0700 Subject: [PATCH] Run discards on the state-driven coordinator --- docs/operations.md | 2 +- internal/agents/compact.go | 7 +- internal/agents/coordinator.go | 46 +++++++------ internal/agents/manager.go | 50 ++------------ internal/agents/summary.go | 115 +++++++++------------------------ internal/store/coordinate.go | 69 ++++++++++++++++++++ internal/store/discard.go | 15 ++--- internal/store/merges.go | 28 -------- 8 files changed, 141 insertions(+), 191 deletions(-) create mode 100644 internal/store/coordinate.go diff --git a/docs/operations.md b/docs/operations.md index b6ec5b8..f0171fe 100644 --- a/docs/operations.md +++ b/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 diff --git a/internal/agents/compact.go b/internal/agents/compact.go index ec1801d..562c49e 100644 --- a/internal/agents/compact.go +++ b/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 diff --git a/internal/agents/coordinator.go b/internal/agents/coordinator.go index aa6bf59..5487c80 100644 --- a/internal/agents/coordinator.go +++ b/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) } - current := map[string]bool{} - for _, f := range work.Finalize { - current[f.Task.Id] = true - if finalizing[f.Task.Id] { - continue - } - finalizing[f.Task.Id] = true - go m.finalizeMerge(f) + for _, id := range work.StartDiscards { + m.startDiscard(ctx, id) } - for id := range finalizing { - if !current[id] { - delete(finalizing, id) + + current := map[string]bool{} + once := func(key string, run func()) { + current[key] = true + if started[key] { + return + } + started[key] = true + go run() + } + 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) } } } diff --git a/internal/agents/manager.go b/internal/agents/manager.go index 3d35c10..bafad8b 100644 --- a/internal/agents/manager.go +++ b/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) diff --git a/internal/agents/summary.go b/internal/agents/summary.go index 70ddf47..e5c142a 100644 --- a/internal/agents/summary.go +++ b/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 -} diff --git a/internal/store/coordinate.go b/internal/store/coordinate.go new file mode 100644 index 0000000..90dc6ad --- /dev/null +++ b/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() +} diff --git a/internal/store/discard.go b/internal/store/discard.go index 3e2bd92..fb44fd8 100644 --- a/internal/store/discard.go +++ b/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) { - state := DiscardNone - err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) 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 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)) -} diff --git a/internal/store/merges.go b/internal/store/merges.go index 588fbec..94f9c00 100644 --- a/internal/store/merges.go +++ b/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