Merge pull request #17 from jacisjake/task/10-post-nest-messages

(#10) POST nest messages + event-driven response

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 {

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

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

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

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

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

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 {