Advance workflow stages from 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
 
-Coordination of merges, discards, reviews and ready-for-merge prompts 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.
+Coordination of merges, discards, reviews, ready-for-merge prompts and workflow stages 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/coordinator.go

@@ -72,6 +72,9 @@ func (m *Manager) coordinatePass(ctx context.Context, started map[string]bool) {
 	for _, r := range work.SettleReviews {
 		once(fmt.Sprintf("settle:%s:%d", r.TaskID, r.N), func() { m.settleReviews(r) })
 	}
+	for _, id := range work.AdvanceStages {
+		once("advance:"+id, func() { m.advanceStage(id) })
+	}
 	for _, p := range work.ReadyPrompts {
 		once("ready:"+p.TaskID, func() { m.promptReady(p) })
 	}

internal/agents/manager.go

@@ -367,11 +367,6 @@ func (m *Manager) Recover(ctx context.Context) error {
 		return err
 	}
 
-	staged, err := m.recoverStages(ctx)
-	if err != nil {
-		return err
-	}
-
 	err = m.recoverPreviews(ctx)
 	if err != nil {
 		return err
@@ -399,9 +394,6 @@ func (m *Manager) Recover(ctx context.Context) error {
 	}
 
 	for _, sess := range sessions {
-		if staged[sess.Id] {
-			continue
-		}
 		task, err := m.taskFor(ctx, sess)
 		if err != nil {
 			slog.Error("resume session: load task", "session", sess.Id, "task", sess.TaskId, "err", err)
@@ -412,6 +404,23 @@ func (m *Manager) Recover(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
+		started, err := m.store.SessionStarted(ctx, sess.Id)
+		if err != nil {
+			return err
+		}
+		finished, err := m.store.StageFinished(ctx, sess.Id)
+		if err != nil {
+			return err
+		}
+		if finished || !pending && !started {
+			m.endMergeTurn(sess.Id, true)
+			err = m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateIdle, "byre", nil)
+			if err != nil {
+				return err
+			}
+			slog.Info("session left idle at startup: its stage finished or it never got its first message", "session", sess.Id, "task", task.Id)
+			continue
+		}
 		if pending {
 			found, err := m.deliverRecorded(ctx, sess, task)
 			if err != nil {
@@ -1307,9 +1316,6 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
 
 	if end.Discard != store.DiscardNone {
 		m.logUndelivered(ctx, l.id, "turn ended in discard")
-		if end.Discard == store.DiscardHeld && stageFinished {
-			go m.advanceStage(l)
-		}
 		return nil
 	}
 	if handoff {
@@ -1322,7 +1328,6 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
 		return nil
 	}
 	if stageFinished {
-		go m.advanceStage(l)
 		return nil
 	}
 	go m.afterTurn(context.WithoutCancel(ctx), l)

internal/agents/workflows.go

@@ -362,34 +362,24 @@ func (m *Manager) recordStage(ctx context.Context, l *live) error {
 	return m.store.RecordStage(ctx, l.taskID, output, head)
 }
 
-func (m *Manager) advanceStage(l *live) {
+func (m *Manager) advanceStage(sessionID string) {
 	ctx := context.Background()
-	_, err := m.nextStage(ctx, l.id)
-	if err != nil {
-		slog.Error("advance workflow stage", "session", l.id, "err", err)
-		m.recordError(l, fmt.Sprintf("The workflow didn't move on from this stage: %v. byre tries again when it next starts.", err))
-	}
-	m.afterTurn(ctx, l)
-}
-
-func (m *Manager) recoverStages(ctx context.Context) (map[string]bool, error) {
-	ids, err := m.store.FinishedStages(ctx)
-	if err != nil {
-		return nil, err
-	}
-	started := map[string]bool{}
-	for _, id := range ids {
-		next, err := m.nextStage(ctx, id)
-		if err != nil {
-			slog.Error("advance workflow stage", "session", id, "err", err)
-			continue
+	next, err := m.nextStage(ctx, sessionID)
+	if err == nil {
+		slog.Info("advanced workflow stage", "session", sessionID, "next", next)
+	} else {
+		slog.Error("advance workflow stage", "session", sessionID, "err", err)
+		sess, serr := m.store.GetSession(ctx, sessionID)
+		if serr == nil {
+			m.fileError(ctx, sess.TaskId, fmt.Sprintf("The workflow didn't move on from this stage: %v. byre tries again when it next starts.", err))
 		}
-		for _, n := range next {
-			started[n] = true
-		}
-		slog.Info("advanced workflow stage", "session", id, "next", next)
 	}
-	return started, nil
+	m.mu.Lock()
+	l := m.live[sessionID]
+	m.mu.Unlock()
+	if l != nil {
+		m.afterTurn(ctx, l)
+	}
 }
 
 func (m *Manager) nextStage(ctx context.Context, sessionID string) ([]string, error) {

internal/store/coordinate.go

@@ -16,6 +16,7 @@ type Work struct {
 	StartStages     []StageStart
 	SettleReviews   []SnapshotRef
 	ReadyPrompts    []ReadyPrompt
+	AdvanceStages   []string
 }
 
 func (s *Store) Coordinate(ctx context.Context) (Work, error) {
@@ -69,6 +70,10 @@ func (s *Store) Coordinate(ctx context.Context) (Work, error) {
 			return err
 		}
 		work.ReadyPrompts, err = readyPrompts(ctx, tx, emit)
+		if err != nil {
+			return err
+		}
+		work.AdvanceStages, err = finishedStages(ctx, tx)
 		return err
 	})
 	return work, err

internal/store/workflow.go

@@ -260,8 +260,14 @@ func (s *Store) StageOutput(ctx context.Context, sessionID string) (string, erro
 	return output, err
 }
 
-func (s *Store) FinishedStages(ctx context.Context) ([]string, error) {
-	return queryIDs(ctx, s.db, `SELECT id FROM agent_sessions WHERE stage_finished = 1 ORDER BY created_at`)
+func (s *Store) StageFinished(ctx context.Context, sessionID string) (bool, error) {
+	finished := false
+	err := s.db.QueryRowContext(ctx, `SELECT stage_finished FROM agent_sessions WHERE id = ?`, sessionID).Scan(&finished)
+	return finished, err
+}
+
+func finishedStages(ctx context.Context, tx *sql.Tx) ([]string, error) {
+	return queryIDs(ctx, tx, `SELECT id FROM agent_sessions WHERE stage_finished = 1 AND state != 'running' ORDER BY created_at`)
 }
 
 func (s *Store) ClearStage(ctx context.Context, sessionID string) error {