From defdbf217a51ff04cecb92d3b9d0ad885c9a4d57 Mon Sep 17 00:00:00 2001 From: Lakshman Patel Date: Mon, 24 Aug 2026 08:29:56 +0530 Subject: [PATCH] feat(prime): continual harness, autonomous budgets, family messaging Three adoptions from Prime Agent's RLM agent. Continual harness (internal/intelligence/harness): - Versioned, evidence-backed refinement store for supplemental agent state (prompts, memories, skill descriptions, subagent specs). - Refine bumps entry version and records a refinement event (trigger -> changes -> evidence -> outcome); every change is auditable and reversible via snapshot-based rollback. Never rewrites the immutable base system prompt. Durable JSON persistence. Bounded autonomous budgets (internal/engine/autonomous_budget.go): - Tracks turns/tokens/time/continuations within configured limits and a status machine that reports WHY a run stopped: budget_limited: vs gate_passed vs error (a passed gate checks only what it verifies; reaching a limit does not imply success). First budget hit wins and terminal reasons stick. Agent family messaging (internal/multiagent/family_messaging.go): - Direct inter-agent messages scoped to a family graph (parent / sibling / child) rather than global, with per-agent pending caps and token- bucket rate limiting to prevent flooding. Verification: new suites green (harness 8, autonomous 8, family 7); engine + multiagent suites pass; golangci-lint 0 issues; gofmt clean. --- README.md | 3 + internal/engine/autonomous_budget.go | 121 +++++++ internal/engine/autonomous_budget_test.go | 87 +++++ internal/intelligence/harness/harness.go | 322 ++++++++++++++++++ internal/intelligence/harness/harness_test.go | 115 +++++++ internal/multiagent/family_messaging.go | 186 ++++++++++ internal/multiagent/family_messaging_test.go | 113 ++++++ 7 files changed, 947 insertions(+) create mode 100644 internal/engine/autonomous_budget.go create mode 100644 internal/engine/autonomous_budget_test.go create mode 100644 internal/intelligence/harness/harness.go create mode 100644 internal/intelligence/harness/harness_test.go create mode 100644 internal/multiagent/family_messaging.go create mode 100644 internal/multiagent/family_messaging_test.go diff --git a/README.md b/README.md index c45d8376..247db7bc 100644 --- a/README.md +++ b/README.md @@ -279,6 +279,9 @@ Features adopted from open-source agent projects. All are off by default unless | Voice transcription | Telegram voice notes + `stt` package | Transcribe Telegram voice/audio into the prompt. Backend via `stt.SetTranscriber`; an OpenAI-compatible client ships in `eyrie/client` (`AudioClient`), wired by the host | | Git-tree file snapshots | `internal/gitsnapshot` | Content-addressed tree capture/diff/preview/restore | | Turn-boundary rewind | `internal/filestate` | Per-prompt before/after snapshots with durable store | +| Continual harness | `internal/intelligence/harness` | Versioned, evidence-backed refinement of supplemental prompts/memories/skills/subagents with rollback | +| Bounded autonomous budgets | `internal/engine` (`AutonomousBudget`) | Track turns/tokens/time/continuations; report why a run stopped (budget vs gate-passed vs error) | +| Agent family messaging | `internal/multiagent` (`FamilyMessenger`) | Direct parent/sibling/child messages with pending caps + rate limits | | Path reservations | `internal/multiagent` ledger | Detect overlapping-file changes between parallel branches | | Live agent status | `GET /v1/agent/status` (daemon) | Machine-readable working/idle/stale per session | diff --git a/internal/engine/autonomous_budget.go b/internal/engine/autonomous_budget.go new file mode 100644 index 00000000..bfcfb6de --- /dev/null +++ b/internal/engine/autonomous_budget.go @@ -0,0 +1,121 @@ +package engine + +import ( + "time" +) + +// Bounded autonomous budget tracking, adopted from Prime Agent's autonomous +// mode: the agent runs within configured turn/token/time/continuation budgets +// and a status machine reports WHY a run stopped — budget_limited (with the +// specific budget hit) vs gate_passed vs error — so completion is honest +// ("a passed gate checks only what that gate verifies; reaching a limit does +// not imply task success"). + +// AutonomousReason classifies why a bounded run stopped. +type AutonomousReason string + +const ( + ReasonRunning AutonomousReason = "running" + ReasonBudgetTokens AutonomousReason = "budget_limited:tokens" + ReasonBudgetTurns AutonomousReason = "budget_limited:turns" + ReasonBudgetTime AutonomousReason = "budget_limited:time" + ReasonBudgetConcurr AutonomousReason = "budget_limited:continuations" + ReasonGatePassed AutonomousReason = "gate_passed" + ReasonError AutonomousReason = "error" +) + +// AutonomousLimits are the configured bounds. +type AutonomousLimits struct { + MaxTurns int `json:"max_turns,omitempty"` // 0 = unlimited + MaxTokens int `json:"max_tokens,omitempty"` // 0 = unlimited + MaxTime time.Duration `json:"max_time,omitempty"` // 0 = unlimited + MaxContinuations int `json:"max_continuations,omitempty"` // 0 = unlimited +} + +// AutonomousStatus is the live tracking state. +type AutonomousStatus struct { + Limits AutonomousLimits `json:"limits"` + TurnsUsed int `json:"turns_used"` + TokensUsed int `json:"tokens_used"` + TimeUsedSeconds float64 `json:"time_used_seconds"` + ContinuationsUsed int `json:"continuations_used"` + Reason AutonomousReason `json:"reason"` + LastError string `json:"last_error,omitempty"` + StartedAt time.Time `json:"started_at"` +} + +// AutonomousBudget tracks a bounded autonomous run. +type AutonomousBudget struct { + limits AutonomousLimits + status AutonomousStatus + started time.Time +} + +// NewAutonomousBudget starts a bounded run with the given limits. +func NewAutonomousBudget(limits AutonomousLimits) *AutonomousBudget { + return &AutonomousBudget{ + limits: limits, + status: AutonomousStatus{Limits: limits, Reason: ReasonRunning, StartedAt: time.Now()}, + started: time.Now(), + } +} + +// RecordTurn increments the turn counter and reports the current reason. +func (b *AutonomousBudget) RecordTurn() AutonomousReason { + b.status.TurnsUsed++ + return b.Check() +} + +// RecordToken adds usage and reports the current reason. +func (b *AutonomousBudget) RecordToken(n int) AutonomousReason { + b.status.TokensUsed += n + return b.Check() +} + +// RecordContinuation increments the continuation counter. +func (b *AutonomousBudget) RecordContinuation() AutonomousReason { + b.status.ContinuationsUsed++ + return b.Check() +} + +// Check recomputes the status from budgets + elapsed time. Once a terminal +// reason is reached it stays terminal (first budget hit wins). +func (b *AutonomousBudget) Check() AutonomousReason { + if b.status.Reason != ReasonRunning { + return b.status.Reason + } + b.status.TimeUsedSeconds = time.Since(b.started).Seconds() + switch { + case b.limits.MaxTurns > 0 && b.status.TurnsUsed >= b.limits.MaxTurns: + b.status.Reason = ReasonBudgetTurns + case b.limits.MaxTokens > 0 && b.status.TokensUsed >= b.limits.MaxTokens: + b.status.Reason = ReasonBudgetTokens + case b.limits.MaxContinuations > 0 && b.status.ContinuationsUsed >= b.limits.MaxContinuations: + b.status.Reason = ReasonBudgetConcurr + case b.limits.MaxTime > 0 && time.Since(b.started) >= b.limits.MaxTime: + b.status.Reason = ReasonBudgetTime + } + return b.status.Reason +} + +// MarkGatePassed records that a configured quality gate succeeded. +func (b *AutonomousBudget) MarkGatePassed() { + if b.status.Reason == ReasonRunning { + b.status.Reason = ReasonGatePassed + } +} + +// MarkError records a run error. +func (b *AutonomousBudget) MarkError(errMsg string) { + b.status.Reason = ReasonError + b.status.LastError = errMsg +} + +// Status returns a copy of the live status. +func (b *AutonomousBudget) Status() AutonomousStatus { return b.status } + +// Running reports whether the run should continue (not terminal). +func (b *AutonomousBudget) Running() bool { return b.status.Reason == ReasonRunning } + +// StopReason returns the terminal reason, or ReasonRunning if still running. +func (b *AutonomousBudget) StopReason() AutonomousReason { return b.status.Reason } diff --git a/internal/engine/autonomous_budget_test.go b/internal/engine/autonomous_budget_test.go new file mode 100644 index 00000000..88d1d397 --- /dev/null +++ b/internal/engine/autonomous_budget_test.go @@ -0,0 +1,87 @@ +package engine + +import ( + "testing" + "time" +) + +func TestBudgetTurnLimit(t *testing.T) { + b := NewAutonomousBudget(AutonomousLimits{MaxTurns: 3}) + if b.Running() != true { + t.Fatal("should start running") + } + b.RecordTurn() + b.RecordTurn() + if r := b.RecordTurn(); r != ReasonBudgetTurns { + t.Fatalf("reason = %q, want budget_limited:turns", r) + } + if b.Running() { + t.Fatal("should be stopped") + } + if s := b.Status(); s.TurnsUsed != 3 || s.Reason != ReasonBudgetTurns { + t.Fatalf("status = %+v", s) + } +} + +func TestBudgetTokenLimit(t *testing.T) { + b := NewAutonomousBudget(AutonomousLimits{MaxTokens: 1000}) + b.RecordToken(600) + if r := b.RecordToken(500); r != ReasonBudgetTokens { + t.Fatalf("reason = %q, want budget_limited:tokens", r) + } +} + +func TestBudgetTimeLimit(t *testing.T) { + b := NewAutonomousBudget(AutonomousLimits{MaxTime: 20 * time.Millisecond}) + time.Sleep(30 * time.Millisecond) + if r := b.Check(); r != ReasonBudgetTime { + t.Fatalf("reason = %q, want budget_limited:time", r) + } +} + +func TestGatePassedBeatsBudgetsOnlyIfRunning(t *testing.T) { + b := NewAutonomousBudget(AutonomousLimits{MaxTurns: 2}) + b.RecordTurn() + b.MarkGatePassed() + if b.StopReason() != ReasonGatePassed { + t.Fatalf("reason = %q, want gate_passed", b.StopReason()) + } + // After a terminal reason, budgets do not override it. + b.RecordTurn() + if b.StopReason() != ReasonGatePassed { + t.Fatalf("terminal should stick, got %q", b.StopReason()) + } +} + +func TestErrorSticks(t *testing.T) { + b := NewAutonomousBudget(AutonomousLimits{MaxTurns: 10}) + b.MarkError("boom") + if b.StopReason() != ReasonError || b.Status().LastError != "boom" { + t.Fatalf("status = %+v", b.Status()) + } + if b.Running() { + t.Fatal("should stop on error") + } +} + +func TestFirstBudgetHitWins(t *testing.T) { + b := NewAutonomousBudget(AutonomousLimits{MaxTurns: 2, MaxTokens: 100}) + b.RecordTurn() + b.RecordToken(120) // tokens hit first -> terminal reason sticks + b.RecordTurn() // turns now also exceeded, but reason is already terminal + if b.StopReason() != ReasonBudgetTokens { + t.Fatalf("reason = %q, want budget_limited:tokens (first hit wins)", b.StopReason()) + } +} + +func TestUnlimitedRun(t *testing.T) { + b := NewAutonomousBudget(AutonomousLimits{}) + for i := 0; i < 100; i++ { + b.RecordTurn() + b.RecordToken(1) + b.RecordContinuation() + } + if !b.Running() || b.StopReason() != ReasonRunning { + t.Fatalf("unlimited run should keep running, got %q", b.StopReason()) + } +} diff --git a/internal/intelligence/harness/harness.go b/internal/intelligence/harness/harness.go new file mode 100644 index 00000000..5aadd76e --- /dev/null +++ b/internal/intelligence/harness/harness.go @@ -0,0 +1,322 @@ +// Package harness implements a continual, evidence-backed refinement store +// adopted from Prime Agent's Continual Harness: supplemental agent state +// (prompts, memories, skill descriptions, subagent specs) is stored as +// versioned entries that the agent can refine through small, evidence-backed +// updates, with a recorded refinement history and snapshot-based rollback. +// +// It never rewrites the immutable base system prompt — only this supplemental +// state. Entries are keyed by kind; each update bumps the entry version and +// appends a refinement event (trigger -> changes -> evidence -> outcome) so +// every change is auditable and reversible. +package harness + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" +) + +// Kind is the category of supplemental state. +type Kind string + +const ( + KindPrompt Kind = "prompt" + KindMemory Kind = "memory" + KindSkill Kind = "skill" + KindSubagent Kind = "subagent" +) + +func (k Kind) Valid() bool { + switch k { + case KindPrompt, KindMemory, KindSkill, KindSubagent: + return true + } + return false +} + +// Scope is where an entry applies. +type Scope string + +const ( + ScopeLocal Scope = "local" + ScopeGlobal Scope = "global" +) + +// Entry is one versioned supplemental state item. +type Entry struct { + ID string `json:"id"` + Kind Kind `json:"kind"` + Title string `json:"title"` + Content string `json:"content"` + Path string `json:"path,omitempty"` + Scope Scope `json:"scope,omitempty"` + Version int `json:"version"` + Evidence string `json:"evidence,omitempty"` + Source string `json:"source,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// Refinement is one recorded update event. +type Refinement struct { + ID string `json:"id"` + Trigger string `json:"trigger"` + Changes []string `json:"changes"` + Evidence string `json:"evidence"` + Outcome string `json:"outcome,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +// Snapshot is a point-in-time copy of all entries, used for rollback. +type Snapshot struct { + Schema int `json:"schema"` + Entries []Entry `json:"entries"` +} + +// Store is a thread-safe continual harness. +type Store struct { + mu sync.Mutex + entries map[Kind]map[string]*Entry + refinements []Refinement + dir string +} + +// New creates a Store persisted under dir ("" disables persistence). +func New(dir string) (*Store, error) { + s := &Store{ + entries: map[Kind]map[string]*Entry{ + KindPrompt: {}, KindMemory: {}, KindSkill: {}, KindSubagent: {}, + }, + dir: dir, + } + if dir != "" { + if err := s.load(); err != nil { + return nil, err + } + } + return s, nil +} + +func (s *Store) path() string { + if s.dir == "" { + return "" + } + return filepath.Join(s.dir, "harness.json") +} + +func (s *Store) load() error { + raw, err := os.ReadFile(s.path()) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("harness: load: %w", err) + } + var st struct { + Schema int `json:"schema"` + Entries []Entry `json:"entries"` + History []Refinement `json:"history"` + } + if err := json.Unmarshal(raw, &st); err != nil { + return fmt.Errorf("harness: parse: %w", err) + } + for _, e := range st.Entries { + if s.entries[e.Kind] == nil { + s.entries[e.Kind] = map[string]*Entry{} + } + cp := e + s.entries[e.Kind][e.ID] = &cp + } + s.refinements = st.History + return nil +} + +func (s *Store) save() error { + if s.dir == "" { + return nil + } + if err := os.MkdirAll(s.dir, 0o750); err != nil { + return err + } + var all []Entry + for _, m := range s.entries { + for _, e := range m { + all = append(all, *e) + } + } + sort.Slice(all, func(i, j int) bool { + if all[i].Kind != all[j].Kind { + return all[i].Kind < all[j].Kind + } + return all[i].Title < all[j].Title + }) + data, err := json.MarshalIndent(struct { + Schema int `json:"schema"` + Entries []Entry `json:"entries"` + History []Refinement `json:"history"` + }{Schema: 1, Entries: all, History: s.refinements}, "", " ") + if err != nil { + return err + } + tmp := s.path() + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return err + } + return os.Rename(tmp, s.path()) +} + +// Create adds a new entry and records a refinement event. +func (s *Store) Create(kind Kind, title, content, evidence, source string) (*Entry, error) { + if !kind.Valid() { + return nil, fmt.Errorf("harness: invalid kind %q", kind) + } + s.mu.Lock() + defer s.mu.Unlock() + id := slug(kind, title) + e := &Entry{ + ID: id, Kind: kind, Title: title, Content: content, + Scope: ScopeLocal, Version: 1, Evidence: evidence, Source: source, + CreatedAt: time.Now(), UpdatedAt: time.Now(), + } + s.entries[kind][id] = e + s.recordLocked("create", kind, id, evidence, "") + return s.saveThen(e) +} + +// Refine updates an existing entry (or creates it if absent), bumping the +// version and recording an evidence-backed refinement event. +func (s *Store) Refine(kind Kind, title, content, evidence string) (*Entry, error) { + if !kind.Valid() { + return nil, fmt.Errorf("harness: invalid kind %q", kind) + } + s.mu.Lock() + defer s.mu.Unlock() + id := slug(kind, title) + now := time.Now() + if e, ok := s.entries[kind][id]; ok { + e.Content = content + e.Version++ + e.Evidence = evidence + e.UpdatedAt = now + s.recordLocked("update", kind, id, evidence, "") + return s.saveThen(e) + } + e := &Entry{ + ID: id, Kind: kind, Title: title, Content: content, + Scope: ScopeLocal, Version: 1, Evidence: evidence, CreatedAt: now, UpdatedAt: now, + } + s.entries[kind][id] = e + s.recordLocked("create", kind, id, evidence, "") + return s.saveThen(e) +} + +// Delete removes an entry, recording the event. +func (s *Store) Delete(kind Kind, title, evidence string) error { + if !kind.Valid() { + return fmt.Errorf("harness: invalid kind %q", kind) + } + s.mu.Lock() + defer s.mu.Unlock() + id := slug(kind, title) + if _, ok := s.entries[kind][id]; !ok { + return fmt.Errorf("harness: %s %q not found", kind, title) + } + delete(s.entries[kind], id) + s.recordLocked("delete", kind, id, evidence, "") + return s.save() +} + +// Get returns a copy of an entry. +func (s *Store) Get(kind Kind, title string) (*Entry, bool) { + s.mu.Lock() + defer s.mu.Unlock() + e, ok := s.entries[kind][slug(kind, title)] + if !ok { + return nil, false + } + cp := *e + return &cp, true +} + +// List returns all entries of a kind, sorted by title. +func (s *Store) List(kind Kind) []Entry { + s.mu.Lock() + defer s.mu.Unlock() + var out []Entry + for _, e := range s.entries[kind] { + out = append(out, *e) + } + sort.Slice(out, func(i, j int) bool { return out[i].Title < out[j].Title }) + return out +} + +// History returns the recorded refinement events, oldest first. +func (s *Store) History() []Refinement { + s.mu.Lock() + defer s.mu.Unlock() + return append([]Refinement{}, s.refinements...) +} + +// Snapshot returns a point-in-time copy of all entries (for rollback). +func (s *Store) Snapshot() Snapshot { + s.mu.Lock() + defer s.mu.Unlock() + var all []Entry + for _, m := range s.entries { + for _, e := range m { + all = append(all, *e) + } + } + return Snapshot{Schema: 1, Entries: all} +} + +// Restore replaces all entries with a snapshot, recording a rollback event. +func (s *Store) Restore(snap Snapshot, evidence string) error { + s.mu.Lock() + defer s.mu.Unlock() + for k := range s.entries { + s.entries[k] = map[string]*Entry{} + } + for i := range snap.Entries { + e := snap.Entries[i] + if s.entries[e.Kind] == nil { + s.entries[e.Kind] = map[string]*Entry{} + } + cp := e + s.entries[e.Kind][e.ID] = &cp + } + s.recordLocked("rollback", "", "", evidence, fmt.Sprintf("%d entries", len(snap.Entries))) + return s.save() +} + +func (s *Store) recordLocked(action string, kind Kind, id, evidence, outcome string) { + s.refinements = append(s.refinements, Refinement{ + ID: fmt.Sprintf("r-%d", len(s.refinements)+1), + Trigger: action + " " + string(kind) + " " + id, + Changes: []string{action + ":" + string(kind) + ":" + id}, + Evidence: evidence, + Outcome: outcome, + CreatedAt: time.Now(), + }) +} + +func (s *Store) saveThen(e *Entry) (*Entry, error) { + if err := s.save(); err != nil { + return nil, err + } + cp := *e + return &cp, nil +} + +// slug builds a stable entry id from kind + title. +func slug(kind Kind, title string) string { + t := strings.ToLower(strings.TrimSpace(title)) + t = strings.ReplaceAll(t, " ", "-") + t = strings.ReplaceAll(t, "/", "-") + return string(kind) + ":" + t +} diff --git a/internal/intelligence/harness/harness_test.go b/internal/intelligence/harness/harness_test.go new file mode 100644 index 00000000..d18471df --- /dev/null +++ b/internal/intelligence/harness/harness_test.go @@ -0,0 +1,115 @@ +package harness + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestCreateAndGet(t *testing.T) { + s, _ := New("") + e, err := s.Create(KindMemory, "user-prefers-rust", "prefers Rust over Go", "user stated in session", "session-1") + if err != nil { + t.Fatal(err) + } + if e.Version != 1 { + t.Fatalf("version = %d", e.Version) + } + got, ok := s.Get(KindMemory, "user-prefers-rust") + if !ok || got.Content != "prefers Rust over Go" { + t.Fatalf("get = %+v ok=%v", got, ok) + } +} + +func TestRefineBumpsVersionAndRecordsEvidence(t *testing.T) { + s, _ := New("") + _, _ = s.Create(KindSkill, "deploy", "step one", "initial", "s") + refined, err := s.Refine(KindSkill, "deploy", "step one, step two", "observed new step") + if err != nil { + t.Fatal(err) + } + if refined.Version != 2 { + t.Fatalf("version = %d, want 2", refined.Version) + } + if refined.Evidence != "observed new step" { + t.Fatalf("evidence = %q", refined.Evidence) + } + hist := s.History() + if len(hist) != 2 || hist[1].Evidence != "observed new step" { + t.Fatalf("history = %+v", hist) + } +} + +func TestSnapshotRestoreRollback(t *testing.T) { + s, _ := New("") + _, _ = s.Create(KindMemory, "m1", "old value", "e1", "s") + snap := s.Snapshot() + _, _ = s.Refine(KindMemory, "m1", "new value", "e2") + if got, _ := s.Get(KindMemory, "m1"); got.Content != "new value" { + t.Fatalf("after refine = %q", got.Content) + } + if err := s.Restore(snap, "revert bad change"); err != nil { + t.Fatal(err) + } + got, _ := s.Get(KindMemory, "m1") + if got.Content != "old value" { + t.Fatalf("after rollback = %q", got.Content) + } +} + +func TestDelete(t *testing.T) { + s, _ := New("") + _, _ = s.Create(KindSubagent, "reviewer", "prompt", "e", "s") + if err := s.Delete(KindSubagent, "reviewer", "no longer used"); err != nil { + t.Fatal(err) + } + if _, ok := s.Get(KindSubagent, "reviewer"); ok { + t.Fatal("entry should be deleted") + } +} + +func TestPersistence(t *testing.T) { + dir := t.TempDir() + s1, _ := New(dir) + _, _ = s1.Create(KindPrompt, "style", "be concise", "evidence", "s") + s1.Refine(KindPrompt, "style", "be very concise", "more evidence") + + s2, err := New(dir) + if err != nil { + t.Fatal(err) + } + got, ok := s2.Get(KindPrompt, "style") + if !ok || got.Version != 2 || got.Content != "be very concise" { + t.Fatalf("reloaded = %+v ok=%v", got, ok) + } + if len(s2.History()) != 2 { + t.Fatalf("history not persisted: %d", len(s2.History())) + } +} + +func TestInvalidKind(t *testing.T) { + s, _ := New("") + if _, err := s.Create("bogus", "x", "y", "", ""); err == nil || !strings.Contains(err.Error(), "invalid kind") { + t.Fatalf("err = %v", err) + } +} + +func TestListSorted(t *testing.T) { + s, _ := New("") + _, _ = s.Create(KindMemory, "zeta", "1", "", "") + _, _ = s.Create(KindMemory, "alpha", "2", "", "") + items := s.List(KindMemory) + if len(items) != 2 || items[0].Title != "alpha" || items[1].Title != "zeta" { + t.Fatalf("items = %+v", items) + } +} + +func TestFilePersistenceRoundTrip(t *testing.T) { + dir := t.TempDir() + s, _ := New(dir) + _, _ = s.Create(KindSkill, "s1", "c", "", "") + if _, err := os.Stat(filepath.Join(dir, "harness.json")); err != nil { + t.Fatalf("harness.json not written: %v", err) + } +} diff --git a/internal/multiagent/family_messaging.go b/internal/multiagent/family_messaging.go new file mode 100644 index 00000000..052cde99 --- /dev/null +++ b/internal/multiagent/family_messaging.go @@ -0,0 +1,186 @@ +package mission + +import ( + "sync" + "time" +) + +// Family messaging adopted from Prime Agent's agent-messages: running agents +// can exchange messages directly, but reach is scoped to a family graph +// (parent / sibling / child) rather than global, with per-session pending caps +// and token-bucket rate limiting to prevent one agent flooding another. + +// FamilyRole is an agent's relationship to another. +type FamilyRole string + +const ( + FamilyParent FamilyRole = "parent" + FamilySibling FamilyRole = "sibling" + FamilyChild FamilyRole = "child" +) + +// FamilyLinks describes one agent's known family. +type FamilyLinks struct { + Parent string `json:"parent,omitempty"` + Siblings []string `json:"siblings,omitempty"` + Children []string `json:"children,omitempty"` +} + +// FamilyMessengerConfig controls rate limiting. +type FamilyMessengerConfig struct { + MaxPendingPerAgent int `json:"max_pending_per_agent,omitempty"` + RefillPerSec float64 `json:"refill_per_sec,omitempty"` // token bucket refill + Capacity float64 `json:"capacity,omitempty"` +} + +func (c FamilyMessengerConfig) withDefaults() FamilyMessengerConfig { + if c.MaxPendingPerAgent <= 0 { + c.MaxPendingPerAgent = 20 + } + if c.RefillPerSec <= 0 { + c.RefillPerSec = 3 + } + if c.Capacity <= 0 { + c.Capacity = 3 + } + return c +} + +// FamilyMessage is one inter-agent message. +type FamilyMessage struct { + From string `json:"from"` + To string `json:"to"` + Role FamilyRole `json:"role"` + Content string `json:"content"` + SentAt time.Time `json:"sent_at"` +} + +// bucket is a token bucket for one destination. +type bucket struct { + tokens float64 + last time.Time +} + +// FamilyMessenger delivers messages within a family graph with rate limits. +type FamilyMessenger struct { + mu sync.Mutex + links map[string]FamilyLinks + inbox map[string][]FamilyMessage + buckets map[string]*bucket + cfg FamilyMessengerConfig +} + +// NewFamilyMessenger creates a messenger. +func NewFamilyMessenger(cfg FamilyMessengerConfig) *FamilyMessenger { + return &FamilyMessenger{ + links: map[string]FamilyLinks{}, + inbox: map[string][]FamilyMessage{}, + buckets: map[string]*bucket{}, + cfg: cfg.withDefaults(), + } +} + +// Register sets an agent's family links. +func (m *FamilyMessenger) Register(agentID string, links FamilyLinks) { + m.mu.Lock() + defer m.mu.Unlock() + m.links[agentID] = links +} + +// RoleBetween returns the family relationship from -> to, or "" if none. +func (m *FamilyMessenger) RoleBetween(from, to string) FamilyRole { + m.mu.Lock() + defer m.mu.Unlock() + links := m.links[from] + if links.Parent == to { + return FamilyParent + } + for _, s := range links.Siblings { + if s == to { + return FamilySibling + } + } + for _, c := range links.Children { + if c == to { + return FamilyChild + } + } + return "" +} + +// Allowed reports whether from may message to within the family graph. +func (m *FamilyMessenger) Allowed(from, to string) bool { return m.RoleBetween(from, to) != "" } + +// Send delivers a message if relationship, rate limit, and pending cap allow. +// Returns accepted=true when delivered. +func (m *FamilyMessenger) Send(from, to, content string) (bool, FamilyRole) { + m.mu.Lock() + defer m.mu.Unlock() + role := m.roleBetweenLocked(from, to) + if role == "" { + return false, "" + } + if len(m.inbox[to]) >= m.cfg.MaxPendingPerAgent { + return false, role // pending cap hit + } + b := m.bucketForLocked(to) + if b.tokens < 1 { + return false, role // rate limited + } + b.tokens-- + m.inbox[to] = append(m.inbox[to], FamilyMessage{ + From: from, To: to, Role: role, Content: content, SentAt: time.Now(), + }) + return true, role +} + +// Receive drains all pending messages for an agent. +func (m *FamilyMessenger) Receive(agentID string) []FamilyMessage { + m.mu.Lock() + defer m.mu.Unlock() + out := m.inbox[agentID] + delete(m.inbox, agentID) + return out +} + +// Pending returns the count of undelivered messages for an agent. +func (m *FamilyMessenger) Pending(agentID string) int { + m.mu.Lock() + defer m.mu.Unlock() + return len(m.inbox[agentID]) +} + +func (m *FamilyMessenger) roleBetweenLocked(from, to string) FamilyRole { + links := m.links[from] + if links.Parent == to { + return FamilyParent + } + for _, s := range links.Siblings { + if s == to { + return FamilySibling + } + } + for _, c := range links.Children { + if c == to { + return FamilyChild + } + } + return "" +} + +func (m *FamilyMessenger) bucketForLocked(dest string) *bucket { + b, ok := m.buckets[dest] + if !ok { + b = &bucket{tokens: m.cfg.Capacity, last: time.Now()} + m.buckets[dest] = b + return b + } + elapsed := time.Since(b.last).Seconds() + refill := elapsed * m.cfg.RefillPerSec + b.tokens += refill + if b.tokens > m.cfg.Capacity { + b.tokens = m.cfg.Capacity + } + b.last = time.Now() + return b +} diff --git a/internal/multiagent/family_messaging_test.go b/internal/multiagent/family_messaging_test.go new file mode 100644 index 00000000..6d70ed87 --- /dev/null +++ b/internal/multiagent/family_messaging_test.go @@ -0,0 +1,113 @@ +package mission + +import ( + "strings" + "testing" + "time" +) + +func TestRoleBetweenAndAllowed(t *testing.T) { + m := NewFamilyMessenger(FamilyMessengerConfig{}) + m.Register("parent", FamilyLinks{Children: []string{"childA", "childB"}}) + m.Register("childA", FamilyLinks{Parent: "parent", Siblings: []string{"childB"}}) + m.Register("childB", FamilyLinks{Parent: "parent", Siblings: []string{"childA"}}) + + if m.RoleBetween("childA", "parent") != FamilyParent { + t.Fatalf("childA->parent role = %q", m.RoleBetween("childA", "parent")) + } + if m.RoleBetween("childA", "childB") != FamilySibling { + t.Fatalf("childA->childB role = %q", m.RoleBetween("childA", "childB")) + } + if m.RoleBetween("parent", "childA") != FamilyChild { + t.Fatalf("parent->childA role = %q", m.RoleBetween("parent", "childA")) + } + // Unrelated: not allowed. + if m.Allowed("childA", "parentX") || m.Allowed("parent", "outsider") { + t.Fatal("unrelated agents should not be allowed") + } +} + +func TestSendWithinFamily(t *testing.T) { + m := NewFamilyMessenger(FamilyMessengerConfig{Capacity: 100, RefillPerSec: 100}) + m.Register("childA", FamilyLinks{Parent: "parent"}) + m.Register("parent", FamilyLinks{Children: []string{"childA"}}) + + ok, role := m.Send("childA", "parent", "done") + if !ok || role != FamilyParent { + t.Fatalf("send = %v %q", ok, role) + } + msgs := m.Receive("parent") + if len(msgs) != 1 || msgs[0].From != "childA" || msgs[0].Content != "done" { + t.Fatalf("msgs = %+v", msgs) + } +} + +func TestSendRejectsUnrelated(t *testing.T) { + m := NewFamilyMessenger(FamilyMessengerConfig{}) + m.Register("a", FamilyLinks{}) + m.Register("b", FamilyLinks{}) + if ok, role := m.Send("a", "b", "hi"); ok || role != "" { + t.Fatalf("unrelated send should be rejected: %v %q", ok, role) + } +} + +func TestPendingCap(t *testing.T) { + m := NewFamilyMessenger(FamilyMessengerConfig{MaxPendingPerAgent: 2, Capacity: 100, RefillPerSec: 100}) + m.Register("a", FamilyLinks{Children: []string{"b"}}) + for i := 0; i < 3; i++ { + ok, _ := m.Send("a", "b", "m") + if i < 2 && !ok { + t.Fatalf("send %d should be accepted", i) + } + if i == 2 && ok { + t.Fatal("3rd send should exceed pending cap") + } + } + if m.Pending("b") != 2 { + t.Fatalf("pending = %d, want 2", m.Pending("b")) + } +} + +func TestRateLimit(t *testing.T) { + m := NewFamilyMessenger(FamilyMessengerConfig{Capacity: 2, RefillPerSec: 1, MaxPendingPerAgent: 100}) + m.Register("a", FamilyLinks{Children: []string{"b"}}) + // First 2 accepted (capacity 2). + for i := 0; i < 2; i++ { + if ok, _ := m.Send("a", "b", "x"); !ok { + t.Fatalf("send %d should be accepted", i) + } + } + // Third rejected by rate limit. + if ok, _ := m.Send("a", "b", "x"); ok { + t.Fatal("third send should be rate-limited") + } + // After refill elapses, send is accepted again. + time.Sleep(1100 * time.Millisecond) + if ok, _ := m.Send("a", "b", "x"); !ok { + t.Fatal("send after refill should be accepted") + } +} + +func TestReceiveDrains(t *testing.T) { + m := NewFamilyMessenger(FamilyMessengerConfig{Capacity: 100, RefillPerSec: 100}) + m.Register("a", FamilyLinks{Children: []string{"b"}}) + m.Send("a", "b", "1") + m.Send("a", "b", "2") + if m.Pending("b") != 2 { + t.Fatalf("pending = %d", m.Pending("b")) + } + msgs := m.Receive("b") + if len(msgs) != 2 || m.Pending("b") != 0 { + t.Fatalf("receive drained wrong: %d msgs, pending=%d", len(msgs), m.Pending("b")) + } +} + +func TestMessageContentContains(t *testing.T) { + m := NewFamilyMessenger(FamilyMessengerConfig{Capacity: 100, RefillPerSec: 100}) + m.Register("p", FamilyLinks{Children: []string{"c"}}) + m.Send("p", "c", "please run the tests") + msgs := m.Receive("c") + if len(msgs) != 1 || !strings.Contains(msgs[0].Content, "run the tests") { + t.Fatalf("msgs = %+v", msgs) + } +}