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 }