diff --git a/flightlog.go b/flightlog.go index fb71a0c..01a6033 100644 --- a/flightlog.go +++ b/flightlog.go @@ -24,6 +24,7 @@ type Event struct { Output string `json:"output,omitempty"` Error string `json:"error,omitempty"` Note string `json:"note,omitempty"` + Prob any `json:"probs,omitempty"` // question probabilities, when a call returns them (calibration data) } type FlightLog struct { diff --git a/hub.go b/hub.go index dae1755..1830b90 100644 --- a/hub.go +++ b/hub.go @@ -116,6 +116,22 @@ func (h *Hub) Publish(listID, name string, fields map[string]any) { } } +// OpenNest lazily opens a stream for a nest's conversation. Nest ids are +// slugs and list ids are hex, so the same map serves both without collision. +func (h *Hub) OpenNest(nestID string) { + h.mu.Lock() + defer h.mu.Unlock() + if h.lists[nestID] == nil { + h.lists[nestID] = &listStream{created: time.Now(), subs: map[chan sseMsg]struct{}{}} + } +} + +// PublishNest records one nest event and sends it to subscribers. +func (h *Hub) PublishNest(nestID, name string, fields map[string]any) { + h.OpenNest(nestID) + h.Publish(nestID, name, fields) +} + // Subscribe returns the history after lastID and a channel for what comes next. // The channel closes if the subscriber falls too far behind. func (h *Hub) Subscribe(listID string, lastID int) ([]sseMsg, chan sseMsg, bool) { diff --git a/jev.go b/jev.go index fe6f7cb..fef8523 100644 --- a/jev.go +++ b/jev.go @@ -159,6 +159,68 @@ func (c JevCall) confident(threshold float64) bool { return c.ModeConf >= threshold && c.TaskTypeConf >= threshold } +// kindQuestions tags a nest message with its kind (A1: Jev first). The escape +// option "say" keeps Jev from forcing a wrong label: any message it isn't sure +// about falls back to the safest kind. +var kindQuestions = map[string]any{ + "kind": map[string]any{ + "type": "choice", + "instructions": "What kind of chatroom message is this?", + "criteria": map[string]string{ + "request": "Asks a member or the room for work.", + "claim": "Takes ownership of a request.", + "deliver": "Posts the result of a request.", + "refer": "Points to a file or artifact instead of pasting it.", + "say": "Ordinary talk, or unclear.", + }, + }, +} + +// ClassifyKind returns the kind when Jev is set and its choice sits at or +// above the confidence band, with the choice probabilities for the log. Not ok +// means the caller falls back to the poster's header (EVENTS.md tagging rule). +func (j *JevClient) ClassifyKind(ctx context.Context, text string) (string, map[string]float64, bool) { + if j == nil { + return "", nil, false + } + answers, _, _, err := j.SystemOne(ctx, text, kindQuestions) + if err != nil { + return "", nil, false + } + a := answers["kind"] + if a.Choice == "" || a.Confidence == nil || *a.Confidence < j.threshold { + return "", nil, false + } + return a.Choice, a.Probabilities, true +} + +// ClassifyKind tags a nest message under the same guardrails as any call: +// flock cap, timeout and flight log, with the probabilities recorded. +func (f *Flock) ClassifyKind(ctx context.Context, text string, budgets ...*Budget) (string, map[string]float64, bool) { + if f.jev == nil { + return "", nil, false + } + for _, b := range budgets { + if err := b.Check(); err != nil { + return "", nil, false + } + } + f.sem <- struct{}{} + defer func() { <-f.sem }() + cctx, cancel := context.WithTimeout(ctx, f.g.CrowTimeout) + defer cancel() + start := time.Now() + kind, probs, ok := f.jev.ClassifyKind(cctx, text) + e := Event{Kind: "tag", Item: text, Crow: "Sentry", Model: f.jev.model, Ms: time.Since(start).Milliseconds(), Prob: probs, Note: kind} + if !ok { + e.Note = "" // unsure: the header decides + } + if f.log != nil { + f.log.Write(e) + } + return kind, probs, ok +} + func (c JevCall) summary() string { return fmt.Sprintf("Jev: %s %.0f%%, %s %.0f%%", c.Mode, c.ModeConf*100, c.TaskType, c.TaskTypeConf*100) } diff --git a/nestapi.go b/nestapi.go new file mode 100644 index 0000000..07658cd --- /dev/null +++ b/nestapi.go @@ -0,0 +1,184 @@ +package main + +import ( + "fmt" + "net/http" + "slices" + "strings" + "time" + "unicode/utf8" +) + +var crowPostable = map[string]bool{ + "say": true, + "request": true, + "claim": true, + "deliver": true, + "refer": true, +} + +func (s *Server) getNestMessages(w http.ResponseWriter, r *http.Request) { + nestID := r.PathValue("nest_id") + if _, found := s.nests.Get(nestID); !found { + httpError(w, http.StatusNotFound, fmt.Sprintf("There is no nest called %q.", nestID)) + return + } + all := s.msgs.All(nestID) + out := make([]map[string]any, 0, len(all)) + for _, m := range all { + out = append(out, map[string]any{ + "msg_id": m.MsgID(), + "kind": m.Kind, + "author": m.Author, + "text": m.Text, + "to": m.To, + "refers_to": m.RefersTo, + "tagged_by": m.TaggedBy, + "ts": m.Timestamp, + }) + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) postNestMessage(w http.ResponseWriter, r *http.Request) { + nestID := r.PathValue("nest_id") + nest, found := s.nests.Get(nestID) + if !found { + httpError(w, http.StatusNotFound, fmt.Sprintf("There is no nest called %q.", nestID)) + return + } + + var body struct { + Author string `json:"author"` + Text string `json:"text"` + Kind string `json:"kind"` + To string `json:"to"` + RefersTo string `json:"refers_to"` + } + if !decode(w, r, &body) { + return + } + + body.Text = strings.TrimSpace(body.Text) + body.To = strings.TrimSpace(body.To) + + // Guardrails (EVENTS.md: Nest messages → Guardrails) + if n := utf8.RuneCountInString(body.Text); n < 1 || n > s.p.G.MaxItemChars { + httpError(w, http.StatusBadRequest, fmt.Sprintf("A message is 1 to %d characters.", s.p.G.MaxItemChars)) + return + } + if body.Author == "owner" { + // owner can always post + } else if !slices.Contains(nest.Crows, body.Author) { + httpError(w, http.StatusBadRequest, fmt.Sprintf("%q is not a member of %s.", body.Author, nestID)) + return + } + if body.Author != "owner" { + if ok, msg, wait := s.p.Ledger.Admit(body.Author, time.Now(), s.p.G); !ok { + cooldown := int(wait.Seconds()) + writeJSON(w, http.StatusTooManyRequests, map[string]any{"message": msg, "cooldown_s": cooldown}) + return + } + } + if body.Kind != "" && body.Author != "owner" && !crowPostable[body.Kind] { + httpError(w, http.StatusBadRequest, fmt.Sprintf("%q is not a kind crows may post.", body.Kind)) + return + } + + // Tagging (A1, EVENTS.md): Jev first, header fallback, say as last resort. + kind, taggedBy := s.tagKind(body.Kind, body.Text) + + msg := Message{ + NestID: nestID, + Kind: kind, + Author: body.Author, + Text: body.Text, + To: body.To, + RefersTo: body.RefersTo, + TaggedBy: taggedBy, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + record, err := s.msgs.Append(msg) + if err != nil { + httpError(w, http.StatusInternalServerError, "could not save the message: "+err.Error()) + return + } + + s.emitNestMessage(record) + + // Event-driven trigger: if kind is request and To is a crow, respond. + if kind == "request" && record.To != "" && record.To != "owner" { + go s.crowResponds(nest, record) + } + + writeJSON(w, http.StatusCreated, map[string]any{ + "msg_id": record.MsgID(), + "kind": record.Kind, + "author": record.Author, + "text": record.Text, + "to": record.To, + "refers_to": record.RefersTo, + "tagged_by": record.TaggedBy, + "ts": record.Timestamp, + }) +} + +// tagKind runs the A1 rule: Jev first; Jev unsure -> the poster's header; +// neither -> say. tagged_by records which source won, and a Jev override of a +// header logs as a disagreement (fuel for calibration). +func (s *Server) tagKind(header, text string) (string, string) { + kind, _, ok := s.p.Flock.ClassifyKind(s.ctx, text) + if ok { + if crowPostable[header] && header != kind { + return kind, "disagreement" + } + return kind, "jev" + } + if crowPostable[header] { + return header, "header" + } + return "say", "default" +} + +func (s *Server) emitNestMessage(m Message) { + s.hub.PublishNest(m.NestID, "nest.message", map[string]any{ + "nest_id": m.NestID, + "msg_id": m.MsgID(), + "kind": m.Kind, + "author": m.Author, + "text": m.Text, + "to": m.To, + "refers_to": m.RefersTo, + "tagged_by": m.TaggedBy, + }) +} + +func (s *Server) crowResponds(nest Nest, req Message) { + crows := s.p.Flock.pick(nest, 1) + if len(crows) == 0 { + return + } + crow := crows[0] // B4 (@-addressing) makes this the addressed crow + system := systemFor(crow, crowSystem, true) + prompt := "NEST: " + nest.Name + "\nREQUEST FROM: " + req.Author + "\nTEXT: " + req.Text + budget := &Budget{name: "nest-message", limit: s.p.G.TokenBudgetPerItem} + answer, err := s.p.Flock.ask(s.ctx, nil, crow, system, prompt, budget) + if err != nil { + return + } + reply := Message{ + NestID: nest.ID, + Kind: "say", + Author: crow, + Text: strings.TrimSpace(answer), + To: req.Author, + RefersTo: req.MsgID(), + TaggedBy: "harness", + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + record, err := s.msgs.Append(reply) + if err != nil { + return + } + s.emitNestMessage(record) +} diff --git a/nestapi_test.go b/nestapi_test.go new file mode 100644 index 0000000..acef3f4 --- /dev/null +++ b/nestapi_test.go @@ -0,0 +1,126 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" +) + +// newTestNestServer is newTestServer plus a message store. +func newTestNestServer(t *testing.T) (*httptest.Server, *Server) { + t.Helper() + ts, s := newTestServer(t, DefaultGuardrails(), time.Minute) + ms, err := LoadMessageStore(filepath.Join(t.TempDir(), "messages")) + if err != nil { + t.Fatal(err) + } + s.WithMessages(ms) + return ts, s +} + +func newNest(t *testing.T, ts *httptest.Server, name, crows string) string { + t.Helper() + code, out := post(t, ts.URL+"/api/nests", `{"name":"`+name+`","crows":[`+crows+`]}`) + if code != http.StatusCreated { + t.Fatalf("create nest: %d %v", code, out) + } + return out["nest_id"].(string) +} + +func nestMessages(t *testing.T, ts *httptest.Server, nestID string) []map[string]any { + t.Helper() + resp, err := http.Get(ts.URL + "/api/nests/" + nestID + "/messages") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("get messages: %d", resp.StatusCode) + } + var msgs []map[string]any + if err := json.NewDecoder(resp.Body).Decode(&msgs); err != nil { + t.Fatal(err) + } + return msgs +} + +// awaitMessages polls until the nest has at least n messages or the deadline passes. +func awaitMessages(t *testing.T, ts *httptest.Server, nestID string, n int) []map[string]any { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for { + msgs := nestMessages(t, ts, nestID) + if len(msgs) >= n || time.Now().After(deadline) { + return msgs + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestPostMessageToNest(t *testing.T) { + ts, _ := newTestNestServer(t) + id := newNest(t, ts, "Lab", `"Hugin"`) + code, msg := post(t, ts.URL+"/api/nests/"+id+"/messages", `{"author":"owner","text":"hello, flock"}`) + if code != http.StatusCreated { + t.Fatalf("want 201, got %d %v", code, msg) + } + if msg["kind"] != "say" || msg["author"] != "owner" || msg["text"] != "hello, flock" { + t.Fatalf("bad record: %v", msg) + } + if msg["msg_id"] != id+"-1" { + t.Fatalf("want msg_id %s-1, got %v", id, msg["msg_id"]) + } + got := nestMessages(t, ts, id) + if len(got) != 1 || got[0]["text"] != "hello, flock" { + t.Fatalf("log missing the message: %v", got) + } +} + +func TestPostMessageUnknownNest(t *testing.T) { + ts, _ := newTestNestServer(t) + code, out := post(t, ts.URL+"/api/nests/nope/messages", `{"author":"owner","text":"hi"}`) + if code != http.StatusNotFound || out["message"] == nil { + t.Fatalf("want 404 with message, got %d %v", code, out) + } +} + +func TestPostMessageTriggersOneResponse(t *testing.T) { + ts, _ := newTestNestServer(t) + id := newNest(t, ts, "Lab", `"Hugin"`) + code, req := post(t, ts.URL+"/api/nests/"+id+"/messages", `{"author":"owner","kind":"request","text":"count the eggs","to":"Hugin"}`) + if code != http.StatusCreated { + t.Fatalf("request post: %d %v", code, req) + } + msgs := awaitMessages(t, ts, id, 2) + replies := 0 + for _, m := range msgs { + if m["author"] == "Hugin" { + replies++ + if m["kind"] != "say" && m["kind"] != "claim" && m["kind"] != "deliver" { + t.Fatalf("crow reply has a bad kind: %v", m["kind"]) + } + if m["refers_to"] != req["msg_id"] { + t.Fatalf("crow reply must point at the request, got %v", m["refers_to"]) + } + } + } + if replies != 1 { + t.Fatalf("want exactly one Hugin message, got %d in %v", replies, msgs) + } +} + +func TestNoIdleResponse(t *testing.T) { + ts, _ := newTestNestServer(t) + id := newNest(t, ts, "Lab", `"Hugin"`) + code, out := post(t, ts.URL+"/api/nests/"+id+"/messages", `{"author":"owner","text":"just thinking out loud"}`) + if code != http.StatusCreated { + t.Fatalf("say post: %d %v", code, out) + } + time.Sleep(300 * time.Millisecond) + if got := nestMessages(t, ts, id); len(got) != 1 { + t.Fatalf("an unaddressed say must not summon a crow, log: %v", got) + } +} diff --git a/nestsse_test.go b/nestsse_test.go new file mode 100644 index 0000000..5197306 --- /dev/null +++ b/nestsse_test.go @@ -0,0 +1,53 @@ +package main + +import ( + "bufio" + "net/http" + "testing" + "time" +) + +// TestNestStreamSSE checks GET /api/events?nest_id= streams nest.message events +// and replays history from Last-Event-ID (the EVENTS.md nest SSE contract). +func TestNestStreamSSE(t *testing.T) { + ts, s := newTestNestServer(t) + id := newNest(t, ts, "Lab", `"Hugin"`) + s.hub.OpenNest(id) + if _, out := post(t, ts.URL+"/api/nests/"+id+"/messages", `{"author":"owner","text":"hello"}`); out["msg_id"] == nil { + t.Fatalf("post failed: %v", out) + } + + req, err := http.NewRequest("GET", ts.URL+"/api/events?nest_id="+id, nil) + if err != nil { + t.Fatal(err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" { + t.Fatalf("content type %q", ct) + } + + // Read in a goroutine with a deadline so a stuck stream fails, not hangs. + got := make(chan bool, 1) + go func() { + sc := bufio.NewScanner(resp.Body) + for sc.Scan() { + if sc.Text() == "event: nest.message" { + got <- true + return + } + } + got <- false + }() + select { + case ok := <-got: + if !ok { + t.Fatal("stream ended before nest.message appeared") + } + case <-time.After(2 * time.Second): + t.Fatal("no nest.message event within 2s") + } +} diff --git a/server.go b/server.go index a5fe7b3..ed1c4e9 100644 --- a/server.go +++ b/server.go @@ -32,6 +32,7 @@ type Server struct { approver *WebApprover hub *Hub nests *NestStore + msgs *MessageStore ctx context.Context // cancelled on shutdown; lists and streams end with it beat time.Duration } @@ -40,6 +41,9 @@ func NewServer(ctx context.Context, p *Pipeline, approver *WebApprover, nests *N return &Server{p: p, approver: approver, hub: NewHub(), nests: nests, ctx: ctx, beat: heartbeat} } +// WithMessages attaches the nest conversation log (chatroom path). +func (s *Server) WithMessages(m *MessageStore) *Server { s.msgs = m; return s } + func (s *Server) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("POST /api/lists", s.postList) @@ -49,6 +53,8 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("GET /api/crows", s.crows) mux.HandleFunc("GET /api/nests", s.getNests) mux.HandleFunc("POST /api/nests", s.postNest) + mux.HandleFunc("GET /api/nests/{nest_id}/messages", s.getNestMessages) + mux.HandleFunc("POST /api/nests/{nest_id}/messages", s.postNestMessage) if st, err := os.Stat(webDir); err == nil && st.IsDir() { mux.Handle("GET /", http.FileServer(http.Dir(webDir))) } else { @@ -73,6 +79,8 @@ func cmdServe(args []string) { defer flog.Close() nests, err := LoadNests(filepath.Join(dataDir, "nests.json")) must(err) + msgs, err := LoadMessageStore(filepath.Join(dataDir, "messages")) + must(err) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) defer stop() @@ -84,7 +92,7 @@ func cmdServe(args []string) { if jev != nil { fmt.Printf("Sentry triage: %s, LLM fallback\n", jev.Name()) } - s := NewServer(ctx, p, approver, nests) + s := NewServer(ctx, p, approver, nests).WithMessages(msgs) srv := &http.Server{Addr: *addr, Handler: s.Handler(), ReadHeaderTimeout: 10 * time.Second} go func() { @@ -190,6 +198,15 @@ func newListID() string { func (s *Server) events(w http.ResponseWriter, r *http.Request) { listID := r.URL.Query().Get("list_id") + nestID := r.URL.Query().Get("nest_id") + if listID == "" && nestID != "" { + if _, found := s.nests.Get(nestID); !found { + httpError(w, http.StatusNotFound, "no such nest_id") + return + } + s.hub.OpenNest(nestID) + listID = nestID + } lastID, _ := strconv.Atoi(r.Header.Get("Last-Event-ID")) past, ch, ok := s.hub.Subscribe(listID, lastID) if !ok {