From a090f90cf3a85ef2790841f29408af164c05c96f Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 20:21:56 -0500 Subject: [PATCH 1/9] feat(daemon): wire researcher role into daemon lifecycle Watch researcher/inbox/ for incoming tasks. Resolve researcher config (model, session_timeout) from pool.toml [researcher] section. Drain researcher inbox at startup. Register researcher tools stub in MCP server. Role-aware MCP tool name selection via ToolNamesForRole() replaces hardcoded ExpertToolNames in --allowedTools construction. Adds researcher/logs/ to ensureDirs for log archival. Also adds v0.8 implementation plan to docs/plans/. Co-Authored-By: Claude Opus 4.6 (1M context) --- docs/plans/2026-04-06-researcher-curation.md | 246 +++++++++++++++++++ internal/daemon/daemon.go | 42 +++- internal/daemon/daemon_test.go | 222 +++++++++++++++++ internal/mcp/config.go | 38 ++- internal/mcp/config_test.go | 81 ++++++ internal/mcp/researcher_tools.go | 18 ++ internal/mcp/server.go | 4 +- 7 files changed, 640 insertions(+), 11 deletions(-) create mode 100644 docs/plans/2026-04-06-researcher-curation.md create mode 100644 internal/mcp/researcher_tools.go diff --git a/docs/plans/2026-04-06-researcher-curation.md b/docs/plans/2026-04-06-researcher-curation.md new file mode 100644 index 0000000..8a169bb --- /dev/null +++ b/docs/plans/2026-04-06-researcher-curation.md @@ -0,0 +1,246 @@ +# v0.8 — Researcher + Curation + +## Context + +v0.7 (shared experts + multi-pool) is complete on `feat/v0.7-shared-experts`. The researcher role has been scaffolded across prior versions — config parsing, mail routing, directory creation, builtin role registration — but nothing actually watches its inbox, spawns sessions, or provides tools. v0.8 brings the researcher to life as the pool's knowledge curator: it reads expert state/logs, distills knowledge, promotes patterns to identity, and keeps state.md lean over time. + +The architecture doc validates: "Does curation keep state.md lean over time? Does cold-start seeding produce useful initial state?" + +## Phase 1: Researcher Daemon Wiring + +Wire the researcher into the daemon's event loop so it can receive and execute tasks. + +### Files to modify + +**`internal/daemon/daemon.go`** — 5 targeted changes: +1. `Run()` (~line 148): Add `watcher.Add()` for researcher inbox, mirroring architect +2. `drainAllInboxes()` (~line 860): Add researcher drain goroutine +3. `resolveExpertName()` (~line 1163): Add researcher inbox → "researcher" mapping +4. `resolveExpertConfig()` (~line 972): Add `d.cfg.Researcher.Model` case +5. `resolveSessionTimeout()` (~line 1126): Add `d.cfg.Researcher.SessionTimeout` case + +**`internal/mcp/server.go`** — Replace TODO at line 62: +```go +if cfg.Role == "researcher" { + RegisterResearcherTools(srv, cfg) +} +``` + +**`internal/mcp/config.go`** — Add `ResearcherToolNames` slice + `ToolNamesForRole(role string)` helper that returns the correct tool name list per role. Update `ExpertToolNames` comment for clarity. + +**`internal/daemon/daemon.go`** — `processInboxMessage` (~line 624): Replace hardcoded `agentmcp.ExpertToolNames` with role-aware `agentmcp.ToolNamesForRole(expertName)`. + +### Tests + +- `TestResearcherInboxWatched` — drop message in researcher/inbox/, verify spawner called +- `TestResolveExpertConfig_Researcher` — researcher model from pool config +- `TestResolveSessionTimeout_Researcher` — researcher timeout parsing + +### Commit +`feat(daemon): wire researcher role into daemon lifecycle` + +--- + +## Phase 2: Researcher MCP Tools + +Six tools for cross-expert state management. New file: `internal/mcp/researcher_tools.go`. + +| Tool | R/W | Purpose | +|------|-----|---------| +| `list_experts` | R | All experts with state sizes, log counts, last task time | +| `read_expert_state` | R | Read another expert's identity/state/errors files | +| `read_expert_logs` | R | Last N log index entries, optional query filter | +| `enrich_state` | R | Full context assembly for curation (state + recent logs) | +| `write_expert_state` | W | Write curated state back to an expert | +| `promote_pattern` | W | Append graduated pattern to an expert's identity.md | + +### Tool details + +**`list_experts`** — Loads pool config, stats each expert dir (pool-scoped + shared), returns name/type/state_bytes/log_count/last_task. Richer than concierge's list (which only returns names). + +**`read_expert_state`** — Params: `expert` (required), `file` (optional: identity/state/errors/all). Resolves expert dir via `mail.ResolveExpertDir`. Reuses `expert.ReadState()`. + +**`read_expert_logs`** — Params: `expert` (required), `count` (optional, default 10), `query` (optional). Reads `logs/index.md`, returns last N entries. Reuses `expert.SearchIndex()` for query. + +**`enrich_state`** — Params: `expert` (required). Returns identity + state + errors + last 10 index entries + last 3 full log file contents. This is the "read everything" step before the researcher reasons about curation. + +**`write_expert_state`** — Params: `expert` (required), `content` (required), `file` (optional: state/errors, default state). Validates size via `expert.MaxStateSize`. Uses `expert.WriteState()` or `expert.WriteErrors()`. + +**`promote_pattern`** — Params: `expert` (required), `pattern` (required), `section` (optional heading, default "## Graduated Patterns"). Reads identity.md, finds/creates section, appends pattern. Atomic write. + +### Files + +- **Create:** `internal/mcp/researcher_tools.go` (~250 lines) +- **Create:** `internal/mcp/researcher_tools_test.go` (~300 lines) +- **Modify:** `internal/mcp/testhelp_test.go` — add researcher case to `buildMCPTestServer` + +### Commit +`feat(mcp): implement researcher tools for cross-expert curation` + +--- + +## Phase 3: Extract `mail.Post()` + Curation Scheduler + +### Phase 3a: Extract `mail.Post()` + +Move `postMessage` logic from `internal/mcp/postoffice.go` to `internal/mail/post.go` as exported `Post(poolDir string, msg *Message) error`. Thin `mcp.postMessage` becomes `mail.Post(cfg.PoolDir, msg)`. This enables CLI and daemon to post without importing `mcp`. + +**Files:** +- **Create:** `internal/mail/post.go` +- **Modify:** `internal/mcp/postoffice.go` — delegate to `mail.Post()` +- **Create:** `internal/mail/post_test.go` + +### Phase 3b: Curation Scheduler + +New file `internal/daemon/curation.go`: + +```go +type curationScheduler struct { + intervalTasks int + intervalHours int + poolDir string + logger *slog.Logger + taskCount int + lastCuration time.Time + mu sync.Mutex +} + +func newCurationScheduler(cfg *config.CurationSection, poolDir string, logger *slog.Logger) *curationScheduler +func (cs *curationScheduler) RecordTaskCompletion() bool // returns true when threshold hit +func (cs *curationScheduler) Reset() +``` + +**Daemon integration:** +- Add `curation *curationScheduler` field to `Daemon` +- Initialize in `New()` +- In `Run()`: start ticker goroutine for time-based trigger (period = `intervalHours`) +- In `markTaskCompleted()`: call `RecordTaskCompletion()`, trigger curation if true +- New method `triggerCuration(reason string)`: compose curation task message, post via `mail.Post()` + +**Event:** Add `EventCurationTriggered` to `internal/daemon/events.go`. + +**Curation task body** includes: list of experts, their state sizes, reason for trigger, instructions for the researcher (prune stale state, promote patterns, check sizes). + +**Files:** +- **Create:** `internal/daemon/curation.go` +- **Create:** `internal/daemon/curation_test.go` +- **Modify:** `internal/daemon/daemon.go` — integrate scheduler +- **Modify:** `internal/daemon/events.go` — add event type + +### Commits +``` +refactor(mail): extract Post() for CLI and daemon reuse +feat(daemon): add curation scheduler with task and time triggers +``` + +--- + +## Phase 4: Cold-Start Seeding + +`agent-pool seed --pool --expert ` — sends a seed task to the researcher. + +**`cmd/agent-pool/main.go`:** +- Add `seed` case to command switch +- `cmdSeed()`: parse flags, discover pool dir, validate expert exists, load identity.md for context, compose seed task message (from: "cli", to: "researcher", type: task), call `mail.Post()`, print confirmation +- Update `printUsage()` + +**Seed task body:** structured instructions telling the researcher to explore the project codebase (via `project_dir` from pool.toml) and create initial state.md for the named expert based on its identity.md. + +### Tests +- `TestCmdSeed_WritesPostoffice` — temp pool, verify message in postoffice +- `TestCmdSeed_UnknownExpert` — verify error + +### Commit +`feat(cli): add 'seed' command for cold-start expert bootstrapping` + +--- + +## Phase 5: Log Rotation + +New file `internal/expert/rotate.go`: + +```go +const DefaultLogRetention = 50 + +func RotateLogs(expertDir string, retention int) (archived int, err error) +``` + +Implementation: list `.json` files in `logs/`, sort by mtime newest-first, archive files beyond threshold into `logs/archive-{timestamp}.tar.gz` (stdlib `archive/tar` + `compress/gzip`), delete archived files + matching `.stderr` companions. `index.md` untouched (remains searchable). + +**Config:** Add `LogRetention int` to `config.DefaultsSection`, default 50 in `LoadPool()`. + +**Integration:** `triggerCuration()` in daemon runs rotation for all experts before generating the researcher task. + +### Files +- **Create:** `internal/expert/rotate.go` +- **Create:** `internal/expert/rotate_test.go` +- **Modify:** `internal/config/config.go` — add `LogRetention` +- **Modify:** `internal/daemon/curation.go` — call rotation + +### Commit +`feat(expert): add log rotation with configurable retention` + +--- + +## Phase 6: Shared Expert Enrichment + +Update Phase 2 tools to handle shared experts with layered state. + +**Changes in `researcher_tools.go`:** +- `read_expert_state`: detect shared expert, return both user-level and project overlay state +- `write_expert_state`: add `layer` param ("user"/"project") for shared experts +- `enrich_state`: return both layers clearly labeled +- `promote_pattern`: target user-level identity.md for shared experts (patterns are cross-pool) +- `list_experts`: include shared expert metadata (user + overlay sizes) + +Detection: load pool config, check `cfg.Shared.Include`, resolve paths via existing `config.SharedExpertDir()` and `{poolDir}/shared-state/{name}/`. + +### Tests +- `TestReadExpertState_SharedExpert` — both layers returned +- `TestWriteExpertState_SharedExpert_UserLayer` / `_ProjectLayer` +- `TestPromotePattern_SharedExpert` — writes user-level identity.md + +### Commit +`feat(researcher): enable shared expert enrichment with layered state` + +--- + +## Dependency Graph + +``` +Phase 1 (daemon wiring) ──┐ + v +Phase 2 (tools) ──────────┬──> Phase 6 (shared enrichment) + │ +Phase 3a (mail.Post) ─────┤ + v +Phase 3b (scheduler) ─────┤ + v +Phase 4 (seed CLI) ───────┘ + +Phase 5 (log rotation) ── independent, can parallel with 3-4-6 +``` + +## Deferred + +| Item | Why | +|------|-----| +| Auto-seed on empty state.md | Prevents duplicate seeds; manual `seed` command sufficient | +| Per-expert log retention | Global default covers it; trivial to add later | +| Archive extraction in `recall` | index.md still searchable; manual extraction if needed | +| Multi-pool researcher | Requires cross-pool coordination; single-pool is v0.8 scope | +| Curation metrics dashboard | Manual state size checks sufficient for validation | + +## Verification + +After each phase: +1. `make test` — all existing + new tests pass +2. `make build` — compiles cleanly + +End-to-end validation after all phases: +1. Create a test pool with 2 experts and a researcher section in pool.toml +2. Start daemon, send tasks to experts, verify they complete +3. After `interval_tasks` completions, verify curation task appears in researcher/inbox/ +4. Spawn researcher manually (`agent-pool mcp --pool --role researcher`), call `list_experts`, `enrich_state`, `write_expert_state` +5. Run `agent-pool seed --pool --expert auth`, verify seed message in postoffice +6. Create 60 log files in an expert dir, trigger rotation, verify archive + 50 remaining diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 6bf5fd5..7b9c4bd 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -151,11 +151,15 @@ func (d *Daemon) Run(ctx context.Context) error { return fmt.Errorf("watching postoffice: %w", err) } - // Watch architect inbox + // Watch built-in role inboxes (architect, researcher) architectInbox := mail.ResolveInbox(d.poolDir, "architect") if err := watcher.Add(architectInbox); err != nil { return fmt.Errorf("watching architect inbox: %w", err) } + researcherInbox := mail.ResolveInbox(d.poolDir, "researcher") + if err := watcher.Add(researcherInbox); err != nil { + return fmt.Errorf("watching researcher inbox: %w", err) + } // Watch approvals directory for human approval requests approvalsDir := filepath.Join(d.poolDir, "approvals") @@ -630,8 +634,9 @@ func (d *Daemon) processInboxMessage(ctx context.Context, expertName string, pat } }() - // Append pool MCP tool names so they're pre-approved in headless mode - allTools := append(tools, agentmcp.ExpertToolNames...) + // Append pool MCP tool names so they're pre-approved in headless mode. + // Built-in roles get their role-specific tools in addition to expert tools. + allTools := append(tools, agentmcp.ToolNamesForRole(expertName)...) cfg := &expert.SpawnConfig{ Name: expertName, @@ -865,9 +870,11 @@ func (d *Daemon) markTaskFailed(taskID string, exitCode int) { // drainAllInboxes processes any files sitting in expert and architect inboxes // when the daemon starts. Each drains in its own goroutine via handleInbox. func (d *Daemon) drainAllInboxes(ctx context.Context) { - // Drain architect inbox + // Drain built-in role inboxes d.wg.Add(1) go func() { defer d.wg.Done(); d.handleInbox(ctx, "architect", "") }() + d.wg.Add(1) + go func() { defer d.wg.Done(); d.handleInbox(ctx, "researcher", "") }() for name := range d.cfg.Experts { d.wg.Add(1) @@ -981,6 +988,13 @@ func (d *Daemon) resolveExpertConfig(name string) (model string, tools []string) return model, tools } + if name == "researcher" { + if d.cfg.Researcher.Model != "" { + model = d.cfg.Researcher.Model + } + return model, tools + } + if ec, ok := d.cfg.Experts[name]; ok { if ec.Model != "" { model = ec.Model @@ -1132,6 +1146,13 @@ func (d *Daemon) resolveSessionTimeout(name string) (time.Duration, error) { } return dur, nil } + if name == "researcher" && d.cfg.Researcher.SessionTimeout != "" { + dur, err := time.ParseDuration(d.cfg.Researcher.SessionTimeout) + if err != nil { + return 0, fmt.Errorf("parsing researcher.session_timeout %q: %w", d.cfg.Researcher.SessionTimeout, err) + } + return dur, nil + } return d.cfg.Defaults.ParseSessionTimeout() } @@ -1162,12 +1183,14 @@ func (d *Daemon) resolveProjectDir() string { } // resolveExpertName extracts the expert name from an inbox directory path. -// Checks built-in roles (architect) first, then pool-scoped experts. +// Checks built-in roles first, then pool-scoped experts. func (d *Daemon) resolveExpertName(inboxDir string) string { - // Check architect first — it's the only built-in role we spawn in v0.4 - architectInbox := mail.ResolveInbox(d.poolDir, "architect") - if absEqual(inboxDir, architectInbox) { - return "architect" + // Check built-in roles first + for _, role := range []string{"architect", "researcher"} { + roleInbox := mail.ResolveInbox(d.poolDir, role) + if absEqual(inboxDir, roleInbox) { + return role + } } for name := range d.cfg.Experts { @@ -1199,6 +1222,7 @@ func (d *Daemon) ensureDirs() error { filepath.Join(d.poolDir, "architect", "logs"), filepath.Join(d.poolDir, "architect", "verifications"), filepath.Join(d.poolDir, "researcher", "inbox"), + filepath.Join(d.poolDir, "researcher", "logs"), filepath.Join(d.poolDir, "concierge", "inbox"), } diff --git a/internal/daemon/daemon_test.go b/internal/daemon/daemon_test.go index 4c696d7..0aaee0d 100644 --- a/internal/daemon/daemon_test.go +++ b/internal/daemon/daemon_test.go @@ -32,6 +32,11 @@ // - ArchitectConfigResolution: architect uses opus, auth uses haiku // - ArchitectInboxDrainOnStart: pre-existing inbox message processed on startup // - NotifyRoutedNotRegistered: notify messages route to inbox but skip taskboard +// +// Researcher: +// - ResearcherSpawn: message to researcher routes and spawns with configured model +// - ResearcherConfigResolution: researcher uses haiku, auth uses sonnet +// - ResearcherInboxDrainOnStart: pre-existing inbox message processed on startup package daemon_test import ( @@ -2261,3 +2266,220 @@ approval_mode = "none" shutdownDaemon(t, cancel, errCh) } + +// Researcher: +// - ResearcherSpawn: message to researcher routes and spawns with configured model +// - ResearcherConfigResolution: researcher uses haiku, architect uses opus +// - ResearcherInboxDrainOnStart: pre-existing inbox message processed on startup + +func TestDaemon_ResearcherSpawn(t *testing.T) { + poolDir := t.TempDir() + + poolToml := `[pool] +name = "test-pool" +project_dir = "` + poolDir + `" + +[researcher] +model = "haiku" + +[experts.auth] +` + os.WriteFile(filepath.Join(poolDir, "pool.toml"), []byte(poolToml), 0o644) + + cfg, err := config.LoadPool(poolDir) + if err != nil { + t.Fatalf("LoadPool: %v", err) + } + + fake := &fakeSpawner{} + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) + d := daemon.New(cfg, poolDir, logger, daemon.WithSpawner(fake)) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- d.Run(ctx) }() + + time.Sleep(500 * time.Millisecond) + + // Route a message to researcher via postoffice + writeMessage(t, filepath.Join(poolDir, "postoffice"), "task-res-001", "architect", "researcher") + + // Wait for spawn + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if len(fake.getCalls()) > 0 { + break + } + time.Sleep(100 * time.Millisecond) + } + + calls := fake.getCalls() + if len(calls) == 0 { + t.Fatal("expected researcher spawn, got none") + } + + call := calls[0] + if call.Name != "researcher" { + t.Errorf("spawn name = %q, want researcher", call.Name) + } + if call.Model != "haiku" { + t.Errorf("spawn model = %q, want haiku", call.Model) + } + if call.TaskMessage.ID != "task-res-001" { + t.Errorf("task ID = %q, want task-res-001", call.TaskMessage.ID) + } + + // Verify log file written to researcher dir (not experts/researcher) + logPath := filepath.Join(poolDir, "researcher", "logs", "task-res-001.json") + deadline = time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if _, err := os.Stat(logPath); err == nil { + break + } + time.Sleep(100 * time.Millisecond) + } + + if _, err := os.Stat(logPath); os.IsNotExist(err) { + t.Error("log should be written to researcher/logs/, not experts/researcher/logs/") + } + + shutdownDaemon(t, cancel, errCh) +} + +func TestDaemon_ResearcherConfigResolution(t *testing.T) { + poolDir := t.TempDir() + + poolToml := `[pool] +name = "test-pool" +project_dir = "` + poolDir + `" + +[defaults] +model = "sonnet" + +[architect] +model = "opus" + +[researcher] +model = "haiku" +session_timeout = "15m" + +[experts.auth] +model = "sonnet" +` + os.WriteFile(filepath.Join(poolDir, "pool.toml"), []byte(poolToml), 0o644) + + cfg, err := config.LoadPool(poolDir) + if err != nil { + t.Fatalf("LoadPool: %v", err) + } + + fake := &fakeSpawner{} + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) + d := daemon.New(cfg, poolDir, logger, daemon.WithSpawner(fake)) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- d.Run(ctx) }() + + time.Sleep(500 * time.Millisecond) + + // Route to researcher + writeMessage(t, filepath.Join(poolDir, "postoffice"), "task-cfg-res", "architect", "researcher") + // Route to auth + writeMessage(t, filepath.Join(poolDir, "postoffice"), "task-cfg-auth2", "architect", "auth") + + // Wait for both spawns + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if len(fake.getCalls()) >= 2 { + break + } + time.Sleep(100 * time.Millisecond) + } + + calls := fake.getCalls() + if len(calls) < 2 { + t.Fatalf("expected 2 spawns, got %d", len(calls)) + } + + for _, c := range calls { + switch c.Name { + case "researcher": + if c.Model != "haiku" { + t.Errorf("researcher model = %q, want haiku", c.Model) + } + case "auth": + if c.Model != "sonnet" { + t.Errorf("auth model = %q, want sonnet", c.Model) + } + } + } + + shutdownDaemon(t, cancel, errCh) +} + +func TestDaemon_ResearcherInboxDrainOnStart(t *testing.T) { + poolDir := t.TempDir() + + poolToml := `[pool] +name = "test-pool" +project_dir = "` + poolDir + `" + +[researcher] +model = "haiku" + +[experts.auth] +` + os.WriteFile(filepath.Join(poolDir, "pool.toml"), []byte(poolToml), 0o644) + + cfg, err := config.LoadPool(poolDir) + if err != nil { + t.Fatalf("LoadPool: %v", err) + } + + // Pre-create inbox and put a message there before daemon starts + researcherInbox := filepath.Join(poolDir, "researcher", "inbox") + os.MkdirAll(researcherInbox, 0o755) + os.MkdirAll(filepath.Join(poolDir, "researcher", "logs"), 0o755) + os.MkdirAll(filepath.Join(poolDir, "architect", "inbox"), 0o755) + os.MkdirAll(filepath.Join(poolDir, "architect", "logs"), 0o755) + os.MkdirAll(filepath.Join(poolDir, "postoffice"), 0o755) + os.MkdirAll(filepath.Join(poolDir, "experts", "auth", "inbox"), 0o755) + os.MkdirAll(filepath.Join(poolDir, "experts", "auth", "logs"), 0o755) + + writeMessage(t, researcherInbox, "task-predrain-res", "architect", "researcher") + + fake := &fakeSpawner{} + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) + d := daemon.New(cfg, poolDir, logger, daemon.WithSpawner(fake)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- d.Run(ctx) }() + + // Wait for drain to process pre-existing message + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if len(fake.getCalls()) > 0 { + break + } + time.Sleep(100 * time.Millisecond) + } + + calls := fake.getCalls() + if len(calls) == 0 { + t.Fatal("expected pre-existing researcher inbox message to be processed on startup") + } + + if calls[0].Name != "researcher" { + t.Errorf("spawn name = %q, want researcher", calls[0].Name) + } + + shutdownDaemon(t, cancel, errCh) +} diff --git a/internal/mcp/config.go b/internal/mcp/config.go index 00ef8ed..a27585f 100644 --- a/internal/mcp/config.go +++ b/internal/mcp/config.go @@ -7,7 +7,7 @@ import ( "os/exec" ) -// ExpertToolNames returns the --allowedTools names for pool MCP tools. +// ExpertToolNames lists the --allowedTools names for base expert MCP tools. // Claude Code requires MCP tools to be explicitly allowed in headless mode. // Format: mcp____ var ExpertToolNames = []string{ @@ -19,6 +19,42 @@ var ExpertToolNames = []string{ "mcp__agent-pool__search_index", } +// ArchitectToolNames lists the architect-specific MCP tool names (in addition +// to ExpertToolNames). +var ArchitectToolNames = []string{ + "mcp__agent-pool__define_contract", + "mcp__agent-pool__send_task", + "mcp__agent-pool__verify_result", + "mcp__agent-pool__amend_contract", +} + +// ResearcherToolNames lists the researcher-specific MCP tool names (in addition +// to ExpertToolNames). +var ResearcherToolNames = []string{ + "mcp__agent-pool__list_experts", + "mcp__agent-pool__read_expert_state", + "mcp__agent-pool__read_expert_logs", + "mcp__agent-pool__enrich_state", + "mcp__agent-pool__write_expert_state", + "mcp__agent-pool__promote_pattern", +} + +// ToolNamesForRole returns the full set of MCP tool names for the given role. +// Built-in roles get their role-specific tools appended to ExpertToolNames. +// Unknown roles (including regular experts) get ExpertToolNames only. +func ToolNamesForRole(name string) []string { + base := make([]string, len(ExpertToolNames)) + copy(base, ExpertToolNames) + switch name { + case "architect": + return append(base, ArchitectToolNames...) + case "researcher": + return append(base, ResearcherToolNames...) + default: + return base + } +} + // MCPConfig is the JSON structure claude expects for --mcp-config. type MCPConfig struct { MCPServers map[string]MCPServerEntry `json:"mcpServers"` diff --git a/internal/mcp/config_test.go b/internal/mcp/config_test.go index 1b4efd2..0b2416e 100644 --- a/internal/mcp/config_test.go +++ b/internal/mcp/config_test.go @@ -8,6 +8,12 @@ // // WriteTempConfigShared: // - Creates valid JSON with --shared true in args +// +// ToolNamesForRole: +// - Architect: expert + architect tools +// - Researcher: expert + researcher tools +// - Unknown/expert: expert tools only +// - Returned slice is a copy (mutation-safe) package mcp_test @@ -111,3 +117,78 @@ func TestWriteTempConfigShared_IncludesSharedFlag(t *testing.T) { } } } + +func TestToolNamesForRole_Architect(t *testing.T) { + names := agentmcp.ToolNamesForRole("architect") + + // Should include all expert tools + for _, tool := range agentmcp.ExpertToolNames { + if !contains(names, tool) { + t.Errorf("missing expert tool %q", tool) + } + } + + // Should include all architect tools + for _, tool := range agentmcp.ArchitectToolNames { + if !contains(names, tool) { + t.Errorf("missing architect tool %q", tool) + } + } + + // Should NOT include researcher tools + for _, tool := range agentmcp.ResearcherToolNames { + if contains(names, tool) { + t.Errorf("unexpected researcher tool %q in architect role", tool) + } + } +} + +func TestToolNamesForRole_Researcher(t *testing.T) { + names := agentmcp.ToolNamesForRole("researcher") + + for _, tool := range agentmcp.ExpertToolNames { + if !contains(names, tool) { + t.Errorf("missing expert tool %q", tool) + } + } + + for _, tool := range agentmcp.ResearcherToolNames { + if !contains(names, tool) { + t.Errorf("missing researcher tool %q", tool) + } + } + + for _, tool := range agentmcp.ArchitectToolNames { + if contains(names, tool) { + t.Errorf("unexpected architect tool %q in researcher role", tool) + } + } +} + +func TestToolNamesForRole_Expert(t *testing.T) { + names := agentmcp.ToolNamesForRole("auth") + + if len(names) != len(agentmcp.ExpertToolNames) { + t.Errorf("got %d tools, want %d (expert tools only)", len(names), len(agentmcp.ExpertToolNames)) + } +} + +func TestToolNamesForRole_ReturnsCopy(t *testing.T) { + names1 := agentmcp.ToolNamesForRole("researcher") + names2 := agentmcp.ToolNamesForRole("researcher") + + // Mutating one should not affect the other + names1[0] = "mutated" + if names2[0] == "mutated" { + t.Error("ToolNamesForRole should return independent copies") + } +} + +func contains(ss []string, s string) bool { + for _, v := range ss { + if v == s { + return true + } + } + return false +} diff --git a/internal/mcp/researcher_tools.go b/internal/mcp/researcher_tools.go new file mode 100644 index 0000000..5d1a8f2 --- /dev/null +++ b/internal/mcp/researcher_tools.go @@ -0,0 +1,18 @@ +package mcp + +import ( + "github.com/mark3labs/mcp-go/server" +) + +// RegisterResearcherTools adds researcher-scope tools to the MCP server. +// These are registered in addition to the expert tools when running as +// the researcher role. The researcher reads cross-expert state and logs, +// writes curated state back, and promotes patterns to identity. +func RegisterResearcherTools(srv *server.MCPServer, cfg *ServerConfig) { + if cfg == nil { + return + } + + // Phase 2 will add: list_experts, read_expert_state, read_expert_logs, + // enrich_state, write_expert_state, promote_pattern +} diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 3759b28..0ce41c3 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -59,7 +59,9 @@ func Run(ctx context.Context, cfg *ServerConfig) error { if cfg.Role == "concierge" { RegisterConciergeTools(srv, cfg) } - // TODO(v0.6): RegisterResearcherTools when researcher role is implemented + if cfg.Role == "researcher" { + RegisterResearcherTools(srv, cfg) + } cfg.Logger.Info("Preparing to serve MCP tools", "pool_dir", cfg.PoolDir, From 859174c1094009cebe96fefa89e03b292a6f9d57 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 20:26:12 -0500 Subject: [PATCH 2/9] feat(mcp): implement researcher tools for cross-expert curation Six tools for the researcher role: - list_experts: all experts with state sizes, log counts, last task - read_expert_state: read another expert's identity/state/errors - read_expert_logs: last N log index entries with optional query - enrich_state: full context assembly (state + recent logs + summaries) - write_expert_state: write curated state or errors back to an expert - promote_pattern: append graduated pattern to identity.md The researcher reads any expert's state and logs, reasons about what to keep/prune/promote, then writes curated results back. Two-step pattern gives the LLM control over curation decisions. Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/mcp/researcher_tools.go | 458 ++++++++++++++++++++++++- internal/mcp/researcher_tools_test.go | 463 ++++++++++++++++++++++++++ internal/mcp/testhelp_test.go | 2 + 3 files changed, 921 insertions(+), 2 deletions(-) create mode 100644 internal/mcp/researcher_tools_test.go diff --git a/internal/mcp/researcher_tools.go b/internal/mcp/researcher_tools.go index 5d1a8f2..cee1b39 100644 --- a/internal/mcp/researcher_tools.go +++ b/internal/mcp/researcher_tools.go @@ -1,7 +1,21 @@ package mcp import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + + "github.com/mark3labs/mcp-go/mcp" "github.com/mark3labs/mcp-go/server" + + "github.com/cameronsjo/agent-pool/internal/atomicfile" + "github.com/cameronsjo/agent-pool/internal/config" + "github.com/cameronsjo/agent-pool/internal/expert" + "github.com/cameronsjo/agent-pool/internal/mail" ) // RegisterResearcherTools adds researcher-scope tools to the MCP server. @@ -13,6 +27,446 @@ func RegisterResearcherTools(srv *server.MCPServer, cfg *ServerConfig) { return } - // Phase 2 will add: list_experts, read_expert_state, read_expert_logs, - // enrich_state, write_expert_state, promote_pattern + srv.AddTool( + mcp.NewTool("list_experts", + mcp.WithDescription("List all experts in the pool with state file sizes, log counts, and last task time. Use to triage which experts need curation."), + mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), + ), + handleResearcherListExperts(cfg), + ) + + srv.AddTool( + mcp.NewTool("read_expert_state", + mcp.WithDescription("Read another expert's state files (identity.md, state.md, errors.md). Returns all files by default, or a specific file."), + mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), + mcp.WithString("expert", mcp.Required(), mcp.Description("Expert name to read state from")), + mcp.WithString("file", mcp.Description("Specific file to read: 'identity', 'state', 'errors', or 'all' (default)")), + ), + handleReadExpertState(cfg), + ) + + srv.AddTool( + mcp.NewTool("read_expert_logs", + mcp.WithDescription("Read another expert's recent log index entries. Returns the last N entries, optionally filtered by a search query."), + mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), + mcp.WithString("expert", mcp.Required(), mcp.Description("Expert name to read logs from")), + mcp.WithString("count", mcp.Description("Number of recent entries to return (default: 10)")), + mcp.WithString("query", mcp.Description("Optional search query to filter entries (case-insensitive substring)")), + ), + handleReadExpertLogs(cfg), + ) + + srv.AddTool( + mcp.NewTool("enrich_state", + mcp.WithDescription("Assemble an expert's full context for curation analysis. Returns identity, state, errors, recent log index entries, and the last 3 full log file contents."), + mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), + mcp.WithString("expert", mcp.Required(), mcp.Description("Expert name to assemble context for")), + ), + handleEnrichState(cfg), + ) + + srv.AddTool( + mcp.NewTool("write_expert_state", + mcp.WithDescription("Write curated state back to an expert. Targets state.md by default. Content must be non-empty and under 50KB."), + mcp.WithString("expert", mcp.Required(), mcp.Description("Expert name to write state to")), + mcp.WithString("content", mcp.Required(), mcp.Description("New file content")), + mcp.WithString("file", mcp.Description("Target file: 'state' (default) or 'errors'")), + ), + handleWriteExpertState(cfg), + ) + + srv.AddTool( + mcp.NewTool("promote_pattern", + mcp.WithDescription("Append a graduated pattern to an expert's identity.md. Patterns promoted to identity become permanent expert knowledge."), + mcp.WithString("expert", mcp.Required(), mcp.Description("Expert name")), + mcp.WithString("pattern", mcp.Required(), mcp.Description("Pattern text to append (markdown)")), + mcp.WithString("section", mcp.Description("Heading to append under (default: '## Graduated Patterns')")), + ), + handlePromotePattern(cfg), + ) +} + +// expertInfo holds metadata about an expert for list_experts. +type expertInfo struct { + Name string `json:"name"` + Type string `json:"type"` // "pool" or "shared" + StateBytes int64 `json:"state_bytes"` + LogCount int `json:"log_count"` + LastTask string `json:"last_task,omitempty"` +} + +func handleResearcherListExperts(cfg *ServerConfig) server.ToolHandlerFunc { + return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) { + poolCfg, err := config.LoadPool(cfg.PoolDir) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("loading pool config: %v", err)), nil + } + + var experts []expertInfo + + // Pool-scoped experts + var names []string + for name := range poolCfg.Experts { + names = append(names, name) + } + sort.Strings(names) + + for _, name := range names { + dir := mail.ResolveExpertDir(cfg.PoolDir, name) + experts = append(experts, gatherExpertInfo(name, "pool", dir)) + } + + // Shared experts + for _, name := range poolCfg.Shared.Include { + dir, err := config.SharedExpertDir(name) + if err != nil { + continue + } + experts = append(experts, gatherExpertInfo(name, "shared", dir)) + } + + data, err := json.MarshalIndent(experts, "", " ") + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("marshaling result: %v", err)), nil + } + + return mcp.NewToolResultText(string(data)), nil + } +} + +// gatherExpertInfo stats an expert directory for metadata. +func gatherExpertInfo(name, expertType, dir string) expertInfo { + info := expertInfo{Name: name, Type: expertType} + + if fi, err := os.Stat(filepath.Join(dir, "state.md")); err == nil { + info.StateBytes = fi.Size() + } + + logsDir := filepath.Join(dir, "logs") + if entries, err := os.ReadDir(logsDir); err == nil { + for _, e := range entries { + if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { + info.LogCount++ + } + } + } + + // Last task from index.md (last non-empty line) + if data, err := os.ReadFile(filepath.Join(logsDir, "index.md")); err == nil { + lines := strings.Split(strings.TrimSpace(string(data)), "\n") + for i := len(lines) - 1; i >= 2; i-- { // skip header rows + line := strings.TrimSpace(lines[i]) + if line != "" && strings.HasPrefix(line, "|") { + // Extract task ID from first column + parts := strings.SplitN(line, "|", 3) + if len(parts) >= 3 { + info.LastTask = strings.TrimSpace(parts[1]) + } + break + } + } + } + + return info +} + +func handleReadExpertState(cfg *ServerConfig) server.ToolHandlerFunc { + return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) { + expertName := request.GetString("expert", "") + if expertName == "" { + return mcp.NewToolResultError("expert parameter is required"), nil + } + + dir := resolveTargetExpertDir(cfg.PoolDir, expertName) + file := request.GetString("file", "all") + + switch file { + case "identity": + content, err := readFileOr(filepath.Join(dir, "identity.md"), "") + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("reading identity.md: %v", err)), nil + } + return mcp.NewToolResultText(content), nil + + case "state": + content, err := readFileOr(filepath.Join(dir, "state.md"), "") + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("reading state.md: %v", err)), nil + } + return mcp.NewToolResultText(content), nil + + case "errors": + content, err := readFileOr(filepath.Join(dir, "errors.md"), "") + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("reading errors.md: %v", err)), nil + } + return mcp.NewToolResultText(content), nil + + case "all", "": + identity, state, errors, err := expert.ReadState(dir) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("reading state: %v", err)), nil + } + result := map[string]string{ + "identity": identity, + "state": state, + "errors": errors, + } + data, err := json.MarshalIndent(result, "", " ") + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("marshaling: %v", err)), nil + } + return mcp.NewToolResultText(string(data)), nil + + default: + return mcp.NewToolResultError(fmt.Sprintf("invalid file %q: use 'identity', 'state', 'errors', or 'all'", file)), nil + } + } +} + +func handleReadExpertLogs(cfg *ServerConfig) server.ToolHandlerFunc { + return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) { + expertName := request.GetString("expert", "") + if expertName == "" { + return mcp.NewToolResultError("expert parameter is required"), nil + } + + dir := resolveTargetExpertDir(cfg.PoolDir, expertName) + query := request.GetString("query", "") + + countStr := request.GetString("count", "10") + count := 10 + if _, err := fmt.Sscanf(countStr, "%d", &count); err != nil || count <= 0 { + count = 10 + } + + if query != "" { + matches, err := expert.SearchIndex(dir, query) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("searching index: %v", err)), nil + } + if len(matches) == 0 { + return mcp.NewToolResultText("no matching entries found"), nil + } + if len(matches) > count { + matches = matches[len(matches)-count:] + } + return mcp.NewToolResultText(strings.Join(matches, "\n")), nil + } + + // No query — return last N entries from index + indexPath := filepath.Join(dir, "logs", "index.md") + data, err := os.ReadFile(indexPath) + if err != nil { + if os.IsNotExist(err) { + return mcp.NewToolResultText("no log entries yet"), nil + } + return mcp.NewToolResultError(fmt.Sprintf("reading index: %v", err)), nil + } + + lines := strings.Split(strings.TrimSpace(string(data)), "\n") + // Skip header rows (lines 0 and 1) + var entries []string + for i := 2; i < len(lines); i++ { + line := strings.TrimSpace(lines[i]) + if line != "" { + entries = append(entries, line) + } + } + + if len(entries) == 0 { + return mcp.NewToolResultText("no log entries yet"), nil + } + + // Return last N + start := 0 + if len(entries) > count { + start = len(entries) - count + } + return mcp.NewToolResultText(strings.Join(entries[start:], "\n")), nil + } +} + +func handleEnrichState(cfg *ServerConfig) server.ToolHandlerFunc { + return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) { + expertName := request.GetString("expert", "") + if expertName == "" { + return mcp.NewToolResultError("expert parameter is required"), nil + } + + dir := resolveTargetExpertDir(cfg.PoolDir, expertName) + + identity, state, errors, err := expert.ReadState(dir) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("reading state: %v", err)), nil + } + + // Read last 10 index entries + var recentIndex []string + indexPath := filepath.Join(dir, "logs", "index.md") + if data, err := os.ReadFile(indexPath); err == nil { + lines := strings.Split(strings.TrimSpace(string(data)), "\n") + for i := 2; i < len(lines); i++ { + line := strings.TrimSpace(lines[i]) + if line != "" { + recentIndex = append(recentIndex, line) + } + } + if len(recentIndex) > 10 { + recentIndex = recentIndex[len(recentIndex)-10:] + } + } + + // Read last 3 full log files (newest first by filename sort) + var recentLogs []map[string]string + logsDir := filepath.Join(dir, "logs") + if entries, err := os.ReadDir(logsDir); err == nil { + var jsonFiles []string + for _, e := range entries { + if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { + jsonFiles = append(jsonFiles, e.Name()) + } + } + sort.Strings(jsonFiles) // lexicographic ≈ chronological for task IDs + if len(jsonFiles) > 3 { + jsonFiles = jsonFiles[len(jsonFiles)-3:] + } + for _, f := range jsonFiles { + content, readErr := os.ReadFile(filepath.Join(logsDir, f)) + if readErr == nil { + taskID := strings.TrimSuffix(f, ".json") + summary := expert.ExtractSummary(content) + recentLogs = append(recentLogs, map[string]string{ + "task_id": taskID, + "summary": summary, + }) + } + } + } + + result := map[string]any{ + "identity": identity, + "state": state, + "errors": errors, + "recent_index": recentIndex, + "recent_logs": recentLogs, + } + + data, err := json.MarshalIndent(result, "", " ") + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("marshaling: %v", err)), nil + } + + return mcp.NewToolResultText(string(data)), nil + } +} + +func handleWriteExpertState(cfg *ServerConfig) server.ToolHandlerFunc { + return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) { + expertName := request.GetString("expert", "") + if expertName == "" { + return mcp.NewToolResultError("expert parameter is required"), nil + } + + content := request.GetString("content", "") + if content == "" { + return mcp.NewToolResultError("content parameter is required"), nil + } + + dir := resolveTargetExpertDir(cfg.PoolDir, expertName) + file := request.GetString("file", "state") + + switch file { + case "state", "": + if err := expert.WriteState(dir, content); err != nil { + return mcp.NewToolResultError(fmt.Sprintf("writing state.md: %v", err)), nil + } + return mcp.NewToolResultText(fmt.Sprintf("%s/state.md updated (%d bytes)", expertName, len(content))), nil + + case "errors": + // Overwrite errors.md entirely (not append — researcher is curating) + content = strings.TrimSpace(content) + if len(content) > expert.MaxStateSize { + return mcp.NewToolResultError(fmt.Sprintf("content exceeds maximum size (%d > %d bytes)", len(content), expert.MaxStateSize)), nil + } + path := filepath.Join(dir, "errors.md") + if err := atomicfile.WriteFile(path, []byte(content+"\n")); err != nil { + return mcp.NewToolResultError(fmt.Sprintf("writing errors.md: %v", err)), nil + } + return mcp.NewToolResultText(fmt.Sprintf("%s/errors.md updated (%d bytes)", expertName, len(content))), nil + + default: + return mcp.NewToolResultError(fmt.Sprintf("invalid file %q: use 'state' or 'errors'", file)), nil + } + } +} + +func handlePromotePattern(cfg *ServerConfig) server.ToolHandlerFunc { + return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) { + expertName := request.GetString("expert", "") + if expertName == "" { + return mcp.NewToolResultError("expert parameter is required"), nil + } + + pattern := request.GetString("pattern", "") + if pattern == "" { + return mcp.NewToolResultError("pattern parameter is required"), nil + } + + section := request.GetString("section", "## Graduated Patterns") + + dir := resolveTargetExpertDir(cfg.PoolDir, expertName) + identityPath := filepath.Join(dir, "identity.md") + + existing, err := os.ReadFile(identityPath) + if err != nil && !os.IsNotExist(err) { + return mcp.NewToolResultError(fmt.Sprintf("reading identity.md: %v", err)), nil + } + + content := string(existing) + + // Find or create the target section + if strings.Contains(content, section) { + // Append after the section heading + idx := strings.Index(content, section) + insertAt := idx + len(section) + // Skip to end of heading line + if nl := strings.Index(content[insertAt:], "\n"); nl >= 0 { + insertAt += nl + } else { + insertAt = len(content) + } + content = content[:insertAt] + "\n\n" + strings.TrimSpace(pattern) + "\n" + content[insertAt:] + } else { + // Append section at end + if content != "" && !strings.HasSuffix(content, "\n") { + content += "\n" + } + content += "\n" + section + "\n\n" + strings.TrimSpace(pattern) + "\n" + } + + if err := atomicfile.WriteFile(identityPath, []byte(content)); err != nil { + return mcp.NewToolResultError(fmt.Sprintf("writing identity.md: %v", err)), nil + } + + return mcp.NewToolResultText(fmt.Sprintf("pattern promoted to %s/identity.md under %q", expertName, section)), nil + } +} + +// resolveTargetExpertDir returns the state directory for a target expert. +// Built-in roles use {poolDir}/{role}/, pool-scoped use {poolDir}/experts/{name}/. +func resolveTargetExpertDir(poolDir, name string) string { + return mail.ResolveExpertDir(poolDir, name) +} + +// readFileOr reads a file and returns its trimmed content, or the fallback if not found. +func readFileOr(path, fallback string) (string, error) { + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return fallback, nil + } + return "", err + } + return strings.TrimSpace(string(data)), nil } diff --git a/internal/mcp/researcher_tools_test.go b/internal/mcp/researcher_tools_test.go new file mode 100644 index 0000000..487ab6c --- /dev/null +++ b/internal/mcp/researcher_tools_test.go @@ -0,0 +1,463 @@ +// Test plan for researcher_tools.go: +// +// RegisterResearcherTools (INTEGRATION) +// [x] All 6 researcher tools + 6 expert tools registered +// +// list_experts (FILESYSTEM I/O) +// [x] Happy: returns experts with state sizes and log counts +// [x] Edge: empty pool returns empty list +// +// read_expert_state (FILESYSTEM I/O) +// [x] Happy: reads all state files +// [x] Happy: reads single file (identity only) +// [x] Error: missing expert param +// +// read_expert_logs (FILESYSTEM I/O) +// [x] Happy: returns last N log entries +// [x] Happy: query filters entries +// [x] Edge: empty logs dir returns no entries +// +// enrich_state (FILESYSTEM I/O) +// [x] Happy: returns assembled context with logs +// +// write_expert_state (FILESYSTEM I/O) +// [x] Happy: writes curated state.md +// [x] Happy: writes curated errors.md +// [x] Error: empty content +// +// promote_pattern (FILESYSTEM I/O) +// [x] Happy: pattern appended to existing identity.md +// [x] Happy: creates section if absent +// [x] Edge: empty identity.md gets header + pattern + +package mcp_test + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/cameronsjo/agent-pool/internal/expert" +) + +// setupResearcherPool creates a pool directory with a researcher and target experts. +func setupResearcherPool(t *testing.T) (poolDir string) { + t.Helper() + poolDir = makePoolDirs(t, + "researcher/inbox", + "researcher/logs", + "postoffice", + "experts/auth/inbox", + "experts/auth/logs", + "experts/billing/inbox", + "experts/billing/logs", + ) + + // Write pool.toml + poolToml := `[pool] +name = "test-pool" +project_dir = "` + poolDir + `" + +[experts.auth] +model = "sonnet" + +[experts.billing] +model = "haiku" +` + os.WriteFile(filepath.Join(poolDir, "pool.toml"), []byte(poolToml), 0o644) + + // Write state files for auth expert + authDir := filepath.Join(poolDir, "experts", "auth") + os.WriteFile(filepath.Join(authDir, "identity.md"), []byte("# Auth Expert\n\nHandles authentication.\n"), 0o644) + os.WriteFile(filepath.Join(authDir, "state.md"), []byte("OAuth tokens cached.\n"), 0o644) + os.WriteFile(filepath.Join(authDir, "errors.md"), []byte("### 2026-04-01T00:00:00Z\n\nToken refresh failed.\n"), 0o644) + + // Write some log files for auth + expert.WriteLog(authDir, "task-001", []byte(`{"type":"result","result":"Built auth endpoint"}`)) + expert.WriteLog(authDir, "task-002", []byte(`{"type":"result","result":"Fixed OAuth bug"}`)) + expert.AppendIndex(authDir, &expert.LogEntry{TaskID: "task-001", From: "architect", ExitCode: 0, Summary: "Built auth endpoint"}) + expert.AppendIndex(authDir, &expert.LogEntry{TaskID: "task-002", From: "architect", ExitCode: 0, Summary: "Fixed OAuth bug"}) + + return poolDir +} + +func TestResearcherTools_Registered(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + names := listToolNames(t, srv) + + // Expert tools (6) + for _, tool := range []string{"read_state", "update_state", "append_error", "send_response", "recall", "search_index"} { + if !names[tool] { + t.Errorf("missing expert tool %q", tool) + } + } + + // Researcher tools (6) + for _, tool := range []string{"list_experts", "read_expert_state", "read_expert_logs", "enrich_state", "write_expert_state", "promote_pattern"} { + if !names[tool] { + t.Errorf("missing researcher tool %q", tool) + } + } + + if len(names) != 12 { + t.Errorf("expected 12 tools, got %d: %v", len(names), names) + } +} + +func TestResearcherListExperts(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "list_experts", nil) + text := resultText(t, result) + + var experts []struct { + Name string `json:"name"` + Type string `json:"type"` + StateBytes int64 `json:"state_bytes"` + LogCount int `json:"log_count"` + LastTask string `json:"last_task"` + } + if err := json.Unmarshal([]byte(text), &experts); err != nil { + t.Fatalf("unmarshaling: %v\nraw: %s", err, text) + } + + if len(experts) != 2 { + t.Fatalf("expected 2 experts, got %d", len(experts)) + } + + // Find auth + var auth *struct { + Name string `json:"name"` + Type string `json:"type"` + StateBytes int64 `json:"state_bytes"` + LogCount int `json:"log_count"` + LastTask string `json:"last_task"` + } + for i := range experts { + if experts[i].Name == "auth" { + auth = &experts[i] + break + } + } + if auth == nil { + t.Fatal("auth expert not found") + } + if auth.Type != "pool" { + t.Errorf("auth type = %q, want pool", auth.Type) + } + if auth.StateBytes == 0 { + t.Error("auth state_bytes should be > 0") + } + if auth.LogCount != 2 { + t.Errorf("auth log_count = %d, want 2", auth.LogCount) + } + if auth.LastTask != "task-002" { + t.Errorf("auth last_task = %q, want task-002", auth.LastTask) + } +} + +func TestResearcherListExperts_Empty(t *testing.T) { + poolDir := makePoolDirs(t, "researcher/inbox", "postoffice") + poolToml := `[pool] +name = "empty-pool" +project_dir = "` + poolDir + `" +` + os.WriteFile(filepath.Join(poolDir, "pool.toml"), []byte(poolToml), 0o644) + + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + result := callTool(t, srv, "list_experts", nil) + text := resultText(t, result) + + var experts []any + json.Unmarshal([]byte(text), &experts) + if len(experts) != 0 { + t.Errorf("expected empty list, got %d", len(experts)) + } +} + +func TestResearcherReadExpertState_All(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "read_expert_state", map[string]any{ + "expert": "auth", + }) + text := resultText(t, result) + + var state map[string]string + if err := json.Unmarshal([]byte(text), &state); err != nil { + t.Fatalf("unmarshaling: %v", err) + } + + if !strings.Contains(state["identity"], "Auth Expert") { + t.Errorf("identity should contain 'Auth Expert', got: %s", state["identity"]) + } + if !strings.Contains(state["state"], "OAuth tokens") { + t.Errorf("state should contain 'OAuth tokens', got: %s", state["state"]) + } + if !strings.Contains(state["errors"], "Token refresh") { + t.Errorf("errors should contain 'Token refresh', got: %s", state["errors"]) + } +} + +func TestResearcherReadExpertState_SingleFile(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "read_expert_state", map[string]any{ + "expert": "auth", + "file": "identity", + }) + text := resultText(t, result) + + if !strings.Contains(text, "Auth Expert") { + t.Errorf("expected identity content, got: %s", text) + } +} + +func TestResearcherReadExpertState_MissingExpert(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "read_expert_state", map[string]any{}) + if !result.IsError { + t.Error("expected error for missing expert param") + } +} + +func TestResearcherReadExpertLogs(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "read_expert_logs", map[string]any{ + "expert": "auth", + }) + text := resultText(t, result) + + if !strings.Contains(text, "task-001") { + t.Errorf("should contain task-001, got: %s", text) + } + if !strings.Contains(text, "task-002") { + t.Errorf("should contain task-002, got: %s", text) + } +} + +func TestResearcherReadExpertLogs_Query(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "read_expert_logs", map[string]any{ + "expert": "auth", + "query": "OAuth", + }) + text := resultText(t, result) + + if !strings.Contains(text, "task-002") { + t.Errorf("should contain task-002 (OAuth bug), got: %s", text) + } + if strings.Contains(text, "task-001") { + t.Errorf("should NOT contain task-001 (not matching OAuth), got: %s", text) + } +} + +func TestResearcherReadExpertLogs_Empty(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "read_expert_logs", map[string]any{ + "expert": "billing", + }) + text := resultText(t, result) + + if !strings.Contains(text, "no log entries") { + t.Errorf("expected 'no log entries', got: %s", text) + } +} + +func TestResearcherEnrichState(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "enrich_state", map[string]any{ + "expert": "auth", + }) + text := resultText(t, result) + + var enriched map[string]any + if err := json.Unmarshal([]byte(text), &enriched); err != nil { + t.Fatalf("unmarshaling: %v", err) + } + + if id, ok := enriched["identity"].(string); !ok || !strings.Contains(id, "Auth Expert") { + t.Errorf("identity missing or wrong: %v", enriched["identity"]) + } + if s, ok := enriched["state"].(string); !ok || !strings.Contains(s, "OAuth") { + t.Errorf("state missing or wrong: %v", enriched["state"]) + } + + recentIndex, ok := enriched["recent_index"].([]any) + if !ok { + t.Fatalf("recent_index not an array: %T", enriched["recent_index"]) + } + if len(recentIndex) != 2 { + t.Errorf("expected 2 index entries, got %d", len(recentIndex)) + } + + recentLogs, ok := enriched["recent_logs"].([]any) + if !ok { + t.Fatalf("recent_logs not an array: %T", enriched["recent_logs"]) + } + if len(recentLogs) != 2 { + t.Errorf("expected 2 log files, got %d", len(recentLogs)) + } +} + +func TestResearcherWriteExpertState(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "write_expert_state", map[string]any{ + "expert": "auth", + "content": "Curated: OAuth tokens managed via refresh flow.", + }) + text := resultText(t, result) + + if !strings.Contains(text, "state.md updated") { + t.Errorf("expected update confirmation, got: %s", text) + } + + // Verify file was written + data, err := os.ReadFile(filepath.Join(poolDir, "experts", "auth", "state.md")) + if err != nil { + t.Fatalf("reading state.md: %v", err) + } + if !strings.Contains(string(data), "Curated") { + t.Errorf("state.md content wrong: %s", string(data)) + } +} + +func TestResearcherWriteExpertState_Errors(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "write_expert_state", map[string]any{ + "expert": "auth", + "content": "Curated error log.", + "file": "errors", + }) + text := resultText(t, result) + + if !strings.Contains(text, "errors.md updated") { + t.Errorf("expected update confirmation, got: %s", text) + } + + data, err := os.ReadFile(filepath.Join(poolDir, "experts", "auth", "errors.md")) + if err != nil { + t.Fatalf("reading errors.md: %v", err) + } + if !strings.Contains(string(data), "Curated error log") { + t.Errorf("errors.md content wrong: %s", string(data)) + } +} + +func TestResearcherWriteExpertState_EmptyContent(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "write_expert_state", map[string]any{ + "expert": "auth", + "content": "", + }) + if !result.IsError { + t.Error("expected error for empty content") + } +} + +func TestResearcherPromotePattern_ExistingIdentity(t *testing.T) { + poolDir := setupResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "promote_pattern", map[string]any{ + "expert": "auth", + "pattern": "- Always validate token expiry before API calls", + }) + text := resultText(t, result) + + if !strings.Contains(text, "promoted") { + t.Errorf("expected promotion confirmation, got: %s", text) + } + + data, err := os.ReadFile(filepath.Join(poolDir, "experts", "auth", "identity.md")) + if err != nil { + t.Fatalf("reading identity.md: %v", err) + } + content := string(data) + + if !strings.Contains(content, "## Graduated Patterns") { + t.Error("missing Graduated Patterns section") + } + if !strings.Contains(content, "Always validate token expiry") { + t.Error("missing promoted pattern") + } + if !strings.Contains(content, "Auth Expert") { + t.Error("original content should be preserved") + } +} + +func TestResearcherPromotePattern_CreatesSection(t *testing.T) { + poolDir := setupResearcherPool(t) + + // Write identity without the section + authDir := filepath.Join(poolDir, "experts", "auth") + os.WriteFile(filepath.Join(authDir, "identity.md"), []byte("# Auth Expert\n\nHandles auth.\n"), 0o644) + + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + callTool(t, srv, "promote_pattern", map[string]any{ + "expert": "auth", + "pattern": "- Use refresh tokens, not long-lived access tokens", + }) + + data, _ := os.ReadFile(filepath.Join(authDir, "identity.md")) + content := string(data) + + if !strings.Contains(content, "## Graduated Patterns") { + t.Error("section should be created") + } + if !strings.Contains(content, "refresh tokens") { + t.Error("pattern should be present") + } +} + +func TestResearcherPromotePattern_EmptyIdentity(t *testing.T) { + poolDir := setupResearcherPool(t) + + // Remove identity file to test cold-start + authDir := filepath.Join(poolDir, "experts", "auth") + os.Remove(filepath.Join(authDir, "identity.md")) + + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + callTool(t, srv, "promote_pattern", map[string]any{ + "expert": "auth", + "pattern": "- First promoted pattern", + }) + + data, err := os.ReadFile(filepath.Join(authDir, "identity.md")) + if err != nil { + t.Fatalf("identity.md should exist after promotion: %v", err) + } + content := string(data) + + if !strings.Contains(content, "## Graduated Patterns") { + t.Error("section should be created") + } + if !strings.Contains(content, "First promoted pattern") { + t.Error("pattern should be present") + } +} diff --git a/internal/mcp/testhelp_test.go b/internal/mcp/testhelp_test.go index d2a6ee6..6e51561 100644 --- a/internal/mcp/testhelp_test.go +++ b/internal/mcp/testhelp_test.go @@ -56,6 +56,8 @@ func buildMCPTestServer(t *testing.T, poolDir, expertName, role string) *server. agentmcp.RegisterArchitectTools(srv, cfg) case "concierge": agentmcp.RegisterConciergeTools(srv, cfg) + case "researcher": + agentmcp.RegisterResearcherTools(srv, cfg) } initMsg := mustJSON(t, map[string]any{ From 4556f723bc5ace5f351c18bebaabc89717ab2d6f Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 20:27:08 -0500 Subject: [PATCH 3/9] refactor(mail): extract Post() for CLI and daemon reuse Move postMessage logic from internal/mcp to mail.Post() so the CLI (seed command) and daemon (curation scheduler) can post messages without importing the mcp package. mcp.postMessage becomes a thin delegate to mail.Post. Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/mail/post.go | 31 ++++++++++++++++ internal/mail/post_test.go | 76 ++++++++++++++++++++++++++++++++++++++ internal/mcp/postoffice.go | 25 +------------ 3 files changed, 109 insertions(+), 23 deletions(-) create mode 100644 internal/mail/post.go create mode 100644 internal/mail/post_test.go diff --git a/internal/mail/post.go b/internal/mail/post.go new file mode 100644 index 0000000..9e2c067 --- /dev/null +++ b/internal/mail/post.go @@ -0,0 +1,31 @@ +package mail + +import ( + "fmt" + "os" + "path/filepath" + + "github.com/cameronsjo/agent-pool/internal/atomicfile" +) + +// Post composes a mail message and writes it atomically to the pool's postoffice. +// Creates the postoffice directory if it doesn't exist. Uses atomic writes +// to prevent partial files from being picked up by the daemon's watcher. +func Post(poolDir string, msg *Message) error { + composed, err := Compose(msg) + if err != nil { + return fmt.Errorf("composing message: %w", err) + } + + postoffice := filepath.Join(poolDir, "postoffice") + if err := os.MkdirAll(postoffice, 0o755); err != nil { + return fmt.Errorf("creating postoffice dir: %w", err) + } + + path := filepath.Join(postoffice, msg.ID+".md") + if err := atomicfile.WriteFile(path, []byte(composed)); err != nil { + return fmt.Errorf("writing to postoffice: %w", err) + } + + return nil +} diff --git a/internal/mail/post_test.go b/internal/mail/post_test.go new file mode 100644 index 0000000..814b136 --- /dev/null +++ b/internal/mail/post_test.go @@ -0,0 +1,76 @@ +// Test plan for post.go: +// +// Post: +// [x] Happy: message written to postoffice/{id}.md +// [x] Happy: creates postoffice dir if missing +// [x] File is valid parseable mail message + +package mail + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestPost_WritesMessage(t *testing.T) { + poolDir := t.TempDir() + + msg := &Message{ + ID: "task-post-001", + From: "daemon", + To: "researcher", + Type: TypeTask, + Priority: PriorityNormal, + Timestamp: time.Now().UTC(), + Body: "Curate all experts.", + } + + if err := Post(poolDir, msg); err != nil { + t.Fatalf("Post: %v", err) + } + + path := filepath.Join(poolDir, "postoffice", "task-post-001.md") + if _, err := os.Stat(path); os.IsNotExist(err) { + t.Fatal("message file not created") + } + + // Verify it's parseable + parsed, err := ParseFile(path) + if err != nil { + t.Fatalf("ParseFile: %v", err) + } + if parsed.ID != "task-post-001" { + t.Errorf("ID = %q, want task-post-001", parsed.ID) + } + if parsed.To != "researcher" { + t.Errorf("To = %q, want researcher", parsed.To) + } + if parsed.Body != "Curate all experts." { + t.Errorf("Body = %q, want 'Curate all experts.'", parsed.Body) + } +} + +func TestPost_CreatesPostofficeDir(t *testing.T) { + poolDir := t.TempDir() + // Don't pre-create postoffice/ + + msg := &Message{ + ID: "task-mkdir-001", + From: "cli", + To: "researcher", + Type: TypeTask, + Priority: PriorityNormal, + Timestamp: time.Now().UTC(), + Body: "Seed expert state.", + } + + if err := Post(poolDir, msg); err != nil { + t.Fatalf("Post: %v", err) + } + + if _, err := os.Stat(filepath.Join(poolDir, "postoffice")); os.IsNotExist(err) { + t.Error("postoffice dir not created") + } +} diff --git a/internal/mcp/postoffice.go b/internal/mcp/postoffice.go index 2270b2f..e829e91 100644 --- a/internal/mcp/postoffice.go +++ b/internal/mcp/postoffice.go @@ -1,32 +1,11 @@ package mcp import ( - "fmt" - "os" - "path/filepath" - - "github.com/cameronsjo/agent-pool/internal/atomicfile" "github.com/cameronsjo/agent-pool/internal/mail" ) // postMessage composes a mail message and writes it to the pool's postoffice. -// Creates the postoffice directory if it doesn't exist. Uses atomic writes -// to prevent partial files from being picked up by the daemon's watcher. +// Delegates to mail.Post which handles directory creation and atomic writes. func postMessage(poolDir string, msg *mail.Message) error { - composed, err := mail.Compose(msg) - if err != nil { - return fmt.Errorf("composing message: %w", err) - } - - postoffice := filepath.Join(poolDir, "postoffice") - if err := os.MkdirAll(postoffice, 0o755); err != nil { - return fmt.Errorf("creating postoffice dir: %w", err) - } - - path := filepath.Join(postoffice, msg.ID+".md") - if err := atomicfile.WriteFile(path, []byte(composed)); err != nil { - return fmt.Errorf("writing to postoffice: %w", err) - } - - return nil + return mail.Post(poolDir, msg) } From 538a62c68843d296452dc6aec495ff4b18d7303a Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 20:30:41 -0500 Subject: [PATCH 4/9] feat(daemon): add curation scheduler with task and time triggers Triggers researcher curation after interval_tasks completions (default 10) or interval_hours elapsed (default 168h). Generates a structured task message with per-expert metadata (state sizes, log counts) to guide the researcher's curation decisions. The scheduler integrates at two points: - markTaskCompleted: increments counter, triggers on threshold - Run: starts a time-based ticker goroutine Adds EventCurationTriggered event type to the event bus. Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/daemon/curation.go | 167 +++++++++++++++++++++++++++++++ internal/daemon/curation_test.go | 103 +++++++++++++++++++ internal/daemon/daemon.go | 30 ++++++ internal/daemon/events.go | 9 +- 4 files changed, 307 insertions(+), 2 deletions(-) create mode 100644 internal/daemon/curation.go create mode 100644 internal/daemon/curation_test.go diff --git a/internal/daemon/curation.go b/internal/daemon/curation.go new file mode 100644 index 0000000..0e30e5a --- /dev/null +++ b/internal/daemon/curation.go @@ -0,0 +1,167 @@ +package daemon + +import ( + "fmt" + "log/slog" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" + + "github.com/cameronsjo/agent-pool/internal/config" + "github.com/cameronsjo/agent-pool/internal/mail" +) + +// curationScheduler tracks task completions and triggers researcher curation +// when thresholds are reached (task count or time elapsed). +type curationScheduler struct { + intervalTasks int + intervalHours int + poolDir string + logger *slog.Logger + + mu sync.Mutex + taskCount int + lastCuration time.Time +} + +func newCurationScheduler(cfg *config.CurationSection, poolDir string, logger *slog.Logger) *curationScheduler { + return &curationScheduler{ + intervalTasks: cfg.IntervalTasks, + intervalHours: cfg.IntervalHours, + poolDir: poolDir, + logger: logger, + lastCuration: time.Now(), + } +} + +// RecordTaskCompletion increments the completed task counter. Returns true +// when the threshold is reached, signaling that curation should be triggered. +func (cs *curationScheduler) RecordTaskCompletion() bool { + cs.mu.Lock() + defer cs.mu.Unlock() + + cs.taskCount++ + return cs.taskCount >= cs.intervalTasks +} + +// ShouldTriggerByTime returns true if enough time has elapsed since the last +// curation trigger. +func (cs *curationScheduler) ShouldTriggerByTime() bool { + cs.mu.Lock() + defer cs.mu.Unlock() + + return time.Since(cs.lastCuration) >= time.Duration(cs.intervalHours)*time.Hour +} + +// Reset zeroes the task counter and updates the last curation timestamp. +func (cs *curationScheduler) Reset() { + cs.mu.Lock() + defer cs.mu.Unlock() + + cs.taskCount = 0 + cs.lastCuration = time.Now() +} + +// triggerCuration composes a curation task and posts it to the researcher's +// inbox via the postoffice. The task body includes expert names and their +// state sizes to guide the researcher's curation decisions. +func (d *Daemon) triggerCuration(reason string) { + d.logger.Info("Triggering curation", + "reason", reason, + ) + + body := buildCurationTaskBody(d.cfg, d.poolDir, reason) + + msg := &mail.Message{ + ID: fmt.Sprintf("curation-%d", time.Now().UnixMilli()), + From: "daemon", + To: "researcher", + Type: mail.TypeTask, + Priority: mail.PriorityNormal, + Timestamp: time.Now().UTC(), + Body: body, + } + + if err := mail.Post(d.poolDir, msg); err != nil { + d.logger.Error("Failed to post curation task", + "error", err, + ) + return + } + + d.events.emit(Event{ + Type: EventCurationTriggered, + Timestamp: time.Now(), + Data: CurationTriggeredData{Reason: reason}, + }) +} + +// buildCurationTaskBody assembles the structured task body for a curation +// request. Includes per-expert metadata to help the researcher prioritize. +func buildCurationTaskBody(cfg *config.PoolConfig, poolDir, reason string) string { + var b strings.Builder + + b.WriteString("## Curation Task\n\n") + b.WriteString(fmt.Sprintf("**Trigger:** %s\n\n", reason)) + b.WriteString("Review each expert's state and logs. Prune stale information from state.md, ") + b.WriteString("promote recurring patterns to identity.md, and check state sizes.\n\n") + b.WriteString("### Experts to Curate\n\n") + b.WriteString("| Expert | Type | State Size | Log Count |\n") + b.WriteString("|--------|------|------------|----------:|\n") + + // Pool-scoped experts + var names []string + for name := range cfg.Experts { + names = append(names, name) + } + sort.Strings(names) + + for _, name := range names { + dir := mail.ResolveExpertDir(poolDir, name) + stateSize := fileSize(filepath.Join(dir, "state.md")) + logCount := countLogFiles(filepath.Join(dir, "logs")) + b.WriteString(fmt.Sprintf("| %s | pool | %d bytes | %d |\n", name, stateSize, logCount)) + } + + // Shared experts + for _, name := range cfg.Shared.Include { + dir, err := config.SharedExpertDir(name) + if err != nil { + continue + } + stateSize := fileSize(filepath.Join(dir, "state.md")) + logCount := countLogFiles(filepath.Join(dir, "logs")) + b.WriteString(fmt.Sprintf("| %s | shared | %d bytes | %d |\n", name, stateSize, logCount)) + } + + b.WriteString("\nUse `enrich_state` to read each expert's full context, then ") + b.WriteString("`write_expert_state` to write curated state back. Use `promote_pattern` ") + b.WriteString("for patterns that should become permanent identity.\n") + + return b.String() +} + +func fileSize(path string) int64 { + fi, err := os.Stat(path) + if err != nil { + return 0 + } + return fi.Size() +} + +func countLogFiles(logsDir string) int { + entries, err := os.ReadDir(logsDir) + if err != nil { + return 0 + } + count := 0 + for _, e := range entries { + if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { + count++ + } + } + return count +} diff --git a/internal/daemon/curation_test.go b/internal/daemon/curation_test.go new file mode 100644 index 0000000..4eed843 --- /dev/null +++ b/internal/daemon/curation_test.go @@ -0,0 +1,103 @@ +// Test plan for curation.go: +// +// Daemon integration: +// [x] Curation triggered after N task completions — researcher spawned +// [x] Curation task body contains expert metadata + +package daemon_test + +import ( + "context" + "log/slog" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/cameronsjo/agent-pool/internal/config" + "github.com/cameronsjo/agent-pool/internal/daemon" +) + +func TestCurationScheduler_TaskThreshold(t *testing.T) { + poolDir := t.TempDir() + poolToml := `[pool] +name = "curation-test" +project_dir = "` + poolDir + `" + +[curation] +interval_tasks = 3 +interval_hours = 168 + +[experts.auth] +` + os.WriteFile(filepath.Join(poolDir, "pool.toml"), []byte(poolToml), 0o644) + + cfg, err := config.LoadPool(poolDir) + if err != nil { + t.Fatalf("LoadPool: %v", err) + } + + fake := &fakeSpawner{} + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) + d := daemon.New(cfg, poolDir, logger, daemon.WithSpawner(fake)) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- d.Run(ctx) }() + + time.Sleep(500 * time.Millisecond) + + // Send 3 tasks to auth (threshold is 3) + for i := 1; i <= 3; i++ { + id := "task-cur-00" + string(rune('0'+i)) + writeMessage(t, filepath.Join(poolDir, "postoffice"), id, "architect", "auth") + time.Sleep(1 * time.Second) + } + + // Wait for the researcher to be spawned (curation task triggers researcher) + deadline := time.Now().Add(10 * time.Second) + var researcherSpawned bool + for time.Now().Before(deadline) { + for _, c := range fake.getCalls() { + if c.Name == "researcher" && strings.HasPrefix(c.TaskMessage.ID, "curation-") { + researcherSpawned = true + break + } + } + if researcherSpawned { + break + } + time.Sleep(200 * time.Millisecond) + } + + if !researcherSpawned { + calls := fake.getCalls() + var names []string + for _, c := range calls { + names = append(names, c.Name+":"+c.TaskMessage.ID) + } + t.Fatalf("expected researcher spawn with curation task, got spawns: %v", names) + } + + // Verify the curation task body contains expert metadata + for _, c := range fake.getCalls() { + if c.Name == "researcher" && strings.HasPrefix(c.TaskMessage.ID, "curation-") { + body := c.TaskMessage.Body + if !strings.Contains(body, "auth") { + t.Error("curation task body should mention auth expert") + } + if !strings.Contains(body, "task_threshold") { + t.Error("curation task body should contain trigger reason") + } + if !strings.Contains(body, "enrich_state") { + t.Error("curation task body should reference enrich_state tool") + } + break + } + } + + shutdownDaemon(t, cancel, errCh) +} diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 7b9c4bd..bd15142 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -57,6 +57,7 @@ type Daemon struct { sockPathOver string // overrides default socket path (for tests with long TempDir paths) events *eventBus sharedSet map[string]bool // cached shared.include lookup set, built once in New + curation *curationScheduler } // Option configures a Daemon. @@ -112,6 +113,7 @@ func New(cfg *config.PoolConfig, poolDir string, logger *slog.Logger, opts ...Op spawner: defaultSpawner{}, drainTimeout: 30 * time.Second, events: newEventBus(), + curation: newCurationScheduler(&cfg.Curation, poolDir, logger), } // Build cached shared expert lookup set @@ -207,6 +209,27 @@ func (d *Daemon) Run(ctx context.Context) error { d.drainPostoffice(childCtx) d.drainAllInboxes(childCtx) + // Start time-based curation ticker + if d.curation.intervalHours > 0 { + d.wg.Add(1) + go func() { + defer d.wg.Done() + ticker := time.NewTicker(time.Duration(d.curation.intervalHours) * time.Hour) + defer ticker.Stop() + for { + select { + case <-childCtx.Done(): + return + case <-ticker.C: + if d.curation.ShouldTriggerByTime() { + d.curation.Reset() + d.triggerCuration("time_interval") + } + } + } + }() + } + // Main event loop for { select { @@ -848,6 +871,13 @@ func (d *Daemon) markTaskCompleted(ctx context.Context, taskID string, exitCode d.wg.Add(1) go func(e string) { defer d.wg.Done(); d.handleInbox(ctx, e, "") }(expert) } + + // Check curation threshold after task completion + if d.curation.RecordTaskCompletion() { + d.curation.Reset() + d.wg.Add(1) + go func() { defer d.wg.Done(); d.triggerCuration("task_threshold") }() + } } // markTaskFailed updates a task's status to failed, propagates failure to diff --git a/internal/daemon/events.go b/internal/daemon/events.go index a284aed..b69380d 100644 --- a/internal/daemon/events.go +++ b/internal/daemon/events.go @@ -13,8 +13,9 @@ const ( EventExpertSpawning EventType = "expert.spawning" EventExpertCompleted EventType = "expert.completed" EventExpertFailed EventType = "expert.failed" - EventTaskCancelled EventType = "task.cancelled" - EventTaskUnblocked EventType = "task.unblocked" + EventTaskCancelled EventType = "task.cancelled" + EventTaskUnblocked EventType = "task.unblocked" + EventCurationTriggered EventType = "curation.triggered" ) // Event is a structured daemon event emitted at state transitions. @@ -63,6 +64,10 @@ type TaskUnblockedData struct { Expert string `json:"expert"` } +type CurationTriggeredData struct { + Reason string `json:"reason"` +} + // EventBufSize is the subscriber channel buffer capacity. Subscribers that // can't keep up will miss events once the buffer fills (non-blocking emit). const EventBufSize = 64 From cd2a6aeee6d5ac05585baccb76539903b4a34c98 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 20:32:29 -0500 Subject: [PATCH 5/9] feat(cli): add 'seed' command for cold-start expert bootstrapping Composes a seed task for the researcher that includes the target expert name, project directory, and existing identity context. Validates the expert exists in pool config before posting. Usage: agent-pool seed --pool --expert The seed task instructs the researcher to explore the project codebase and create initial state.md for the named expert. Co-Authored-By: Claude Opus 4.6 (1M context) --- cmd/agent-pool/main.go | 95 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index e94c2f1..74a1c53 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -19,6 +19,7 @@ import ( "github.com/cameronsjo/agent-pool/internal/config" "github.com/cameronsjo/agent-pool/internal/daemon" "github.com/cameronsjo/agent-pool/internal/hooks" + "github.com/cameronsjo/agent-pool/internal/mail" agentmcp "github.com/cameronsjo/agent-pool/internal/mcp" ) @@ -43,6 +44,8 @@ func main() { cmdFlush() case "guard": cmdGuard() + case "seed": + cmdSeed() case "version": fmt.Println("agent-pool v0.6.0-dev") case "help", "--help", "-h": @@ -563,6 +566,97 @@ func parseFlagsFromArgs(args []string, names ...string) map[string]string { return result } +func cmdSeed() { + flags := parseFlags(2, "pool", "expert") + + poolDir := flags["pool"] + expertName := flags["expert"] + + if expertName == "" { + fmt.Fprintf(os.Stderr, "usage: agent-pool seed --pool --expert \n") + os.Exit(1) + } + + var err error + if poolDir == "" { + poolDir, err = config.DiscoverPoolDir("") + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + } + + cfg, err := config.LoadPool(poolDir) + if err != nil { + fmt.Fprintf(os.Stderr, "error loading pool config: %v\n", err) + os.Exit(1) + } + + // Validate expert exists in pool config or shared includes + found := false + if _, ok := cfg.Experts[expertName]; ok { + found = true + } + for _, name := range cfg.Shared.Include { + if name == expertName { + found = true + break + } + } + if !found { + fmt.Fprintf(os.Stderr, "error: expert %q not found in pool config (check [experts] or [shared] sections)\n", expertName) + os.Exit(1) + } + + // Read identity if available (gives the researcher context) + var identityContext string + expertDir := mail.ResolveExpertDir(poolDir, expertName) + if data, readErr := os.ReadFile(filepath.Join(expertDir, "identity.md")); readErr == nil { + identityContext = string(data) + } + + // Compose seed task + var body strings.Builder + body.WriteString("## Cold-Start Seed Task\n\n") + body.WriteString(fmt.Sprintf("**Target expert:** %s\n\n", expertName)) + + if cfg.Pool.ProjectDir != "" { + body.WriteString(fmt.Sprintf("**Project directory:** %s\n\n", cfg.Pool.ProjectDir)) + } + + if identityContext != "" { + body.WriteString("### Expert Identity\n\n") + body.WriteString(identityContext) + body.WriteString("\n\n") + } + + body.WriteString("### Instructions\n\n") + body.WriteString("Explore the project codebase and create initial state.md for this expert.\n") + body.WriteString("Focus on:\n") + body.WriteString("- Key files and patterns relevant to this expert's domain\n") + body.WriteString("- Important APIs, endpoints, or interfaces\n") + body.WriteString("- Current status of the domain (what works, what's in progress)\n") + body.WriteString("- Any conventions or gotchas specific to this area\n\n") + body.WriteString("Use `write_expert_state` to save the initial state.\n") + + msg := &mail.Message{ + ID: fmt.Sprintf("seed-%s-%d", expertName, time.Now().UnixMilli()), + From: "cli", + To: "researcher", + Type: mail.TypeTask, + Priority: mail.PriorityNormal, + Timestamp: time.Now().UTC(), + Body: body.String(), + } + + if err := mail.Post(poolDir, msg); err != nil { + fmt.Fprintf(os.Stderr, "error posting seed task: %v\n", err) + os.Exit(1) + } + + fmt.Printf("Seed task posted for expert %q (id: %s)\n", expertName, msg.ID) +} + func printUsage() { fmt.Println(`agent-pool — process supervisor for Claude Code expert sessions @@ -573,6 +667,7 @@ Usage: agent-pool watch [pool-dir] Stream daemon events agent-pool mcp --pool --expert Start expert MCP server (stdio) agent-pool mcp --pool --role Start built-in role MCP server + agent-pool seed --pool --expert Cold-start expert state via researcher agent-pool flush --pool --expert --task Stop hook: verify state agent-pool guard --pool --expert --path PreToolUse hook: ownership guard agent-pool version Print version From 25ac2474e182909b20debd0e019ed5271f08cc91 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 20:35:11 -0500 Subject: [PATCH 6/9] feat(expert): add log rotation with configurable retention Archives log files beyond defaults.log_retention (default 50) into tar.gz bundles during curation triggers. Matching .stderr companion files are included. index.md remains untouched for searchability. Uses archive/tar + compress/gzip from stdlib. The daemon's rotateAllLogs runs for all experts (pool-scoped and shared) before generating the researcher's curation task. Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/config/config.go | 4 + internal/daemon/curation.go | 42 +++++++ internal/expert/rotate.go | 141 +++++++++++++++++++++++ internal/expert/rotate_test.go | 199 +++++++++++++++++++++++++++++++++ 4 files changed, 386 insertions(+) create mode 100644 internal/expert/rotate.go create mode 100644 internal/expert/rotate_test.go diff --git a/internal/config/config.go b/internal/config/config.go index d4f70e4..b786f8c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -52,6 +52,7 @@ type DefaultsSection struct { Model string `toml:"model"` AllowedTools []string `toml:"allowed_tools"` SessionTimeout string `toml:"session_timeout"` + LogRetention int `toml:"log_retention"` } // CurationSection controls the researcher's curation schedule. @@ -215,6 +216,9 @@ func LoadPool(poolDir string) (*PoolConfig, error) { if cfg.Architect.HumanInbox == "" { cfg.Architect.HumanInbox = "stdout" } + if cfg.Defaults.LogRetention == 0 { + cfg.Defaults.LogRetention = 50 + } if cfg.Curation.IntervalTasks == 0 { cfg.Curation.IntervalTasks = 10 } diff --git a/internal/daemon/curation.go b/internal/daemon/curation.go index 0e30e5a..adf0625 100644 --- a/internal/daemon/curation.go +++ b/internal/daemon/curation.go @@ -11,6 +11,7 @@ import ( "time" "github.com/cameronsjo/agent-pool/internal/config" + "github.com/cameronsjo/agent-pool/internal/expert" "github.com/cameronsjo/agent-pool/internal/mail" ) @@ -73,6 +74,9 @@ func (d *Daemon) triggerCuration(reason string) { "reason", reason, ) + // Rotate logs for all experts before curation + d.rotateAllLogs() + body := buildCurationTaskBody(d.cfg, d.poolDir, reason) msg := &mail.Message{ @@ -144,6 +148,44 @@ func buildCurationTaskBody(cfg *config.PoolConfig, poolDir, reason string) strin return b.String() } +// rotateAllLogs runs log rotation for all experts (pool-scoped and shared). +func (d *Daemon) rotateAllLogs() { + retention := d.cfg.Defaults.LogRetention + + for name := range d.cfg.Experts { + dir := mail.ResolveExpertDir(d.poolDir, name) + if archived, err := expert.RotateLogs(dir, retention); err != nil { + d.logger.Warn("Failed to rotate logs", + "expert", name, + "error", err, + ) + } else if archived > 0 { + d.logger.Info("Rotated expert logs", + "expert", name, + "archived", archived, + ) + } + } + + for _, name := range d.cfg.Shared.Include { + dir, err := config.SharedExpertDir(name) + if err != nil { + continue + } + if archived, rotErr := expert.RotateLogs(dir, retention); rotErr != nil { + d.logger.Warn("Failed to rotate shared expert logs", + "expert", name, + "error", rotErr, + ) + } else if archived > 0 { + d.logger.Info("Rotated shared expert logs", + "expert", name, + "archived", archived, + ) + } + } +} + func fileSize(path string) int64 { fi, err := os.Stat(path) if err != nil { diff --git a/internal/expert/rotate.go b/internal/expert/rotate.go new file mode 100644 index 0000000..5b52544 --- /dev/null +++ b/internal/expert/rotate.go @@ -0,0 +1,141 @@ +package expert + +import ( + "archive/tar" + "compress/gzip" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "time" +) + +// DefaultLogRetention is the default number of log files to keep before archiving. +const DefaultLogRetention = 50 + +// RotateLogs archives log files beyond the retention count into a tar.gz bundle. +// Only .json files count toward the threshold. Archived .json and their matching +// .stderr companion files are deleted after successful archival. index.md is +// never modified — it retains all entries for searchability. +// +// Returns the number of files archived, or 0 if below threshold. +func RotateLogs(expertDir string, retention int) (int, error) { + if retention <= 0 { + retention = DefaultLogRetention + } + + logsDir := filepath.Join(expertDir, "logs") + entries, err := os.ReadDir(logsDir) + if err != nil { + if os.IsNotExist(err) { + return 0, nil + } + return 0, fmt.Errorf("reading logs dir: %w", err) + } + + // Collect .json files and sort by mtime (oldest first) + type logFile struct { + name string + mtime time.Time + } + var jsonFiles []logFile + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".json") { + continue + } + info, err := e.Info() + if err != nil { + continue + } + jsonFiles = append(jsonFiles, logFile{name: e.Name(), mtime: info.ModTime()}) + } + + if len(jsonFiles) <= retention { + return 0, nil + } + + // Sort oldest first + sort.Slice(jsonFiles, func(i, j int) bool { + return jsonFiles[i].mtime.Before(jsonFiles[j].mtime) + }) + + // Files to archive: everything before the retention cutoff + toArchive := jsonFiles[:len(jsonFiles)-retention] + + // Create archive + archiveName := fmt.Sprintf("archive-%s.tar.gz", time.Now().UTC().Format("20060102T150405Z")) + archivePath := filepath.Join(logsDir, archiveName) + + f, err := os.Create(archivePath) + if err != nil { + return 0, fmt.Errorf("creating archive: %w", err) + } + defer f.Close() + + gw := gzip.NewWriter(f) + defer gw.Close() + tw := tar.NewWriter(gw) + defer tw.Close() + + var archived int + var toDelete []string + + for _, lf := range toArchive { + jsonPath := filepath.Join(logsDir, lf.name) + if err := addToTar(tw, jsonPath, lf.name); err != nil { + return archived, fmt.Errorf("adding %s to archive: %w", lf.name, err) + } + toDelete = append(toDelete, jsonPath) + archived++ + + // Also archive matching .stderr if it exists + stderrName := strings.TrimSuffix(lf.name, ".json") + ".stderr" + stderrPath := filepath.Join(logsDir, stderrName) + if _, err := os.Stat(stderrPath); err == nil { + if err := addToTar(tw, stderrPath, stderrName); err != nil { + return archived, fmt.Errorf("adding %s to archive: %w", stderrName, err) + } + toDelete = append(toDelete, stderrPath) + } + } + + // Close writers before deleting source files + tw.Close() + gw.Close() + f.Close() + + // Delete archived files + for _, path := range toDelete { + os.Remove(path) // best-effort; files are safely in the archive + } + + return archived, nil +} + +// addToTar writes a single file to a tar archive. +func addToTar(tw *tar.Writer, path, name string) error { + data, err := os.ReadFile(path) + if err != nil { + return err + } + + info, err := os.Stat(path) + if err != nil { + return err + } + + header := &tar.Header{ + Name: name, + Size: info.Size(), + Mode: 0o644, + ModTime: info.ModTime(), + } + + if err := tw.WriteHeader(header); err != nil { + return err + } + + _, err = tw.Write(data) + return err +} diff --git a/internal/expert/rotate_test.go b/internal/expert/rotate_test.go new file mode 100644 index 0000000..d0e18e9 --- /dev/null +++ b/internal/expert/rotate_test.go @@ -0,0 +1,199 @@ +// Test plan for rotate.go: +// +// RotateLogs: +// [x] BelowThreshold: 5 files, retention=10, returns 0 +// [x] AboveThreshold: 15 files, retention=5, archives 10, keeps 5 +// [x] ArchiveIsValidTarGz: decompress and verify file list +// [x] StderrIncluded: .stderr companion files archived with .json +// [x] EmptyDir: returns 0, no error +// [x] IndexUntouched: index.md not modified after rotation + +package expert + +import ( + "archive/tar" + "compress/gzip" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestRotateLogs_BelowThreshold(t *testing.T) { + dir := t.TempDir() + logsDir := filepath.Join(dir, "logs") + os.MkdirAll(logsDir, 0o755) + + for i := 0; i < 5; i++ { + os.WriteFile(filepath.Join(logsDir, fmt.Sprintf("task-%03d.json", i)), []byte("{}"), 0o644) + } + + archived, err := RotateLogs(dir, 10) + if err != nil { + t.Fatalf("RotateLogs: %v", err) + } + if archived != 0 { + t.Errorf("archived = %d, want 0 (below threshold)", archived) + } +} + +func TestRotateLogs_AboveThreshold(t *testing.T) { + dir := t.TempDir() + logsDir := filepath.Join(dir, "logs") + os.MkdirAll(logsDir, 0o755) + + for i := 0; i < 15; i++ { + os.WriteFile(filepath.Join(logsDir, fmt.Sprintf("task-%03d.json", i)), []byte("{}"), 0o644) + } + + archived, err := RotateLogs(dir, 5) + if err != nil { + t.Fatalf("RotateLogs: %v", err) + } + if archived != 10 { + t.Errorf("archived = %d, want 10", archived) + } + + // Count remaining .json files + entries, _ := os.ReadDir(logsDir) + jsonCount := 0 + for _, e := range entries { + if strings.HasSuffix(e.Name(), ".json") { + jsonCount++ + } + } + if jsonCount != 5 { + t.Errorf("remaining .json files = %d, want 5", jsonCount) + } +} + +func TestRotateLogs_ArchiveIsValidTarGz(t *testing.T) { + dir := t.TempDir() + logsDir := filepath.Join(dir, "logs") + os.MkdirAll(logsDir, 0o755) + + for i := 0; i < 8; i++ { + os.WriteFile(filepath.Join(logsDir, fmt.Sprintf("task-%03d.json", i)), []byte(fmt.Sprintf(`{"id":%d}`, i)), 0o644) + } + + archived, err := RotateLogs(dir, 3) + if err != nil { + t.Fatalf("RotateLogs: %v", err) + } + if archived != 5 { + t.Errorf("archived = %d, want 5", archived) + } + + // Find and open the archive + entries, _ := os.ReadDir(logsDir) + var archivePath string + for _, e := range entries { + if strings.HasPrefix(e.Name(), "archive-") && strings.HasSuffix(e.Name(), ".tar.gz") { + archivePath = filepath.Join(logsDir, e.Name()) + break + } + } + if archivePath == "" { + t.Fatal("no archive file found") + } + + // Verify it's valid tar.gz + f, err := os.Open(archivePath) + if err != nil { + t.Fatalf("opening archive: %v", err) + } + defer f.Close() + + gr, err := gzip.NewReader(f) + if err != nil { + t.Fatalf("gzip reader: %v", err) + } + defer gr.Close() + + tr := tar.NewReader(gr) + var archivedNames []string + for { + hdr, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("tar next: %v", err) + } + archivedNames = append(archivedNames, hdr.Name) + } + + if len(archivedNames) != 5 { + t.Errorf("archive contains %d files, want 5: %v", len(archivedNames), archivedNames) + } +} + +func TestRotateLogs_StderrIncluded(t *testing.T) { + dir := t.TempDir() + logsDir := filepath.Join(dir, "logs") + os.MkdirAll(logsDir, 0o755) + + // Create 6 log files, 3 with stderr companions + for i := 0; i < 6; i++ { + os.WriteFile(filepath.Join(logsDir, fmt.Sprintf("task-%03d.json", i)), []byte("{}"), 0o644) + if i < 3 { + os.WriteFile(filepath.Join(logsDir, fmt.Sprintf("task-%03d.stderr", i)), []byte("err"), 0o644) + } + } + + archived, err := RotateLogs(dir, 3) + if err != nil { + t.Fatalf("RotateLogs: %v", err) + } + if archived != 3 { + t.Errorf("archived = %d, want 3", archived) + } + + // Verify .stderr files for archived tasks are gone + for i := 0; i < 3; i++ { + stderrPath := filepath.Join(logsDir, fmt.Sprintf("task-%03d.stderr", i)) + if _, err := os.Stat(stderrPath); !os.IsNotExist(err) { + t.Errorf("stderr file should be deleted: %s", stderrPath) + } + } +} + +func TestRotateLogs_EmptyDir(t *testing.T) { + dir := t.TempDir() + // No logs dir at all + + archived, err := RotateLogs(dir, 10) + if err != nil { + t.Fatalf("RotateLogs: %v", err) + } + if archived != 0 { + t.Errorf("archived = %d, want 0", archived) + } +} + +func TestRotateLogs_IndexUntouched(t *testing.T) { + dir := t.TempDir() + logsDir := filepath.Join(dir, "logs") + os.MkdirAll(logsDir, 0o755) + + // Write index with entries + indexContent := indexHeader + "| task-000 | 2026-04-01T00:00:00Z | architect | 0 | First task |\n" + os.WriteFile(filepath.Join(logsDir, "index.md"), []byte(indexContent), 0o644) + + for i := 0; i < 8; i++ { + os.WriteFile(filepath.Join(logsDir, fmt.Sprintf("task-%03d.json", i)), []byte("{}"), 0o644) + } + + RotateLogs(dir, 3) + + // Verify index.md is unchanged + data, err := os.ReadFile(filepath.Join(logsDir, "index.md")) + if err != nil { + t.Fatalf("reading index.md: %v", err) + } + if string(data) != indexContent { + t.Errorf("index.md was modified:\n%s", string(data)) + } +} From e95e8f80d2eed2145297a8972a3b27e481fa728f Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 20:38:46 -0500 Subject: [PATCH 7/9] feat(researcher): enable shared expert enrichment with layered state Researcher tools now detect shared experts via pool config and handle both user-level and project overlay state: - read_expert_state: returns project_state field for shared experts - enrich_state: includes project overlay in assembled context - write_expert_state: layer param ("user"/"project") for shared experts - promote_pattern: targets user-level identity.md (cross-pool knowledge) - list_experts: shared experts reported with type "shared" Detection uses config.LoadPool to check shared.include, then config.SharedExpertDir for user-level paths and poolDir/shared-state for project overlay paths. Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/mcp/researcher_tools.go | 67 ++++++++++- internal/mcp/researcher_tools_test.go | 166 ++++++++++++++++++++++++++ 2 files changed, 230 insertions(+), 3 deletions(-) diff --git a/internal/mcp/researcher_tools.go b/internal/mcp/researcher_tools.go index cee1b39..bcd4ea8 100644 --- a/internal/mcp/researcher_tools.go +++ b/internal/mcp/researcher_tools.go @@ -67,10 +67,11 @@ func RegisterResearcherTools(srv *server.MCPServer, cfg *ServerConfig) { srv.AddTool( mcp.NewTool("write_expert_state", - mcp.WithDescription("Write curated state back to an expert. Targets state.md by default. Content must be non-empty and under 50KB."), + mcp.WithDescription("Write curated state back to an expert. Targets state.md by default. Content must be non-empty and under 50KB. For shared experts, use layer to target user-level or project-level state."), mcp.WithString("expert", mcp.Required(), mcp.Description("Expert name to write state to")), mcp.WithString("content", mcp.Required(), mcp.Description("New file content")), mcp.WithString("file", mcp.Description("Target file: 'state' (default) or 'errors'")), + mcp.WithString("layer", mcp.Description("For shared experts: 'user' (default) or 'project'. Ignored for pool-scoped experts.")), ), handleWriteExpertState(cfg), ) @@ -212,6 +213,14 @@ func handleReadExpertState(cfg *ServerConfig) server.ToolHandlerFunc { "state": state, "errors": errors, } + // For shared experts, include the project overlay state + overlayDir := resolveSharedOverlayDir(cfg.PoolDir, expertName) + if overlayDir != "" { + overlayState := readOverlayState(overlayDir) + if overlayState != "" { + result["project_state"] = overlayState + } + } data, err := json.MarshalIndent(result, "", " ") if err != nil { return mcp.NewToolResultError(fmt.Sprintf("marshaling: %v", err)), nil @@ -352,6 +361,15 @@ func handleEnrichState(cfg *ServerConfig) server.ToolHandlerFunc { "recent_logs": recentLogs, } + // For shared experts, include project overlay state + overlayDir := resolveSharedOverlayDir(cfg.PoolDir, expertName) + if overlayDir != "" { + overlayState := readOverlayState(overlayDir) + if overlayState != "" { + result["project_state"] = overlayState + } + } + data, err := json.MarshalIndent(result, "", " ") if err != nil { return mcp.NewToolResultError(fmt.Sprintf("marshaling: %v", err)), nil @@ -373,15 +391,28 @@ func handleWriteExpertState(cfg *ServerConfig) server.ToolHandlerFunc { return mcp.NewToolResultError("content parameter is required"), nil } - dir := resolveTargetExpertDir(cfg.PoolDir, expertName) file := request.GetString("file", "state") + layer := request.GetString("layer", "user") + + // Resolve write directory: shared experts route by layer + dir := resolveTargetExpertDir(cfg.PoolDir, expertName) + overlayDir := resolveSharedOverlayDir(cfg.PoolDir, expertName) + if overlayDir != "" && layer == "project" && file == "state" { + dir = overlayDir + // Ensure overlay dir exists + os.MkdirAll(dir, 0o755) + } switch file { case "state", "": if err := expert.WriteState(dir, content); err != nil { return mcp.NewToolResultError(fmt.Sprintf("writing state.md: %v", err)), nil } - return mcp.NewToolResultText(fmt.Sprintf("%s/state.md updated (%d bytes)", expertName, len(content))), nil + layerLabel := "" + if overlayDir != "" { + layerLabel = fmt.Sprintf(" (layer: %s)", layer) + } + return mcp.NewToolResultText(fmt.Sprintf("%s/state.md updated (%d bytes)%s", expertName, len(content), layerLabel)), nil case "errors": // Overwrite errors.md entirely (not append — researcher is curating) @@ -455,10 +486,40 @@ func handlePromotePattern(cfg *ServerConfig) server.ToolHandlerFunc { // resolveTargetExpertDir returns the state directory for a target expert. // Built-in roles use {poolDir}/{role}/, pool-scoped use {poolDir}/experts/{name}/. +// For shared experts, returns the user-level directory. func resolveTargetExpertDir(poolDir, name string) string { + if isSharedExpert(poolDir, name) { + dir, err := config.SharedExpertDir(name) + if err == nil { + return dir + } + } return mail.ResolveExpertDir(poolDir, name) } +// resolveSharedOverlayDir returns the pool-scoped overlay directory for a shared expert. +// Returns empty string if the expert is not shared. +func resolveSharedOverlayDir(poolDir, name string) string { + if !isSharedExpert(poolDir, name) { + return "" + } + return filepath.Join(poolDir, "shared-state", name) +} + +// isSharedExpert checks whether an expert is in the pool's shared.include list. +func isSharedExpert(poolDir, name string) bool { + poolCfg, err := config.LoadPool(poolDir) + if err != nil { + return false + } + for _, n := range poolCfg.Shared.Include { + if n == name { + return true + } + } + return false +} + // readFileOr reads a file and returns its trimmed content, or the fallback if not found. func readFileOr(path, fallback string) (string, error) { data, err := os.ReadFile(path) diff --git a/internal/mcp/researcher_tools_test.go b/internal/mcp/researcher_tools_test.go index 487ab6c..384039b 100644 --- a/internal/mcp/researcher_tools_test.go +++ b/internal/mcp/researcher_tools_test.go @@ -36,6 +36,7 @@ import ( "encoding/json" "os" "path/filepath" + "sort" "strings" "testing" @@ -461,3 +462,168 @@ func TestResearcherPromotePattern_EmptyIdentity(t *testing.T) { t.Error("pattern should be present") } } + +// --- Shared expert tests --- +// +// Shared Expert Enrichment: +// [x] ReadExpertState_SharedExpert: returns both user and project state +// [x] WriteExpertState_SharedExpert_ProjectLayer: writes to pool overlay +// [x] PromotePattern_SharedExpert: writes to user-level identity.md +// [x] ListExperts_IncludesShared: shared experts in list with correct type + +func setupSharedResearcherPool(t *testing.T) (poolDir string, fakeHome string) { + t.Helper() + + // Fake HOME so config.SharedExpertDir resolves to a temp dir + fakeHome = t.TempDir() + origHome := os.Getenv("HOME") + os.Setenv("HOME", fakeHome) + t.Cleanup(func() { os.Setenv("HOME", origHome) }) + + // Create user-level shared expert directory + sharedDir := filepath.Join(fakeHome, ".agent-pool", "experts", "security-standards") + os.MkdirAll(filepath.Join(sharedDir, "logs"), 0o755) + os.WriteFile(filepath.Join(sharedDir, "identity.md"), []byte("# Security Standards\n\nCross-project security expert.\n"), 0o644) + os.WriteFile(filepath.Join(sharedDir, "state.md"), []byte("User-level security knowledge.\n"), 0o644) + + // Create pool with shared include + poolDir = makePoolDirs(t, + "researcher/inbox", + "researcher/logs", + "postoffice", + "shared-state/security-standards", + "experts/auth/inbox", + "experts/auth/logs", + ) + + // Write project overlay state + overlayDir := filepath.Join(poolDir, "shared-state", "security-standards") + os.WriteFile(filepath.Join(overlayDir, "state.md"), []byte("Project-specific security rules.\n"), 0o644) + + poolToml := `[pool] +name = "test-shared-pool" +project_dir = "` + poolDir + `" + +[shared] +include = ["security-standards"] + +[experts.auth] +` + os.WriteFile(filepath.Join(poolDir, "pool.toml"), []byte(poolToml), 0o644) + + return poolDir, fakeHome +} + +func TestResearcherReadExpertState_SharedExpert(t *testing.T) { + poolDir, _ := setupSharedResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "read_expert_state", map[string]any{ + "expert": "security-standards", + }) + text := resultText(t, result) + + var state map[string]string + if err := json.Unmarshal([]byte(text), &state); err != nil { + t.Fatalf("unmarshaling: %v\nraw: %s", err, text) + } + + if !strings.Contains(state["identity"], "Security Standards") { + t.Errorf("identity should contain user-level content, got: %s", state["identity"]) + } + if !strings.Contains(state["state"], "User-level security") { + t.Errorf("state should contain user-level content, got: %s", state["state"]) + } + if !strings.Contains(state["project_state"], "Project-specific") { + t.Errorf("project_state should contain overlay content, got: %s", state["project_state"]) + } +} + +func TestResearcherWriteExpertState_SharedExpert_ProjectLayer(t *testing.T) { + poolDir, _ := setupSharedResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "write_expert_state", map[string]any{ + "expert": "security-standards", + "content": "Curated project security rules.", + "layer": "project", + }) + text := resultText(t, result) + + if !strings.Contains(text, "project") { + t.Errorf("expected project layer confirmation, got: %s", text) + } + + // Verify written to overlay dir, not user dir + overlayPath := filepath.Join(poolDir, "shared-state", "security-standards", "state.md") + data, err := os.ReadFile(overlayPath) + if err != nil { + t.Fatalf("reading overlay state.md: %v", err) + } + if !strings.Contains(string(data), "Curated project") { + t.Errorf("overlay state.md = %q, want curated content", string(data)) + } +} + +func TestResearcherPromotePattern_SharedExpert(t *testing.T) { + poolDir, fakeHome := setupSharedResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "promote_pattern", map[string]any{ + "expert": "security-standards", + "pattern": "- Always validate CORS headers", + }) + text := resultText(t, result) + + if !strings.Contains(text, "promoted") { + t.Errorf("expected promotion confirmation, got: %s", text) + } + + // Verify written to user-level identity.md (not pool overlay) + userIdentity := filepath.Join(fakeHome, ".agent-pool", "experts", "security-standards", "identity.md") + data, err := os.ReadFile(userIdentity) + if err != nil { + t.Fatalf("reading user-level identity.md: %v", err) + } + content := string(data) + + if !strings.Contains(content, "CORS headers") { + t.Error("pattern should be promoted to user-level identity.md") + } + if !strings.Contains(content, "Security Standards") { + t.Error("original identity content should be preserved") + } +} + +func TestResearcherListExperts_IncludesShared(t *testing.T) { + poolDir, _ := setupSharedResearcherPool(t) + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + result := callTool(t, srv, "list_experts", nil) + text := resultText(t, result) + + var experts []struct { + Name string `json:"name"` + Type string `json:"type"` + } + if err := json.Unmarshal([]byte(text), &experts); err != nil { + t.Fatalf("unmarshaling: %v", err) + } + + // Sort for deterministic checking + sort.Slice(experts, func(i, j int) bool { return experts[i].Name < experts[j].Name }) + + if len(experts) != 2 { + t.Fatalf("expected 2 experts, got %d: %+v", len(experts), experts) + } + + // auth should be pool type + if experts[0].Name != "auth" || experts[0].Type != "pool" { + t.Errorf("expected auth/pool, got %s/%s", experts[0].Name, experts[0].Type) + } + + // security-standards should be shared type + if experts[1].Name != "security-standards" || experts[1].Type != "shared" { + t.Errorf("expected security-standards/shared, got %s/%s", experts[1].Name, experts[1].Type) + } +} From 3bc5973058c4bebcef36955694be060511b5faa2 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 20:48:46 -0500 Subject: [PATCH 8/9] docs: add v0.8 and v0.9 development prompts v0.8 documents the researcher + curation implementation. v0.9 covers formulas (TOML workflow templates), config hot-reload, and partial-write hardening. Co-Authored-By: Claude Opus 4.6 (1M context) --- docs/prompts/v08-researcher-curation.md | 62 ++++++ docs/prompts/v09-formulas-polish.md | 258 ++++++++++++++++++++++++ 2 files changed, 320 insertions(+) create mode 100644 docs/prompts/v08-researcher-curation.md create mode 100644 docs/prompts/v09-formulas-polish.md diff --git a/docs/prompts/v08-researcher-curation.md b/docs/prompts/v08-researcher-curation.md new file mode 100644 index 0000000..9cf2229 --- /dev/null +++ b/docs/prompts/v08-researcher-curation.md @@ -0,0 +1,62 @@ +# v0.8 — Researcher + Curation + +## Where We Are + +v0.7 is merged on main. Shared experts work across pools: user-level +identity in `~/.agent-pool/experts/`, per-pool project overlays in +`shared-state/`, layered prompt assembly, scope-aware `update_state`. +Config parsing, mail routing, and directory creation for the researcher +role were scaffolded in prior versions but nothing watched the inbox +or provided tools. + +## What v0.8 Adds + +Per docs/plans/architecture.md § Implementation Phasing: + +▎ Knowledge enrichment and hygiene via a dedicated researcher role. + +Scope: +- Researcher role wired into daemon lifecycle (inbox watch, spawn, config) +- Six researcher MCP tools for cross-expert state management +- Curation scheduling (task count + time interval triggers) +- Cold-start seeding via `agent-pool seed` CLI command +- Log rotation (tar.gz archival beyond retention threshold) +- Cross-pool shared expert enrichment with layered state awareness + +Validates: Does curation keep state.md lean over time? Does cold-start +seeding produce useful initial state? + +## Key Design Context + +### The Curation Model + +The researcher is the pool's knowledge curator. It reads expert state +and logs, reasons about what to keep/prune/promote, and writes curated +results back. The two-step pattern (enrich_state reads everything, +write_expert_state writes back) keeps the LLM in control — tools +provide the data plane, not the decision logic. + +### Researcher Tools + +| Tool | R/W | Purpose | +|------|-----|---------| +| list_experts | R | Triage: state sizes, log counts, last task | +| read_expert_state | R | Read another expert's identity/state/errors | +| read_expert_logs | R | Recent log index entries with query filter | +| enrich_state | R | Full context assembly for curation analysis | +| write_expert_state | W | Write curated state back to any expert | +| promote_pattern | W | Graduate patterns from state to identity | + +### Curation Scheduling + +The daemon tracks task completions. After `interval_tasks` (default 10) +or `interval_hours` (default 168h), it generates a structured curation +task describing which experts to curate, their state sizes, and +instructions. Log rotation runs before each curation trigger. + +### Pattern Promotion + +`promote_pattern` moves knowledge from state.md (working memory) to +identity.md (permanent knowledge). This is the key semantic transition: +patterns that recur across tasks graduate from ephemeral state to +durable identity. The researcher decides what crosses this boundary. diff --git a/docs/prompts/v09-formulas-polish.md b/docs/prompts/v09-formulas-polish.md new file mode 100644 index 0000000..1edfafd --- /dev/null +++ b/docs/prompts/v09-formulas-polish.md @@ -0,0 +1,258 @@ +# v0.9 — Formulas + Polish + +## Where We Are + +v0.8 is merged on main. The researcher role is fully operational: daemon +watches researcher inbox, spawns sessions with role-specific MCP tools, +and a curation scheduler auto-triggers after configurable task or time +thresholds. Six researcher tools (list_experts, read_expert_state, +read_expert_logs, enrich_state, write_expert_state, promote_pattern) +enable cross-expert state curation with shared-expert-aware layered state. +Cold-start seeding via `agent-pool seed` bootstraps expert state through +the researcher. Log rotation archives old logs into tar.gz bundles. + +All four roles are now wired: concierge (v0.5), architect (v0.4), +expert (v0.2), researcher (v0.8). The pool is a complete system. + +## What v0.9 Adds + +Per docs/plans/architecture.md § Implementation Phasing: + +▎ Workflow templates and operational hardening. + +Scope: +- TOML formula parsing (`{poolDir}/formulas/*.toml`) +- Formula instantiation by architect (new MCP tool) +- Config hot-reload (watch pool.toml for changes, apply without restart) +- Partial-write detection hardening on mail files + +Validates: Can common workflows be templated and reused across pools? + +## Key Design Context + +### Workflow Formulas + +Today the architect manually decomposes work: define contracts, send +tasks with `depends_on` edges, verify results. This works but is +repetitive for recurring patterns (feature implementation, bug triage, +code review). Formulas codify these patterns as reusable TOML templates. + +The critical design choice: **the daemon evaluates dependencies and +dispatches — no LLM needed for sequencing.** Formulas are deterministic +DAGs. The architect provides the creative decomposition (what to ask +each expert); the formula provides the structure (which roles in what +order). The architect _instantiates_ a formula, filling in the +task-specific details for each step. + +```toml +# formulas/feature-impl.toml +description = "Standard feature implementation flow" + +[[steps]] +id = "gather" +role = "concierge" +title = "Gather expert input" +description = "Ask targeted questions to relevant experts" + +[[steps]] +id = "plan" +role = "concierge" +title = "Build plan" +description = "Synthesize expert input into plan/spec" +depends_on = ["gather"] + +[[steps]] +id = "review" +role = "architect" +title = "Review plan + define contracts" +description = "Review plan, identify boundaries, define contracts" +depends_on = ["plan"] + +[[steps]] +id = "implement" +role = "experts" +title = "Implementation" +description = "Experts execute in parallel, building to contracts" +depends_on = ["review"] + +[[steps]] +id = "verify" +role = "architect" +title = "Verification" +description = "Verify each expert's output against contracts" +depends_on = ["implement"] +``` + +When the architect calls `instantiate_formula`, the daemon: +1. Parses the formula TOML +2. Creates a task for each step, with the formula's `depends_on` edges +3. Fills in architect-provided overrides (specific expert names, task bodies) +4. Registers all tasks in the taskboard +5. Dispatches the first ready step (no dependencies) + +The taskboard already handles dependency evaluation (`EvaluateDeps`). +Formulas just bulk-register tasks with pre-defined dependency graphs. + +### Formula Structure + +``` +{poolDir}/formulas/ +├── feature-impl.toml # Standard feature flow +├── bug-triage.toml # Bug investigation + fix +├── code-review.toml # Review + feedback cycle +└── index.md # Auto-generated summary +``` + +Each formula is a standalone TOML file. The `[[steps]]` array defines +the DAG. Each step has: +- `id` — unique within the formula (used in `depends_on`) +- `role` — "concierge", "architect", or a specific expert name +- `title` — short description (becomes task summary) +- `description` — detailed instructions for the role +- `depends_on` — list of step IDs that must complete first + +### Instantiation + +The architect calls `instantiate_formula` with: +- `formula` — formula filename (without .toml) +- `prefix` — ID prefix for generated tasks (e.g., "feat-auth" → "feat-auth-gather") +- `overrides` — JSON map of step ID → custom body text +- `experts` — JSON map of step ID → specific expert name (for steps with `role = "experts"`) + +This produces N tasks in the postoffice, all with correct dependency edges. + +### Config Hot-Reload + +Today, changing `pool.toml` requires restarting the daemon. v0.9 adds +an fsnotify watcher on `pool.toml` that: +1. Detects writes to pool.toml +2. Re-parses with `config.LoadPool` +3. Validates the new config +4. Updates `d.cfg` under lock +5. Adjusts watchers if experts were added/removed +6. Logs what changed + +This is useful for adding experts to a running pool without downtime. +The daemon already watches directories — extending to watch a single +file is straightforward. The main risk is partial-write detection: +pool.toml could be mid-write when fsnotify fires. + +### Partial-Write Detection + +The watcher already has `waitForStable` for mail files (polls until +file size stops changing). Two enhancements: +1. Apply `waitForStable` to pool.toml reloads +2. Validate TOML parse succeeds before accepting the new config +3. If parse fails, log a warning and keep the old config (don't crash) + +For mail files, the `.routing-*` temp file pattern plus atomic rename +already prevents partial reads. The enhancement here is defensive — +verify that `atomicfile.WriteFile` is used consistently for all +daemon-consumed files, and add a TOML validation step to the reload. + +## What Already Exists + +**Taskboard (internal/taskboard/):** +- `Board.Add(task)` registers tasks with `DependsOn` edges +- `Board.EvaluateDeps()` returns newly-ready task IDs +- `Board.Save(path)` persists to JSON +- Tasks have `Status`, `Expert`, `DependsOn`, `ID` fields +- The daemon calls `registerTask()` for each postoffice message + +**Config (internal/config/config.go):** +- `LoadPool(poolDir)` parses and validates pool.toml +- `PoolConfig` has all role sections, curation section, defaults +- No formula-related fields yet + +**Watcher (internal/daemon/watcher.go):** +- `Watcher.Add(dir)` watches a directory for Create events on .md files +- `waitForStable(path)` polls for file size stability +- Filters `.routing-*` temp files +- Currently only watches directories, not individual files + +**Architect tools (internal/mcp/architect_tools.go):** +- `send_task` dispatches a single task with contracts + depends_on +- `define_contract` creates a versioned interface spec +- No formula-related tools yet + +**Mail (internal/mail/):** +- `Post(poolDir, msg)` composes + atomic write to postoffice +- Messages have `DependsOn []string` field +- Router handles delivery to inboxes + +**Gap: No formula parsing.** No TOML formula struct, no loader. + +**Gap: No formula instantiation.** Architect can't bulk-create tasks. + +**Gap: No file watcher for pool.toml.** Watcher only handles directories. + +**Gap: No config reload path.** `d.cfg` is set once in `New()`. + +## What to Read First + +1. docs/plans/architecture.md — §Workflow Formulas, §Resolved Decisions +2. internal/taskboard/ — Board, Task struct, Add, EvaluateDeps +3. internal/daemon/daemon.go — registerTask, resolveExpertConfig, Run +4. internal/daemon/watcher.go — Watcher, waitForStable, Run +5. internal/mcp/architect_tools.go — handleSendTask (single-task dispatch pattern) +6. internal/config/config.go — LoadPool, PoolConfig struct + +## Approach Suggestion + +**Phase 1: Formula Parsing** +New package `internal/formula/` with `Formula` and `Step` structs. +`Load(path)` parses a single TOML file. `LoadAll(formulasDir)` scans +the directory. `Validate(formula)` checks for DAG cycles, missing +step IDs in `depends_on`, and duplicate IDs. Add `formula_test.go` +with cycle detection, happy path, and malformed input tests. + +**Phase 2: Formula Instantiation** +New architect tool `instantiate_formula`. Takes formula name, prefix, +overrides, and expert assignments. Generates N `mail.Message`s with +correct `DependsOn` edges (prefixed IDs). Posts all to postoffice. +The daemon's existing taskboard + dependency evaluation handles the rest. +This is the highest-value phase — it turns the formula into running tasks. + +**Phase 3: Config Hot-Reload** +Add `pool.toml` to the watcher (may need a file-level watch, not just +directory). On change: waitForStable, re-parse, validate, swap under +lock. Diff old vs new config to log changes. Adjust expert inbox +watchers if experts were added or removed. Add a new event type +`config.reloaded`. Test: modify pool.toml while daemon runs, verify +new expert becomes spawnable. + +**Phase 4: Hardening** +- Ensure all daemon-consumed file writes use `atomicfile.WriteFile` +- Add TOML validation step to config reload (parse failure = keep old) +- Formula directory auto-created in `ensureDirs` +- Formula index auto-generated (like contracts/index.md) + +## Design Questions to Resolve + +1. **Formula location:** `{poolDir}/formulas/` (pool-scoped) vs + `~/.agent-pool/formulas/` (user-scoped, like shared experts)? + Recommendation: pool-scoped. Formulas reference pool-specific roles + and experts. Sharing between pools is a copy-paste concern, not a + runtime concern. + +2. **Role = "experts" expansion:** When a step has `role = "experts"`, + does the architect provide a specific expert name at instantiation, + or does the daemon resolve it? Recommendation: architect provides + via the `experts` map. The daemon doesn't know which expert is right + for a given task — that's the architect's judgment. + +3. **Formula versioning:** Should formulas track version numbers like + contracts? Recommendation: no. Formulas are templates, not agreements + between parties. Git versioning is sufficient. + +4. **Watcher type for pool.toml:** fsnotify.Write event on a specific + file vs watching the parent directory + filtering for pool.toml? + Recommendation: watch the pool directory (already watched for + postoffice) and filter for `pool.toml` by filename. Avoids needing + a separate watcher for a single file. + +5. **Expert add/remove during runtime:** When config reload detects a + new expert, should the daemon auto-create inbox dirs and start + watching? Recommendation: yes — call `ensureDirs` again and add + the new inbox to the watcher. For removals, stop watching but don't + delete directories (data preservation). From 8b31437823a8bfa728ecedad807d3fc887096d97 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Mon, 6 Apr 2026 21:19:50 -0500 Subject: [PATCH 9/9] fix: address review findings (round 1) Critical: - Check tar/gzip/file Close errors before deleting source logs Major: - Atomic check-and-reset in curation scheduler prevents double-fire - Guard intervalTasks <= 0 (disabled, not "always fire") - Shared expert log rotation now targets pool overlay, not user dir - Shared expert list_experts/read_expert_logs/enrich_state now read logs from pool overlay (shared-state//logs) Minor: - Normalize non-positive log_retention at config load time - waitForTaskStatus before shutdown in researcher/curation tests - Handle non-ENOENT os.Stat errors in post_test assertions - Check RotateLogs error in TestRotateLogs_IndexUntouched - Fix "builtin" typo in plan doc - Add language tag to fenced code block in v0.9 prompt Nitpick: - Guard empty poolDir in mail.Post - Add shared overlay log integration test Co-Authored-By: Claude Opus 4.6 (1M context) --- docs/plans/2026-04-06-researcher-curation.md | 2 +- docs/prompts/v09-formulas-polish.md | 2 +- internal/config/config.go | 2 +- internal/daemon/curation.go | 50 +++++++++++-------- internal/daemon/curation_test.go | 7 +++ internal/daemon/daemon.go | 2 - internal/daemon/daemon_test.go | 4 ++ internal/expert/rotate.go | 22 +++++---- internal/expert/rotate_test.go | 4 +- internal/mail/post.go | 4 ++ internal/mail/post_test.go | 4 +- internal/mcp/researcher_tools.go | 39 ++++++++++----- internal/mcp/researcher_tools_test.go | 52 ++++++++++++++++++++ 13 files changed, 145 insertions(+), 49 deletions(-) diff --git a/docs/plans/2026-04-06-researcher-curation.md b/docs/plans/2026-04-06-researcher-curation.md index 8a169bb..efd3735 100644 --- a/docs/plans/2026-04-06-researcher-curation.md +++ b/docs/plans/2026-04-06-researcher-curation.md @@ -2,7 +2,7 @@ ## Context -v0.7 (shared experts + multi-pool) is complete on `feat/v0.7-shared-experts`. The researcher role has been scaffolded across prior versions — config parsing, mail routing, directory creation, builtin role registration — but nothing actually watches its inbox, spawns sessions, or provides tools. v0.8 brings the researcher to life as the pool's knowledge curator: it reads expert state/logs, distills knowledge, promotes patterns to identity, and keeps state.md lean over time. +v0.7 (shared experts + multi-pool) is complete on `feat/v0.7-shared-experts`. The researcher role has been scaffolded across prior versions — config parsing, mail routing, directory creation, built-in role registration — but nothing actually watches its inbox, spawns sessions, or provides tools. v0.8 brings the researcher to life as the pool's knowledge curator: it reads expert state/logs, distills knowledge, promotes patterns to identity, and keeps state.md lean over time. The architecture doc validates: "Does curation keep state.md lean over time? Does cold-start seeding produce useful initial state?" diff --git a/docs/prompts/v09-formulas-polish.md b/docs/prompts/v09-formulas-polish.md index 1edfafd..058c7dd 100644 --- a/docs/prompts/v09-formulas-polish.md +++ b/docs/prompts/v09-formulas-polish.md @@ -95,7 +95,7 @@ Formulas just bulk-register tasks with pre-defined dependency graphs. ### Formula Structure -``` +```text {poolDir}/formulas/ ├── feature-impl.toml # Standard feature flow ├── bug-triage.toml # Bug investigation + fix diff --git a/internal/config/config.go b/internal/config/config.go index b786f8c..b0e6bac 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -216,7 +216,7 @@ func LoadPool(poolDir string) (*PoolConfig, error) { if cfg.Architect.HumanInbox == "" { cfg.Architect.HumanInbox = "stdout" } - if cfg.Defaults.LogRetention == 0 { + if cfg.Defaults.LogRetention <= 0 { cfg.Defaults.LogRetention = 50 } if cfg.Curation.IntervalTasks == 0 { diff --git a/internal/daemon/curation.go b/internal/daemon/curation.go index adf0625..7713f10 100644 --- a/internal/daemon/curation.go +++ b/internal/daemon/curation.go @@ -40,30 +40,41 @@ func newCurationScheduler(cfg *config.CurationSection, poolDir string, logger *s // RecordTaskCompletion increments the completed task counter. Returns true // when the threshold is reached, signaling that curation should be triggered. +// Atomically resets the counter when the threshold fires, preventing double-fire. +// Returns false if intervalTasks <= 0 (disabled). func (cs *curationScheduler) RecordTaskCompletion() bool { cs.mu.Lock() defer cs.mu.Unlock() + if cs.intervalTasks <= 0 { + return false + } + cs.taskCount++ - return cs.taskCount >= cs.intervalTasks + if cs.taskCount >= cs.intervalTasks { + cs.taskCount = 0 + cs.lastCuration = time.Now() + return true + } + return false } // ShouldTriggerByTime returns true if enough time has elapsed since the last -// curation trigger. +// curation trigger. Atomically resets the timer when triggered. +// Returns false if intervalHours <= 0 (disabled). func (cs *curationScheduler) ShouldTriggerByTime() bool { cs.mu.Lock() defer cs.mu.Unlock() - return time.Since(cs.lastCuration) >= time.Duration(cs.intervalHours)*time.Hour -} - -// Reset zeroes the task counter and updates the last curation timestamp. -func (cs *curationScheduler) Reset() { - cs.mu.Lock() - defer cs.mu.Unlock() + if cs.intervalHours <= 0 { + return false + } - cs.taskCount = 0 - cs.lastCuration = time.Now() + if time.Since(cs.lastCuration) >= time.Duration(cs.intervalHours)*time.Hour { + cs.lastCuration = time.Now() + return true + } + return false } // triggerCuration composes a curation task and posts it to the researcher's @@ -130,14 +141,15 @@ func buildCurationTaskBody(cfg *config.PoolConfig, poolDir, reason string) strin b.WriteString(fmt.Sprintf("| %s | pool | %d bytes | %d |\n", name, stateSize, logCount)) } - // Shared experts + // Shared experts — logs live in pool overlay, state in user dir for _, name := range cfg.Shared.Include { - dir, err := config.SharedExpertDir(name) + overlayDir := filepath.Join(poolDir, "shared-state", name) + userDir, err := config.SharedExpertDir(name) if err != nil { continue } - stateSize := fileSize(filepath.Join(dir, "state.md")) - logCount := countLogFiles(filepath.Join(dir, "logs")) + stateSize := fileSize(filepath.Join(userDir, "state.md")) + logCount := countLogFiles(filepath.Join(overlayDir, "logs")) b.WriteString(fmt.Sprintf("| %s | shared | %d bytes | %d |\n", name, stateSize, logCount)) } @@ -168,11 +180,9 @@ func (d *Daemon) rotateAllLogs() { } for _, name := range d.cfg.Shared.Include { - dir, err := config.SharedExpertDir(name) - if err != nil { - continue - } - if archived, rotErr := expert.RotateLogs(dir, retention); rotErr != nil { + // Shared expert logs live in pool overlay, not user-level dir + overlayDir := filepath.Join(d.poolDir, "shared-state", name) + if archived, rotErr := expert.RotateLogs(overlayDir, retention); rotErr != nil { d.logger.Warn("Failed to rotate shared expert logs", "expert", name, "error", rotErr, diff --git a/internal/daemon/curation_test.go b/internal/daemon/curation_test.go index 4eed843..f157158 100644 --- a/internal/daemon/curation_test.go +++ b/internal/daemon/curation_test.go @@ -17,6 +17,7 @@ import ( "github.com/cameronsjo/agent-pool/internal/config" "github.com/cameronsjo/agent-pool/internal/daemon" + "github.com/cameronsjo/agent-pool/internal/taskboard" ) func TestCurationScheduler_TaskThreshold(t *testing.T) { @@ -83,8 +84,10 @@ interval_hours = 168 } // Verify the curation task body contains expert metadata + var curationTaskID string for _, c := range fake.getCalls() { if c.Name == "researcher" && strings.HasPrefix(c.TaskMessage.ID, "curation-") { + curationTaskID = c.TaskMessage.ID body := c.TaskMessage.Body if !strings.Contains(body, "auth") { t.Error("curation task body should mention auth expert") @@ -99,5 +102,9 @@ interval_hours = 168 } } + // Wait for curation task to complete before shutdown + if curationTaskID != "" { + waitForTaskStatus(t, poolDir, curationTaskID, taskboard.StatusCompleted) + } shutdownDaemon(t, cancel, errCh) } diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index bd15142..c04e2d9 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -222,7 +222,6 @@ func (d *Daemon) Run(ctx context.Context) error { return case <-ticker.C: if d.curation.ShouldTriggerByTime() { - d.curation.Reset() d.triggerCuration("time_interval") } } @@ -874,7 +873,6 @@ func (d *Daemon) markTaskCompleted(ctx context.Context, taskID string, exitCode // Check curation threshold after task completion if d.curation.RecordTaskCompletion() { - d.curation.Reset() d.wg.Add(1) go func() { defer d.wg.Done(); d.triggerCuration("task_threshold") }() } diff --git a/internal/daemon/daemon_test.go b/internal/daemon/daemon_test.go index 0aaee0d..bfbd9c4 100644 --- a/internal/daemon/daemon_test.go +++ b/internal/daemon/daemon_test.go @@ -2345,6 +2345,7 @@ model = "haiku" t.Error("log should be written to researcher/logs/, not experts/researcher/logs/") } + waitForTaskStatus(t, poolDir, "task-res-001", taskboard.StatusCompleted) shutdownDaemon(t, cancel, errCh) } @@ -2419,6 +2420,8 @@ model = "sonnet" } } + waitForTaskStatus(t, poolDir, "task-cfg-res", taskboard.StatusCompleted) + waitForTaskStatus(t, poolDir, "task-cfg-auth2", taskboard.StatusCompleted) shutdownDaemon(t, cancel, errCh) } @@ -2481,5 +2484,6 @@ model = "haiku" t.Errorf("spawn name = %q, want researcher", calls[0].Name) } + waitForTaskStatus(t, poolDir, "task-predrain-res", taskboard.StatusCompleted) shutdownDaemon(t, cancel, errCh) } diff --git a/internal/expert/rotate.go b/internal/expert/rotate.go index 5b52544..1120445 100644 --- a/internal/expert/rotate.go +++ b/internal/expert/rotate.go @@ -71,12 +71,9 @@ func RotateLogs(expertDir string, retention int) (int, error) { if err != nil { return 0, fmt.Errorf("creating archive: %w", err) } - defer f.Close() gw := gzip.NewWriter(f) - defer gw.Close() tw := tar.NewWriter(gw) - defer tw.Close() var archived int var toDelete []string @@ -100,14 +97,21 @@ func RotateLogs(expertDir string, retention int) (int, error) { } } - // Close writers before deleting source files - tw.Close() - gw.Close() - f.Close() + // Close writers before deleting source files. Check errors — if + // finalization fails the archive is incomplete and sources must be kept. + if err := tw.Close(); err != nil { + return 0, fmt.Errorf("finalizing tar: %w", err) + } + if err := gw.Close(); err != nil { + return 0, fmt.Errorf("finalizing gzip: %w", err) + } + if err := f.Close(); err != nil { + return 0, fmt.Errorf("closing archive file: %w", err) + } - // Delete archived files + // Delete archived files (best-effort — sources are safely in the archive) for _, path := range toDelete { - os.Remove(path) // best-effort; files are safely in the archive + os.Remove(path) } return archived, nil diff --git a/internal/expert/rotate_test.go b/internal/expert/rotate_test.go index d0e18e9..f5b6fad 100644 --- a/internal/expert/rotate_test.go +++ b/internal/expert/rotate_test.go @@ -186,7 +186,9 @@ func TestRotateLogs_IndexUntouched(t *testing.T) { os.WriteFile(filepath.Join(logsDir, fmt.Sprintf("task-%03d.json", i)), []byte("{}"), 0o644) } - RotateLogs(dir, 3) + if _, err := RotateLogs(dir, 3); err != nil { + t.Fatalf("RotateLogs: %v", err) + } // Verify index.md is unchanged data, err := os.ReadFile(filepath.Join(logsDir, "index.md")) diff --git a/internal/mail/post.go b/internal/mail/post.go index 9e2c067..530c48c 100644 --- a/internal/mail/post.go +++ b/internal/mail/post.go @@ -12,6 +12,10 @@ import ( // Creates the postoffice directory if it doesn't exist. Uses atomic writes // to prevent partial files from being picked up by the daemon's watcher. func Post(poolDir string, msg *Message) error { + if poolDir == "" { + return fmt.Errorf("pool directory is empty") + } + composed, err := Compose(msg) if err != nil { return fmt.Errorf("composing message: %w", err) diff --git a/internal/mail/post_test.go b/internal/mail/post_test.go index 814b136..79b075a 100644 --- a/internal/mail/post_test.go +++ b/internal/mail/post_test.go @@ -32,8 +32,8 @@ func TestPost_WritesMessage(t *testing.T) { } path := filepath.Join(poolDir, "postoffice", "task-post-001.md") - if _, err := os.Stat(path); os.IsNotExist(err) { - t.Fatal("message file not created") + if _, err := os.Stat(path); err != nil { + t.Fatalf("message file not created: %v", err) } // Verify it's parseable diff --git a/internal/mcp/researcher_tools.go b/internal/mcp/researcher_tools.go index bcd4ea8..6c04a83 100644 --- a/internal/mcp/researcher_tools.go +++ b/internal/mcp/researcher_tools.go @@ -114,16 +114,17 @@ func handleResearcherListExperts(cfg *ServerConfig) server.ToolHandlerFunc { for _, name := range names { dir := mail.ResolveExpertDir(cfg.PoolDir, name) - experts = append(experts, gatherExpertInfo(name, "pool", dir)) + experts = append(experts, gatherExpertInfo(name, "pool", dir, dir)) } - // Shared experts + // Shared experts — state in user dir, logs in pool overlay for _, name := range poolCfg.Shared.Include { dir, err := config.SharedExpertDir(name) if err != nil { continue } - experts = append(experts, gatherExpertInfo(name, "shared", dir)) + overlayDir := filepath.Join(cfg.PoolDir, "shared-state", name) + experts = append(experts, gatherExpertInfo(name, "shared", dir, overlayDir)) } data, err := json.MarshalIndent(experts, "", " ") @@ -135,15 +136,17 @@ func handleResearcherListExperts(cfg *ServerConfig) server.ToolHandlerFunc { } } -// gatherExpertInfo stats an expert directory for metadata. -func gatherExpertInfo(name, expertType, dir string) expertInfo { +// gatherExpertInfo stats an expert directory for metadata. The logBaseDir +// parameter specifies where logs live (same as dir for pool-scoped experts, +// but the pool overlay for shared experts). +func gatherExpertInfo(name, expertType, dir, logBaseDir string) expertInfo { info := expertInfo{Name: name, Type: expertType} if fi, err := os.Stat(filepath.Join(dir, "state.md")); err == nil { info.StateBytes = fi.Size() } - logsDir := filepath.Join(dir, "logs") + logsDir := filepath.Join(logBaseDir, "logs") if entries, err := os.ReadDir(logsDir); err == nil { for _, e := range entries { if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { @@ -240,7 +243,8 @@ func handleReadExpertLogs(cfg *ServerConfig) server.ToolHandlerFunc { return mcp.NewToolResultError("expert parameter is required"), nil } - dir := resolveTargetExpertDir(cfg.PoolDir, expertName) + // Logs dir may differ from state dir for shared experts + dir := resolveLogsDir(cfg.PoolDir, expertName) query := request.GetString("query", "") countStr := request.GetString("count", "10") @@ -303,16 +307,17 @@ func handleEnrichState(cfg *ServerConfig) server.ToolHandlerFunc { return mcp.NewToolResultError("expert parameter is required"), nil } - dir := resolveTargetExpertDir(cfg.PoolDir, expertName) + stateDir := resolveTargetExpertDir(cfg.PoolDir, expertName) + logDir := resolveLogsDir(cfg.PoolDir, expertName) - identity, state, errors, err := expert.ReadState(dir) + identity, state, errors, err := expert.ReadState(stateDir) if err != nil { return mcp.NewToolResultError(fmt.Sprintf("reading state: %v", err)), nil } // Read last 10 index entries var recentIndex []string - indexPath := filepath.Join(dir, "logs", "index.md") + indexPath := filepath.Join(logDir, "logs", "index.md") if data, err := os.ReadFile(indexPath); err == nil { lines := strings.Split(strings.TrimSpace(string(data)), "\n") for i := 2; i < len(lines); i++ { @@ -328,7 +333,7 @@ func handleEnrichState(cfg *ServerConfig) server.ToolHandlerFunc { // Read last 3 full log files (newest first by filename sort) var recentLogs []map[string]string - logsDir := filepath.Join(dir, "logs") + logsDir := filepath.Join(logDir, "logs") if entries, err := os.ReadDir(logsDir); err == nil { var jsonFiles []string for _, e := range entries { @@ -486,7 +491,7 @@ func handlePromotePattern(cfg *ServerConfig) server.ToolHandlerFunc { // resolveTargetExpertDir returns the state directory for a target expert. // Built-in roles use {poolDir}/{role}/, pool-scoped use {poolDir}/experts/{name}/. -// For shared experts, returns the user-level directory. +// For shared experts, returns the user-level directory (identity + state). func resolveTargetExpertDir(poolDir, name string) string { if isSharedExpert(poolDir, name) { dir, err := config.SharedExpertDir(name) @@ -497,6 +502,16 @@ func resolveTargetExpertDir(poolDir, name string) string { return mail.ResolveExpertDir(poolDir, name) } +// resolveLogsDir returns the directory containing logs for an expert. +// For shared experts, logs live in the pool overlay (shared-state//). +// For pool-scoped experts, logs are in the expert dir itself. +func resolveLogsDir(poolDir, name string) string { + if isSharedExpert(poolDir, name) { + return filepath.Join(poolDir, "shared-state", name) + } + return mail.ResolveExpertDir(poolDir, name) +} + // resolveSharedOverlayDir returns the pool-scoped overlay directory for a shared expert. // Returns empty string if the expert is not shared. func resolveSharedOverlayDir(poolDir, name string) string { diff --git a/internal/mcp/researcher_tools_test.go b/internal/mcp/researcher_tools_test.go index 384039b..93db04e 100644 --- a/internal/mcp/researcher_tools_test.go +++ b/internal/mcp/researcher_tools_test.go @@ -627,3 +627,55 @@ func TestResearcherListExperts_IncludesShared(t *testing.T) { t.Errorf("expected security-standards/shared, got %s/%s", experts[1].Name, experts[1].Type) } } + +func TestResearcherReadExpertLogs_SharedOverlay(t *testing.T) { + poolDir, _ := setupSharedResearcherPool(t) + + // Seed logs in the pool overlay (where the daemon writes them) + overlayDir := filepath.Join(poolDir, "shared-state", "security-standards") + expert.WriteLog(overlayDir, "task-sec-001", []byte(`{"type":"result","result":"Audited CORS policy"}`)) + expert.AppendIndex(overlayDir, &expert.LogEntry{TaskID: "task-sec-001", From: "architect", ExitCode: 0, Summary: "Audited CORS policy"}) + + srv := buildMCPTestServer(t, poolDir, "researcher", "researcher") + + // read_expert_logs should find logs from the overlay + result := callTool(t, srv, "read_expert_logs", map[string]any{ + "expert": "security-standards", + }) + text := resultText(t, result) + + if !strings.Contains(text, "task-sec-001") { + t.Errorf("expected shared overlay log entry, got: %s", text) + } + + // enrich_state should include overlay logs + result = callTool(t, srv, "enrich_state", map[string]any{ + "expert": "security-standards", + }) + text = resultText(t, result) + + var enriched map[string]any + if err := json.Unmarshal([]byte(text), &enriched); err != nil { + t.Fatalf("unmarshaling: %v", err) + } + + recentIndex, ok := enriched["recent_index"].([]any) + if !ok || len(recentIndex) == 0 { + t.Error("enrich_state should include overlay log index entries") + } + + // list_experts should show log count from overlay + result = callTool(t, srv, "list_experts", nil) + text = resultText(t, result) + + var allExperts []struct { + Name string `json:"name"` + LogCount int `json:"log_count"` + } + json.Unmarshal([]byte(text), &allExperts) + for _, e := range allExperts { + if e.Name == "security-standards" && e.LogCount != 1 { + t.Errorf("shared expert log_count = %d, want 1", e.LogCount) + } + } +}