feat: implement global workspace, working memory and the cognitive cycle
Phase 4 (ADR-0001). internal/attention scores thought candidates as a normalised weighted sum over ten named features with a per-term breakdown, ranked by (score, id) so ordering is total and replayable. internal/workspace holds the capacity-limited workspace (refresh by subject, decayed competition, explained admit/reject outcomes) and the decaying working memory. internal/brain runs one runtime per active mind: debounced wake-ups from the event bus (own output never re-triggers), a settle window so concurrently committing events are not skipped, bounded concurrent Observe calls with per-process isolation and timeouts, deterministic scoring, a loop detector that suppresses recurring thoughts and raises a metacognition warning, executive dispatch by thought kind, and one transaction per cycle whose events are published only after commit. Every candidate, selection, attention shift and broadcast is an event with provenance back to the triggering observation. Operators can run a cycle, pause, resume, reset transient state and change attention weights live; the manager reconciles runtimes with mind status. cognition/attend is the first process: deterministic candidates from external events using an urgency lexicon, novelty against context and trust-based operator priority.
This commit is contained in:
Andy Dixon
@@ -0,0 +1,192 @@
|
||||
package brain
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/andydixon/noema/internal/attention"
|
||||
evstore "github.com/andydixon/noema/internal/events"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/scheduler"
|
||||
"github.com/andydixon/noema/internal/storage"
|
||||
"github.com/andydixon/noema/internal/workspace"
|
||||
"github.com/andydixon/noema/migrations"
|
||||
"github.com/andydixon/noema/pkg/events"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
)
|
||||
|
||||
// echo proposes one candidate per new external event; repeat proposes the
|
||||
// same subject every cycle to exercise the loop detector; boom panics.
|
||||
type echo struct{ calls atomic.Int32 }
|
||||
|
||||
func (e *echo) Name() string { return "echo" }
|
||||
func (e *echo) Observe(_ context.Context, s *State) ([]Candidate, error) {
|
||||
e.calls.Add(1)
|
||||
var out []Candidate
|
||||
for _, ev := range s.NewEvents {
|
||||
if ev.Type == events.Observation {
|
||||
out = append(out, Candidate{Kind: "attend", Subject: string(ev.ID), Summary: "noticed: " + ev.Summary, Features: attention.Features{Salience: ev.Salience, Urgency: 0.5}, Confidence: 1, SourceEvents: []ids.ID{ev.ID}})
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
type repeat struct{}
|
||||
|
||||
func (repeat) Name() string { return "repeat" }
|
||||
func (repeat) Observe(_ context.Context, s *State) ([]Candidate, error) {
|
||||
return []Candidate{{Kind: "note", Subject: "same-thing", Summary: "the same thought again", Features: attention.Features{Salience: 0.9}, Confidence: 0.5}}, nil
|
||||
}
|
||||
|
||||
type boom struct{}
|
||||
|
||||
func (boom) Name() string { return "boom" }
|
||||
func (boom) Observe(context.Context, *State) ([]Candidate, error) { panic("malformed input") }
|
||||
|
||||
type failing struct{}
|
||||
|
||||
func (failing) Name() string { return "failing" }
|
||||
func (failing) Observe(context.Context, *State) ([]Candidate, error) {
|
||||
return nil, errors.New("upstream down")
|
||||
}
|
||||
|
||||
func setup(t *testing.T, reg *Registry) (Config, minds.Mind, *evstore.Store) {
|
||||
t.Helper()
|
||||
pool := storage.TestPool(t, migrations.FS)
|
||||
ctx := context.Background()
|
||||
ms := minds.NewStore(pool)
|
||||
m, err := ms.Create(ctx, minds.CreateParams{Slug: "brain"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
es := evstore.NewStore(pool, evstore.NewBus(16), nil)
|
||||
cfg := Config{DB: pool, Events: es, Minds: ms, Registry: reg, Pool: scheduler.NewPool(2), Log: slog.New(slog.NewTextHandler(os.Stderr, nil)), Services: Services{Events: es}}
|
||||
return cfg, m, es
|
||||
}
|
||||
|
||||
func TestCycleRecordsProvenanceAndPersists(t *testing.T) {
|
||||
e := &echo{}
|
||||
cfg, m, es := setup(t, NewRegistry().AddProcess(e).AddProcess(boom{}).AddProcess(failing{}))
|
||||
ctx := context.Background()
|
||||
rt, err := NewRuntime(ctx, cfg, m)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rt.now = func() time.Time { return time.Now().Add(time.Second) } // events are "settled"
|
||||
obs, _ := es.Append(ctx, events.New{MindID: m.ID, Type: events.Observation, Source: "test", Summary: "load spike on atlas", Salience: 0.9, Confidence: 1})
|
||||
rt.runCycle(ctx, "operator")
|
||||
|
||||
if e.calls.Load() != 1 {
|
||||
t.Fatalf("echo called %d times", e.calls.Load())
|
||||
}
|
||||
st := rt.copyState()
|
||||
if st.Cycle != 1 || len(st.Workspace) != 1 || st.Workspace[0].Subject != string(obs[0].ID) || len(st.WorkingMemory) != 1 {
|
||||
t.Fatalf("workspace wrong: %+v", st)
|
||||
}
|
||||
all, _ := es.Query(ctx, evstore.Filter{MindID: m.ID, Ascending: true})
|
||||
types := map[events.Type]int{}
|
||||
var selected events.Event
|
||||
for _, ev := range all {
|
||||
types[ev.Type]++
|
||||
if ev.Type == events.ThoughtSelected {
|
||||
selected = ev
|
||||
}
|
||||
}
|
||||
for _, want := range []events.Type{events.ThoughtCandidate, events.ThoughtSelected, events.WorkspaceBroadcast, events.AttentionShift, events.CycleCompleted} {
|
||||
if types[want] == 0 {
|
||||
t.Fatalf("missing %s in %v", want, types)
|
||||
}
|
||||
}
|
||||
// Provenance: selected thought → candidate event → observation.
|
||||
if selected.CausationID.IsZero() {
|
||||
t.Fatal("thought_selected has no causation")
|
||||
}
|
||||
cand, _ := es.Get(ctx, selected.CausationID)
|
||||
if cand.Type != events.ThoughtCandidate || cand.CausationID != obs[0].ID {
|
||||
t.Fatalf("provenance chain broken: %+v", cand)
|
||||
}
|
||||
if cur, _ := es.Cursor(ctx, cfg.DB, m.ID, cursorConsumer); cur != obs[0].Seq {
|
||||
t.Fatalf("cursor %d, want %d", cur, obs[0].Seq)
|
||||
}
|
||||
// Restart continuity: a fresh runtime loads the same state.
|
||||
rt2, err := NewRuntime(ctx, cfg, m)
|
||||
if err != nil || rt2.state.Cycle != 1 || len(rt2.state.Workspace) != 1 {
|
||||
t.Fatalf("state not persisted: %v %+v", err, rt2.state)
|
||||
}
|
||||
// A second cycle with nothing new triggered by stimulus does nothing.
|
||||
rt2.now = rt.now
|
||||
rt2.runCycle(ctx, "stimulus")
|
||||
if rt2.state.Cycle != 1 {
|
||||
t.Fatal("cycle ran without input")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopDetectorSuppressesRepeats(t *testing.T) {
|
||||
cfg, m, es := setup(t, NewRegistry().AddProcess(repeat{}))
|
||||
ctx := context.Background()
|
||||
m.Config.LoopDetectThreshold, m.Config.LoopDetectWindow = 3, 10
|
||||
rt, err := NewRuntime(ctx, cfg, m)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rt.now = func() time.Time { return time.Now().Add(time.Second) }
|
||||
for i := 0; i < 6; i++ {
|
||||
es.Append(ctx, events.New{MindID: m.ID, Type: events.Observation, Source: "test", Summary: "tick", Salience: 0.1, Confidence: 1}) //nolint:errcheck
|
||||
rt.runCycle(ctx, "operator")
|
||||
}
|
||||
meta, _ := es.Query(ctx, evstore.Filter{MindID: m.ID, Types: []events.Type{events.Metacognition}})
|
||||
if len(meta) != 1 {
|
||||
t.Fatalf("expected exactly one loop warning, got %d", len(meta))
|
||||
}
|
||||
cands, _ := es.Query(ctx, evstore.Filter{MindID: m.ID, Types: []events.Type{events.ThoughtCandidate}})
|
||||
var suppressed int
|
||||
for _, c := range cands {
|
||||
if c.Metadata["fate"] == "suppressed" {
|
||||
suppressed++
|
||||
}
|
||||
}
|
||||
if suppressed != 3 {
|
||||
t.Fatalf("expected 3 suppressed candidates, got %d", suppressed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPauseResetAndCancellation(t *testing.T) {
|
||||
cfg, m, es := setup(t, NewRegistry().AddProcess(&echo{}))
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
rt, _ := NewRuntime(ctx, cfg, m)
|
||||
done := make(chan struct{})
|
||||
go func() { rt.Run(ctx); close(done) }() //nolint:errcheck
|
||||
if err := rt.Command(ctx, "pause", nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := rt.Command(ctx, "cycle", nil); err == nil {
|
||||
t.Fatal("cycle should be refused while paused")
|
||||
}
|
||||
if err := rt.Command(ctx, "resume", nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := rt.Command(ctx, "reset", map[string]bool{"workspace": true, "working_memory": true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
modes, _ := es.Query(ctx, evstore.Filter{MindID: m.ID, Types: []events.Type{events.ModeChanged}})
|
||||
if len(modes) != 2 {
|
||||
t.Fatalf("expected 2 mode changes, got %d", len(modes))
|
||||
}
|
||||
st, err := rt.Snapshot(ctx)
|
||||
if err != nil || st.Mode != string(ModeAwake) {
|
||||
t.Fatalf("snapshot: %v %+v", err, st)
|
||||
}
|
||||
cancel()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("runtime did not stop on cancellation")
|
||||
}
|
||||
_ = workspace.State{}
|
||||
}
|
||||
@@ -0,0 +1,661 @@
|
||||
package brain
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/andydixon/noema/internal/attention"
|
||||
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"
|
||||
"github.com/andydixon/noema/internal/storage"
|
||||
"github.com/andydixon/noema/internal/workspace"
|
||||
"github.com/andydixon/noema/pkg/events"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
)
|
||||
|
||||
// cursorConsumer names the brain's durable read position.
|
||||
const cursorConsumer = "brain"
|
||||
|
||||
// settle is how old an event must be before the brain reads it, so that a
|
||||
// concurrently committing transaction with a lower sequence number is not
|
||||
// skipped forever (see ADR-0002 and PLAN risk R3).
|
||||
const settle = 250 * time.Millisecond
|
||||
|
||||
// processTimeout bounds one process's Observe call.
|
||||
const processTimeout = 15 * time.Second
|
||||
|
||||
// Cycle is the working context of one cognitive cycle, handed to handlers.
|
||||
type Cycle struct {
|
||||
Runtime *Runtime
|
||||
State *State
|
||||
Tx pgx.Tx
|
||||
pending []events.New
|
||||
emitted []events.Event
|
||||
}
|
||||
|
||||
// Emit queues an event for the cycle's transaction. Correlation is the cycle
|
||||
// id unless the caller set one.
|
||||
func (c *Cycle) Emit(n events.New) {
|
||||
n.MindID = c.State.Mind.ID
|
||||
if n.CorrelationID.IsZero() {
|
||||
n.CorrelationID = c.State.CycleID
|
||||
}
|
||||
if n.OccurredAt.IsZero() {
|
||||
n.OccurredAt = c.State.Now
|
||||
}
|
||||
if n.Metadata == nil {
|
||||
n.Metadata = map[string]any{}
|
||||
}
|
||||
n.Metadata["cycle"] = c.State.Cycle
|
||||
c.pending = append(c.pending, n)
|
||||
}
|
||||
|
||||
// flush writes queued events inside the transaction and returns them.
|
||||
func (c *Cycle) flush(ctx context.Context) error {
|
||||
if len(c.pending) == 0 {
|
||||
return nil
|
||||
}
|
||||
evs, err := c.Runtime.events.AppendIn(ctx, c.Tx, c.pending...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c.emitted = append(c.emitted, evs...)
|
||||
c.pending = nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// Runtime is the cognitive loop of one mind.
|
||||
type Runtime struct {
|
||||
mind minds.Mind
|
||||
db *pgxpool.Pool
|
||||
events *evstore.Store
|
||||
minds *minds.Store
|
||||
registry *Registry
|
||||
pool *scheduler.Pool
|
||||
log *slog.Logger
|
||||
metrics *observability.Metrics
|
||||
services Services
|
||||
|
||||
wake chan struct{}
|
||||
commands chan command
|
||||
snaps chan chan workspace.State
|
||||
state workspace.State
|
||||
loops *loopDetector
|
||||
budget *cycleBudget
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
type command struct {
|
||||
kind string // cycle | pause | resume | reset
|
||||
args map[string]bool
|
||||
reply chan error
|
||||
}
|
||||
|
||||
// Config groups runtime dependencies.
|
||||
type Config struct {
|
||||
DB *pgxpool.Pool
|
||||
Events *evstore.Store
|
||||
Minds *minds.Store
|
||||
Registry *Registry
|
||||
Pool *scheduler.Pool
|
||||
Log *slog.Logger
|
||||
Metrics *observability.Metrics
|
||||
Services Services
|
||||
}
|
||||
|
||||
// NewRuntime creates a runtime for a mind and loads its persisted state.
|
||||
func NewRuntime(ctx context.Context, cfg Config, m minds.Mind) (*Runtime, error) {
|
||||
r := &Runtime{mind: m, db: cfg.DB, events: cfg.Events, minds: cfg.Minds, registry: cfg.Registry, pool: cfg.Pool,
|
||||
log: cfg.Log.With("mind", m.Slug), metrics: cfg.Metrics, services: cfg.Services,
|
||||
wake: make(chan struct{}, 1), commands: make(chan command), snaps: make(chan chan workspace.State), now: time.Now}
|
||||
r.loops = newLoopDetector(m.Config.LoopDetectWindow, m.Config.LoopDetectThreshold)
|
||||
r.budget = newCycleBudget(m.Config.MaxCyclesPerMinute)
|
||||
if err := r.loadState(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// persisted wraps the workspace state with runtime bookkeeping.
|
||||
type persisted struct {
|
||||
workspace.State
|
||||
LoopRing []string `json:"loop_ring,omitempty"`
|
||||
}
|
||||
|
||||
func (r *Runtime) loadState(ctx context.Context) error {
|
||||
var raw []byte
|
||||
err := r.db.QueryRow(ctx, `SELECT state FROM workspace_states WHERE mind_id=$1`, r.mind.ID).Scan(&raw)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
r.state = workspace.State{Mode: string(ModeAwake)}
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var p persisted
|
||||
if err := json.Unmarshal(raw, &p); err != nil {
|
||||
return fmt.Errorf("corrupt workspace state: %w", err)
|
||||
}
|
||||
r.state = p.State
|
||||
r.loops.restore(p.LoopRing)
|
||||
if r.state.Mode == "" {
|
||||
r.state.Mode = string(ModeAwake)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Runtime) saveState(ctx context.Context, q storage.Querier) error {
|
||||
raw, _ := json.Marshal(persisted{State: r.state, LoopRing: r.loops.snapshot()})
|
||||
_, err := q.Exec(ctx, `INSERT INTO workspace_states (mind_id, state) VALUES ($1,$2) ON CONFLICT (mind_id) DO UPDATE SET state=excluded.state, updated_at=now()`, r.mind.ID, raw)
|
||||
return err
|
||||
}
|
||||
|
||||
// Wake asks the runtime to run a cycle soon (debounced by CycleMinInterval).
|
||||
func (r *Runtime) Wake() {
|
||||
select {
|
||||
case r.wake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// Command sends an operator command and waits for the result.
|
||||
func (r *Runtime) Command(ctx context.Context, kind string, args map[string]bool) error {
|
||||
reply := make(chan error, 1)
|
||||
select {
|
||||
case r.commands <- command{kind: kind, args: args, reply: reply}:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
select {
|
||||
case err := <-reply:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// Snapshot returns a copy of the current workspace state, taken on the loop
|
||||
// goroutine so no lock is needed.
|
||||
func (r *Runtime) Snapshot(ctx context.Context) (workspace.State, error) {
|
||||
reply := make(chan workspace.State, 1)
|
||||
select {
|
||||
case r.snaps <- reply:
|
||||
case <-ctx.Done():
|
||||
return workspace.State{}, ctx.Err()
|
||||
}
|
||||
select {
|
||||
case st := <-reply:
|
||||
return st, nil
|
||||
case <-ctx.Done():
|
||||
return workspace.State{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// Run is the loop. It returns when ctx ends.
|
||||
func (r *Runtime) Run(ctx context.Context) error {
|
||||
r.log.Info("mind runtime started", "mode", r.state.Mode, "cycle", r.state.Cycle)
|
||||
idle := time.NewTimer(r.mind.Config.IdleCycleInterval.D())
|
||||
defer idle.Stop()
|
||||
var debounce *time.Timer
|
||||
var debounceC <-chan time.Time
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-r.wake:
|
||||
if debounce == nil {
|
||||
debounce = time.NewTimer(r.mind.Config.CycleMinInterval.D() + settle)
|
||||
debounceC = debounce.C
|
||||
}
|
||||
case <-debounceC:
|
||||
debounce, debounceC = nil, nil
|
||||
r.runCycle(ctx, "stimulus")
|
||||
idle.Reset(r.mind.Config.IdleCycleInterval.D())
|
||||
case <-idle.C:
|
||||
r.runCycle(ctx, "idle")
|
||||
idle.Reset(r.mind.Config.IdleCycleInterval.D())
|
||||
case cmd := <-r.commands:
|
||||
cmd.reply <- r.handleCommand(ctx, cmd)
|
||||
case reply := <-r.snaps:
|
||||
reply <- r.copyState()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Runtime) handleCommand(ctx context.Context, cmd command) error {
|
||||
switch cmd.kind {
|
||||
case "cycle":
|
||||
if Mode(r.state.Mode) == ModePaused {
|
||||
return errors.New("mind is paused")
|
||||
}
|
||||
r.runCycle(ctx, "operator")
|
||||
return nil
|
||||
case "pause":
|
||||
return r.setMode(ctx, ModePaused, "operator paused cognition")
|
||||
case "resume":
|
||||
if Mode(r.state.Mode) != ModePaused {
|
||||
return nil
|
||||
}
|
||||
return r.setMode(ctx, ModeAwake, "operator resumed cognition")
|
||||
case "reset":
|
||||
return r.reset(ctx, cmd.args)
|
||||
case "reload":
|
||||
m, err := r.minds.Get(ctx, r.mind.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
r.mind = m
|
||||
r.loops = newLoopDetector(m.Config.LoopDetectWindow, m.Config.LoopDetectThreshold)
|
||||
r.budget = newCycleBudget(m.Config.MaxCyclesPerMinute)
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("unknown command %q", cmd.kind)
|
||||
}
|
||||
|
||||
func (r *Runtime) setMode(ctx context.Context, to Mode, why string) error {
|
||||
from := r.state.Mode
|
||||
if from == string(to) {
|
||||
return nil
|
||||
}
|
||||
r.state.Mode = string(to)
|
||||
return storage.Tx(ctx, r.db, func(tx pgx.Tx) error {
|
||||
if _, err := r.events.AppendIn(ctx, tx, events.New{MindID: r.mind.ID, Type: events.ModeChanged, Source: "brain", Summary: fmt.Sprintf("mode %s → %s: %s", from, to, why),
|
||||
Salience: 0.3, Confidence: 1, Metadata: map[string]any{"from": from, "to": to}, Payload: map[string]any{"from": from, "to": to, "reason": why}}); err != nil {
|
||||
return err
|
||||
}
|
||||
return r.saveState(ctx, tx)
|
||||
})
|
||||
}
|
||||
|
||||
// reset clears transient state on operator request (spec §3). It never touches
|
||||
// events, memories or beliefs.
|
||||
func (r *Runtime) reset(ctx context.Context, what map[string]bool) error {
|
||||
var parts []string
|
||||
if what["workspace"] {
|
||||
r.state.Workspace = nil
|
||||
parts = append(parts, "workspace")
|
||||
}
|
||||
if what["working_memory"] {
|
||||
r.state.WorkingMemory = nil
|
||||
parts = append(parts, "working memory")
|
||||
}
|
||||
if what["loops"] {
|
||||
r.loops = newLoopDetector(r.mind.Config.LoopDetectWindow, r.mind.Config.LoopDetectThreshold)
|
||||
parts = append(parts, "loop detector")
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return nil
|
||||
}
|
||||
return storage.Tx(ctx, r.db, func(tx pgx.Tx) error {
|
||||
if _, err := r.events.AppendIn(ctx, tx, events.New{MindID: r.mind.ID, Type: events.OperatorAction, Source: "operator", Summary: "Operator reset transient state: " + strings.Join(parts, ", "),
|
||||
Salience: 0.5, Confidence: 1, Metadata: map[string]any{"action": "reset"}, Payload: map[string]any{"reset": parts}}); err != nil {
|
||||
return err
|
||||
}
|
||||
return r.saveState(ctx, tx)
|
||||
})
|
||||
}
|
||||
|
||||
// runCycle executes one cycle. Errors are logged, never fatal: a broken
|
||||
// process or malformed event must not stop the mind (spec §46).
|
||||
func (r *Runtime) runCycle(ctx context.Context, trigger string) {
|
||||
if Mode(r.state.Mode) == ModePaused {
|
||||
return
|
||||
}
|
||||
if !r.budget.allow(r.now()) {
|
||||
r.log.Warn("cycle rate limit reached; skipping", "trigger", trigger)
|
||||
return
|
||||
}
|
||||
start := r.now()
|
||||
cctx, cancel := context.WithTimeout(ctx, 60*time.Second)
|
||||
defer cancel()
|
||||
var emitted []events.Event
|
||||
before := r.copyState()
|
||||
err := storage.Tx(cctx, r.db, func(tx pgx.Tx) error {
|
||||
var err error
|
||||
emitted, err = r.cycle(cctx, tx, trigger, start)
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
r.state = before // the transaction rolled back; forget in-memory changes too
|
||||
r.log.Error("cycle failed", "trigger", trigger, "err", err)
|
||||
return
|
||||
}
|
||||
r.events.Publish(emitted...) // only after commit, so subscribers never see phantom events
|
||||
}
|
||||
|
||||
func (r *Runtime) cycle(ctx context.Context, tx pgx.Tx, trigger string, start time.Time) ([]events.Event, error) {
|
||||
cursor, err := r.events.Cursor(ctx, tx, r.mind.ID, cursorConsumer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Read settled events after the cursor. The query is done on the pool (not
|
||||
// the tx) so a long transaction does not hold a snapshot; ordering by seq.
|
||||
newEvents, err := r.events.Query(ctx, evstore.Filter{MindID: r.mind.ID, AfterSeq: cursor, Ascending: true, Limit: 200})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Recorded-at settle: Query filters by occurred_at, so also drop events
|
||||
// recorded too recently (occurred_at may be backdated by integrations).
|
||||
filtered := newEvents[:0]
|
||||
for _, ev := range newEvents {
|
||||
if start.Sub(ev.RecordedAt) >= settle {
|
||||
filtered = append(filtered, ev)
|
||||
}
|
||||
}
|
||||
newEvents = filtered
|
||||
stimulus := false
|
||||
for _, ev := range newEvents {
|
||||
if !strings.HasPrefix(ev.Source, "brain") {
|
||||
stimulus = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if trigger == "stimulus" && !stimulus {
|
||||
// Woken by our own output or nothing has settled yet: consume and wait.
|
||||
if len(newEvents) > 0 {
|
||||
if err := r.events.SetCursor(ctx, tx, r.mind.ID, cursorConsumer, newEvents[len(newEvents)-1].Seq); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
if trigger == "idle" && !stimulus && len(r.state.Workspace) == 0 {
|
||||
// Nothing to think about: just move to idle mode.
|
||||
if Mode(r.state.Mode) != ModeIdle && r.state.LastStimulus.Add(5*time.Minute).Before(start) {
|
||||
r.state.Mode = string(ModeIdle)
|
||||
}
|
||||
return nil, r.saveState(ctx, tx)
|
||||
}
|
||||
|
||||
cycleID := ids.New()
|
||||
if stimulus {
|
||||
r.state.LastStimulus = start
|
||||
if Mode(r.state.Mode) == ModeIdle {
|
||||
r.state.Mode = string(ModeAwake)
|
||||
}
|
||||
}
|
||||
sm, _ := r.minds.GetSelfModel(ctx, r.mind.ID)
|
||||
st := &State{Mind: r.mind, SelfModel: sm, Mode: Mode(r.state.Mode), Cycle: r.state.Cycle + 1, CycleID: cycleID, Now: start, NewEvents: newEvents, Workspace: r.state, Services: r.services}
|
||||
c := &Cycle{Runtime: r, State: st, Tx: tx}
|
||||
|
||||
// 1. Ask every process, concurrently and bounded.
|
||||
cands := r.observe(ctx, st)
|
||||
if len(cands) == 0 && len(r.state.Workspace) == 0 {
|
||||
// Nothing proposed and nothing held: consume the events without
|
||||
// counting a cycle, so self-generated events cannot keep the loop busy.
|
||||
if len(newEvents) > 0 {
|
||||
if err := r.events.SetCursor(ctx, tx, r.mind.ID, cursorConsumer, newEvents[len(newEvents)-1].Seq); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return nil, r.saveState(ctx, tx)
|
||||
}
|
||||
r.state.Cycle++
|
||||
|
||||
// 2. Score deterministically; suppress loops.
|
||||
weights := r.mind.Config.AttentionWeights
|
||||
var scored []attention.Scored
|
||||
byID := map[ids.ID]*Candidate{}
|
||||
suppressed := map[ids.ID]string{}
|
||||
for i := range cands {
|
||||
cand := &cands[i]
|
||||
cand.ID = ids.New()
|
||||
byID[cand.ID] = cand
|
||||
b := attention.Score(cand.Features, weights)
|
||||
fp := fingerprint(*cand)
|
||||
if r.loops.looping(fp) {
|
||||
suppressed[cand.ID] = fp
|
||||
b.Total = 0
|
||||
if r.loops.shouldWarn(fp) {
|
||||
c.Emit(events.New{Type: events.Metacognition, Source: "brain.loop-detector", Summary: "suppressed recurring thought without progress: " + cand.Summary,
|
||||
Salience: 0.6, Confidence: 0.9, Metadata: map[string]any{"fingerprint": fp, "check": "loop"}, Payload: map[string]any{"fingerprint": fp, "kind": cand.Kind, "subject": cand.Subject}})
|
||||
}
|
||||
}
|
||||
scored = append(scored, attention.Scored{ID: string(cand.ID), Breakdown: b})
|
||||
}
|
||||
attention.Rank(scored)
|
||||
if r.metrics != nil {
|
||||
r.metrics.Candidates.Add(float64(len(scored)))
|
||||
}
|
||||
|
||||
// 3. Decay, then admit.
|
||||
prevTop, _ := r.state.Top()
|
||||
r.state.Decay(r.mind.Config.WorkingMemoryDecay)
|
||||
var thoughts []workspace.Thought
|
||||
for _, sc := range scored {
|
||||
cand := byID[ids.ID(sc.ID)]
|
||||
if _, isSuppressed := suppressed[cand.ID]; isSuppressed {
|
||||
continue
|
||||
}
|
||||
thoughts = append(thoughts, workspace.Thought{ID: cand.ID, Process: cand.Process, Kind: cand.Kind, Subject: cand.Subject, Summary: cand.Summary, Content: cand.Content,
|
||||
Score: sc.Breakdown.Total, Breakdown: sc.Breakdown, Evidence: cand.Evidence, Confidence: cand.Confidence})
|
||||
}
|
||||
outcomes, displaced := r.state.Admit(thoughts, r.mind.Config.WorkspaceCapacity, r.state.Cycle, start)
|
||||
outcomeByID := map[ids.ID]workspace.Outcome{}
|
||||
for _, o := range outcomes {
|
||||
outcomeByID[o.ID] = o
|
||||
}
|
||||
|
||||
// 4. Record every candidate with its arithmetic and fate.
|
||||
admittedThoughts := map[ids.ID]workspace.Thought{}
|
||||
for _, t := range r.state.Workspace {
|
||||
if t.AdmittedCycle == r.state.Cycle {
|
||||
admittedThoughts[t.ID] = t
|
||||
}
|
||||
}
|
||||
for _, sc := range scored {
|
||||
cand := byID[ids.ID(sc.ID)]
|
||||
o, has := outcomeByID[cand.ID]
|
||||
fate := "rejected"
|
||||
reason := o.Reason
|
||||
if fp, s := suppressed[cand.ID]; s {
|
||||
fate, reason = "suppressed", "loop detector: "+fp
|
||||
} else if has && o.Admitted {
|
||||
fate = "admitted"
|
||||
}
|
||||
var causation ids.ID
|
||||
if len(cand.SourceEvents) > 0 {
|
||||
causation = cand.SourceEvents[0]
|
||||
}
|
||||
c.Emit(events.New{Type: events.ThoughtCandidate, Source: "brain." + cand.Process, Summary: fmt.Sprintf("[%s %.2f] %s", fate, sc.Breakdown.Total, cand.Summary),
|
||||
Salience: sc.Breakdown.Total, Confidence: cand.Confidence, CausationID: causation, ParentIDs: cand.SourceEvents, Evidence: cand.Evidence,
|
||||
Metadata: map[string]any{"candidate_id": cand.ID, "process": cand.Process, "kind": cand.Kind, "subject": cand.Subject, "fate": fate, "rank": o.Rank},
|
||||
Payload: map[string]any{"candidate_id": cand.ID, "kind": cand.Kind, "subject": cand.Subject, "features": cand.Features, "breakdown": sc.Breakdown, "fate": fate, "reason": reason, "rank": o.Rank, "content": cand.Content}})
|
||||
}
|
||||
if err := c.flush(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
candidateEvent := map[ids.ID]ids.ID{}
|
||||
for _, ev := range c.emitted {
|
||||
if ev.Type == events.ThoughtCandidate {
|
||||
if cid, ok := ev.Metadata["candidate_id"].(ids.ID); ok {
|
||||
candidateEvent[cid] = ev.ID
|
||||
} else if s, ok := ev.Metadata["candidate_id"].(string); ok {
|
||||
candidateEvent[ids.ID(s)] = ev.ID
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 5. thought_selected for admissions, then broadcast.
|
||||
for i := range r.state.Workspace {
|
||||
t := &r.state.Workspace[i]
|
||||
if t.AdmittedCycle != r.state.Cycle {
|
||||
continue
|
||||
}
|
||||
cand := byID[t.ID]
|
||||
var parents []ids.ID
|
||||
if pe, ok := candidateEvent[t.ID]; ok {
|
||||
parents = append(parents, pe)
|
||||
}
|
||||
if cand != nil {
|
||||
parents = append(parents, cand.SourceEvents...)
|
||||
}
|
||||
c.Emit(events.New{Type: events.ThoughtSelected, Source: "brain.workspace", Summary: t.Summary, Salience: t.Score, Confidence: t.Confidence,
|
||||
CausationID: firstOr(parents, ids.Nil), ParentIDs: parents, Evidence: t.Evidence,
|
||||
Metadata: map[string]any{"candidate_id": t.ID, "process": t.Process, "kind": t.Kind, "subject": t.Subject, "rank": outcomeByID[t.ID].Rank},
|
||||
Payload: map[string]any{"thought": t, "why": fmt.Sprintf("rank %d of %d with score %.2f; %s", outcomeByID[t.ID].Rank, len(scored), t.Score, outcomeByID[t.ID].Reason)}})
|
||||
if r.metrics != nil {
|
||||
r.metrics.Admissions.Inc()
|
||||
}
|
||||
}
|
||||
if err := c.flush(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, ev := range c.emitted {
|
||||
if ev.Type == events.ThoughtSelected {
|
||||
if s, ok := ev.Metadata["candidate_id"].(string); ok {
|
||||
for i := range r.state.Workspace {
|
||||
if string(r.state.Workspace[i].ID) == s {
|
||||
r.state.Workspace[i].EventID = ev.ID
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 6. Executive: dispatch admitted thoughts to handlers (sequential, ordered by rank).
|
||||
for _, t := range r.state.Workspace {
|
||||
if t.AdmittedCycle != r.state.Cycle {
|
||||
continue
|
||||
}
|
||||
h, ok := r.registry.handlers[t.Kind]
|
||||
if !ok {
|
||||
h = defaultHandler
|
||||
}
|
||||
if err := h(ctx, c, t); err != nil {
|
||||
r.log.Warn("handler failed", "kind", t.Kind, "err", err)
|
||||
c.Emit(events.New{Type: events.Metacognition, Source: "brain.executive", Summary: "handler for " + t.Kind + " failed: " + err.Error(), Salience: 0.4, Confidence: 1,
|
||||
Metadata: map[string]any{"check": "handler_error", "kind": t.Kind}})
|
||||
}
|
||||
r.loops.record(fingerprint(Candidate{Process: t.Process, Kind: t.Kind, Subject: t.Subject}))
|
||||
}
|
||||
pruned := r.state.Prune(0.05)
|
||||
|
||||
// 7. Broadcast the resulting workspace and note attention shifts.
|
||||
if newTop, ok := r.state.Top(); ok && (prevTop.ID == ids.Nil || prevTop.Subject != newTop.Subject || prevTop.Kind != newTop.Kind) {
|
||||
c.Emit(events.New{Type: events.AttentionShift, Source: "brain.workspace", Summary: "attention now on: " + newTop.Summary, Salience: newTop.Score, Confidence: newTop.Confidence,
|
||||
CausationID: newTop.EventID, Metadata: map[string]any{"from": prevTop.Subject, "to": newTop.Subject}, Payload: map[string]any{"from": prevTop.Summary, "to": newTop.Summary}})
|
||||
}
|
||||
if len(scored) > 0 || len(displaced) > 0 || pruned > 0 {
|
||||
c.Emit(events.New{Type: events.WorkspaceBroadcast, Source: "brain.workspace", Summary: fmt.Sprintf("workspace: %d thoughts, %d in working memory", len(r.state.Workspace), len(r.state.WorkingMemory)),
|
||||
Salience: 0.2, Confidence: 1, Metadata: map[string]any{"thoughts": len(r.state.Workspace), "working_memory": len(r.state.WorkingMemory)},
|
||||
Payload: map[string]any{"workspace": r.state.Workspace, "working_memory": r.state.WorkingMemory, "displaced": displaced, "pruned": pruned}})
|
||||
c.Emit(events.New{Type: events.CycleCompleted, Source: "brain", Summary: fmt.Sprintf("cycle %d (%s): %d events, %d candidates, %d admitted, %d suppressed in %dms", r.state.Cycle, trigger, len(newEvents), len(scored), len(admittedThoughts), len(suppressed), time.Since(start).Milliseconds()),
|
||||
Salience: 0.1, Confidence: 1, Metadata: map[string]any{"trigger": trigger, "candidates": len(scored), "admitted": len(admittedThoughts), "suppressed": len(suppressed), "duration_ms": time.Since(start).Milliseconds()}})
|
||||
}
|
||||
if err := c.flush(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 8. Persist and advance the cursor to the last *read* event.
|
||||
r.state.LastCycleID, r.state.LastCycleAt = cycleID, start
|
||||
if len(newEvents) > 0 {
|
||||
if err := r.events.SetCursor(ctx, tx, r.mind.ID, cursorConsumer, newEvents[len(newEvents)-1].Seq); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if err := r.saveState(ctx, tx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.emitted, nil
|
||||
}
|
||||
|
||||
// copyState deep-copies slices so callers cannot alias loop-owned state.
|
||||
func (r *Runtime) copyState() workspace.State {
|
||||
st := r.state
|
||||
st.Workspace = append([]workspace.Thought{}, r.state.Workspace...)
|
||||
st.WorkingMemory = append([]workspace.Item{}, r.state.WorkingMemory...)
|
||||
return st
|
||||
}
|
||||
|
||||
func firstOr(xs []ids.ID, d ids.ID) ids.ID {
|
||||
if len(xs) > 0 {
|
||||
return xs[0]
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// observe runs all processes through the bounded pool with per-process
|
||||
// timeouts and isolation. A failing process yields no candidates.
|
||||
func (r *Runtime) observe(ctx context.Context, st *State) []Candidate {
|
||||
procs := r.registry.Processes()
|
||||
results := make([][]Candidate, len(procs))
|
||||
tasks := make([]func(context.Context) error, len(procs))
|
||||
for i, p := range procs {
|
||||
tasks[i] = func(ctx context.Context) error {
|
||||
pctx, cancel := context.WithTimeout(ctx, processTimeout)
|
||||
defer cancel()
|
||||
out, err := p.Observe(pctx, st)
|
||||
if err != nil {
|
||||
r.log.Warn("process failed", "process", p.Name(), "err", err)
|
||||
return nil
|
||||
}
|
||||
for j := range out {
|
||||
out[j].Process = p.Name()
|
||||
out[j].Features = out[j].Features.Clamp()
|
||||
if out[j].Kind == "" {
|
||||
out[j].Kind = "note"
|
||||
}
|
||||
}
|
||||
results[i] = out
|
||||
return nil
|
||||
}
|
||||
}
|
||||
scheduler.Parallel(ctx, r.pool, tasks) //nolint:errcheck // per-task errors already logged
|
||||
var all []Candidate
|
||||
for _, rs := range results {
|
||||
all = append(all, rs...)
|
||||
}
|
||||
return all
|
||||
}
|
||||
|
||||
// defaultHandler places an admitted thought into working memory.
|
||||
func defaultHandler(_ context.Context, c *Cycle, t workspace.Thought) error {
|
||||
key := t.Kind + ":" + t.Subject
|
||||
if t.Subject == "" {
|
||||
key = t.Kind + ":" + string(t.ID)
|
||||
}
|
||||
ref, _ := t.Content["ref"].(string)
|
||||
var src ids.ID
|
||||
if s, ok := t.Content["source_event"].(string); ok {
|
||||
src = ids.ID(s)
|
||||
}
|
||||
c.Runtime.state.Hold(workspace.Item{Key: key, Content: t.Summary, Kind: t.Kind, SourceEvent: src, Ref: ref}, c.Runtime.mind.Config.WorkingMemoryCapacity, c.State.Cycle, c.State.Now)
|
||||
return nil
|
||||
}
|
||||
|
||||
// cycleBudget is a per-minute token bucket for cycles.
|
||||
type cycleBudget struct {
|
||||
perMinute int
|
||||
tokens float64
|
||||
last time.Time
|
||||
}
|
||||
|
||||
func newCycleBudget(perMinute int) *cycleBudget {
|
||||
return &cycleBudget{perMinute: perMinute, tokens: float64(perMinute)}
|
||||
}
|
||||
|
||||
func (b *cycleBudget) allow(now time.Time) bool {
|
||||
if !b.last.IsZero() {
|
||||
b.tokens += now.Sub(b.last).Minutes() * float64(b.perMinute)
|
||||
}
|
||||
b.last = now
|
||||
if b.tokens > float64(b.perMinute) {
|
||||
b.tokens = float64(b.perMinute)
|
||||
}
|
||||
if b.tokens < 1 {
|
||||
return false
|
||||
}
|
||||
b.tokens--
|
||||
return true
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package brain
|
||||
|
||||
// loopDetector spots candidates that keep being admitted without the world
|
||||
// changing (T20). It keeps a ring of recent admitted fingerprints; when one
|
||||
// recurs threshold times within the window it is suppressed and a
|
||||
// metacognition warning is raised once per streak.
|
||||
type loopDetector struct {
|
||||
window int
|
||||
threshold int
|
||||
ring []string
|
||||
warned map[string]bool
|
||||
}
|
||||
|
||||
func newLoopDetector(window, threshold int) *loopDetector {
|
||||
return &loopDetector{window: window, threshold: threshold, warned: map[string]bool{}}
|
||||
}
|
||||
|
||||
func fingerprint(c Candidate) string { return c.Process + "|" + c.Kind + "|" + c.Subject }
|
||||
|
||||
// count returns how many times fp appears in the window.
|
||||
func (l *loopDetector) count(fp string) int {
|
||||
n := 0
|
||||
for _, x := range l.ring {
|
||||
if x == fp {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// looping reports whether admitting fp again would exceed the threshold.
|
||||
func (l *loopDetector) looping(fp string) bool { return l.count(fp) >= l.threshold }
|
||||
|
||||
// record notes an admission.
|
||||
func (l *loopDetector) record(fp string) {
|
||||
l.ring = append(l.ring, fp)
|
||||
if len(l.ring) > l.window {
|
||||
l.ring = l.ring[len(l.ring)-l.window:]
|
||||
}
|
||||
if l.count(fp) < l.threshold {
|
||||
delete(l.warned, fp)
|
||||
}
|
||||
}
|
||||
|
||||
// shouldWarn returns true the first time a fingerprint is suppressed in a streak.
|
||||
func (l *loopDetector) shouldWarn(fp string) bool {
|
||||
if l.warned[fp] {
|
||||
return false
|
||||
}
|
||||
l.warned[fp] = true
|
||||
return true
|
||||
}
|
||||
|
||||
// snapshot exports the ring for persistence.
|
||||
func (l *loopDetector) snapshot() []string { return append([]string{}, l.ring...) }
|
||||
|
||||
func (l *loopDetector) restore(ring []string) { l.ring = append([]string{}, ring...) }
|
||||
@@ -0,0 +1,138 @@
|
||||
package brain
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/scheduler"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
)
|
||||
|
||||
// Manager owns one Runtime per active mind, wakes them on new events and
|
||||
// reconciles with the minds table so enabling or disabling a mind in the UI
|
||||
// takes effect without a restart.
|
||||
type Manager struct {
|
||||
cfg Config
|
||||
group *scheduler.Group
|
||||
mu sync.RWMutex
|
||||
runtimes map[ids.ID]*entry
|
||||
}
|
||||
|
||||
type entry struct {
|
||||
rt *Runtime
|
||||
cancel context.CancelFunc
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// ErrNotRunning is returned for minds without a live runtime.
|
||||
var ErrNotRunning = errors.New("mind runtime is not running")
|
||||
|
||||
// NewManager creates a manager; call Start to launch runtimes.
|
||||
func NewManager(cfg Config, group *scheduler.Group) *Manager {
|
||||
return &Manager{cfg: cfg, group: group, runtimes: map[ids.ID]*entry{}}
|
||||
}
|
||||
|
||||
// Start launches runtimes for active minds, subscribes to the event bus for
|
||||
// wake-ups and reconciles every 30s.
|
||||
func (m *Manager) Start(ctx context.Context) error {
|
||||
if err := m.Reconcile(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
ch, err := m.cfg.Events.Bus().Subscribe(m.group.Context(), ids.Nil, 1024)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.group.Go("brain.wake", func(ctx context.Context) error {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case ev := <-ch:
|
||||
if strings.HasPrefix(ev.Source, "brain") {
|
||||
continue // our own output never re-triggers a cycle (T20)
|
||||
}
|
||||
if rt := m.Get(ev.MindID); rt != nil {
|
||||
rt.Wake()
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
m.group.Every("brain.reconcile", 30*time.Second, m.Reconcile)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Reconcile starts runtimes for active minds and stops the rest.
|
||||
func (m *Manager) Reconcile(ctx context.Context) error {
|
||||
list, err := m.cfg.Minds.List(ctx, true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
want := map[ids.ID]minds.Mind{}
|
||||
for _, md := range list {
|
||||
if md.Status == minds.StatusActive {
|
||||
want[md.ID] = md
|
||||
}
|
||||
}
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for id, e := range m.runtimes {
|
||||
if _, ok := want[id]; !ok {
|
||||
e.cancel()
|
||||
<-e.done
|
||||
delete(m.runtimes, id)
|
||||
m.cfg.Log.Info("mind runtime stopped", "mind", id)
|
||||
}
|
||||
}
|
||||
for id, md := range want {
|
||||
if _, ok := m.runtimes[id]; ok {
|
||||
continue
|
||||
}
|
||||
rt, err := NewRuntime(ctx, m.cfg, md)
|
||||
if err != nil {
|
||||
m.cfg.Log.Error("cannot start mind runtime", "mind", md.Slug, "err", err)
|
||||
continue
|
||||
}
|
||||
rctx, cancel := context.WithCancel(m.group.Context())
|
||||
e := &entry{rt: rt, cancel: cancel, done: make(chan struct{})}
|
||||
m.runtimes[id] = e
|
||||
m.group.Go("brain."+md.Slug, func(context.Context) error {
|
||||
defer close(e.done)
|
||||
return rt.Run(rctx)
|
||||
})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Get returns the runtime for a mind or nil.
|
||||
func (m *Manager) Get(id ids.ID) *Runtime {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
if e, ok := m.runtimes[id]; ok {
|
||||
return e.rt
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Command forwards an operator command to a mind's runtime.
|
||||
func (m *Manager) Command(ctx context.Context, id ids.ID, kind string, args map[string]bool) error {
|
||||
rt := m.Get(id)
|
||||
if rt == nil {
|
||||
return ErrNotRunning
|
||||
}
|
||||
return rt.Command(ctx, kind, args)
|
||||
}
|
||||
|
||||
// Running lists the ids of minds with live runtimes.
|
||||
func (m *Manager) Running() []ids.ID {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
out := make([]ids.ID, 0, len(m.runtimes))
|
||||
for id := range m.runtimes {
|
||||
out = append(out, id)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
// Package brain runs the cognitive cycle for each mind: it gathers new events,
|
||||
// asks registered cognitive processes for thought candidates, scores them
|
||||
// deterministically, admits the winners to the global workspace, dispatches
|
||||
// them to executive handlers and records every step as events (ADR-0001).
|
||||
package brain
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/andydixon/noema/internal/attention"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/workspace"
|
||||
"github.com/andydixon/noema/pkg/events"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
)
|
||||
|
||||
// Process is a cognitive faculty. Observe must be safe to call concurrently
|
||||
// with other processes, must respect ctx, and must not mutate the state it is
|
||||
// given; it returns candidates and lets the cycle decide.
|
||||
type Process interface {
|
||||
Name() string
|
||||
Observe(ctx context.Context, s *State) ([]Candidate, error)
|
||||
}
|
||||
|
||||
// Candidate is a proposed thought.
|
||||
type Candidate struct {
|
||||
Kind string // attend | hypothesis | question | respond | goal | retrieve | reflect | note | action
|
||||
Subject string // stable key for refresh and loop detection (entity, event id, proposition hash)
|
||||
Summary string // one line, human readable
|
||||
Content map[string]any // structured detail for handlers and the UI
|
||||
Features attention.Features // scoring inputs
|
||||
Confidence float64 // epistemic confidence of the candidate itself
|
||||
Evidence []events.EvidenceRef // what it rests on
|
||||
SourceEvents []ids.ID // events that triggered it (first becomes causation)
|
||||
// filled by the cycle
|
||||
ID ids.ID
|
||||
Process string
|
||||
}
|
||||
|
||||
// Mode is the coarse cognitive mode.
|
||||
type Mode string
|
||||
|
||||
// Modes. Later phases add reflecting, consolidating and dreaming.
|
||||
const (
|
||||
ModeAwake Mode = "awake"
|
||||
ModeFocused Mode = "focused"
|
||||
ModeIdle Mode = "idle"
|
||||
ModePaused Mode = "paused"
|
||||
)
|
||||
|
||||
// State is what processes see: a read-only view of the mind plus service
|
||||
// handles. Nil services mean the capability is not available yet.
|
||||
type State struct {
|
||||
Mind minds.Mind
|
||||
SelfModel minds.SelfModel
|
||||
Mode Mode
|
||||
Cycle int64
|
||||
CycleID ids.ID
|
||||
Now time.Time
|
||||
NewEvents []events.Event // since the last cycle, ascending
|
||||
Workspace workspace.State // snapshot before this cycle
|
||||
Services Services
|
||||
}
|
||||
|
||||
// Services are read handles processes may use. Extended per phase.
|
||||
type Services struct {
|
||||
Events EventReader
|
||||
}
|
||||
|
||||
// EventReader is the subset of the event store processes may call.
|
||||
type EventReader interface {
|
||||
Get(ctx context.Context, id ids.ID) (events.Event, error)
|
||||
}
|
||||
|
||||
// Handler executes an admitted thought of a given kind. Handlers run
|
||||
// sequentially after admission, may write to stores and emit events, and
|
||||
// receive the cycle's emitter so provenance links are consistent.
|
||||
type Handler func(ctx context.Context, c *Cycle, t workspace.Thought) error
|
||||
|
||||
// Registry holds processes and handlers. It is immutable after construction.
|
||||
type Registry struct {
|
||||
processes []Process
|
||||
handlers map[string]Handler
|
||||
}
|
||||
|
||||
// NewRegistry builds a registry.
|
||||
func NewRegistry() *Registry { return &Registry{handlers: map[string]Handler{}} }
|
||||
|
||||
// AddProcess registers a process.
|
||||
func (r *Registry) AddProcess(p Process) *Registry { r.processes = append(r.processes, p); return r }
|
||||
|
||||
// Handle registers the handler for a thought kind.
|
||||
func (r *Registry) Handle(kind string, h Handler) *Registry { r.handlers[kind] = h; return r }
|
||||
|
||||
// Processes returns the registered processes.
|
||||
func (r *Registry) Processes() []Process { return r.processes }
|
||||
Reference in New Issue
Block a user