so-many-crows/pipeline.go

package main

import (
	"context"
	"fmt"
	"math/rand/v2"
	"os"
	"slices"
	"strings"
	"sync"
	"time"
)

// Emitter receives the events in EVENTS.md. The web server streams them; the
// command line prints them. Fields exclude list_id and ts; the emitter adds those.
type Emitter interface {
	Emit(event string, fields map[string]any)
}

// Approval is one question for a human: may this triage or answer stand?
type Approval struct {
	ListID, ItemID, Item string
	Target               string // "triage" or "output"
}

// Approver asks a human. It returns "approved", "rejected" or "timeout".
// Anything other than a clear yes is a no.
type Approver interface {
	Approve(ctx context.Context, a Approval) string
}

// List is one todo list handed to the flock.
type List struct {
	ID, Requester, Source string
	Items                 []string
	Nest                  Nest // zero value means the todo nest
	Parallel              bool // fly items at once (web); the command line goes one by one
}

type itemResult struct {
	Item, Mode, TaskType, Answer, Verdict, Skipped string
	Crows                                          []string
}

// Pipeline is the one path every list takes, from the command line or the web.
type Pipeline struct {
	Flock    *Flock
	Ledger   *Ledger
	Log      *FlightLog
	G        Guardrails
	Approver Approver

	mem     *flockMemory // shared memory banter retrieves from
	memOnce sync.Once
}

func itemID(listID string, i int) string { return fmt.Sprintf("%s-%d", listID, i+1) }

// Run flies a list and returns per-item results and the list budget.
func (p *Pipeline) Run(ctx context.Context, l List, em Emitter) ([]itemResult, *Budget) {
	if l.Nest.ID == "" {
		l.Nest = defaultNest
	}
	listBudget := &Budget{name: "list", limit: p.G.TokenBudgetPerList}
	items := make([]map[string]any, len(l.Items))
	for i, text := range l.Items {
		items[i] = map[string]any{"item_id": itemID(l.ID, i), "text": text}
	}
	p.Log.Write(Event{Kind: "list", Requester: l.Requester, Note: fmt.Sprintf("%s (%d items)", l.Source, len(l.Items))})
	em.Emit("list.started", map[string]any{"items": items, "requester": l.Requester, "nest_id": l.Nest.ID})
	chat := &chatter{f: p.Flock, mem: p.memory(), em: em, budget: listBudget, nest: l.Nest.ID, roll: rand.Float64}
	ctx = withChatter(ctx, chat)

	results := make([]itemResult, len(l.Items))
	if l.Parallel {
		var wg sync.WaitGroup
		for i := range l.Items {
			wg.Add(1)
			go func() {
				defer wg.Done()
				results[i] = p.item(ctx, l, i, listBudget, em)
			}()
		}
		wg.Wait()
	} else {
		for i := range l.Items {
			results[i] = p.item(ctx, l, i, listBudget, em)
		}
	}

	chat.wait(20 * time.Second) // let in-flight banter land before the list closes
	p.save()
	em.Emit("list.finished", map[string]any{"tokens_used": listBudget.Spent(), "token_budget": listBudget.limit})
	return results, listBudget
}

func (p *Pipeline) item(ctx context.Context, l List, i int, listBudget *Budget, em Emitter) itemResult {
	id, item := itemID(l.ID, i), l.Items[i]
	r := itemResult{Item: item}
	skip := func(reason string) itemResult {
		r.Skipped = reason
		em.Emit("item.skipped", map[string]any{"item_id": id, "reason": reason})
		return r
	}
	w := &itemWatcher{em: em, itemID: id, core: l.Nest.Crows}
	itemBudget := &Budget{name: "item", limit: p.G.TokenBudgetPerItem}

	t, err := p.Flock.Triage(ctx, w, item, itemBudget, listBudget)
	if err != nil {
		return skip("triage failed: " + err.Error())
	}
	r.Mode, r.TaskType = t.Mode, t.TaskType
	p.Log.Write(Event{Kind: "triage", Requester: l.Requester, Item: item, Mode: t.Mode, TaskType: t.TaskType, Note: t.Reason})

	// The Sentry earns autonomy on triage separately from the crows' work.
	triageType := "triage/" + t.TaskType
	needTriage := p.Ledger.Level("Sentry", triageType) < Trusted
	em.Emit("item.triaged", map[string]any{
		"item_id": id, "mode": t.Mode, "task_type": t.TaskType,
		"subtasks": nonNil(t.Subtasks), "reason": t.Reason, "needs_approval": needTriage,
	})
	if needTriage {
		v := p.judge(ctx, Approval{ListID: l.ID, ItemID: id, Item: item, Target: "triage"}, []string{"Sentry"}, triageType, em)
		if v != "approved" {
			return skip("triage " + v)
		}
	} else {
		em.Emit("verdict", map[string]any{"item_id": id, "target": "triage", "verdict": "auto"})
	}

	out, err := p.Flock.Execute(ctx, w, item, t, l.Nest, itemBudget, listBudget)
	if err != nil {
		return skip("flight failed: " + err.Error())
	}
	r.Answer, r.Crows = out.Answer, out.Crows

	// The least-trusted crow involved sets the bar.
	lowest := Trusted
	for _, c := range out.Crows {
		lowest = min(lowest, p.Ledger.Level(c, t.TaskType))
	}
	needOutput := lowest < Regular
	em.Emit("item.answered", map[string]any{"item_id": id, "answer": out.Answer, "crows": out.Crows, "needs_approval": needOutput})
	chat := chatterFrom(ctx)
	chat.remember(fmt.Sprintf("%s answered %q", strings.Join(out.Crows, ", "), item))
	for _, pair := range [][2]string{{"Hugin", "Munin"}, {"Corvina", "Edgar"}, {"Pepper", "Jet"}} {
		if slices.Contains(out.Crows, pair[0]) && slices.Contains(out.Crows, pair[1]) {
			chat.moment(ctx, id, pair[1], pair[0], fmt.Sprintf("%s and %s just worked on the same task together: %q.", pair[0], pair[1], item))
			break
		}
	}
	switch {
	case needOutput:
		r.Verdict = p.judge(ctx, Approval{ListID: l.ID, ItemID: id, Item: item, Target: "output"}, out.Crows, t.TaskType, em)
	case lowest < Trusted:
		r.Verdict = "accepted, unreviewed"
		em.Emit("verdict", map[string]any{"item_id": id, "target": "output", "verdict": "auto"})
	default:
		r.Verdict = "accepted, trusted"
		em.Emit("verdict", map[string]any{"item_id": id, "target": "output", "verdict": "auto"})
	}
	p.Log.Write(Event{Kind: "verdict", Item: item, Crow: strings.Join(out.Crows, ","), TaskType: t.TaskType, Verdict: r.Verdict})
	return r
}

// judge asks a human, records the verdict, reports trust changes, and saves.
// A timeout counts as a rejection. A shutdown records nothing.
func (p *Pipeline) judge(ctx context.Context, a Approval, holders []string, taskType string, em Emitter) string {
	v := p.Approver.Approve(ctx, a)
	if ctx.Err() != nil {
		return "rejected" // shutting down: no human said anything, so record nothing
	}
	if v != "approved" && v != "timeout" {
		v = "rejected"
	}
	em.Emit("verdict", map[string]any{"item_id": a.ItemID, "target": a.Target, "verdict": v})
	chat := chatterFrom(ctx)
	what := "answer to"
	if a.Target == "triage" {
		what = "plan for"
	}
	chat.remember(fmt.Sprintf("%s's %s %q was %s by the human", strings.Join(holders, " and "), what, a.Item, v))
	if v != "approved" && len(holders) > 0 {
		subject := holders[rand.IntN(len(holders))]
		chat.moment(ctx, a.ItemID, subject, "", fmt.Sprintf("The human just %s %s's %s %q.", map[string]string{"rejected": "rejected", "timeout": "ignored, until it timed out,"}[v], subject, what, a.Item))
	}
	for _, c := range p.Ledger.Judge(holders, taskType, v == "approved", p.G.PromoteAfter) {
		em.Emit("trust.changed", map[string]any{"holder": c.Holder, "task_type": c.TaskType, "from": c.From, "to": c.To})
		if c.Holder != "flock" && c.To > c.From {
			names := []string{"stranger", "regular", "trusted"}
			chat.remember(fmt.Sprintf("%s leveled up to %s on %s", c.Holder, names[c.To], c.TaskType))
			chat.moment(ctx, a.ItemID, c.Holder, "", fmt.Sprintf("%s just leveled up: now %s on %s tasks.", c.Holder, names[c.To], c.TaskType))
		}
	}
	if a.Target == "triage" {
		p.Log.Write(Event{Kind: "verdict", Item: a.Item, Crow: "Sentry", TaskType: taskType, Verdict: v})
	}
	p.save()
	return v
}

func (p *Pipeline) save() {
	if err := p.Ledger.Save(); err != nil {
		fmt.Fprintln(os.Stderr, "could not save ledger:", err)
	}
}

// itemWatcher turns flock takeoffs and landings into crow events for one item.
type itemWatcher struct {
	em     Emitter
	itemID string
	core   []string // the nest's core crows; empty for todo
}

// Flying marks a crow borrowed when the nest has a core and the crow is a
// roster crow outside it. The Sentry and Elder are never borrowed.
func (w *itemWatcher) Flying(crow string) {
	borrowed := len(w.core) > 0 && slices.Contains(roster, crow) && !slices.Contains(w.core, crow)
	w.em.Emit("crow.flying", map[string]any{"item_id": w.itemID, "crow": crow, "borrowed": borrowed})
}

func (w *itemWatcher) Landed(crow string, tokens int, err error) {
	f := map[string]any{"item_id": w.itemID, "crow": crow, "ok": err == nil, "tokens": tokens}
	if err != nil {
		f["error"] = err.Error()
	}
	w.em.Emit("crow.landed", f)
}

func nonNil(s []string) []string {
	if s == nil {
		return []string{}
	}
	return s
}

// memory is the flock's short shared memory, kept across lists for banter.
func (p *Pipeline) memory() *flockMemory {
	p.memOnce.Do(func() { p.mem = &flockMemory{} })
	return p.mem
}