so-many-crows/watchdog_test.go

package main

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"
)

// fakeWatchJev answers kind/mode as choices below the band (so the poster's
// header always wins) and restates_earlier low (the divergence path stays
// quiet). The three watchdog noul questions answer with the configured values.
// Same server shape as fakeDispatchJev, kept local to this test (no shared
// helper edits).
func fakeWatchJev(t *testing.T, scores map[string]float64) *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": "other", "confidence": 0.2}
			case "restates_earlier":
				out[q] = map[string]any{"type": "noul", "noul": 0.05}
			default:
				if v, ok := scores[q]; ok {
					out[q] = map[string]any{"type": "noul", "noul": v}
				}
			}
		}
		json.NewEncoder(w).Encode(map[string]any{"model": "jev-watch-fake", "answers": out,
			"usage": map[string]int{"input_tokens": 60, "output_tokens": 0}})
	}))
	t.Cleanup(srv.Close)
	return &JevClient{url: srv.URL, key: "tk", model: "jev-watch-fake", threshold: 0.7, client: srv.Client()}
}

func watchScores(fact, contra, answers float64) map[string]float64 {
	return map[string]float64{
		"states_fact_without_source": fact,
		"contradicts_earlier":        contra,
		"answers_request":            answers,
	}
}

// newWatchServer wires a nest test server with a watchdog-capable Jev, in the
// shape of newDispatchServer but local here.
func newWatchServer(t *testing.T, scores map[string]float64, brain Brain) (*httptest.Server, *Server) {
	t.Helper()
	ts, s := newTestNestServer(t)
	if brain != nil {
		s.p.Flock.brain = brain
	}
	s.p.Flock.SetClassifier(fakeWatchJev(t, scores))
	return ts, s
}

func findFlag(ts *httptest.Server, id, target string) map[string]any {
	for _, m := range nestMessagesNoT(ts, id) {
		if m["kind"] == "flag" && m["author"] == "Sentry" && m["refers_to"] == target {
			return m
		}
	}
	return nil
}

func waitForFlag(ts *httptest.Server, id, target string) map[string]any {
	deadline := time.Now().Add(2 * time.Second)
	for {
		if f := findFlag(ts, id, target); f != nil {
			return f
		}
		if time.Now().After(deadline) {
			return nil
		}
		time.Sleep(10 * time.Millisecond)
	}
}

func countSentryFlags(ts *httptest.Server, id string) int {
	n := 0
	for _, m := range nestMessagesNoT(ts, id) {
		if m["kind"] == "flag" && m["author"] == "Sentry" {
			n++
		}
	}
	return n
}

func TestSentryQuarantinesFabrication(t *testing.T) {
	ts, s := newWatchServer(t, watchScores(0.9, 0.05, 0.05), nil)
	id := newNest(t, ts, "Lab", `"Hugin","Munin"`)
	_, req := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"owner","kind":"request","text":"count the eggs"}`)
	d := postDeliver(t, ts, id, "Hugin", req["msg_id"].(string), "twelve eggs, from last spring's ledger")

	f := waitForFlag(ts, id, d)
	if f == nil {
		t.Fatalf("an unsourced-fact deliver must draw a Sentry flag, log: %+v", s.msgs.All(id))
	}
	if f["to"] != "owner" {
		t.Fatalf("the flag must be handed to the owner, got %v", f["to"])
	}
	if !s.p.Ledger.IsQuarantined("Hugin") {
		t.Fatal("a flagged crow must be quarantined")
	}

	code, out := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"Hugin","kind":"claim","refers_to":"`+req["msg_id"].(string)+`","text":"I will take the next one"}`)
	if code != http.StatusForbidden {
		t.Fatalf("a quarantined crow's claim must get 403, got %d %v", code, out)
	}
	if msg, _ := out["message"].(string); !strings.Contains(msg, "Hugin") || !strings.Contains(msg, "quarantined") {
		t.Fatalf("the 403 must name the crow and the quarantine, got %q", msg)
	}
}

func TestSentryBelowBandSilent(t *testing.T) {
	ts, s := newWatchServer(t, watchScores(0.4, 0.4, 0.9), nil) // nothing reaches the 0.7 band
	id := newNest(t, ts, "Lab", `"Hugin","Munin"`)
	_, req := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"owner","kind":"request","text":"count the eggs"}`)
	postDeliver(t, ts, id, "Hugin", req["msg_id"].(string), "I think there are about a dozen")
	time.Sleep(400 * time.Millisecond)

	if n := countSentryFlags(ts, id); n != 0 {
		t.Fatalf("below-band verdicts must stay silent, got %d flags", n)
	}
	if s.p.Ledger.IsQuarantined("Hugin") {
		t.Fatal("a below-band deliver must not quarantine the crow")
	}
	code, out := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"Hugin","kind":"claim","refers_to":"`+req["msg_id"].(string)+`","text":"mine"}`)
	if code == http.StatusForbidden {
		t.Fatalf("an unflagged crow must not be silenced, got 403 %v", out)
	}
	if code != http.StatusCreated {
		t.Fatalf("claim should post, got %d %v", code, out)
	}
	time.Sleep(400 * time.Millisecond)
	if n := countSentryFlags(ts, id); n != 0 {
		t.Fatalf("the claim path stays below band too, got %d flags", n)
	}
}

func TestSentryOneFlagPerExpectation(t *testing.T) {
	ts, s := newWatchServer(t, watchScores(0.9, 0.05, 0.05), nil)
	id := newNest(t, ts, "Lab", `"Hugin","Munin"`)
	_, req := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"owner","kind":"request","text":"count the eggs"}`)
	d := postDeliver(t, ts, id, "Hugin", req["msg_id"].(string), "twelve eggs, sourced nowhere")
	if waitForFlag(ts, id, d) == nil {
		t.Fatal("setup: first fabrication must flag")
	}

	// The quarantined crow's next POST is turned away at the door (403),
	// so it cannot generate a second flag through the API.
	code, _ := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"Hugin","kind":"deliver","refers_to":"`+req["msg_id"].(string)+`","text":"also twelve"}`)
	if code != http.StatusForbidden {
		t.Fatalf("quarantined re-post must get 403, got %d", code)
	}

	// A later message already in the log (posted before quarantine, e.g. by
	// the harness path) must reuse the one flag: later breaches are logged,
	// not posted.
	m2, err := s.msgs.Append(Message{NestID: id, Kind: "deliver", Author: "Hugin", To: "owner",
		Text: "twelve again", RefersTo: req["msg_id"].(string), Timestamp: time.Now().UTC().Format(time.RFC3339)})
	if err != nil {
		t.Fatal(err)
	}
	s.checkWatchdog(nestOf(s, id), m2)
	if n := countSentryFlags(ts, id); n != 1 {
		t.Fatalf("at most one flag per crow per expectation, got %d", n)
	}
}

func TestSentryNeverSeesOwner(t *testing.T) {
	ts, s := newWatchServer(t, watchScores(0.95, 0.9, 0.05), nil)
	id := newNest(t, ts, "Lab", `"Hugin","Munin"`)
	_, req := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"owner","kind":"request","text":"count the eggs"}`)
	code, out := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"owner","kind":"deliver","refers_to":"`+req["msg_id"].(string)+`","text":"twelve eggs, from last spring's ledger"}`)
	if code != http.StatusCreated {
		t.Fatalf("owner post: %d %v", code, out)
	}
	time.Sleep(400 * time.Millisecond)
	if n := countSentryFlags(ts, id); n != 0 {
		t.Fatalf("the Sentry never inspects owner messages, got %d flags: %+v", n, s.msgs.All(id))
	}
}

func TestQuarantineSilencesUnprompted(t *testing.T) {
	rb := &recordBrain{}
	ts, s := newWatchServer(t, watchScores(0.05, 0.05, 0.95), rb)
	id := newNest(t, ts, "Lab", `"Hugin","Munin"`)
	promote(s, "Hugin", "Munin") // both may take open work
	s.p.Ledger.Quarantine("Hugin")
	if err := s.p.Ledger.Save(); err != nil {
		t.Fatal(err)
	}

	post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"owner","kind":"request","text":"tidy the branch notes"}`)
	awaitMessages(t, ts, id, 2)

	rb.mu.Lock()
	defer rb.mu.Unlock()
	for _, pr := range rb.pairs {
		if strings.Contains(pr[0], "You are Hugin,") {
			t.Fatalf("a quarantined crow must not be summoned for open work: %q", pr[0])
		}
	}
	invited := false
	for _, pr := range rb.pairs {
		if strings.Contains(pr[0], "You are Munin,") {
			invited = true
		}
	}
	if !invited {
		t.Fatalf("the unquarantined regular should still be invited, calls: %v", rb.pairs)
	}
}

func TestWatchdogGoesThroughFlock(t *testing.T) {
	ts, s := newWatchServer(t, watchScores(0.95, 0.9, 0.05), nil)
	id := newNest(t, ts, "Lab", `"Hugin","Munin"`)
	_, req := post(t, ts.URL+"/api/nests/"+id+"/messages",
		`{"author":"owner","kind":"request","text":"count the eggs"}`)

	// A saturated flock must block the verdict: it goes through Flock.reserve
	// (a direct client call would still flag). Adapted from the literal
	// criterion (zero-capacity sem alone): the POST path itself tags through
	// Flock.ClassifyKind, so with a live ctx a saturated flock would hang the
	// POST forever; ending the server ctx makes every wrapper call fail fast
	// (header fallback) while proving the watchdog can never obtain a slot.
	s.p.Flock.sem = make(chan struct{}, 0)
	canceled, cancel := context.WithCancel(context.Background())
	cancel()
	s.ctx = canceled

	d := postDeliver(t, ts, id, "Hugin", req["msg_id"].(string), "twelve eggs, unsourced") // 201: the message is unaffected
	time.Sleep(400 * time.Millisecond)
	if n := countSentryFlags(ts, id); n != 0 {
		t.Fatalf("a saturated flock must not flag (watchdog must go through Flock), got %d flags: %+v", n, s.msgs.All(id))
	}
	_ = d
}