From c957bdd3f6b9154e5a073d6583ae78fcc9b59a4b Mon Sep 17 00:00:00 2001 From: Smokey <150986568+jacisjake@users.noreply.github.com> Date: Wed, 28 Oct 2026 06:46:15 -0600 Subject: [PATCH 1/2] =?UTF-8?q?test(dispatch):=20red=20for=20#28=20?= =?UTF-8?q?=E2=80=94=20saturated=20flock=20must=20honor=20context,=20all?= =?UTF-8?q?=20dispatch=20through=20Flock?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dispatch_test.go | 169 +++++++++++++++++++++++++++++++++++++++++++++++ nestapi_test.go | 6 ++ 2 files changed, 175 insertions(+) create mode 100644 dispatch_test.go diff --git a/dispatch_test.go b/dispatch_test.go new file mode 100644 index 0000000..bb6a4ee --- /dev/null +++ b/dispatch_test.go @@ -0,0 +1,169 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +// fakeDispatchJev answers kind/mode as choice and restates_earlier as noul. +func fakeDispatchJev(t *testing.T) *JevClient { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body struct { + Questions map[string]any `json:"questions"` + } + json.NewDecoder(r.Body).Decode(&body) + out := map[string]any{} + for q := range body.Questions { + switch q { + case "kind": + out[q] = map[string]any{"type": "choice", "choice": "say", "confidence": 0.5} + case "mode": + out[q] = map[string]any{"type": "choice", "choice": "conspiracy", "confidence": 0.9, + "probabilities": map[string]float64{"conspiracy": 0.9, "solo": 0.1}} + case "restates_earlier": + out[q] = map[string]any{"type": "noul", "noul": 0.85} + } + } + json.NewEncoder(w).Encode(map[string]any{"model": "jev-fake", "answers": out, + "usage": map[string]int{"input_tokens": 50, "output_tokens": 0}}) + })) + t.Cleanup(srv.Close) + return &JevClient{url: srv.URL, key: "tk", model: "jev-fake", threshold: 0.7, client: srv.Client()} +} + +// peakBrain tracks concurrent model calls. +type peakBrain struct { + mu sync.Mutex + peak, now int32 +} + +func (b *peakBrain) Name() string { return "peak" } +func (b *peakBrain) Ask(ctx context.Context, system, prompt string) (string, int, error) { + c := atomic.AddInt32(&b.now, 1) + if c > atomic.LoadInt32(&b.peak) { + atomic.StoreInt32(&b.peak, c) + } + time.Sleep(30 * time.Millisecond) + atomic.AddInt32(&b.now, -1) + return "KIND: deliver\n[a] and [b] matter", 10, nil +} + +func newDispatchServer(t *testing.T, g Guardrails) (*Server, string, *peakBrain) { + t.Helper() + dir := t.TempDir() + ledger, _ := LoadLedger(filepath.Join(dir, "ledger.json")) + flog, err := OpenFlightLog(filepath.Join(dir, "flight.jsonl")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { flog.Close() }) + nests, _ := LoadNests(filepath.Join(dir, "nests.json")) + msgs, _ := LoadMessageStore(filepath.Join(dir, "messages")) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + pb := &peakBrain{} + appr := NewWebApprover(time.Minute) + p := &Pipeline{Flock: NewFlock(pb, g, flog), Ledger: ledger, Log: flog, G: g, Approver: appr} + p.Flock.SetClassifier(fakeDispatchJev(t)) + s := NewServer(ctx, p, appr, nests).WithMessages(msgs) + return s, filepath.Join(dir, "flight.jsonl"), pb +} + +func TestDispatchCallsGoThroughFlock(t *testing.T) { + g := DefaultGuardrails() + flog := "" + dir := t.TempDir() + lg, err := OpenFlightLog(filepath.Join(dir, "flight.jsonl")) + if err != nil { + t.Fatal(err) + } + defer lg.Close() + flog = filepath.Join(dir, "flight.jsonl") + f := NewFlock(MockBrain{}, g, lg) + f.SetClassifier(fakeDispatchJev(t)) + + mode, ok := f.ClassifyMode(context.Background(), "is this risky") + if !ok || mode != "conspiracy" { + t.Fatalf("want conspiracy from the fake, got %q ok=%v", mode, ok) + } + evs, _ := ReadEvents(flog) + probLogged := false + for _, e := range evs { + if e.Kind == "mode" && e.Prob != nil { + probLogged = true + } + } + if !probLogged { + t.Fatal("ClassifyMode must write a flight-log event with probabilities") + } + + // Saturate the flock semaphore: a wrapper call must wait for a slot, + // proving it goes through Flock (never a direct client call). + for i := 0; i < g.MaxCrowsInFlight; i++ { + f.sem <- struct{}{} + } + cctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond) + defer cancel() + _, blockedOk := f.ClassifyMode(cctx, "risky?") + if blockedOk { + t.Fatal("a saturated flock must not answer (semaphore not enforced)") + } + for i := 0; i < g.MaxCrowsInFlight; i++ { + <-f.sem + } +} + +func TestConspiracyInviteCapInFlight(t *testing.T) { + g := DefaultGuardrails() + g.MaxCrowsInFlight = 1 + s, _, _ := newDispatchServer(t, g) + id := newNest(t, serve(s), "Wide", `"Hugin","Munin","Soot"`) + promote(s, "Hugin", "Munin", "Soot") + // inviteCount caps at MaxCrowsInFlight=1 -> at most one invited + if n := s.inviteCount(mustNest(s, id), 4); n != 1 { + t.Fatalf("want invite capped to 1 by in-flight limit, got %d", n) + } +} + +func TestDispatchGuardrailsEndToEnd(t *testing.T) { + g := DefaultGuardrails() + g.MaxCrowsInFlight = 2 + s, logPath, pb := newDispatchServer(t, g) + ts := serve(s) + id := newNest(t, ts, "Lab", `"Hugin","Munin"`) + promote(s, "Hugin", "Munin") + code, req := post(t, ts.URL+"/api/nests/"+id+"/messages", + `{"author":"owner","kind":"request","text":"is this launch risky"}`) + if code != http.StatusCreated || req["mode"] != "conspiracy" || req["tagged_by"] != "header" { + t.Fatalf("post: %d %v", code, req) + } + awaitMessages(t, ts, id, 3) // request + 2 lens claims + + evs, err := ReadEvents(logPath) + if err != nil { + t.Fatal(err) + } + kinds := map[string]int{} + for _, e := range evs { + kinds[e.Kind]++ + } + if kinds["mode"] == 0 || kinds["tag"] == 0 || kinds["flight"] == 0 { + t.Fatalf("dispatch decisions must be logged (mode/tag/flight), got %v", kinds) + } + if pb.peak > 2 { + t.Fatalf("flock cap violated: peak concurrent calls %d > 2", pb.peak) + } +} + +func mustNest(s *Server, id string) Nest { + n, _ := s.nests.Get(id) + return n +} diff --git a/nestapi_test.go b/nestapi_test.go index ed5b3f9..c38fcce 100644 --- a/nestapi_test.go +++ b/nestapi_test.go @@ -214,3 +214,9 @@ func promote(s *Server, crows ...string) { s.p.Ledger.Crows[c+"|other"] = &Record{Approved: 10, Streak: 10, Level: Regular} } } + +// serve mounts a server's handler on a test HTTP server. +func serve(s *Server) *httptest.Server { + ts := httptest.NewServer(s.Handler()) + return ts +} From acc09331b88202206659da486f2132d3076a2372 Mon Sep 17 00:00:00 2001 From: Smokey <150986568+jacisjake@users.noreply.github.com> Date: Wed, 28 Oct 2026 06:46:15 -0600 Subject: [PATCH 2/2] =?UTF-8?q?feat(guardrails):=20context-aware=20flock?= =?UTF-8?q?=20slot=20=E2=80=94=20saturated=20calls=20give=20up=20instead?= =?UTF-8?q?=20of=20wedging=20(#28)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- flock.go | 18 ++++++++++++++++-- jev.go | 28 ++++++++++++++++++++-------- 2 files changed, 36 insertions(+), 10 deletions(-) diff --git a/flock.go b/flock.go index 52f119b..16ab4a9 100644 --- a/flock.go +++ b/flock.go @@ -128,6 +128,17 @@ func (f *Flock) pick(nest Nest, n int) []string { return out } +// reserve takes a flock slot, honoring ctx: a saturated flock waits for a +// slot, but gives up when the context ends instead of wedging the caller. +func (f *Flock) reserve(ctx context.Context) (func(), bool) { + select { + case f.sem <- struct{}{}: + return func() { <-f.sem }, true + case <-ctx.Done(): + return nil, false + } +} + // ask is the only way any crow touches the model: budget, flock cap, timeout, log. func (f *Flock) ask(ctx context.Context, w Watcher, crow, system, prompt string, budgets ...*Budget) (string, error) { for _, b := range budgets { @@ -136,8 +147,11 @@ func (f *Flock) ask(ctx context.Context, w Watcher, crow, system, prompt string, return "", err } } - f.sem <- struct{}{} - defer func() { <-f.sem }() + release, slot := f.reserve(ctx) + if !slot { + return "", fmt.Errorf("flock at cap: %w", ctx.Err()) + } + defer release() if w != nil { w.Flying(crow) diff --git a/jev.go b/jev.go index ea7343d..80b8660 100644 --- a/jev.go +++ b/jev.go @@ -200,8 +200,11 @@ func (f *Flock) ClassifyMode(ctx context.Context, text string, budgets ...*Budge return "", false } } - f.sem <- struct{}{} - defer func() { <-f.sem }() + release, slot := f.reserve(ctx) + if !slot { + return "", false + } + defer release() cctx, cancel := context.WithTimeout(ctx, f.g.CrowTimeout) defer cancel() start := time.Now() @@ -295,8 +298,11 @@ func (f *Flock) ClassifyDivergence(ctx context.Context, state any, budgets ...*B return 0, false } } - f.sem <- struct{}{} - defer func() { <-f.sem }() + release, slot := f.reserve(ctx) + if !slot { + return 0, false + } + defer release() cctx, cancel := context.WithTimeout(ctx, f.g.CrowTimeout) defer cancel() start := time.Now() @@ -318,8 +324,11 @@ func (f *Flock) ClassifyKind(ctx context.Context, text string, budgets ...*Budge return "", nil, false } } - f.sem <- struct{}{} - defer func() { <-f.sem }() + release, slot := f.reserve(ctx) + if !slot { + return "", nil, false + } + defer release() cctx, cancel := context.WithTimeout(ctx, f.g.CrowTimeout) defer cancel() start := time.Now() @@ -346,8 +355,11 @@ func (f *Flock) classify(ctx context.Context, w Watcher, item string, budgets .. return JevCall{}, err } } - f.sem <- struct{}{} - defer func() { <-f.sem }() + release, slot := f.reserve(ctx) + if !slot { + return JevCall{}, fmt.Errorf("flock at cap: %w", ctx.Err()) + } + defer release() if w != nil { w.Flying("Sentry") }