diff --git a/CHANGELOG.md b/CHANGELOG.md index e0bcbfa..8f87e3b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ All notable changes to Noema are recorded here. Format follows Keep a Changelog; ## [Unreleased] ### Added +- Phase 3 events: append-only cognitive event store with provenance links, filtered queries, durable cursors, live SSE streams with resume, forensic event explorer UI, thought stream page, operator observations and CLI. - Phase 2 minds: structured personalities with append-only version history, diff, restore, templates with preserved overrides, self-model, mind lifecycle (create, clone, disable, archive), API, UI editor and CLI. - Phase 1 foundation: configuration, structured logging, metrics, migrations, local authentication (Argon2id, sessions, CSRF, lockout, roles, API tokens), audit log, JSON API skeleton, server-rendered admin UI with light/dark theme, `noemad` and `noemactl`. - Repository foundation: plan, architecture, security model, threat model, ADRs, Makefile, container files. diff --git a/PLAN.md b/PLAN.md index b375c29..022090b 100644 --- a/PLAN.md +++ b/PLAN.md @@ -60,14 +60,14 @@ An administrator can: start Noema → log in → create a mind → configure per - ☑ Audit events for every change; docs/personality.md - ☑ Tests: brief determinism, diff/override round trip, normalisation, store lifecycle including append-only history -### Phase 3 – Immutable cognitive event store -- ☐ `pkg/events`: types, categories, evidence refs, payload contracts -- ☐ `migrations/0003_events`: cognitive_events (append-only trigger, indexes), mind_cursors -- ☐ `internal/events`: store (append single/batch in tx), query (cursor, filters), bus (bounded fan-out), SSE endpoint with Last-Event-ID resume -- ☐ API: list/inspect events, causal navigation (cause, parents, children, evidence) -- ☐ UI: event explorer (forensic view, filters, navigation), live thought stream page -- ☐ `noemactl event inspect` -- ☐ Tests: immutability trigger, ordering, cursor resume, bus backpressure, filters +### Phase 3 – Immutable cognitive event store ☑ +- ☑ `pkg/events`: closed type vocabulary with categories, evidence refs, observation payload with trust levels and credibility +- ☑ `migrations/0003_events`: cognitive_events (append-only trigger, seq, indexes on mind/type/correlation/causation/parents/evidence), mind_cursors +- ☑ `internal/events`: validating append (single tx or caller tx + publish after commit), filtered queries with seq cursors, causal navigation (cause, parents, children, cited, derived, correlated), bounded fan-out bus, durable cursors, stats, operator-action helper +- ☑ API: list/filter, stats, get, forensic context, SSE stream (per mind and global) with Last-Event-ID replay, filter matching and heartbeats, operator observation injection +- ☑ UI: live thought stream with filters/paging, forensic event page (causal chain, evidence, derived, correlation, payload), dashboard important events, mind overview activity + observe form; operator changes recorded as `operator_action` events +- ☑ `noemactl event inspect|list|tail`, `noemactl observe` +- ☑ Tests: bus bounds and filtering, append validation, immutability, filters, cursors, causal lookups, SSE end to end ### Phase 4 – Working memory, global workspace, attention - ☐ `internal/brain`: `BrainState`, `Process` interface, registry, cycle loop, bounded pool, per-mind runtime lifecycle, modes, cursor advance, loop detector diff --git a/cmd/noemactl/events.go b/cmd/noemactl/events.go new file mode 100644 index 0000000..0624e8d --- /dev/null +++ b/cmd/noemactl/events.go @@ -0,0 +1,108 @@ +package main + +import ( + "bufio" + "encoding/json" + "flag" + "fmt" + "net/http" + "os" + "strings" + "time" +) + +func init() { + extraCommands = append(extraCommands, + command{"event", "inspect |list [-type t] [-n 50]|tail – cognitive events", cmdEvent}, + command{"observe", " – inject an operator observation", cmdObserve}, + ) +} + +func printEvent(ev map[string]any) { + fmt.Printf("%s %-22s %-12s %s [sal %.2f conf %.2f]\n", fmt.Sprint(ev["occurred_at"])[11:19], ev["type"], ev["source"], ev["summary"], num(ev["salience"]), num(ev["confidence"])) +} + +func num(v any) float64 { f, _ := v.(float64); return f } + +func cmdEvent(args []string) int { + if len(args) < 2 { + fmt.Fprintln(os.Stderr, "usage: noemactl event inspect | list [-type t] [-category c] [-n 50] [-json] | tail ") + return exitUsage + } + switch args[0] { + case "inspect": + var ctx map[string]any + if code, err := apiRequest("GET", "/api/v1/events/"+args[1]+"/context", nil, &ctx); err != nil { + return exitFor(code, err) + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + enc.Encode(ctx) //nolint:errcheck + return exitOK + case "list": + fs := flag.NewFlagSet("event list", flag.ContinueOnError) + typ := fs.String("type", "", "event type") + cat := fs.String("category", "", "category") + n := fs.Int("n", 50, "count") + asJSON := fs.Bool("json", false, "raw JSON") + if fs.Parse(args[2:]) != nil { + return exitUsage + } + var out struct{ Items []map[string]any } + q := fmt.Sprintf("?limit=%d&type=%s&category=%s", *n, *typ, *cat) + if code, err := apiRequest("GET", "/api/v1/minds/"+args[1]+"/events"+q, nil, &out); err != nil { + return exitFor(code, err) + } + if *asJSON { + json.NewEncoder(os.Stdout).Encode(out.Items) //nolint:errcheck + return exitOK + } + for i := len(out.Items) - 1; i >= 0; i-- { + printEvent(out.Items[i]) + } + return exitOK + case "tail": + base := os.Getenv("NOEMA_URL") + if base == "" { + base = "http://127.0.0.1:8080" + } + req, _ := http.NewRequest("GET", strings.TrimRight(base, "/")+"/api/v1/minds/"+args[1]+"/events/stream", nil) + req.Header.Set("Authorization", "Bearer "+os.Getenv("NOEMA_API_TOKEN")) + res, err := (&http.Client{Timeout: 0}).Do(req) + if err != nil { + fmt.Fprintln(os.Stderr, "error:", err) + return exitFail + } + defer res.Body.Close() + if res.StatusCode != 200 { + return exitFor(res.StatusCode, fmt.Errorf("%s", res.Status)) + } + sc := bufio.NewScanner(res.Body) + sc.Buffer(make([]byte, 1<<20), 1<<20) + for sc.Scan() { + line := sc.Text() + if strings.HasPrefix(line, "data: ") { + var ev map[string]any + if json.Unmarshal([]byte(line[6:]), &ev) == nil { + printEvent(ev) + } + } + } + return exitOK + } + fmt.Fprintln(os.Stderr, "unknown event subcommand", args[0]) + return exitUsage +} + +func cmdObserve(args []string) int { + if len(args) < 2 { + fmt.Fprintln(os.Stderr, "usage: noemactl observe ") + return exitUsage + } + var ev map[string]any + if code, err := apiRequest("POST", "/api/v1/minds/"+args[0]+"/observations", map[string]any{"text": strings.Join(args[1:], " "), "source": "noemactl"}, &ev); err != nil { + return exitFor(code, err) + } + fmt.Printf("recorded observation %v (seq %v) at %v\n", ev["id"], ev["seq"], time.Now().Format(time.RFC3339)) + return exitOK +} diff --git a/cmd/noemad/main.go b/cmd/noemad/main.go index 48a9668..287057e 100644 --- a/cmd/noemad/main.go +++ b/cmd/noemad/main.go @@ -18,6 +18,7 @@ import ( "github.com/andydixon/noema/internal/audit" "github.com/andydixon/noema/internal/auth" "github.com/andydixon/noema/internal/config" + evstore "github.com/andydixon/noema/internal/events" "github.com/andydixon/noema/internal/minds" "github.com/andydixon/noema/internal/observability" "github.com/andydixon/noema/internal/scheduler" @@ -25,6 +26,7 @@ import ( "github.com/andydixon/noema/internal/ui" "github.com/andydixon/noema/internal/version" "github.com/andydixon/noema/migrations" + "github.com/andydixon/noema/pkg/ids" ) func main() { @@ -74,14 +76,15 @@ func run() error { metrics := observability.NewMetrics() authSvc := auth.NewService(db, cfg.SessionIdleTTL, cfg.SessionAbsoluteTTL) auditLog := audit.New(db, log) - deps := app.Deps{Config: cfg, Log: log, DB: db, Auth: authSvc, Audit: auditLog, Metrics: metrics, Minds: minds.NewStore(db), Migrations: migrations.FS, Started: time.Now()} + deps := app.Deps{Config: cfg, Log: log, DB: db, Auth: authSvc, Audit: auditLog, Metrics: metrics, Minds: minds.NewStore(db), Events: evstore.NewStore(db, evstore.NewBus(200), metrics), Migrations: migrations.FS, Started: time.Now()} if err := bootstrapAdmin(ctx, deps); err != nil { return err } var srv *api.Server - uiHandler, err := ui.New(deps, func(r *http.Request) any { return srv.Status(r.Context()) }) + uiHandler, err := ui.New(deps, func(r *http.Request) any { return srv.Status(r.Context()) }, + func(r *http.Request, id ids.ID) (api.EventContext, error) { return srv.Context(r, id) }) if err != nil { return err } diff --git a/internal/api/api_test.go b/internal/api/api_test.go index 61f8524..1179071 100644 --- a/internal/api/api_test.go +++ b/internal/api/api_test.go @@ -14,6 +14,7 @@ import ( "github.com/andydixon/noema/internal/audit" "github.com/andydixon/noema/internal/auth" "github.com/andydixon/noema/internal/config" + evstore "github.com/andydixon/noema/internal/events" "github.com/andydixon/noema/internal/minds" "github.com/andydixon/noema/internal/observability" "github.com/andydixon/noema/internal/storage" @@ -27,7 +28,7 @@ func testServer(t *testing.T) (*httptest.Server, string) { cfg.DatabaseURL = "postgres://test" log := slog.New(slog.DiscardHandler) deps := app.Deps{Config: cfg, Log: log, DB: pool, Auth: auth.NewService(pool, time.Hour, 2*time.Hour), - Audit: audit.New(pool, log), Metrics: observability.NewMetrics(), Minds: minds.NewStore(pool), Migrations: migrations.FS, Started: time.Now()} + Audit: audit.New(pool, log), Metrics: observability.NewMetrics(), Minds: minds.NewStore(pool), Events: evstore.NewStore(pool, evstore.NewBus(50), nil), Migrations: migrations.FS, Started: time.Now()} ctx := context.Background() u, err := deps.Auth.CreateUser(ctx, "admin", "administrator-pass", auth.RoleAdministrator) if err != nil { diff --git a/internal/api/events.go b/internal/api/events.go new file mode 100644 index 0000000..6929d8f --- /dev/null +++ b/internal/api/events.go @@ -0,0 +1,335 @@ +package api + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/andydixon/noema/internal/auth" + evstore "github.com/andydixon/noema/internal/events" + "github.com/andydixon/noema/pkg/events" + "github.com/andydixon/noema/pkg/ids" +) + +func (s *Server) eventRoutes() { + m := s.mux + m.Handle("GET /api/v1/minds/{mind}/events", require(auth.RoleObserver, s.eventsList)) + m.Handle("GET /api/v1/minds/{mind}/events/stream", require(auth.RoleObserver, s.eventsStream)) + m.Handle("GET /api/v1/minds/{mind}/events/stats", require(auth.RoleObserver, s.eventsStats)) + m.Handle("POST /api/v1/minds/{mind}/observations", require(auth.RoleOperator, s.observationCreate)) + m.Handle("GET /api/v1/events/stream", require(auth.RoleObserver, s.eventsStream)) + m.Handle("GET /api/v1/events/{id}", require(auth.RoleObserver, s.eventGet)) + m.Handle("GET /api/v1/events/{id}/context", require(auth.RoleObserver, s.eventContext)) + m.Handle("GET /api/v1/events/types", require(auth.RoleObserver, func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]any{"types": events.AllTypes()}) + })) +} + +// ParseEventFilter reads the shared query-string filter vocabulary used by +// the API, SSE stream and UI pages. +func ParseEventFilter(q url.Values) evstore.Filter { + f := evstore.Filter{} + for _, t := range q["type"] { + for _, part := range strings.Split(t, ",") { + if part = strings.TrimSpace(part); part != "" { + f.Types = append(f.Types, events.Type(part)) + } + } + } + for _, c := range q["category"] { + for _, part := range strings.Split(c, ",") { + if part = strings.TrimSpace(part); part != "" { + f.Categories = append(f.Categories, events.Category(part)) + } + } + } + for _, src := range q["source"] { + if src = strings.TrimSpace(src); src != "" { + f.Sources = append(f.Sources, src) + } + } + f.MinSalience, _ = strconv.ParseFloat(q.Get("min_salience"), 64) + f.MinConfidence, _ = strconv.ParseFloat(q.Get("min_confidence"), 64) + f.MaxConfidence, _ = strconv.ParseFloat(q.Get("max_confidence"), 64) + if v := q.Get("since"); v != "" { + if t, err := time.Parse(time.RFC3339, v); err == nil { + f.Since = t + } else if d, err := time.ParseDuration(v); err == nil { + f.Since = time.Now().Add(-d) + } + } + if v := q.Get("until"); v != "" { + if t, err := time.Parse(time.RFC3339, v); err == nil { + f.Until = t + } + } + f.AfterSeq, _ = strconv.ParseInt(q.Get("after"), 10, 64) + f.BeforeSeq, _ = strconv.ParseInt(q.Get("before"), 10, 64) + if v := q.Get("correlation"); v != "" { + if id, err := ids.Parse(v); err == nil { + f.CorrelationID = id + } + } + f.Search = q.Get("q") + f.Limit, _ = strconv.Atoi(q.Get("limit")) + f.Ascending = q.Get("order") == "asc" + return f +} + +func (s *Server) eventsList(w http.ResponseWriter, r *http.Request) { + m, ok := s.mind(w, r) + if !ok { + return + } + f := ParseEventFilter(r.URL.Query()) + f.MindID = m.ID + list, err := s.deps.Events.Query(r.Context(), f) + if err != nil { + fail(w, r, err) + return + } + if list == nil { + list = []events.Event{} + } + out := page[events.Event]{Items: list} + if len(list) > 0 && (f.Limit == 0 && len(list) == 100 || f.Limit > 0 && len(list) == f.Limit) { + out.NextCursor = strconv.FormatInt(list[len(list)-1].Seq, 10) + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) eventsStats(w http.ResponseWriter, r *http.Request) { + m, ok := s.mind(w, r) + if !ok { + return + } + var since time.Time + if d, err := time.ParseDuration(r.URL.Query().Get("window")); err == nil { + since = time.Now().Add(-d) + } + st, err := s.deps.Events.Stats(r.Context(), m.ID, since) + if err != nil { + fail(w, r, err) + return + } + writeJSON(w, http.StatusOK, st) +} + +func (s *Server) eventGet(w http.ResponseWriter, r *http.Request) { + id, err := ids.Parse(r.PathValue("id")) + if err != nil { + fail(w, r, ValidationError("bad id")) + return + } + ev, err := s.deps.Events.Get(r.Context(), id) + if err != nil { + fail(w, r, err) + return + } + writeJSON(w, http.StatusOK, ev) +} + +// EventContext is the forensic view: the event, what caused it, what it +// caused, what it cites and what cites it. +type EventContext struct { + Event events.Event `json:"event"` + Cause *events.Event `json:"cause,omitempty"` + Parents []events.Event `json:"parents"` + Children []events.Event `json:"children"` + Evidence []events.Event `json:"evidence_events"` + Derived []events.Event `json:"derived"` + Correlated []events.Event `json:"correlated"` + EvidenceAll []events.EvidenceRef `json:"evidence"` +} + +// Context assembles the forensic context for an event; shared with the UI. +func (s *Server) Context(r *http.Request, id ids.ID) (EventContext, error) { + ctx := r.Context() + ev, err := s.deps.Events.Get(ctx, id) + if err != nil { + return EventContext{}, err + } + out := EventContext{Event: ev, Parents: []events.Event{}, Children: []events.Event{}, Evidence: []events.Event{}, Derived: []events.Event{}, Correlated: []events.Event{}, EvidenceAll: ev.Evidence} + if !ev.CausationID.IsZero() { + if c, err := s.deps.Events.Get(ctx, ev.CausationID); err == nil { + out.Cause = &c + } + } + if len(ev.ParentIDs) > 0 { + out.Parents, _ = s.deps.Events.GetMany(ctx, ev.ParentIDs) + } + out.Children, _ = s.deps.Events.Children(ctx, ev.ID, 50) + var evIDs []ids.ID + for _, ref := range ev.Evidence { + if ref.Kind == "event" { + if eid, err := ids.Parse(ref.ID); err == nil { + evIDs = append(evIDs, eid) + } + } + } + if len(evIDs) > 0 { + out.Evidence, _ = s.deps.Events.GetMany(ctx, evIDs) + } + out.Derived, _ = s.deps.Events.DerivedFrom(ctx, "event", string(ev.ID), 50) + if !ev.CorrelationID.IsZero() { + out.Correlated, _ = s.deps.Events.Query(ctx, evstore.Filter{MindID: ev.MindID, CorrelationID: ev.CorrelationID, Ascending: true, Limit: 200}) + } + for _, l := range []*[]events.Event{&out.Parents, &out.Children, &out.Evidence, &out.Derived, &out.Correlated} { + if *l == nil { + *l = []events.Event{} + } + } + return out, nil +} + +func (s *Server) eventContext(w http.ResponseWriter, r *http.Request) { + id, err := ids.Parse(r.PathValue("id")) + if err != nil { + fail(w, r, ValidationError("bad id")) + return + } + out, err := s.Context(r, id) + if err != nil { + fail(w, r, err) + return + } + writeJSON(w, http.StatusOK, out) +} + +// observationCreate lets an operator inject an observation directly. It is +// tagged with operator trust and used by demos, tests and the CLI. +func (s *Server) observationCreate(w http.ResponseWriter, r *http.Request) { + m, ok := s.mind(w, r) + if !ok { + return + } + var in struct { + Text string `json:"text"` + Source string `json:"source"` + Reference string `json:"reference"` + Salience float64 `json:"salience"` + } + if err := decode(r, &in); err != nil { + fail(w, r, err) + return + } + in.Text = strings.TrimSpace(in.Text) + if in.Text == "" || len(in.Text) > 20000 { + fail(w, r, ValidationError("text required (max 20000 chars)")) + return + } + if in.Source == "" { + in.Source = "operator" + } + if in.Salience == 0 { + in.Salience = 0.5 + } + summary := in.Text + if len(summary) > 160 { + summary = summary[:157] + "..." + } + evs, err := s.deps.Events.Append(r.Context(), events.New{MindID: m.ID, Type: events.Observation, Source: in.Source, Summary: summary, + Salience: in.Salience, Confidence: 1, Metadata: map[string]any{"trust": events.TrustOperator, "actor": actor(r)}, + Payload: events.ObservationPayload{Text: in.Text, ContentType: "text/plain", Source: in.Source, Trust: events.TrustOperator, Method: "api", Reference: in.Reference, Actor: actor(r)}}) + if err != nil { + fail(w, r, err) + return + } + s.deps.Audit.Log(r.Context(), "observation.inject", "mind", m.ID.String(), map[string]any{"event_id": evs[0].ID, "source": in.Source}) + writeJSON(w, http.StatusCreated, evs[0]) +} + +// eventsStream is the Server-Sent Events feed. It subscribes before replaying +// history after Last-Event-ID so no event is missed, then relays live events +// that match the filter. Heartbeats keep proxies from closing idle streams. +func (s *Server) eventsStream(w http.ResponseWriter, r *http.Request) { + f := ParseEventFilter(r.URL.Query()) + if r.PathValue("mind") != "" { + m, ok := s.mind(w, r) + if !ok { + return + } + f.MindID = m.ID + } + flusher, ok := w.(http.Flusher) + if !ok { + writeError(w, r, http.StatusInternalServerError, "internal", "streaming unsupported") + return + } + ctx := r.Context() + ch, err := s.deps.Events.Bus().Subscribe(ctx, f.MindID, 256) + if err != nil { + var tooMany evstore.ErrTooManySubscribers + if errors.As(err, &tooMany) { + writeError(w, r, http.StatusServiceUnavailable, "busy", "too many live streams") + return + } + fail(w, r, err) + return + } + h := w.Header() + h.Set("Content-Type", "text/event-stream") + h.Set("Cache-Control", "no-cache") + h.Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, "retry: 3000\n\n") + flusher.Flush() + + var lastSeq int64 + if v := r.Header.Get("Last-Event-ID"); v != "" { + lastSeq, _ = strconv.ParseInt(v, 10, 64) + } else if f.AfterSeq > 0 { + lastSeq = f.AfterSeq + } + send := func(ev events.Event) bool { + if ev.Seq <= lastSeq { + return true + } + lastSeq = ev.Seq + data, _ := json.Marshal(ev) + if _, err := fmt.Fprintf(w, "id: %d\nevent: cognitive_event\ndata: %s\n\n", ev.Seq, data); err != nil { + return false + } + flusher.Flush() + return true + } + if lastSeq > 0 { + replay := f + replay.AfterSeq, replay.Ascending, replay.Limit = lastSeq, true, 500 + if hist, err := s.deps.Events.Query(ctx, replay); err == nil { + for _, ev := range hist { + if !send(ev) { + return + } + } + } + } + heartbeat := time.NewTicker(15 * time.Second) + defer heartbeat.Stop() + for { + select { + case <-ctx.Done(): + return + case <-heartbeat.C: + if _, err := fmt.Fprint(w, ": ping\n\n"); err != nil { + return + } + flusher.Flush() + case ev, ok := <-ch: + if !ok { + return + } + if !f.Matches(ev) { + continue + } + if !send(ev) { + return + } + } + } +} diff --git a/internal/api/events_test.go b/internal/api/events_test.go new file mode 100644 index 0000000..4c79cc5 --- /dev/null +++ b/internal/api/events_test.go @@ -0,0 +1,69 @@ +package api + +import ( + "bufio" + "context" + "encoding/json" + "net/http" + "strings" + "testing" + "time" + + "github.com/andydixon/noema/pkg/events" +) + +func TestObservationAndSSEStream(t *testing.T) { + srv, tok := testServer(t) + do := func(method, path, body string) (*http.Response, error) { + req, _ := http.NewRequest(method, srv.URL+path, strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+tok) + req.Header.Set("Content-Type", "application/json") + return http.DefaultClient.Do(req) + } + if res, _ := do("POST", "/api/v1/minds", `{"slug":"sse"}`); res.StatusCode != 201 { + t.Fatalf("create mind: %d", res.StatusCode) + } + // Open the stream first. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, "GET", srv.URL+"/api/v1/minds/sse/events/stream?type=observation", nil) + req.Header.Set("Authorization", "Bearer "+tok) + res, err := http.DefaultClient.Do(req) + if err != nil || res.StatusCode != 200 || !strings.HasPrefix(res.Header.Get("Content-Type"), "text/event-stream") { + t.Fatalf("stream open failed: %v %v", err, res) + } + defer res.Body.Close() + + if r, _ := do("POST", "/api/v1/minds/sse/observations", `{"text":"disk latency rising"}`); r.StatusCode != 201 { + t.Fatalf("observation: %d", r.StatusCode) + } + if r, _ := do("POST", "/api/v1/minds/sse/observations", `{"text":""}`); r.StatusCode != 400 { + t.Fatalf("empty observation should be 400, got %d", r.StatusCode) + } + sc := bufio.NewScanner(res.Body) + var got events.Event + for sc.Scan() { + line := sc.Text() + if strings.HasPrefix(line, "data: ") { + if err := json.Unmarshal([]byte(line[6:]), &got); err != nil { + t.Fatal(err) + } + break + } + } + if got.Type != events.Observation || got.Summary != "disk latency rising" || got.Seq == 0 { + t.Fatalf("unexpected stream event: %+v", got) + } + var payload events.ObservationPayload + json.Unmarshal(got.Payload, &payload) //nolint:errcheck + if payload.Trust != events.TrustOperator || payload.Method != "api" { + t.Fatalf("observation provenance missing: %+v", payload) + } + // Forensic context endpoint. + r, _ := do("GET", "/api/v1/events/"+string(got.ID)+"/context", "") + var c EventContext + json.NewDecoder(r.Body).Decode(&c) //nolint:errcheck + if c.Event.ID != got.ID || c.Children == nil { + t.Fatalf("context wrong: %+v", c) + } +} diff --git a/internal/api/minds.go b/internal/api/minds.go index d1b837e..179d5be 100644 --- a/internal/api/minds.go +++ b/internal/api/minds.go @@ -2,6 +2,7 @@ package api import ( "errors" + "fmt" "net/http" "strconv" @@ -119,6 +120,9 @@ func (s *Server) mindPatch(w http.ResponseWriter, r *http.Request) { return } s.deps.Audit.Log(r.Context(), "mind.update", "mind", m.ID.String(), in) + if in.Status != nil { + s.deps.Events.OperatorAction(r.Context(), m.ID, actor(r), "mind.status", "Operator "+actor(r)+" set status to "+string(*in.Status), map[string]any{"status": *in.Status}) + } writeJSON(w, http.StatusOK, out) } @@ -195,6 +199,7 @@ func (s *Server) personalityPut(w http.ResponseWriter, r *http.Request) { return } s.deps.Audit.Log(r.Context(), "personality.update", "mind", m.ID.String(), map[string]any{"version": v.Version, "note": in.ChangeNote, "changes": minds.Diff(m.Personality, v.Personality)}) + s.deps.Events.OperatorAction(r.Context(), m.ID, actor(r), "personality.update", fmt.Sprintf("Operator %s updated personality to version %d: %s", actor(r), v.Version, in.ChangeNote), map[string]any{"version": v.Version, "changes": minds.Diff(m.Personality, v.Personality)}) writeJSON(w, http.StatusOK, v) } @@ -245,6 +250,7 @@ func (s *Server) personalityRestore(w http.ResponseWriter, r *http.Request) { return } s.deps.Audit.Log(r.Context(), "personality.restore", "mind", m.ID.String(), map[string]any{"restored": n, "version": v.Version}) + s.deps.Events.OperatorAction(r.Context(), m.ID, actor(r), "personality.restore", fmt.Sprintf("Operator %s restored personality version %d as version %d", actor(r), n, v.Version), map[string]any{"restored": n, "version": v.Version}) writeJSON(w, http.StatusOK, v) } @@ -290,6 +296,7 @@ func (s *Server) personalityTemplate(w http.ResponseWriter, r *http.Request) { return } s.deps.Audit.Log(r.Context(), "personality.template", "mind", m.ID.String(), map[string]any{"template_id": in.TemplateID, "version": v.Version}) + s.deps.Events.OperatorAction(r.Context(), m.ID, actor(r), "personality.template", "Operator "+actor(r)+": "+v.ChangeNote, map[string]any{"version": v.Version}) writeJSON(w, http.StatusOK, v) } diff --git a/internal/api/server.go b/internal/api/server.go index d16d736..43cd43c 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -52,6 +52,7 @@ func (s *Server) routes() { m.Handle("DELETE /api/v1/me/tokens/{id}", require(auth.RoleObserver, s.myTokenRevoke)) s.mindRoutes() + s.eventRoutes() m.HandleFunc("/api/", func(w http.ResponseWriter, r *http.Request) { writeError(w, r, http.StatusNotFound, "not_found", "no such endpoint") diff --git a/internal/app/app.go b/internal/app/app.go index 880f276..4ad70c9 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -13,6 +13,7 @@ import ( "github.com/andydixon/noema/internal/audit" "github.com/andydixon/noema/internal/auth" "github.com/andydixon/noema/internal/config" + "github.com/andydixon/noema/internal/events" "github.com/andydixon/noema/internal/minds" "github.com/andydixon/noema/internal/observability" ) @@ -26,6 +27,7 @@ type Deps struct { Audit *audit.Logger Metrics *observability.Metrics Minds *minds.Store + Events *events.Store Migrations fs.FS Started time.Time } diff --git a/internal/events/bus.go b/internal/events/bus.go new file mode 100644 index 0000000..b044d1d --- /dev/null +++ b/internal/events/bus.go @@ -0,0 +1,91 @@ +// Package events stores cognitive events append-only and fans them out to +// live subscribers. Durable consumers read by sequence cursor; the in-memory +// bus is only a wake-up and a feed for the UI. +package events + +import ( + "context" + "sync" + "sync/atomic" + + "github.com/andydixon/noema/pkg/events" + "github.com/andydixon/noema/pkg/ids" +) + +// Bus fans events out to subscribers with bounded buffers. A slow subscriber +// loses events (counted in Dropped) rather than slowing the publisher; SSE +// clients recover via Last-Event-ID and durable consumers via cursors. +type Bus struct { + mu sync.RWMutex + subs map[uint64]*subscriber + nextID uint64 + dropped atomic.Int64 + max int +} + +type subscriber struct { + mind ids.ID // Nil = all minds + ch chan events.Event +} + +// NewBus creates a bus allowing at most max subscribers (T18). +func NewBus(max int) *Bus { + return &Bus{subs: map[uint64]*subscriber{}, max: max} +} + +// ErrTooManySubscribers is returned when the subscriber cap is reached. +type ErrTooManySubscribers struct{} + +func (ErrTooManySubscribers) Error() string { return "too many event subscribers" } + +// Subscribe returns a channel of events for one mind (or all when mind is +// Nil). The subscription ends when ctx is cancelled. +func (b *Bus) Subscribe(ctx context.Context, mind ids.ID, buffer int) (<-chan events.Event, error) { + if buffer < 1 { + buffer = 64 + } + b.mu.Lock() + if len(b.subs) >= b.max { + b.mu.Unlock() + return nil, ErrTooManySubscribers{} + } + id := b.nextID + b.nextID++ + s := &subscriber{mind: mind, ch: make(chan events.Event, buffer)} + b.subs[id] = s + b.mu.Unlock() + context.AfterFunc(ctx, func() { + b.mu.Lock() + delete(b.subs, id) + b.mu.Unlock() + }) + return s.ch, nil +} + +// Publish delivers events to matching subscribers without blocking. +func (b *Bus) Publish(evs ...events.Event) { + b.mu.RLock() + defer b.mu.RUnlock() + for _, ev := range evs { + for _, s := range b.subs { + if !s.mind.IsZero() && s.mind != ev.MindID { + continue + } + select { + case s.ch <- ev: + default: + b.dropped.Add(1) + } + } + } +} + +// Subscribers returns the current subscriber count. +func (b *Bus) Subscribers() int { + b.mu.RLock() + defer b.mu.RUnlock() + return len(b.subs) +} + +// Dropped returns the number of events dropped for slow subscribers. +func (b *Bus) Dropped() int64 { return b.dropped.Load() } diff --git a/internal/events/events_test.go b/internal/events/events_test.go new file mode 100644 index 0000000..1d18fbb --- /dev/null +++ b/internal/events/events_test.go @@ -0,0 +1,114 @@ +package events + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/andydixon/noema/internal/minds" + "github.com/andydixon/noema/internal/storage" + "github.com/andydixon/noema/migrations" + "github.com/andydixon/noema/pkg/events" + "github.com/andydixon/noema/pkg/ids" +) + +func TestBusBoundedFanout(t *testing.T) { + b := NewBus(2) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + m1, m2 := ids.New(), ids.New() + all, _ := b.Subscribe(ctx, ids.Nil, 1) + only1, _ := b.Subscribe(ctx, m1, 8) + if _, err := b.Subscribe(ctx, ids.Nil, 1); err == nil { + t.Fatal("subscriber cap not enforced") + } + b.Publish(events.Event{MindID: m1, Type: events.Observation}, events.Event{MindID: m2, Type: events.Observation}, events.Event{MindID: m1, Type: events.Reflection}) + if len(only1) != 2 { + t.Fatalf("mind filter wrong: %d", len(only1)) + } + if len(all) != 1 || b.Dropped() != 2 { + t.Fatalf("expected 1 buffered and 2 dropped, got %d / %d", len(all), b.Dropped()) + } + cancel() + time.Sleep(5 * time.Millisecond) + if b.Subscribers() != 0 { + t.Fatal("subscribers not removed on cancel") + } +} + +func TestStoreAppendQueryImmutableAndCursors(t *testing.T) { + pool := storage.TestPool(t, migrations.FS) + ctx := context.Background() + m, err := minds.NewStore(pool).Create(ctx, minds.CreateParams{Slug: "ev"}) + if err != nil { + t.Fatal(err) + } + other, _ := minds.NewStore(pool).Create(ctx, minds.CreateParams{Slug: "other"}) + s := NewStore(pool, NewBus(10), nil) + sub, _ := s.Bus().Subscribe(ctx, m.ID, 16) + + if _, err := s.Append(ctx, events.New{MindID: m.ID, Type: "bogus", Summary: "x"}); !errors.Is(err, ErrInvalid) { + t.Fatalf("unknown type accepted: %v", err) + } + obs, err := s.Append(ctx, events.New{MindID: m.ID, Type: events.Observation, Source: "test", Summary: "load spike observed", Salience: 0.9, Confidence: 1.2, Payload: map[string]any{"cpu": 97}}) + if err != nil { + t.Fatal(err) + } + if obs[0].Confidence != 1 || obs[0].Seq == 0 || obs[0].Category != events.CatPerception { + t.Fatalf("normalisation or seq wrong: %+v", obs[0]) + } + hyp, _ := s.Append(ctx, events.New{MindID: m.ID, Type: events.HypothesisCreated, Source: "reasoning", Summary: "disk contention hypothesis", + CausationID: obs[0].ID, ParentIDs: []ids.ID{obs[0].ID}, CorrelationID: obs[0].ID, Confidence: 0.4, Salience: 0.7, + Evidence: []events.EvidenceRef{{Kind: "event", ID: string(obs[0].ID)}}}) + s.Append(ctx, events.New{MindID: other.ID, Type: events.Observation, Source: "test", Summary: "other mind sees something else"}) //nolint:errcheck + + // Live delivery. + if len(sub) != 2 { + t.Fatalf("expected 2 live events for mind, got %d", len(sub)) + } + // Mind scoping and filters. + got, _ := s.Query(ctx, Filter{MindID: m.ID}) + if len(got) != 2 || got[0].Type != events.HypothesisCreated { + t.Fatalf("query wrong: %+v", got) + } + got, _ = s.Query(ctx, Filter{MindID: m.ID, Types: []events.Type{events.Observation}, MinSalience: 0.5}) + if len(got) != 1 { + t.Fatalf("type filter wrong: %d", len(got)) + } + got, _ = s.Query(ctx, Filter{MindID: m.ID, Search: "contention"}) + if len(got) != 1 { + t.Fatalf("search wrong: %d", len(got)) + } + got, _ = s.Query(ctx, Filter{MindID: m.ID, AfterSeq: obs[0].Seq, Ascending: true}) + if len(got) != 1 || got[0].ID != hyp[0].ID { + t.Fatalf("cursor paging wrong: %+v", got) + } + // Causal navigation. + kids, _ := s.Children(ctx, obs[0].ID, 10) + if len(kids) != 1 || kids[0].ID != hyp[0].ID { + t.Fatalf("children wrong: %+v", kids) + } + derived, _ := s.DerivedFrom(ctx, "event", string(obs[0].ID), 10) + if len(derived) != 1 { + t.Fatalf("evidence lookup wrong: %d", len(derived)) + } + // Immutability. + if _, err := pool.Exec(ctx, `UPDATE cognitive_events SET summary='rewritten' WHERE id=$1`, obs[0].ID); err == nil { + t.Fatal("events must be append-only") + } + if _, err := pool.Exec(ctx, `DELETE FROM cognitive_events`); err == nil { + t.Fatal("events must not be deletable") + } + // Cursors never move backwards. + s.SetCursor(ctx, pool, m.ID, "brain", hyp[0].Seq) //nolint:errcheck + s.SetCursor(ctx, pool, m.ID, "brain", obs[0].Seq) //nolint:errcheck + cur, _ := s.Cursor(ctx, pool, m.ID, "brain") + if cur != hyp[0].Seq { + t.Fatalf("cursor regressed: %d", cur) + } + st, _ := s.Stats(ctx, m.ID, time.Time{}) + if st.Total != 2 || st.ByCategory["belief"] != 1 { + t.Fatalf("stats wrong: %+v", st) + } +} diff --git a/internal/events/helpers.go b/internal/events/helpers.go new file mode 100644 index 0000000..cbc919a --- /dev/null +++ b/internal/events/helpers.go @@ -0,0 +1,77 @@ +package events + +import ( + "context" + "strings" + + "github.com/andydixon/noema/pkg/events" + "github.com/andydixon/noema/pkg/ids" +) + +// OperatorAction records an administrative change as a cognitive event so the +// mind's history shows when and why it was altered from outside. +func (s *Store) OperatorAction(ctx context.Context, mind ids.ID, actor, action, summary string, details map[string]any) { + if details == nil { + details = map[string]any{} + } + details["action"], details["actor"] = action, actor + s.Append(ctx, events.New{MindID: mind, Type: events.OperatorAction, Source: "operator", Summary: summary, Salience: 0.6, Confidence: 1, //nolint:errcheck + Metadata: map[string]any{"action": action, "actor": actor}, Payload: details}) +} + +// Matches applies the parts of a Filter that can be evaluated in memory, for +// live streams. +func (f Filter) Matches(ev events.Event) bool { + if !f.MindID.IsZero() && ev.MindID != f.MindID { + return false + } + if len(f.Types) > 0 && !containsType(f.Types, ev.Type) { + return false + } + if len(f.Categories) > 0 && !containsCat(f.Categories, ev.Category) { + return false + } + if len(f.Sources) > 0 && !containsStr(f.Sources, ev.Source) { + return false + } + if ev.Salience < f.MinSalience || ev.Confidence < f.MinConfidence { + return false + } + if f.MaxConfidence > 0 && ev.Confidence > f.MaxConfidence { + return false + } + if !f.CorrelationID.IsZero() && ev.CorrelationID != f.CorrelationID { + return false + } + if q := strings.TrimSpace(f.Search); q != "" && !strings.Contains(strings.ToLower(ev.Summary), strings.ToLower(q)) { + return false + } + return true +} + +func containsType(xs []events.Type, x events.Type) bool { + for _, y := range xs { + if y == x { + return true + } + } + return false +} + +func containsCat(xs []events.Category, x events.Category) bool { + for _, y := range xs { + if y == x { + return true + } + } + return false +} + +func containsStr(xs []string, x string) bool { + for _, y := range xs { + if y == x { + return true + } + } + return false +} diff --git a/internal/events/store.go b/internal/events/store.go new file mode 100644 index 0000000..82e5fea --- /dev/null +++ b/internal/events/store.go @@ -0,0 +1,434 @@ +package events + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/andydixon/noema/internal/observability" + "github.com/andydixon/noema/internal/storage" + "github.com/andydixon/noema/pkg/events" + "github.com/andydixon/noema/pkg/ids" +) + +// Store appends and queries cognitive events. +type Store struct { + db *pgxpool.Pool + bus *Bus + metrics *observability.Metrics +} + +// NewStore creates the store. bus and metrics may be nil in tests. +func NewStore(db *pgxpool.Pool, bus *Bus, metrics *observability.Metrics) *Store { + if bus == nil { + bus = NewBus(1000) + } + return &Store{db: db, bus: bus, metrics: metrics} +} + +// Bus exposes the fan-out bus for subscribers. +func (s *Store) Bus() *Bus { return s.bus } + +// ErrInvalid marks malformed events. +var ErrInvalid = errors.New("invalid event") + +func validate(n *events.New) error { + if n.MindID.IsZero() { + return fmt.Errorf("%w: mind id required", ErrInvalid) + } + if !n.Type.Valid() { + return fmt.Errorf("%w: unknown type %q", ErrInvalid, n.Type) + } + if strings.TrimSpace(n.Summary) == "" { + return fmt.Errorf("%w: summary required", ErrInvalid) + } + if n.Source == "" { + n.Source = "unknown" + } + if len(n.Summary) > 2000 { + n.Summary = n.Summary[:2000] + } + n.Salience = clamp(n.Salience, 0, 1) + n.Confidence = clamp(n.Confidence, 0, 1) + if n.Valence != nil { + v := clamp(*n.Valence, -1, 1) + n.Valence = &v + } + if n.Arousal != nil { + a := clamp(*n.Arousal, 0, 1) + n.Arousal = &a + } + if n.OccurredAt.IsZero() { + n.OccurredAt = time.Now() + } + return nil +} + +func clamp(v, lo, hi float64) float64 { + if v != v { // NaN + return lo + } + if v < lo { + return lo + } + if v > hi { + return hi + } + return v +} + +// Append writes events in one transaction and publishes them after commit. +func (s *Store) Append(ctx context.Context, news ...events.New) ([]events.Event, error) { + var out []events.Event + err := storage.Tx(ctx, s.db, func(tx pgx.Tx) error { + var err error + out, err = s.AppendIn(ctx, tx, news...) + return err + }) + if err != nil { + return nil, err + } + s.Publish(out...) + return out, nil +} + +// AppendIn writes events inside the caller's transaction. The caller must call +// Publish after committing so subscribers never see uncommitted events. +func (s *Store) AppendIn(ctx context.Context, q storage.Querier, news ...events.New) ([]events.Event, error) { + out := make([]events.Event, 0, len(news)) + for i := range news { + n := &news[i] + if err := validate(n); err != nil { + return nil, err + } + payload, err := json.Marshal(n.Payload) + if err != nil { + return nil, fmt.Errorf("%w: payload: %v", ErrInvalid, err) + } + if n.Payload == nil { + payload = []byte("{}") + } + if n.Evidence == nil { + n.Evidence = []events.EvidenceRef{} + } + if n.Metadata == nil { + n.Metadata = map[string]any{} + } + evidence, _ := json.Marshal(n.Evidence) + metadata, _ := json.Marshal(n.Metadata) + if n.ParentIDs == nil { + n.ParentIDs = []ids.ID{} + } + ev := events.Event{ID: ids.New(), MindID: n.MindID, Type: n.Type, Category: events.CategoryOf(n.Type), Source: n.Source, + OccurredAt: n.OccurredAt, ParentIDs: n.ParentIDs, CausationID: n.CausationID, CorrelationID: n.CorrelationID, + Salience: n.Salience, Confidence: n.Confidence, Valence: n.Valence, Arousal: n.Arousal, Evidence: n.Evidence, + Summary: n.Summary, Metadata: n.Metadata, Payload: payload} + err = q.QueryRow(ctx, `INSERT INTO cognitive_events (id, mind_id, type, category, source, occurred_at, parent_ids, causation_id, correlation_id, + salience, confidence, valence, arousal, evidence, summary, metadata, payload) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) RETURNING seq, recorded_at`, + ev.ID, ev.MindID, string(ev.Type), string(ev.Category), ev.Source, ev.OccurredAt, uuidArray(ev.ParentIDs), nullID(ev.CausationID), nullID(ev.CorrelationID), + ev.Salience, ev.Confidence, ev.Valence, ev.Arousal, evidence, ev.Summary, metadata, payload).Scan(&ev.Seq, &ev.RecordedAt) + if err != nil { + return nil, fmt.Errorf("append event: %w", err) + } + out = append(out, ev) + } + return out, nil +} + +// Publish fans committed events out to live subscribers and metrics. +func (s *Store) Publish(evs ...events.Event) { + if s.metrics != nil { + for _, ev := range evs { + s.metrics.Events.WithLabelValues(string(ev.Type)).Inc() + } + } + s.bus.Publish(evs...) +} + +func uuidArray(in []ids.ID) []string { + out := make([]string, len(in)) + for i, id := range in { + out[i] = string(id) + } + return out +} + +func nullID(id ids.ID) *string { + if id.IsZero() { + return nil + } + s := string(id) + return &s +} + +// Filter selects events. Zero values mean "no constraint". +type Filter struct { + MindID ids.ID + Types []events.Type + Categories []events.Category + Sources []string + MinSalience float64 + MinConfidence float64 + MaxConfidence float64 // 0 = unset + Since, Until time.Time + AfterSeq int64 // exclusive lower bound (forward paging / cursors) + BeforeSeq int64 // exclusive upper bound (backward paging) + CorrelationID ids.ID + Search string // substring of summary, case-insensitive + Limit int + Ascending bool +} + +const cols = `id, seq, mind_id, type, category, source, occurred_at, recorded_at, parent_ids, causation_id, correlation_id, salience, confidence, valence, arousal, evidence, summary, metadata, payload` + +// Query returns events matching f ordered by seq (descending unless Ascending). +func (s *Store) Query(ctx context.Context, f Filter) ([]events.Event, error) { + var where []string + var args []any + arg := func(v any) string { args = append(args, v); return fmt.Sprintf("$%d", len(args)) } + if !f.MindID.IsZero() { + where = append(where, "mind_id = "+arg(f.MindID)) + } + if len(f.Types) > 0 { + ts := make([]string, len(f.Types)) + for i, t := range f.Types { + ts[i] = string(t) + } + where = append(where, "type = ANY("+arg(ts)+")") + } + if len(f.Categories) > 0 { + cs := make([]string, len(f.Categories)) + for i, c := range f.Categories { + cs[i] = string(c) + } + where = append(where, "category = ANY("+arg(cs)+")") + } + if len(f.Sources) > 0 { + where = append(where, "source = ANY("+arg(f.Sources)+")") + } + if f.MinSalience > 0 { + where = append(where, "salience >= "+arg(f.MinSalience)) + } + if f.MinConfidence > 0 { + where = append(where, "confidence >= "+arg(f.MinConfidence)) + } + if f.MaxConfidence > 0 { + where = append(where, "confidence <= "+arg(f.MaxConfidence)) + } + if !f.Since.IsZero() { + where = append(where, "occurred_at >= "+arg(f.Since)) + } + if !f.Until.IsZero() { + where = append(where, "occurred_at <= "+arg(f.Until)) + } + if f.AfterSeq > 0 { + where = append(where, "seq > "+arg(f.AfterSeq)) + } + if f.BeforeSeq > 0 { + where = append(where, "seq < "+arg(f.BeforeSeq)) + } + if !f.CorrelationID.IsZero() { + where = append(where, "correlation_id = "+arg(f.CorrelationID)) + } + if q := strings.TrimSpace(f.Search); q != "" { + where = append(where, "summary ILIKE "+arg("%"+escapeLike(q)+"%")) + } + limit := f.Limit + if limit <= 0 || limit > 1000 { + limit = 100 + } + order := "DESC" + if f.Ascending { + order = "ASC" + } + sql := "SELECT " + cols + " FROM cognitive_events" + if len(where) > 0 { + sql += " WHERE " + strings.Join(where, " AND ") + } + sql += " ORDER BY seq " + order + " LIMIT " + arg(limit) + rows, err := s.db.Query(ctx, sql, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var out []events.Event + for rows.Next() { + ev, err := scan(rows) + if err != nil { + return nil, err + } + out = append(out, ev) + } + return out, rows.Err() +} + +func escapeLike(s string) string { + r := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`) + return r.Replace(s) +} + +// Get fetches one event by id. +func (s *Store) Get(ctx context.Context, id ids.ID) (events.Event, error) { + ev, err := scan(s.db.QueryRow(ctx, "SELECT "+cols+" FROM cognitive_events WHERE id=$1", id)) + return ev, storage.Translate(err) +} + +// GetMany fetches events by id, preserving no particular order. +func (s *Store) GetMany(ctx context.Context, idList []ids.ID) ([]events.Event, error) { + if len(idList) == 0 { + return nil, nil + } + rows, err := s.db.Query(ctx, "SELECT "+cols+" FROM cognitive_events WHERE id = ANY($1) ORDER BY seq", uuidArray(idList)) + if err != nil { + return nil, err + } + defer rows.Close() + var out []events.Event + for rows.Next() { + ev, err := scan(rows) + if err != nil { + return nil, err + } + out = append(out, ev) + } + return out, rows.Err() +} + +// Children returns events directly caused by, or listing as parent, the given event. +func (s *Store) Children(ctx context.Context, id ids.ID, limit int) ([]events.Event, error) { + if limit <= 0 || limit > 500 { + limit = 100 + } + rows, err := s.db.Query(ctx, "SELECT "+cols+" FROM cognitive_events WHERE causation_id=$1 OR $1 = ANY(parent_ids) ORDER BY seq LIMIT $2", id, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []events.Event + for rows.Next() { + ev, err := scan(rows) + if err != nil { + return nil, err + } + out = append(out, ev) + } + return out, rows.Err() +} + +// DerivedFrom returns events that cite id as evidence. +func (s *Store) DerivedFrom(ctx context.Context, kind string, id string, limit int) ([]events.Event, error) { + if limit <= 0 || limit > 500 { + limit = 100 + } + ref, _ := json.Marshal([]map[string]string{{"kind": kind, "id": id}}) + rows, err := s.db.Query(ctx, "SELECT "+cols+" FROM cognitive_events WHERE evidence @> $1::jsonb ORDER BY seq LIMIT $2", ref, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []events.Event + for rows.Next() { + ev, err := scan(rows) + if err != nil { + return nil, err + } + out = append(out, ev) + } + return out, rows.Err() +} + +func scan(row pgx.Row) (events.Event, error) { + var ev events.Event + var typ, cat string + var parents []string + var causation, correlation *string + var evidence, metadata []byte + err := row.Scan(&ev.ID, &ev.Seq, &ev.MindID, &typ, &cat, &ev.Source, &ev.OccurredAt, &ev.RecordedAt, &parents, &causation, &correlation, + &ev.Salience, &ev.Confidence, &ev.Valence, &ev.Arousal, &evidence, &ev.Summary, &metadata, &ev.Payload) + if err != nil { + return ev, err + } + ev.Type, ev.Category = events.Type(typ), events.Category(cat) + ev.ParentIDs = make([]ids.ID, len(parents)) + for i, p := range parents { + ev.ParentIDs[i] = ids.ID(p) + } + if causation != nil { + ev.CausationID = ids.ID(*causation) + } + if correlation != nil { + ev.CorrelationID = ids.ID(*correlation) + } + json.Unmarshal(evidence, &ev.Evidence) //nolint:errcheck + json.Unmarshal(metadata, &ev.Metadata) //nolint:errcheck + if ev.Evidence == nil { + ev.Evidence = []events.EvidenceRef{} + } + if ev.Metadata == nil { + ev.Metadata = map[string]any{} + } + return ev, nil +} + +// Cursor returns the last sequence consumed by consumer for a mind. +func (s *Store) Cursor(ctx context.Context, q storage.Querier, mind ids.ID, consumer string) (int64, error) { + var seq int64 + err := q.QueryRow(ctx, `SELECT last_seq FROM mind_cursors WHERE mind_id=$1 AND consumer=$2`, mind, consumer).Scan(&seq) + if errors.Is(err, pgx.ErrNoRows) { + return 0, nil + } + return seq, err +} + +// SetCursor advances a consumer cursor (never backwards). +func (s *Store) SetCursor(ctx context.Context, q storage.Querier, mind ids.ID, consumer string, seq int64) error { + _, err := q.Exec(ctx, `INSERT INTO mind_cursors (mind_id, consumer, last_seq) VALUES ($1,$2,$3) + ON CONFLICT (mind_id, consumer) DO UPDATE SET last_seq = GREATEST(mind_cursors.last_seq, excluded.last_seq), updated_at = now()`, mind, consumer, seq) + return err +} + +// Stats summarises a mind's events for overview pages. +type Stats struct { + Total int64 `json:"total"` + ByCategory map[string]int64 `json:"by_category"` + LastSeq int64 `json:"last_seq"` + LastAt *time.Time `json:"last_at,omitempty"` +} + +// Stats counts events for a mind (optionally since a time). +func (s *Store) Stats(ctx context.Context, mind ids.ID, since time.Time) (Stats, error) { + st := Stats{ByCategory: map[string]int64{}} + rows, err := s.db.Query(ctx, `SELECT category, count(*), max(seq), max(occurred_at) FROM cognitive_events WHERE mind_id=$1 AND ($2::timestamptz IS NULL OR occurred_at >= $2) GROUP BY category`, mind, nullTime(since)) + if err != nil { + return st, err + } + defer rows.Close() + for rows.Next() { + var cat string + var n, maxSeq int64 + var at *time.Time + if err := rows.Scan(&cat, &n, &maxSeq, &at); err != nil { + return st, err + } + st.ByCategory[cat] = n + st.Total += n + if maxSeq > st.LastSeq { + st.LastSeq, st.LastAt = maxSeq, at + } + } + return st, rows.Err() +} + +func nullTime(t time.Time) *time.Time { + if t.IsZero() { + return nil + } + return &t +} diff --git a/internal/ui/events.go b/internal/ui/events.go new file mode 100644 index 0000000..426e9e4 --- /dev/null +++ b/internal/ui/events.go @@ -0,0 +1,120 @@ +package ui + +import ( + "encoding/json" + "net/http" + "net/url" + "strconv" + + "github.com/andydixon/noema/internal/api" + "github.com/andydixon/noema/internal/auth" + "github.com/andydixon/noema/pkg/events" + "github.com/andydixon/noema/pkg/ids" +) + +func (u *UI) eventRoutes(m *http.ServeMux) { + m.Handle("GET /minds/{mind}/thoughts", u.require(auth.RoleObserver, u.thoughts)) + m.Handle("POST /minds/{mind}/observe", u.require(auth.RoleOperator, u.observe)) + m.Handle("GET /events/{id}", u.require(auth.RoleObserver, u.event)) +} + +var sinceWindows = map[string]string{"15m": "15 minutes", "1h": "1 hour", "24h": "24 hours", "168h": "7 days"} + +func (u *UI) thoughts(w http.ResponseWriter, r *http.Request) { + m, ok := u.mind(w, r) + if !ok { + return + } + q := r.URL.Query() + f := api.ParseEventFilter(q) + f.MindID = m.ID + f.Limit = 100 + list, err := u.deps.Events.Query(r.Context(), f) + if err != nil { + u.errorPage(w, r, 500, "Could not load events.") + return + } + var topSeq, nextBefore int64 + if len(list) > 0 { + topSeq = list[0].Seq + if len(list) == 100 { + nextBefore = list[len(list)-1].Seq + } + } + // The stream carries the same filter but never paging parameters. + sq := url.Values{} + for _, k := range []string{"category", "type", "source", "min_salience", "min_confidence", "q", "correlation"} { + if v := q.Get(k); v != "" { + sq.Set(k, v) + } + } + streamQuery := "" + if len(sq) > 0 { + streamQuery = "?" + sq.Encode() + } + flat := map[string]string{} + for k := range q { + flat[k] = q.Get(k) + } + cats := []events.Category{events.CatPerception, events.CatCognition, events.CatMemory, events.CatBelief, events.CatPrediction, events.CatGoal, events.CatAffect, events.CatAction, events.CatLLM, events.CatOperator, events.CatSystem} + u.render(w, r, "thoughts", m.DisplayName+" · Thoughts", map[string]any{"Mind": m, "Events": list, "Q": flat, "Types": events.AllTypes(), "Categories": cats, + "Windows": sinceWindows, "StreamQuery": streamQuery, "RawQuery": sq.Encode(), "TopSeq": topSeq, "NextBefore": nextBefore}) +} + +func (u *UI) observe(w http.ResponseWriter, r *http.Request) { + m, ok := u.mind(w, r) + if !ok { + return + } + text := r.PostFormValue("text") + if len(text) == 0 || len(text) > 20000 { + u.flash(w, r, "error", "Observation text required.", "/minds/"+m.Slug) + return + } + summary := text + if len(summary) > 160 { + summary = summary[:157] + "..." + } + evs, err := u.deps.Events.Append(r.Context(), events.New{MindID: m.ID, Type: events.Observation, Source: "operator", Summary: summary, Salience: 0.6, Confidence: 1, + Metadata: map[string]any{"trust": events.TrustOperator, "actor": username(r)}, + Payload: events.ObservationPayload{Text: text, ContentType: "text/plain", Source: "operator", Trust: events.TrustOperator, Method: "ui", Actor: username(r)}}) + if err != nil { + u.flash(w, r, "error", "Could not record observation.", "/minds/"+m.Slug) + return + } + u.deps.Audit.Log(r.Context(), "observation.inject", "mind", m.ID.String(), map[string]any{"event_id": evs[0].ID}) + u.flash(w, r, "ok", "Observation recorded.", "/minds/"+m.Slug+"/thoughts") +} + +func (u *UI) event(w http.ResponseWriter, r *http.Request) { + id, err := ids.Parse(r.PathValue("id")) + if err != nil { + u.errorPage(w, r, 404, "No such event.") + return + } + c, err := u.ctxF(r, id) + if err != nil { + u.errorPage(w, r, 404, "No such event.") + return + } + m, err := u.deps.Minds.Get(r.Context(), c.Event.MindID) + if err != nil { + u.errorPage(w, r, 500, "Mind missing for event.") + return + } + u.render(w, r, "event", "Event "+strconv.FormatInt(c.Event.Seq, 10), map[string]any{"Ctx": c, "Mind": m, "Payload": pretty(c.Event.Payload), "Metadata": prettyAny(c.Event.Metadata)}) +} + +func pretty(raw json.RawMessage) string { + var v any + if json.Unmarshal(raw, &v) != nil { + return string(raw) + } + b, _ := json.MarshalIndent(v, "", " ") + return string(b) +} + +func prettyAny(v any) string { + b, _ := json.MarshalIndent(v, "", " ") + return string(b) +} diff --git a/internal/ui/minds.go b/internal/ui/minds.go index 78474ea..b6882f5 100644 --- a/internal/ui/minds.go +++ b/internal/ui/minds.go @@ -6,6 +6,7 @@ import ( "net/http" "strconv" "strings" + "time" "github.com/andydixon/noema/internal/auth" "github.com/andydixon/noema/internal/minds" @@ -92,7 +93,8 @@ func (u *UI) mindOverview(w http.ResponseWriter, r *http.Request) { return } sm, _ := u.deps.Minds.GetSelfModel(r.Context(), m.ID) - u.render(w, r, "mind", m.DisplayName, map[string]any{"Mind": m, "Brief": m.Personality.Brief(), "SelfModel": sm}) + st, _ := u.deps.Events.Stats(r.Context(), m.ID, time.Now().Add(-24*time.Hour)) + u.render(w, r, "mind", m.DisplayName, map[string]any{"Mind": m, "Brief": m.Personality.Brief(), "SelfModel": sm, "Stats": st}) } func (u *UI) mindStatus(w http.ResponseWriter, r *http.Request) { @@ -106,6 +108,7 @@ func (u *UI) mindStatus(w http.ResponseWriter, r *http.Request) { return } u.deps.Audit.Log(r.Context(), "mind.status", "mind", m.ID.String(), map[string]any{"status": st}) + u.deps.Events.OperatorAction(r.Context(), m.ID, username(r), "mind.status", "Operator "+username(r)+" set status to "+string(st), map[string]any{"status": st}) u.flash(w, r, "ok", "Mind is now "+string(st)+".", "/minds/"+m.Slug) } @@ -232,6 +235,7 @@ func (u *UI) personalitySave(w http.ResponseWriter, r *http.Request) { return } u.deps.Audit.Log(r.Context(), "personality.update", "mind", m.ID.String(), map[string]any{"version": v.Version, "note": v.ChangeNote, "changes": minds.Diff(m.Personality, v.Personality)}) + u.deps.Events.OperatorAction(r.Context(), m.ID, username(r), "personality.update", "Operator "+username(r)+" updated personality to version "+strconv.Itoa(v.Version)+": "+v.ChangeNote, map[string]any{"version": v.Version, "changes": minds.Diff(m.Personality, v.Personality)}) u.flash(w, r, "ok", "Saved as version "+strconv.Itoa(v.Version)+".", "/minds/"+m.Slug+"/personality") } @@ -278,6 +282,7 @@ func (u *UI) personalityImport(w http.ResponseWriter, r *http.Request) { return } u.deps.Audit.Log(r.Context(), "personality.import", "mind", m.ID.String(), map[string]any{"version": v.Version}) + u.deps.Events.OperatorAction(r.Context(), m.ID, username(r), "personality.import", "Operator "+username(r)+" imported personality as version "+strconv.Itoa(v.Version), map[string]any{"version": v.Version}) u.flash(w, r, "ok", "Imported as version "+strconv.Itoa(v.Version)+".", "/minds/"+m.Slug+"/personality") } @@ -301,6 +306,7 @@ func (u *UI) personalityTemplate(w http.ResponseWriter, r *http.Request) { return } u.deps.Audit.Log(r.Context(), "personality.template", "mind", m.ID.String(), map[string]any{"template_id": tid, "version": v.Version}) + u.deps.Events.OperatorAction(r.Context(), m.ID, username(r), "personality.template", "Operator "+username(r)+": "+v.ChangeNote, map[string]any{"version": v.Version}) u.flash(w, r, "ok", v.ChangeNote+" (version "+strconv.Itoa(v.Version)+").", "/minds/"+m.Slug+"/personality") } @@ -320,6 +326,7 @@ func (u *UI) personalityRestore(w http.ResponseWriter, r *http.Request) { return } u.deps.Audit.Log(r.Context(), "personality.restore", "mind", m.ID.String(), map[string]any{"restored": n, "version": v.Version}) + u.deps.Events.OperatorAction(r.Context(), m.ID, username(r), "personality.restore", "Operator "+username(r)+" restored personality version "+strconv.Itoa(n)+" as version "+strconv.Itoa(v.Version), map[string]any{"restored": n, "version": v.Version}) u.flash(w, r, "ok", "Restored version "+strconv.Itoa(n)+" as version "+strconv.Itoa(v.Version)+".", "/minds/"+m.Slug+"/personality") } diff --git a/internal/ui/pages.go b/internal/ui/pages.go index ea03a61..76f050c 100644 --- a/internal/ui/pages.go +++ b/internal/ui/pages.go @@ -8,6 +8,7 @@ import ( "strings" "github.com/andydixon/noema/internal/auth" + evstore "github.com/andydixon/noema/internal/events" "github.com/andydixon/noema/internal/storage" "github.com/andydixon/noema/pkg/ids" ) @@ -97,7 +98,13 @@ func (u *UI) setupSubmit(w http.ResponseWriter, r *http.Request) { func (u *UI) dashboard(w http.ResponseWriter, r *http.Request) { entries, _ := u.deps.Audit.List(r.Context(), 0, 8, "", "") - u.render(w, r, "dashboard", "Dashboard", map[string]any{"Minds": 0, "Status": u.statusF(r), "Audit": entries}) + mindsList, _ := u.deps.Minds.List(r.Context(), false) + names := map[ids.ID]string{} + for _, m := range mindsList { + names[m.ID] = m.DisplayName + } + evs, _ := u.deps.Events.Query(r.Context(), evstore.Filter{MinSalience: 0.6, Limit: 12}) + u.render(w, r, "dashboard", "Dashboard", map[string]any{"Minds": mindsList, "MindNames": names, "Events": evs, "Status": u.statusF(r), "Audit": entries}) } func (u *UI) system(w http.ResponseWriter, r *http.Request) { diff --git a/internal/ui/ui.go b/internal/ui/ui.go index 2234b3c..be8e932 100644 --- a/internal/ui/ui.go +++ b/internal/ui/ui.go @@ -15,10 +15,12 @@ import ( "strings" "time" + "github.com/andydixon/noema/internal/api" "github.com/andydixon/noema/internal/app" "github.com/andydixon/noema/internal/auth" "github.com/andydixon/noema/internal/minds" "github.com/andydixon/noema/internal/version" + "github.com/andydixon/noema/pkg/ids" "github.com/andydixon/noema/web" ) @@ -27,7 +29,8 @@ type UI struct { deps app.Deps pages map[string]*template.Template static fs.FS - statusF func(r *http.Request) any // system status provider injected by api + statusF func(r *http.Request) any // system status provider injected by api + ctxF func(r *http.Request, id ids.ID) (api.EventContext, error) // event forensic context, shared with api } // View is the data passed to every template. @@ -48,7 +51,7 @@ const flashCookie = "noema_flash" // New parses templates. In development, when ./web exists on disk, templates // and static files are read from disk so edits show without a rebuild. -func New(deps app.Deps, statusF func(r *http.Request) any) (*UI, error) { +func New(deps app.Deps, statusF func(r *http.Request) any, ctxF func(r *http.Request, id ids.ID) (api.EventContext, error)) (*UI, error) { tmplFS, staticFS := web.Templates(), web.Static() if !deps.Config.Production() { if st, err := os.Stat("web/templates"); err == nil && st.IsDir() { @@ -56,7 +59,7 @@ func New(deps app.Deps, statusF func(r *http.Request) any) (*UI, error) { deps.Log.Info("ui: serving templates and static files from ./web (development)") } } - u := &UI{deps: deps, static: staticFS, statusF: statusF} + u := &UI{deps: deps, static: staticFS, statusF: statusF, ctxF: ctxF} if err := u.parse(tmplFS); err != nil { return nil, err } @@ -74,6 +77,12 @@ func (u *UI) parse(fsys fs.FS) error { "pct": func(f float64) string { return fmt.Sprintf("%.0f%%", f*100) }, "conf": func(f float64) string { return fmt.Sprintf("%.2f", f) }, "inc": func(i int) int { return i + 1 }, + "deref": func(f *float64) float64 { + if f == nil { + return 0 + } + return *f + }, "title": func(s string) string { s = strings.ReplaceAll(s, "_", " "); return strings.ToUpper(s[:1]) + s[1:] }, "valueKeys": func() []string { return minds.ValueKeys }, "traitKeys": func() []string { return minds.TraitKeys }, @@ -198,6 +207,7 @@ func (u *UI) Register(m *http.ServeMux) { m.Handle("POST /admin/users/{id}", u.require(auth.RoleAdministrator, u.usersUpdate)) u.mindRoutes(m) + u.eventRoutes(m) for _, p := range []struct{ path, title, note string }{ {"/approvals", "Approvals", "The approval centre arrives with the capability system."}, diff --git a/migrations/0003_events.down.sql b/migrations/0003_events.down.sql new file mode 100644 index 0000000..c516be9 --- /dev/null +++ b/migrations/0003_events.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS mind_cursors; +DROP TABLE IF EXISTS cognitive_events; diff --git a/migrations/0003_events.up.sql b/migrations/0003_events.up.sql new file mode 100644 index 0000000..b2e55ce --- /dev/null +++ b/migrations/0003_events.up.sql @@ -0,0 +1,40 @@ +-- Immutable cognitive events and durable consumer cursors. + +CREATE TABLE cognitive_events ( + id uuid PRIMARY KEY, + seq bigserial NOT NULL UNIQUE, + mind_id uuid NOT NULL REFERENCES minds(id), + type text NOT NULL, + category text NOT NULL, + source text NOT NULL, + occurred_at timestamptz NOT NULL, + recorded_at timestamptz NOT NULL DEFAULT now(), + parent_ids uuid[] NOT NULL DEFAULT '{}', + causation_id uuid, + correlation_id uuid, + salience real NOT NULL DEFAULT 0 CHECK (salience >= 0 AND salience <= 1), + confidence real NOT NULL DEFAULT 0 CHECK (confidence >= 0 AND confidence <= 1), + valence real CHECK (valence IS NULL OR (valence >= -1 AND valence <= 1)), + arousal real CHECK (arousal IS NULL OR (arousal >= 0 AND arousal <= 1)), + evidence jsonb NOT NULL DEFAULT '[]'::jsonb, + summary text NOT NULL, + metadata jsonb NOT NULL DEFAULT '{}'::jsonb, + payload jsonb NOT NULL DEFAULT '{}'::jsonb +); +CREATE INDEX cognitive_events_mind_seq_idx ON cognitive_events (mind_id, seq); +CREATE INDEX cognitive_events_mind_type_seq_idx ON cognitive_events (mind_id, type, seq); +CREATE INDEX cognitive_events_correlation_idx ON cognitive_events (correlation_id) WHERE correlation_id IS NOT NULL; +CREATE INDEX cognitive_events_causation_idx ON cognitive_events (causation_id) WHERE causation_id IS NOT NULL; +CREATE INDEX cognitive_events_parents_idx ON cognitive_events USING gin (parent_ids); +CREATE INDEX cognitive_events_evidence_idx ON cognitive_events USING gin (evidence jsonb_path_ops); +CREATE TRIGGER cognitive_events_append_only BEFORE UPDATE OR DELETE ON cognitive_events + FOR EACH ROW EXECUTE FUNCTION noema_forbid_change(); + +-- Durable read positions for consumers (the mind runtime, consolidation, ...). +CREATE TABLE mind_cursors ( + mind_id uuid NOT NULL REFERENCES minds(id), + consumer text NOT NULL, + last_seq bigint NOT NULL DEFAULT 0, + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (mind_id, consumer) +); diff --git a/pkg/events/events.go b/pkg/events/events.go new file mode 100644 index 0000000..5c893c1 --- /dev/null +++ b/pkg/events/events.go @@ -0,0 +1,236 @@ +// Package events defines the cognitive event vocabulary shared by every part +// of Noema. Events are the append-only record of what a mind perceived, +// thought, believed, predicted and did, and the provenance links between +// those things (ADR-0002). +package events + +import ( + "encoding/json" + "time" + + "github.com/andydixon/noema/pkg/ids" +) + +// Type identifies what kind of thing happened. +type Type string + +// Event types. The list is closed: adding one means adding it here so that +// filters, metrics and the UI vocabulary stay in step. +const ( + Observation Type = "observation" + Perception Type = "perception" + ThoughtCandidate Type = "thought_candidate" + ThoughtSelected Type = "thought_selected" + WorkspaceBroadcast Type = "workspace_broadcast" + AttentionShift Type = "attention_shift" + MemoryCreated Type = "memory_created" + MemoryRetrieved Type = "memory_retrieved" + MemoryReinforced Type = "memory_reinforced" + MemoryDecayed Type = "memory_decayed" + MemoryConsolidated Type = "memory_consolidated" + BeliefCreated Type = "belief_created" + BeliefUpdated Type = "belief_updated" + BeliefChallenged Type = "belief_challenged" + BeliefRetracted Type = "belief_retracted" + HypothesisCreated Type = "hypothesis_created" + HypothesisSupported Type = "hypothesis_supported" + HypothesisWeakened Type = "hypothesis_weakened" + PredictionCreated Type = "prediction_created" + PredictionResolved Type = "prediction_resolved" + PredictionError Type = "prediction_error" + GoalCreated Type = "goal_created" + GoalUpdated Type = "goal_updated" + GoalCompleted Type = "goal_completed" + GoalAbandoned Type = "goal_abandoned" + DriveChanged Type = "drive_changed" + EmotionChanged Type = "emotion_changed" + Reflection Type = "reflection" + ContradictionDetected Type = "contradiction_detected" + ToolRequested Type = "tool_requested" + ToolAuthorised Type = "tool_authorised" + ToolDenied Type = "tool_denied" + ToolExecuted Type = "tool_executed" + ToolFailed Type = "tool_failed" + NotificationCreated Type = "notification_created" + NotificationSuppressed Type = "notification_suppressed" + LLMRequest Type = "llm_request" + LLMResponse Type = "llm_response" + PolicyDecision Type = "policy_decision" + SelfModelUpdated Type = "self_model_updated" + DreamAssociation Type = "dream_association" + OperatorAction Type = "operator_action" + Metacognition Type = "metacognition" + NarrativeUpdated Type = "narrative_updated" + ModeChanged Type = "mode_changed" + ResponseGenerated Type = "response_generated" + CycleCompleted Type = "cycle_completed" +) + +// Category groups types for filtering and colouring in the UI. +type Category string + +// Categories. +const ( + CatPerception Category = "perception" + CatCognition Category = "cognition" + CatMemory Category = "memory" + CatBelief Category = "belief" + CatPrediction Category = "prediction" + CatGoal Category = "goal" + CatAffect Category = "affect" + CatAction Category = "action" + CatLLM Category = "llm" + CatOperator Category = "operator" + CatSystem Category = "system" +) + +var categories = map[Type]Category{ + Observation: CatPerception, Perception: CatPerception, + ThoughtCandidate: CatCognition, ThoughtSelected: CatCognition, WorkspaceBroadcast: CatCognition, AttentionShift: CatCognition, + Reflection: CatCognition, ContradictionDetected: CatCognition, Metacognition: CatCognition, DreamAssociation: CatCognition, + NarrativeUpdated: CatCognition, ResponseGenerated: CatCognition, CycleCompleted: CatSystem, ModeChanged: CatSystem, + MemoryCreated: CatMemory, MemoryRetrieved: CatMemory, MemoryReinforced: CatMemory, MemoryDecayed: CatMemory, MemoryConsolidated: CatMemory, + BeliefCreated: CatBelief, BeliefUpdated: CatBelief, BeliefChallenged: CatBelief, BeliefRetracted: CatBelief, + HypothesisCreated: CatBelief, HypothesisSupported: CatBelief, HypothesisWeakened: CatBelief, + PredictionCreated: CatPrediction, PredictionResolved: CatPrediction, PredictionError: CatPrediction, + GoalCreated: CatGoal, GoalUpdated: CatGoal, GoalCompleted: CatGoal, GoalAbandoned: CatGoal, + DriveChanged: CatAffect, EmotionChanged: CatAffect, + ToolRequested: CatAction, ToolAuthorised: CatAction, ToolDenied: CatAction, ToolExecuted: CatAction, ToolFailed: CatAction, + NotificationCreated: CatAction, NotificationSuppressed: CatAction, PolicyDecision: CatAction, + LLMRequest: CatLLM, LLMResponse: CatLLM, + SelfModelUpdated: CatSystem, OperatorAction: CatOperator, +} + +// CategoryOf returns the category of a type (CatSystem for unknown types). +func CategoryOf(t Type) Category { + if c, ok := categories[t]; ok { + return c + } + return CatSystem +} + +// Valid reports whether t is a known event type. +func (t Type) Valid() bool { _, ok := categories[t]; return ok } + +// AllTypes lists every known type in a stable order. +func AllTypes() []Type { + out := make([]Type, 0, len(categories)) + for t := range categories { + out = append(out, t) + } + sortTypes(out) + return out +} + +func sortTypes(ts []Type) { + for i := 1; i < len(ts); i++ { + for j := i; j > 0 && ts[j] < ts[j-1]; j-- { + ts[j], ts[j-1] = ts[j-1], ts[j] + } + } +} + +// EvidenceRef points at something that supports or explains an event: a +// memory, belief, another event, an entity, a document, an LLM call, a +// prediction, a message. +type EvidenceRef struct { + Kind string `json:"kind"` // event | memory | belief | entity | document | llm_call | prediction | goal | message | approval + ID string `json:"id"` + Note string `json:"note,omitempty"` +} + +// Event is one immutable record. +type Event struct { + ID ids.ID `json:"id"` + Seq int64 `json:"seq"` + MindID ids.ID `json:"mind_id"` + Type Type `json:"type"` + Category Category `json:"category"` + Source string `json:"source"` // process, integration or actor that produced it + OccurredAt time.Time `json:"occurred_at"` + RecordedAt time.Time `json:"recorded_at"` + ParentIDs []ids.ID `json:"parent_ids"` + CausationID ids.ID `json:"causation_id,omitempty"` // the event that directly caused this one + CorrelationID ids.ID `json:"correlation_id,omitempty"` // cycle, conversation or job this belongs to + Salience float64 `json:"salience"` + Confidence float64 `json:"confidence"` + Valence *float64 `json:"valence,omitempty"` + Arousal *float64 `json:"arousal,omitempty"` + Evidence []EvidenceRef `json:"evidence"` + Summary string `json:"summary"` + Metadata map[string]any `json:"metadata"` + Payload json.RawMessage `json:"payload"` +} + +// New describes an event to append. Zero OccurredAt means now. +type New struct { + MindID ids.ID + Type Type + Source string + OccurredAt time.Time + ParentIDs []ids.ID + CausationID ids.ID + CorrelationID ids.ID + Salience float64 + Confidence float64 + Valence *float64 + Arousal *float64 + Evidence []EvidenceRef + Summary string + Metadata map[string]any + Payload any // marshalled to JSON; nil → {} +} + +// ObservationPayload is the payload contract for Observation events. Every +// piece of external data carries its provenance and trust level (spec §25). +type ObservationPayload struct { + Text string `json:"text"` + ContentType string `json:"content_type,omitempty"` // text/plain, text/markdown, application/json ... + Source string `json:"source"` // conversation, operator, research_folder, git, prometheus, webhook ... + Trust Trust `json:"trust"` + Method string `json:"method,omitempty"` // how it was ingested: chat, api, watch, poll, push + Reference string `json:"reference,omitempty"` // original locator: message id, file path, URL, commit + Channel string `json:"channel,omitempty"` // conversation or integration identifier + Actor string `json:"actor,omitempty"` // who said or produced it, if known +} + +// Trust is the credibility class assigned to an external source. It never +// exceeds the trust of the channel the data arrived through. +type Trust string + +// Trust levels, from most to least trusted. +const ( + TrustOperator Trust = "operator" // an authenticated operator speaking as themselves + TrustUser Trust = "user" // an authenticated user in conversation + TrustSystem Trust = "system" // Noema's own instrumentation + TrustTool Trust = "tool" // output of a capability invocation + TrustSource Trust = "source" // a configured integration (metrics, repo, feed) + TrustLow Trust = "low" // arbitrary documents, web pages, unknown senders +) + +// Valid reports whether t is a known trust level. +func (t Trust) Valid() bool { + switch t { + case TrustOperator, TrustUser, TrustSystem, TrustTool, TrustSource, TrustLow: + return true + } + return false +} + +// Credibility maps trust to a base weight used by belief and memory scoring. +func (t Trust) Credibility() float64 { + switch t { + case TrustOperator: + return 0.95 + case TrustUser: + return 0.85 + case TrustSystem: + return 0.9 + case TrustTool: + return 0.8 + case TrustSource: + return 0.7 + default: + return 0.4 + } +} diff --git a/web/static/app.css b/web/static/app.css index 97eafd9..0083178 100644 --- a/web/static/app.css +++ b/web/static/app.css @@ -130,3 +130,24 @@ label.check { flex-direction: row; align-items: center; gap: .5rem; font-weight: .grow { flex: 1; } ul.plain { list-style: none; padding: 0; margin: 0; } .diff-old { color: var(--danger); text-decoration: line-through; } .diff-new { color: var(--ok); } + +/* Event stream */ +.filterbar { display: flex; flex-wrap: wrap; gap: .6rem 1rem; align-items: flex-end; margin-bottom: 1rem; } +.filterbar label { font-size: .8rem; } +.filterbar input, .filterbar select { padding: .35rem .5rem; font-size: .88rem; } +.table .num { text-align: right; font-variant-numeric: tabular-nums; } +.table td.summary { max-width: 60ch; } +.etype { display: inline-block; font-family: var(--mono); font-size: .74rem; padding: .05rem .4rem; border-radius: 4px; background: var(--surface-2); color: var(--muted); white-space: nowrap; border-left: 3px solid var(--muted); } +.etype.cat-perception { border-left-color: #0ea5a4; } .etype.cat-cognition { border-left-color: var(--accent); } .etype.cat-memory { border-left-color: #8b5cf6; } +.etype.cat-belief { border-left-color: #d97706; } .etype.cat-prediction { border-left-color: #db2777; } .etype.cat-goal { border-left-color: #16a34a; } +.etype.cat-affect { border-left-color: #e11d48; } .etype.cat-action { border-left-color: #b45309; } .etype.cat-llm { border-left-color: #64748b; } +.etype.cat-operator { border-left-color: #0f766e; } .etype.cat-system { border-left-color: var(--border); } +tr.fresh { animation: fresh 2.5s ease-out; } +@keyframes fresh { from { background: var(--accent-soft); } to { background: transparent; } } +body.stream-offline [data-stream-status] { background: var(--danger-soft); color: var(--danger); } +pre.json { background: var(--surface-2); padding: .75rem; border-radius: 8px; overflow-x: auto; font-size: .82rem; margin: 0; max-height: 28rem; } +.chain-node { padding: .5rem .7rem; border: 1px solid var(--border); border-radius: 8px; margin: .2rem 0; } +.chain-node.current { border-color: var(--accent); background: var(--accent-soft); } +.chain-arrow { text-align: center; color: var(--muted); } +ol.corr li.current { font-weight: 600; } +.cats { display: flex; flex-wrap: wrap; gap: .3rem; margin: .4rem 0; } diff --git a/web/static/app.js b/web/static/app.js index c546221..fce217b 100644 --- a/web/static/app.js +++ b/web/static/app.js @@ -49,7 +49,55 @@ return { close: () => es && es.close() }; } + // Escape text for safe HTML insertion when building rows from JSON. + const esc = (s) => String(s == null ? "" : s).replace(/[&<>"']/g, (c) => ({ "&": "&", "<": "<", ">": ">", '"': """, "'": "'" }[c])); + const f2 = (n) => (typeof n === "number" ? n.toFixed(2) : ""); + const hms = (iso) => { const d = new Date(iso); return isNaN(d) ? "" : d.toTimeString().slice(0, 8); }; + + function eventRow(ev) { + const tr = document.createElement("tr"); + tr.dataset.seq = ev.seq; tr.dataset.id = ev.id; tr.className = "cat-" + ev.category + " fresh"; + const evid = (ev.evidence && ev.evidence.length) ? ` ◉${ev.evidence.length}` : ""; + const cause = ev.causation_id ? `↑` : ""; + tr.innerHTML = `${hms(ev.occurred_at)}` + + `${esc(ev.type)}` + + `${esc(ev.source)}` + + `${esc(ev.summary)}` + + `${f2(ev.salience)}${f2(ev.confidence)}` + + `${cause}${evid}`; + return tr; + } + + // Live event tables: + function attachStreams() { + document.querySelectorAll("table[data-stream]").forEach((table) => { + const body = table.querySelector("[data-stream-target]"); + const status = document.querySelector("[data-stream-status]"); + const toggle = document.querySelector("[data-live-toggle]"); + let url = table.dataset.stream; + const after = table.dataset.streamAfter; + if (after && after !== "0") url += (url.includes("?") ? "&" : "?") + "after=" + encodeURIComponent(after); + let handle = null; + const start = () => { + handle = stream(url, { + cognitive_event: (ev) => { + const empty = body.querySelector("[data-empty]"); if (empty) empty.remove(); + if (body.querySelector(`tr[data-seq="${ev.seq}"]`)) return; + body.insertBefore(eventRow(ev), body.firstChild); + while (body.children.length > 300) body.removeChild(body.lastChild); + if (status) status.textContent = "live"; + }, + }); + if (status) status.textContent = "live"; + }; + const stop = () => { if (handle) handle.close(); handle = null; if (status) status.textContent = "paused"; }; + if (!toggle || toggle.checked) start(); + if (toggle) toggle.addEventListener("change", () => (toggle.checked ? start() : stop())); + }); + } + document.addEventListener("DOMContentLoaded", () => { + attachStreams(); const tt = document.querySelector("[data-theme-toggle]"); if (tt) tt.addEventListener("click", cycleTheme); const nt = document.querySelector("[data-nav-toggle]"); diff --git a/web/templates/pages/dashboard.html b/web/templates/pages/dashboard.html index 568350d..95df98e 100644 --- a/web/templates/pages/dashboard.html +++ b/web/templates/pages/dashboard.html @@ -1,11 +1,16 @@ {{define "content"}}

Dashboard

System overview. Minds, cognition and approvals appear here as they come online.

-
Running minds
{{.Data.Minds}}
No minds yet. Create one under Minds.
+
Active minds
{{len .Data.Minds}}
{{range .Data.Minds}}{{.DisplayName}} {{else}}No minds yet. Create one under Minds.{{end}}
System health
{{if .Data.Status.Database.OK}}Healthy{{else}}Degraded{{end}}
Database {{if .Data.Status.Database.OK}}connected{{else}}unavailable{{end}} · up {{duration .Data.Status.UptimeSeconds}}
Outstanding approvals
0
Capability layer arrives in a later phase.
+
+

Recent important events salience ≥ 0.6, all minds

+ {{if .Data.Events}}
{{range .Data.Events}}{{end}}
{{.OccurredAt.Local.Format "15:04:05"}}{{.Type}}{{.Summary}}{{index $.Data.MindNames .MindID}}
+ {{else}}

Nothing yet.

{{end}} +

Recent administrative activity

{{if .Data.Audit}} diff --git a/web/templates/pages/event.html b/web/templates/pages/event.html new file mode 100644 index 0000000..13f847a --- /dev/null +++ b/web/templates/pages/event.html @@ -0,0 +1,53 @@ +{{define "content"}} +{{$c := .Data.Ctx}}{{$e := $c.Event}} +
+

{{.Data.Mind.DisplayName}} › thoughts › event #{{$e.Seq}}

+

{{$e.Type}} {{$e.Summary}}

+

{{$e.OccurredAt.Local.Format "2006-01-02 15:04:05.000"}} · source {{$e.Source}} · salience {{conf $e.Salience}} · confidence {{conf $e.Confidence}}{{if $e.Valence}} · valence {{conf (deref $e.Valence)}}{{end}}{{if $e.Arousal}} · arousal {{conf (deref $e.Arousal)}}{{end}}

+
+
+
+
+

Causal chain

+
+ {{if $c.Cause}}
caused by
{{$c.Cause.Type}} {{$c.Cause.Summary}}
↓
{{end}} +
{{$e.Type}} {{$e.Summary}}
+ {{if $c.Children}}
↓
{{range $c.Children}}
{{.Type}} {{.Summary}} {{.Source}}
{{end}}{{else}}
No downstream events yet.
{{end}} +
+
+
+

Evidence what this event rests on

+ {{if $c.EvidenceAll}} + {{range $c.EvidenceAll}}{{end}} +
KindReferenceNote
{{.Kind}}{{if eq .Kind "event"}}{{.ID}}{{else if eq .Kind "memory"}}{{.ID}}{{else if eq .Kind "belief"}}{{.ID}}{{else}}{{.ID}}{{end}}{{.Note}}
+ {{if $c.Evidence}}

Cited events

{{end}} + {{else}}

No evidence references. {{if eq $e.Category "perception"}}Observations are primary evidence themselves.{{end}}

{{end}} +
+
+

Payload

+ 
{{.Data.Payload}}
+ {{if $e.Metadata}}

Metadata

{{.Data.Metadata}}
{{end}} +
+
+
+
+

Identifiers

+
+
Event
{{$e.ID}}
+
Seq
{{$e.Seq}}
+
Recorded
{{$e.RecordedAt.Local.Format "15:04:05.000"}}
+ {{if $e.CorrelationID}}
Correlation
{{$e.CorrelationID}}
{{end}} + {{if $e.ParentIDs}}
Parents
{{range $e.ParentIDs}}{{.}}
{{end}}
{{end}} +
+
+
+

Derived from this events citing it as evidence

+ {{if $c.Derived}}{{else}}

Nothing yet.

{{end}} +
+ {{if $c.Correlated}}
+

Same correlation {{len $c.Correlated}} events

+
    {{range $c.Correlated}}
  1. {{.Type}} {{.Summary}}
    2. {{end}}
+
{{end}} +
+
+{{end}} diff --git a/web/templates/pages/mind.html b/web/templates/pages/mind.html index 9c2c61e..69b5924 100644 --- a/web/templates/pages/mind.html +++ b/web/templates/pages/mind.html @@ -5,11 +5,7 @@

{{$m.Personality.Identity.Role}}{{with $m.Personality.Identity.Description}} · {{.}}{{end}}

{{template "status_badge" $m.Status}}
- +{{template "mind_tabs" (dict "M" $m "Active" "overview")}}

Identity

@@ -24,8 +20,17 @@
-

Current focus

-

Working memory, workspace and narrative appear here once the cognitive runtime is online (phase 4).

+

Recent activity

+ {{$st := .Data.Stats}} +
{{$st.Total}} events in 24h
+
{{range $k, $v := $st.ByCategory}}{{$k}} {{$v}} {{end}}
+

Open thought stream

+ {{if .Principal.User.Role.AtLeast "operator"}} +
+ + +
+
{{end}}

Cognitive configuration

diff --git a/web/templates/pages/personality.html b/web/templates/pages/personality.html index 0c94ba2..980bcf4 100644 --- a/web/templates/pages/personality.html +++ b/web/templates/pages/personality.html @@ -4,7 +4,7 @@

{{$m.DisplayName}} personality

Version {{$m.PersonalityVersion}}{{if .Data.Template}} · based on template {{.Data.Template.Name}}{{end}}. Every save creates a new version; nothing is overwritten.

Export JSON
- +{{template "mind_tabs" (dict "M" $m "Active" "personality")}} {{$admin := .Principal.User.Role.AtLeast "administrator"}}
diff --git a/web/templates/pages/thoughts.html b/web/templates/pages/thoughts.html new file mode 100644 index 0000000..5c5afac --- /dev/null +++ b/web/templates/pages/thoughts.html @@ -0,0 +1,36 @@ +{{define "content"}} +{{$m := .Data.Mind}} +
+

{{$m.DisplayName}} thought stream

Every cognitive event, newest first. connecting…

+
+
+{{template "mind_tabs" (dict "M" $m "Active" "thoughts")}} +
+ + + + + + + + Reset +
+
+ + + + {{range .Data.Events}}{{template "event_row" .}}{{else}}{{end}} + +
TimeEventSourceSummarySalConf
No events yet. Operator actions, observations and cognition will appear here.
+ {{with .Data.NextBefore}}

Older

{{end}} +
+{{end}} +{{define "event_row"}} + {{.OccurredAt.Local.Format "15:04:05"}} + {{.Type}} + {{.Source}} + {{.Summary}} + {{conf .Salience}} + {{conf .Confidence}} + {{if .CausationID}}↑{{end}}{{if .Evidence}} ◉{{len .Evidence}}{{end}} +{{end}} diff --git a/web/templates/partials/mind_tabs.html b/web/templates/partials/mind_tabs.html new file mode 100644 index 0000000..b8943d4 --- /dev/null +++ b/web/templates/partials/mind_tabs.html @@ -0,0 +1,8 @@ +{{define "mind_tabs"}}{{$m := .M}}{{$a := .Active}} + +{{end}}