byre/internal/agents/coordinator.go

package agents

import (
	"context"
	"fmt"
	"log/slog"
	"time"
)

const abandonedMergeText = "The Merge turn ended unfinished (interrupted, failed to start, or with no final message), so byre didn't squash or push the task. Merge again to retry."

func (m *Manager) poke() {
	select {
	case m.pokes <- struct{}{}:
	default:
	}
}

func (m *Manager) coordinate() {
	started := map[string]bool{}
	timer := time.NewTimer(time.Hour)
	timer.Stop()
	for {
		select {
		case <-m.watchStop:
			return
		case <-m.pokes:
		case <-timer.C:
		}
		next := m.coordinatePass(context.Background(), started)
		timer.Stop()
		if next != nil {
			timer.Reset(time.Until(*next))
		}
	}
}

func (m *Manager) coordinatePass(ctx context.Context, started map[string]bool) *time.Time {
	work, err := m.store.Coordinate(ctx)
	if err != nil {
		slog.Error("coordinate", "err", err)
		retry := time.Now().Add(time.Minute)
		return &retry
	}
	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.StartMerges {
		m.startClaimedMerge(ctx, c)
	}
	for _, id := range work.StartDiscards {
		m.startDiscard(ctx, id)
	}
	for _, r := range work.EndReviewers {
		go m.cleanUpReviewer(r)
	}

	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 _, r := range work.StartReviews {
		once(fmt.Sprintf("review:%s:%d", r.TaskID, r.N), func() { m.startReviews(r) })
	}
	for _, s := range work.StartStages {
		once(fmt.Sprintf("stage:%s:%d:%s", s.TaskID, s.SnapshotN, s.Stage), func() { m.startReviewStage(s) })
	}
	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 _, d := range work.LeadDeliveries {
		once(fmt.Sprintf("lead:%s:%d:%d", d.Owner, d.First, d.Last), func() { m.deliverLead(d.Owner) })
	}
	for _, q := range work.QueuedDelivery {
		once(fmt.Sprintf("queue:%s:%d:%s", q.SessionID, q.Input, q.Response), func() { m.deliverQueuedTurn(q.SessionID) })
	}
	for _, p := range work.ReadyPrompts {
		once("ready:"+p.TaskID, func() { m.promptReady(p) })
	}
	for _, sc := range work.DueSchedules {
		once(fmt.Sprintf("schedule:%s:%d", sc.Id, sc.NextRun.UnixNano()), func() { m.runSchedule(sc) })
	}
	for key := range started {
		if !current[key] {
			delete(started, key)
		}
	}
	return work.NextSchedule
}