so-many-crows/pipeline.go

package main

import (
	"context"
	"fmt"
	"os"
	"strings"
	"sync"
)

// 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
	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
}

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) {
	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})

	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)
		}
	}

	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}
	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, 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})
	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})
	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 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
}

func (w *itemWatcher) Flying(crow string) {
	w.em.Emit("crow.flying", map[string]any{"item_id": w.itemID, "crow": crow})
}

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
}