diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index 638ea54..0989432 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -1,14 +1,19 @@ package main import ( + "bufio" "context" + "encoding/json" "fmt" "io" "log/slog" + "net" "os" "os/signal" "path/filepath" + "strings" "syscall" + "time" "github.com/cameronsjo/agent-pool/internal/config" "github.com/cameronsjo/agent-pool/internal/daemon" @@ -25,6 +30,12 @@ func main() { switch os.Args[1] { case "start": cmdStart() + case "stop": + cmdStop() + case "status": + cmdStatus() + case "watch": + cmdWatch() case "mcp": cmdMCP() case "flush": @@ -32,7 +43,7 @@ func main() { case "guard": cmdGuard() case "version": - fmt.Println("agent-pool v0.5.0-dev") + fmt.Println("agent-pool v0.6.0-dev") case "help", "--help", "-h": printUsage() default: @@ -86,6 +97,17 @@ func cmdStart() { syscall.SIGTERM, syscall.SIGINT) defer stop() + // Double-signal: first signal triggers graceful drain, second forces exit. + go func() { + <-ctx.Done() + stop() // reset signal handling to default + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) + <-sigCh + logger.Warn("Received second signal, forcing immediate exit") + os.Exit(1) + }() + d := daemon.New(cfg, poolDir, logger) if err := d.Run(ctx); err != nil { logger.Error("Daemon failed", "error", err) @@ -93,6 +115,312 @@ func cmdStart() { } } +func cmdStop() { + explicit := "" + if len(os.Args) > 2 { + explicit = os.Args[2] + } + + poolDir, err := config.DiscoverPoolDir(explicit) + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + + resp, err := connectAndSend(config.ResolveSockPath(poolDir), "stop") + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + + if resp.Status != "ok" { + fmt.Fprintf(os.Stderr, "error: %s\n", resp.Message) + os.Exit(1) + } + + fmt.Println("Daemon is shutting down.") +} + +func cmdStatus() { + explicit := "" + if len(os.Args) > 2 { + explicit = os.Args[2] + } + + poolDir, err := config.DiscoverPoolDir(explicit) + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + + resp, err := connectAndSend(config.ResolveSockPath(poolDir), "status") + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + + if resp.Status != "ok" { + fmt.Fprintf(os.Stderr, "error: %s\n", resp.Message) + os.Exit(1) + } + + var data map[string]json.RawMessage + if err := json.Unmarshal(resp.Data, &data); err != nil { + // Fallback to raw JSON + fmt.Println(string(resp.Data)) + return + } + + printStatusField := func(label, key string) { + if v, ok := data[key]; ok { + var s string + if err := json.Unmarshal(v, &s); err != nil { + fmt.Printf("%-10s %s\n", label+":", string(v)) + return + } + fmt.Printf("%-10s %s\n", label+":", s) + } + } + + printStatusField("Pool", "pool") + printStatusField("State", "state") + printStatusField("Uptime", "uptime") + + // Experts + if v, ok := data["experts"]; ok { + var experts []string + json.Unmarshal(v, &experts) + fmt.Printf("%-10s %s\n", "Experts:", strings.Join(experts, ", ")) + } + + // Task counts + if v, ok := data["task_counts"]; ok { + var counts map[string]int + json.Unmarshal(v, &counts) + if len(counts) > 0 { + fmt.Println("\nTasks:") + for _, status := range []string{"pending", "blocked", "active", "completed", "failed", "cancelled"} { + if n, ok := counts[status]; ok && n > 0 { + fmt.Printf(" %-12s %d\n", status+":", n) + } + } + } + } + + // Active tasks + if v, ok := data["active_tasks"]; ok { + var tasks []map[string]string + json.Unmarshal(v, &tasks) + if len(tasks) > 0 { + fmt.Println("\nActive Tasks:") + for _, t := range tasks { + started := t["started"] + if started != "" { + started = " (" + started + " ago)" + } + fmt.Printf(" %-20s %s%s\n", t["id"], t["expert"], started) + } + } + } +} + +func cmdWatch() { + explicit := "" + if len(os.Args) > 2 { + explicit = os.Args[2] + } + + poolDir, err := config.DiscoverPoolDir(explicit) + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + + sockPath := config.ResolveSockPath(poolDir) + conn, err := net.DialTimeout("unix", sockPath, 5*time.Second) + if err != nil { + fmt.Fprintf(os.Stderr, "error: connecting to daemon (is it running?): %v\n", err) + os.Exit(1) + } + defer conn.Close() + + // Send subscribe request + conn.SetDeadline(time.Now().Add(5 * time.Second)) + if err := json.NewEncoder(conn).Encode(map[string]string{"method": "subscribe"}); err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + + // Read ack + scanner := bufio.NewScanner(conn) + if !scanner.Scan() { + if err := scanner.Err(); err != nil { + fmt.Fprintf(os.Stderr, "error: reading ack: %v\n", err) + } else { + fmt.Fprintf(os.Stderr, "error: no ack from daemon\n") + } + os.Exit(1) + } + var ack socketResponse + if err := json.Unmarshal(scanner.Bytes(), &ack); err != nil || ack.Status != "ok" { + fmt.Fprintf(os.Stderr, "error: subscribe failed: %s\n", ack.Message) + os.Exit(1) + } + + // Clear deadline for streaming + conn.SetDeadline(time.Time{}) + + // Handle Ctrl-C cleanly + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) + go func() { + <-sigCh + conn.Close() + }() + + fmt.Println("Watching daemon events (Ctrl-C to stop)...") + fmt.Println() + + // ANSI colors + const ( + reset = "\033[0m" + green = "\033[32m" + red = "\033[31m" + yellow = "\033[33m" + cyan = "\033[36m" + ) + + type event struct { + Type string `json:"type"` + Timestamp time.Time `json:"timestamp"` + Data json.RawMessage `json:"data"` + } + + for scanner.Scan() { + var e event + if err := json.Unmarshal(scanner.Bytes(), &e); err != nil { + continue + } + + ts := e.Timestamp.Format("15:04:05") + var color, detail string + + switch e.Type { + case "task.routed": + color = cyan + var d struct { + ID string `json:"id"` + From string `json:"from"` + To string `json:"to"` + } + json.Unmarshal(e.Data, &d) + detail = fmt.Sprintf("%s -> %s (%s)", d.From, d.To, d.ID) + + case "expert.spawning": + color = yellow + var d struct { + Expert string `json:"expert"` + TaskID string `json:"task_id"` + Model string `json:"model"` + } + json.Unmarshal(e.Data, &d) + detail = fmt.Sprintf("%s task=%s model=%s", d.Expert, d.TaskID, d.Model) + + case "expert.completed": + color = green + var d struct { + Expert string `json:"expert"` + TaskID string `json:"task_id"` + Duration string `json:"duration"` + Summary string `json:"summary"` + } + json.Unmarshal(e.Data, &d) + detail = fmt.Sprintf("%s task=%s %s", d.Expert, d.TaskID, d.Duration) + if d.Summary != "" { + if len(d.Summary) > 60 { + d.Summary = d.Summary[:60] + "..." + } + detail += " " + d.Summary + } + + case "expert.failed": + color = red + var d struct { + Expert string `json:"expert"` + TaskID string `json:"task_id"` + ExitCode int `json:"exit_code"` + } + json.Unmarshal(e.Data, &d) + detail = fmt.Sprintf("%s task=%s exit=%d", d.Expert, d.TaskID, d.ExitCode) + + case "task.cancelled": + color = red + var d struct { + TaskID string `json:"task_id"` + } + json.Unmarshal(e.Data, &d) + detail = d.TaskID + + case "task.unblocked": + color = green + var d struct { + TaskID string `json:"task_id"` + Expert string `json:"expert"` + } + json.Unmarshal(e.Data, &d) + detail = fmt.Sprintf("%s -> %s", d.TaskID, d.Expert) + + default: + detail = string(e.Data) + } + + fmt.Printf("[%s] %s%-18s%s %s\n", ts, color, e.Type, reset, detail) + } + + if err := scanner.Err(); err != nil { + fmt.Fprintf(os.Stderr, "error: stream interrupted: %v\n", err) + os.Exit(1) + } +} + +// socketResponse mirrors the daemon's response type for CLI deserialization. +type socketResponse struct { + Status string `json:"status"` + Data json.RawMessage `json:"data,omitempty"` + Message string `json:"message,omitempty"` +} + +// connectAndSend dials the daemon socket, sends a method request, and reads the response. +func connectAndSend(sockPath, method string) (*socketResponse, error) { + conn, err := net.DialTimeout("unix", sockPath, 5*time.Second) + if err != nil { + return nil, fmt.Errorf("connecting to daemon (is it running?): %w", err) + } + defer conn.Close() + + conn.SetDeadline(time.Now().Add(5 * time.Second)) + + req := map[string]string{"method": method} + if err := json.NewEncoder(conn).Encode(req); err != nil { + return nil, fmt.Errorf("sending request: %w", err) + } + + scanner := bufio.NewScanner(conn) + if !scanner.Scan() { + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("reading response: %w", err) + } + return nil, fmt.Errorf("no response from daemon") + } + + var resp socketResponse + if err := json.Unmarshal(scanner.Bytes(), &resp); err != nil { + return nil, fmt.Errorf("parsing response: %w", err) + } + + return &resp, nil +} + // cmdMCP starts the stdio MCP server. Stdout is the MCP transport; logs go to stderr. // // Two invocation modes: @@ -229,6 +557,9 @@ func printUsage() { Usage: agent-pool start [pool-dir] Start the daemon for a pool + agent-pool stop [pool-dir] Stop a running daemon + agent-pool status [pool-dir] Show daemon status + 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 flush --pool --expert --task Stop hook: verify state @@ -243,5 +574,7 @@ Roles: Examples: agent-pool start ~/.agent-pool/pools/api-gateway + agent-pool stop + agent-pool status agent-pool mcp --pool ./my-pool --role concierge`) } diff --git a/docs/launchd.md b/docs/launchd.md new file mode 100644 index 0000000..1fd9cf8 --- /dev/null +++ b/docs/launchd.md @@ -0,0 +1,64 @@ +# Running agent-pool with launchd + +launchd is macOS's service manager. Use it to run the daemon in the background +so it starts automatically on login. + +## Install + +1. Copy the plist template: + +```bash +cp scripts/com.agent-pool.daemon.plist ~/Library/LaunchAgents/ +``` + +2. Edit the plist and replace placeholders: + +```bash +# Replace AGENT_POOL_BINARY with the full path to the binary +sed -i '' "s|AGENT_POOL_BINARY|$(which agent-pool)|g" \ + ~/Library/LaunchAgents/com.agent-pool.daemon.plist + +# Replace POOL_DIR with your pool directory +sed -i '' "s|POOL_DIR|$HOME/.agent-pool/pools/my-pool|g" \ + ~/Library/LaunchAgents/com.agent-pool.daemon.plist +``` + +3. Load the service: + +```bash +launchctl load ~/Library/LaunchAgents/com.agent-pool.daemon.plist +``` + +## Stop + +```bash +# Graceful stop via socket (preferred) +agent-pool stop + +# Or via launchctl (sends SIGTERM, daemon drains gracefully) +launchctl stop com.agent-pool.daemon +``` + +## Unload + +Remove the service entirely: + +```bash +launchctl unload ~/Library/LaunchAgents/com.agent-pool.daemon.plist +rm ~/Library/LaunchAgents/com.agent-pool.daemon.plist +``` + +## Logs + +The daemon writes to `{poolDir}/daemon.log` as usual. launchd captures +any output that bypasses slog to `launchd-stdout.log` and `launchd-stderr.log` +in the pool directory. + +## Notes + +- `KeepAlive` is false — the daemon runs once and exits on stop. Use + `agent-pool start` to restart manually, or set `KeepAlive` to true + if you want launchd to auto-restart on crash. +- `RunAtLoad` is true — the daemon starts when you log in. +- Double-signal works: `launchctl stop` sends SIGTERM (graceful drain), + a second stop during drain forces immediate exit. diff --git a/docs/plans/2026-04-05-daemon-lifecycle.md b/docs/plans/2026-04-05-daemon-lifecycle.md new file mode 100644 index 0000000..ef0d3f9 --- /dev/null +++ b/docs/plans/2026-04-05-daemon-lifecycle.md @@ -0,0 +1,213 @@ +# v0.6: Daemon Lifecycle + Observability + +## Context + +Dogfooding on bosun exposed that operating a pool is painful: no way to stop the daemon cleanly, no visibility into what's happening, no lifecycle management. v0.6 adds a Unix domain socket for CLI->daemon communication, enabling `stop`, `status`, and `watch` commands. Also addresses session timeout defaults and the `pool_` tool name prefix. Issues: #8, #9, #10, #11, #13. + +--- + +## Phase 1: Socket + Stop + +**New:** `internal/daemon/socket.go` +- `socketServer` struct: `net.Listener`, `*Daemon`, `context.CancelFunc` +- `newSocketServer(sockPath, daemon, cancel)` — removes stale socket, listens +- `serve(ctx)` — accept loop, one goroutine per connection +- `handleConn(conn)` — 5s read deadline, reads one NDJSON line, dispatches by method, writes response, closes +- `close()` — closes listener, removes socket file +- Protocol types: `socketRequest{Method}`, `socketResponse{Status, Data, Message}` +- Methods: `stop` (calls cancel), `status` (placeholder), `subscribe` (placeholder) + +**New:** `internal/daemon/socket_test.go` +- `TestSocket_StopMethod` — start daemon, connect, send stop, verify exit +- `TestSocket_UnknownMethod` — error response, daemon stays up +- `TestSocket_StaleSocketCleanup` — pre-existing socket file doesn't block start +- `TestSocket_SocketRemovedOnShutdown` — file cleaned up after exit +- Helper: `connectSocket(t, poolDir) net.Conn` + +**Modify:** `internal/daemon/daemon.go` +- Add `startedAt time.Time`, `sockPath string` to Daemon struct +- In `Run()`: wrap ctx in child context, share cancel with socket server + ``` + childCtx, cancel := context.WithCancel(ctx) + sock := newSocketServer(sockPath, d, cancel) + go sock.serve(childCtx) + ``` +- Socket stop -> cancel() -> childCtx.Done() -> event loop exits + +**Modify:** `cmd/agent-pool/main.go` +- Add `"stop"` to switch, implement `cmdStop()` +- `cmdStop`: discover pool dir, `net.Dial("unix", daemon.sock)`, send `{"method":"stop"}`, print result +- Extract `connectAndSend(sockPath, method) (*socketResponse, error)` helper +- Update `printUsage()`, bump version to `v0.6.0-dev` + +**Commit:** `feat: add unix domain socket and stop command` + +--- + +## Phase 2: Graceful Drain + +**Modify:** `internal/daemon/daemon.go` +- Add `wg sync.WaitGroup` to Daemon struct +- Add `drainTimeout time.Duration` (default 30s) + `WithDrainTimeout` option +- Wrap ALL 5 goroutine dispatch sites with `wg.Add(1)` / `defer wg.Done()`: + 1. `handleApprovalRequest` dispatch (Run event loop, ~line 170) + 2. `handleInbox` dispatch (Run event loop, ~line 179) + 3. Architect drain in `drainAllInboxes` (~line 729) + 4. Expert drains in `drainAllInboxes` (~line 731-733) + 5. Wake experts in `markTaskCompleted` (~line 704) +- Replace shutdown path: cancel -> wg.Wait(30s timeout) -> close socket -> return + +**Modify:** `cmd/agent-pool/main.go` +- Double-signal handler: first signal cancels ctx (graceful), second signal calls `os.Exit(1)` (immediate) + +**Modify:** `internal/daemon/daemon_test.go` +- `TestDaemon_GracefulDrainWaitsForInFlight` — gated spawner, cancel while blocked, verify wait then clean exit + +**Commit:** `feat: add graceful drain with WaitGroup and double-signal` + +--- + +## Phase 3: Status + +**Modify:** `internal/daemon/socket.go` +- Define `statusData` struct: Pool, State, Uptime, Experts, TaskCounts, ActiveTasks +- Define `activeTaskInfo`: ID, Expert, Started + +**Modify:** `internal/daemon/daemon.go` +- Add `Status() statusData` method — reads board under mu, computes counts via `TasksByStatus()`, collects active tasks, expert names from cfg + +**Modify:** `internal/daemon/socket.go` +- Wire status method to return `d.daemon.Status()` as response data + +**Modify:** `cmd/agent-pool/main.go` +- Add `"status"` to switch, implement `cmdStatus()` +- Pretty-print: pool name, state, uptime, expert list, task count table, active task list + +**Modify:** `internal/daemon/socket_test.go` +- `TestSocket_StatusMethod` — verify response has pool name, experts, counts +- `TestSocket_StatusWithActiveTasks` — gated spawn, verify active task entry + +**Commit:** `feat: add status command and socket status method` + +--- + +## Phase 4: Watch (Event Streaming) + +**New:** `internal/daemon/events.go` +- Event types: `task.routed`, `expert.spawning`, `expert.completed`, `expert.failed`, `task.cancelled`, `task.unblocked` +- `Event` struct: Type, Timestamp, Data (typed per event) +- `eventBus` struct: `sync.RWMutex`, subscriber map of `id -> chan Event` +- `subscribe() (id, <-chan Event)` — buffered channel (cap 64) +- `unsubscribe(id)` — removes + closes channel +- `emit(Event)` — non-blocking send under read lock (drop if full) + +**New:** `internal/daemon/events_test.go` +- `TestEventBus_SubscribeReceivesEvents` +- `TestEventBus_MultipleSubscribers` +- `TestEventBus_UnsubscribeCleansUp` +- `TestEventBus_SlowSubscriberDropsEvents` + +**Modify:** `internal/daemon/daemon.go` +- Add `events *eventBus` to struct, init in `New()` +- Add `emit(EventType, data)` helper +- Insert emit calls at 6 existing log points: + 1. `handlePostoffice` after successful route -> `task.routed` + 2. `processInboxMessage` before spawn -> `expert.spawning` + 3. `processInboxMessage` on success -> `expert.completed` + 4. `processInboxMessage` on failure -> `expert.failed` + 5. `handleCancel` on cancel -> `task.cancelled` + 6. `markTaskCompleted` for unblocked tasks -> `task.unblocked` + +**Modify:** `internal/daemon/socket.go` +- `handleSubscribe(conn, ctx)` — subscribe to bus, stream events as NDJSON, unsubscribe on disconnect/cancel + +**Modify:** `cmd/agent-pool/main.go` +- Add `"watch"` to switch, implement `cmdWatch()` +- Connect, send subscribe, read NDJSON stream +- ANSI color per event type: green=completed, red=failed, yellow=spawning, cyan=routed +- Signal handler for clean disconnect on Ctrl-C + +**Modify:** `internal/daemon/socket_test.go` +- `TestSocket_SubscribeStreamsEvents` — subscribe, route message, verify event arrives +- `TestSocket_SubscribeDisconnectCleansUp` — close client, verify no leak + +**Commit:** `feat: add event bus and watch command` + +--- + +## Phase 5: Timeout Removal + Tool Rename + +### 5a: Session timeout optional + +**Modify:** `internal/config/config.go` +- Remove `session_timeout = "10m"` default from `LoadPool()` (~line 153) +- `ParseSessionTimeout()` returns `(0, nil)` when string is empty + +**Modify:** `internal/daemon/daemon.go` +- In `processInboxMessage`: if timeout == 0, use `context.WithCancel(ctx)` instead of `context.WithTimeout` +- Update `resolveSessionTimeout` to return 0 for empty values + +**Commit:** `feat: make session_timeout optional (zero = no timeout)` + +### 5b: Drop pool_ prefix + +Mechanical rename across all files. Tool names: `pool_read_state` -> `read_state`, etc. + +| File | Changes | +|------|---------| +| `internal/mcp/config.go` | ExpertToolNames (6 entries) | +| `internal/mcp/tools.go` | 6 NewTool calls | +| `internal/mcp/architect_tools.go` | 4 NewTool calls | +| `internal/mcp/concierge_tools.go` | 6 NewTool calls + description refs | +| `internal/mcp/tools_test.go` | ~10 callTool refs | +| `internal/mcp/architect_tools_test.go` | ~30 refs | +| `internal/mcp/concierge_tools_test.go` | ~35 refs | +| `plugin/concierge-identity.md` | 4 refs | +| `plugin/skills/pool-ask.md` | 2 refs | +| `plugin/skills/pool-build.md` | 3 refs | +| `plugin/skills/pool-status.md` | 1 ref | +| External: bosun concierge identity | 4 refs | + +**Commit:** `refactor: drop pool_ prefix from MCP tool names` + +--- + +## Phase 6: launchd + +**New:** `scripts/com.agent-pool.daemon.plist` — template with `AGENT_POOL_BINARY` and `POOL_DIR` placeholders + +**New:** `docs/launchd.md` — install/load/unload instructions + +**Commit:** `docs: add launchd plist template and install guide` + +--- + +## Verification + +After each phase: +```bash +make test # all tests pass +make build # binary builds +``` + +End-to-end after all phases: +```bash +# Terminal 1: start daemon +bin/agent-pool start ~/.agent-pool/pools/test + +# Terminal 2: check status +bin/agent-pool status + +# Terminal 3: watch events +bin/agent-pool watch + +# Terminal 2: stop daemon +bin/agent-pool stop +# Verify daemon exits cleanly in terminal 1 +``` + +## Risks + +- **Phase 2 WaitGroup**: Exactly 5 goroutine dispatch sites. Missing one = goroutine leak bypassing drain. Mitigate: grep for `go d.` and `go func` after implementation. +- **Phase 4 subscriber leak**: Disconnected client fills channel. Mitigate: non-blocking emit (select+default), write errors break serve loop. +- **Phase 5 rename**: Breaking change for bosun. Mitigate: update bosun in same session. diff --git a/docs/test-audit-report.md b/docs/test-audit-report.md index eef7686..dfe25f0 100644 --- a/docs/test-audit-report.md +++ b/docs/test-audit-report.md @@ -1,88 +1,105 @@ -# Test Audit Report -- agent-pool v0.5 (branch scope) +# Test Audit Report — agent-pool (full scope) -**Branch:** `feat/v0.5-concierge-plugin` vs `main` -**Date:** 2026-04-04 -**Scope:** Files changed on branch only +**Date:** 2026-04-05 +**Scope:** Full codebase after v0.6 completion +**Go version:** 1.26.1 ## Summary -- **Source files changed:** 5 | **Test files changed:** 2 | **Ratio:** 2.5:1 -- **Package coverage:** `internal/mcp` 78.0%, `internal/expert` 84.8% -- **New functions:** All have tests (none at 0%) -- **Uncovered paths:** 13 (high risk: 2, medium: 4, low: 3, skip: 4) -- **Quality issues:** 4 (P0: 0, P1: 1, P2: 3) +- Source files: 29 | Test files: 25 | Ratio: 1.2:1 +- Overall coverage: **69.9%** (threshold: 65%) +- Untested functions: 14 (high risk: 1, medium: 1, low: 0, skip: 12) +- Quality issues: 14 (P0: 0, P1: 2, P2: 4, P3: 2) ---- +## Coverage Gaps (by risk) -## Coverage Gaps by Risk +### High Risk Untested -### High Risk +| Function | File | Classification | Why It Matters | +|----------|------|---------------|----------------| +| `DiscoverPoolDir` | `internal/config/config.go:89` | Configuration | Used by every CLI command; incorrect discovery could point at wrong pool | -| Function | File:Line | Coverage | Classification | Uncovered Path | Why It Matters | -|----------|-----------|----------|---------------|----------------|----------------| -| `pollForCompletion` | `concierge_tools.go:150` | 75.0% | I/O BOUNDARY | Timeout (context deadline) | Primary failure mode when daemon is down or expert hangs | -| `pollForCompletion` | `concierge_tools.go:150` | 75.0% | STATE MACHINE | `StatusCancelled` branch | Real operational path; error includes cancel note | +### Medium Risk Untested -### Medium Risk +| Function | File | Classification | Why It Matters | +|----------|------|---------------|----------------| +| `connectAndSend` | `cmd/agent-pool/main.go:382` | I/O Boundary | Socket client helper; error handling paths (connection refused, timeout) affect UX | -| Function | File:Line | Coverage | Classification | Uncovered Path | Why It Matters | -|----------|-----------|----------|---------------|----------------|----------------| -| `handleAskExpert` | `concierge_tools.go:73` | 84.6% | I/O BOUNDARY | `os.WriteFile` error | Filesystem full/permissions; user gets opaque error | -| `handleSubmitPlan` | `concierge_tools.go:222` | 84.0% | I/O BOUNDARY | `os.WriteFile` error | Same filesystem failure class | -| `handleListExperts` | `concierge_tools.go:357` | 78.6% | CONFIGURATION | `config.LoadPool` error | Missing pool.toml is common misconfiguration | -| `readExpertResult` | `concierge_tools.go:210` | 83.3% | I/O BOUNDARY | `expert.ReadLog` error | Missing log file (race with daemon) | +### Not Worth Unit Testing (skipped) -### Low Risk +| Function | File | Pattern | Rationale | +|----------|------|---------|-----------| +| `main` | `cmd/agent-pool/main.go:24` | Entry point | Pure dispatch switch, no logic | +| `cmdStart` | `cmd/agent-pool/main.go:56` | Entry point | Orchestration glue (load config, create daemon, block) | +| `cmdStop` | `cmd/agent-pool/main.go:118` | Thin wrapper | Calls `connectAndSend("stop")` + prints result | +| `cmdStatus` | `cmd/agent-pool/main.go:144` | CLI rendering | JSON pretty-print; no logic worth testing | +| `cmdWatch` | `cmd/agent-pool/main.go:224` | CLI rendering | ANSI formatting; no logic worth testing | +| `cmdMCP` | `cmd/agent-pool/main.go:418` | Entry point | Flag parsing + `agentmcp.Run()` delegation | +| `cmdFlush` | `cmd/agent-pool/main.go:465` | Thin wrapper | Flag parsing + `hooks.Flush()` delegation | +| `cmdGuard` | `cmd/agent-pool/main.go:489` | Thin wrapper | Flag parsing + `hooks.Guard()` delegation | +| `newStderrLogger` | `cmd/agent-pool/main.go:515` | Trivial factory | Single expression, zero branching | +| `parseFlags` | `cmd/agent-pool/main.go:522` | Thin wrapper | Delegates to `parseFlagsFromArgs` (100% covered) | +| `printUsage` | `cmd/agent-pool/main.go:543` | Help text | Framework glue, no logic | +| `defaultSpawner.Spawn` | `internal/daemon/daemon.go:37` | Thin wrapper | Single delegation to `expert.Spawn` | + +### Partially Covered (worth noting) + +| Function | File | Coverage | Gap | +|----------|------|----------|-----| +| `Status` | `daemon.go:1031` | 62.5% | Active tasks branch untested | +| `ParseHumanInbox` | `presenter.go:116` | 60.0% | Telegram/file modes untested (future) | +| `WriteFile` | `atomicfile.go:13` | 45.5% | Fsync/rename error paths | +| `resolveProjectDir` | `daemon.go:1103` | 60.0% | Tilde expansion path | -| Function | File:Line | Coverage | Classification | Uncovered Path | Why It Matters | -|----------|-----------|----------|---------------|----------------|----------------| -| `handleAskExpert` | `concierge_tools.go:73` | 84.6% | I/O BOUNDARY | `mail.Compose` error | Requires invalid message fields; validated upstream | -| `handleSubmitPlan` | `concierge_tools.go:222` | 84.0% | I/O BOUNDARY | `mail.Compose` error | Same -- requires nil/invalid fields | -| `handleListExperts` | `concierge_tools.go:357` | 78.6% | CONFIGURATION | `json.MarshalIndent` error | Impossible with valid Go string slices | +## Quality Issues -### Not Worth Unit Testing (skipped) +### P0 — Likely Catching Zero Bugs -| Function | File:Line | Pattern | Rationale | -|----------|-----------|---------|-----------| -| `RegisterConciergeTools` nil guard | `concierge_tools.go:31` | Defensive check | Single-line guard; callers never pass nil | -| `handleCheckStatus` marshal errors | `concierge_tools.go:287` | Framework glue | `json.MarshalIndent` on `*taskboard.Task` can't fail with valid data | -| `handleSubmitPlan` marshal error | `concierge_tools.go:222` | Framework glue | Same -- valid structs don't fail marshal | -| `cmd/agent-pool/main.go` at 8.7% | `main.go` | Entry point | CLI glue; `parseFlagsFromArgs` (the logic) is at 100% | +None found. ---- +### P1 — Masking Real Issues -## Quality Issues +**1. `time.Sleep` for synchronization (30+ occurrences)** +- Files: `daemon_test.go`, `watcher_test.go`, `approval_test.go`, `socket_test.go` +- Pattern: `time.Sleep(50ms)` to `time.Sleep(2s)` for goroutine synchronization +- Risk: Flaky on slow CI runners or under load +- Mitigation: `waitForTaskStatus` polling helper exists but many tests use raw sleeps -### P0 -- Likely Catching Zero Bugs +**2. `time.Now()` in tests without injection** +- Files: `taskboard_test.go:109`, `contract_test.go:173` +- Pattern: Captures before/after timestamps for boundary checks +- Risk: Nanosecond-boundary failures (extremely rare but non-deterministic) -None found. +### P2 — Test Debt -### P1 -- Masking Real Issues +**3. Polling loops instead of event-driven sync** +- File: `daemon_test.go` (6+ locations) +- Pattern: `deadline := time.Now().Add(5*time.Second)` with sleep loop +- Better: Channel-based signaling or sync.WaitGroup -| File | Lines | Pattern | Detail | -|------|-------|---------|--------| -| `concierge_tools_test.go` | 152, 241 | Sleep in test | `time.Sleep(50ms)` in goroutines polling the postoffice. Mitigated by retry loop (polls up to 50 times), but nonzero flake risk on slow CI. | +**4. Existence-only assertions** +- Files: `daemon_test.go`, `router_test.go` +- Pattern: `os.Stat(path)` checks file exists but not content -### P2 -- Test Debt +**5. `connectSocket` retry loop** +- File: `socket_test.go:47-63` +- Pattern: 20 retries with 50ms sleep to wait for socket readiness -| File | Lines | Pattern | Detail | -|------|-------|---------|--------| -| `concierge_tools_test.go` | 397-407 | Existence-only assertion | `TestCheckStatus_SingleTask` uses `strings.Contains` for task ID and status. Doesn't validate JSON structure -- a malformed response containing those substrings would pass. | -| `concierge_tools_test.go` | 165-180 | No subtests | `TestAskExpert_MissingParams` tests two cases sequentially. If the first fails, the second never runs. Should use `t.Run` for isolation. | -| `concierge_tools_test.go` | 172, 255, 412+ | `time.Now()` in fixtures | Used for `CreatedAt`/`CompletedAt` in test data. Not a flake risk (no time-dependent assertions) but a determinism code smell -- prefer fixed time constants. | +**6. Test helper complexity** +- File: `daemon_test.go` +- `startTestDaemon` does socket path override, background goroutine, 500ms sleep -### P3 -- Notes +### P3 — Notes -| File | Lines | Pattern | Detail | -|------|-------|---------|--------| -| `concierge_tools_test.go` | 109 | Cross-file helper | Reuses `listArchitectToolNames` from `architect_tools_test.go`. Works (same package) but creates implicit coupling. | +**7. Repeated pool config setup** +- File: `daemon_test.go` — 15+ tests write nearly identical pool.toml configs ---- +**8. `WithSocketPath` at 0% coverage** +- Used conditionally by `startTestDaemon` path-length check; tool doesn't see it ## Recommended Next Steps -1. **Run `write-tests`** for 2 high-risk gaps -- `pollForCompletion` timeout and cancellation paths. Biggest bang for the buck. -2. **Add `TestListExperts_MissingConfig`** -- don't write pool.toml to temp dir, verify error. Easy win for medium-risk gap. -3. **Refactor `TestAskExpert_MissingParams` into subtests** -- `t.Run("missing_expert", ...)` for isolation. Low-effort P2 fix. -4. `make-testable` is NOT needed -- all functions are testable via the MCP server interface. -5. `setup-coverage` is NOT needed -- Makefile already has `test-cover` and `test-gaps` targets. +1. **Write tests for `DiscoverPoolDir`** — high-risk config function with directory traversal logic +2. **Write tests for `connectAndSend`** — medium-risk socket client with error handling paths +3. **Replace readiness sleep** — the 500ms sleep in `startTestDaemon` is the root of downstream flakiness +4. **No immediate action on P2/P3** — debt indicators, not bugs; address when touching those files diff --git a/internal/config/config.go b/internal/config/config.go index e5ad705..e4b287a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -61,7 +61,11 @@ type CurationSection struct { } // ParseSessionTimeout parses the session timeout string to a time.Duration. +// Returns (0, nil) when the timeout is empty, meaning sessions run to completion. func (d DefaultsSection) ParseSessionTimeout() (time.Duration, error) { + if d.SessionTimeout == "" { + return 0, nil + } dur, err := time.ParseDuration(d.SessionTimeout) if err != nil { return 0, fmt.Errorf("parsing defaults.session_timeout %q: %w", d.SessionTimeout, err) @@ -152,7 +156,8 @@ func LoadPool(poolDir string) (*PoolConfig, error) { cfg.Defaults.Model = "sonnet" } if cfg.Defaults.SessionTimeout == "" { - cfg.Defaults.SessionTimeout = "10m" + // No default session timeout — sessions run to completion. + // Set session_timeout in pool.toml to impose a limit. } if len(cfg.Defaults.AllowedTools) == 0 { cfg.Defaults.AllowedTools = []string{"Read", "Write", "Edit", "Bash", "Grep", "Glob"} diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 836f86a..84f6037 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -132,7 +132,7 @@ project_dir = "/tmp/project" } assertEqual(t, "Defaults.Model", "sonnet", cfg.Defaults.Model) - assertEqual(t, "Defaults.SessionTimeout", "10m", cfg.Defaults.SessionTimeout) + assertEqual(t, "Defaults.SessionTimeout", "", cfg.Defaults.SessionTimeout) // no default — sessions run to completion assertSliceEqual(t, "Defaults.AllowedTools", []string{"Read", "Write", "Edit", "Bash", "Grep", "Glob"}, cfg.Defaults.AllowedTools, @@ -183,7 +183,7 @@ func TestLoadPool_EmptyFile(t *testing.T) { // All defaults should be applied assertEqual(t, "Defaults.Model", "sonnet", cfg.Defaults.Model) - assertEqual(t, "Defaults.SessionTimeout", "10m", cfg.Defaults.SessionTimeout) + assertEqual(t, "Defaults.SessionTimeout", "", cfg.Defaults.SessionTimeout) // no default — sessions run to completion assertSliceEqual(t, "Defaults.AllowedTools", []string{"Read", "Write", "Edit", "Bash", "Grep", "Glob"}, cfg.Defaults.AllowedTools, @@ -352,7 +352,7 @@ func TestDefaultsSection_ParseSessionTimeout(t *testing.T) { {"30 seconds", "30s", 30 * time.Second, false}, {"1 hour", "1h", time.Hour, false}, {"invalid", "invalid", 0, true}, - {"empty", "", 0, true}, + {"empty", "", 0, false}, // empty = no timeout, sessions run to completion } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { diff --git a/internal/config/socket.go b/internal/config/socket.go new file mode 100644 index 0000000..991278e --- /dev/null +++ b/internal/config/socket.go @@ -0,0 +1,28 @@ +package config + +import ( + "fmt" + "hash/fnv" + "os" + "path/filepath" +) + +// ResolveSockPath returns the unix socket path for a pool directory. +// Defaults to {poolDir}/daemon.sock. Falls back to a hashed path under +// os.TempDir() when the default would exceed the macOS Unix socket path +// limit (104 bytes). Both the daemon and CLI must use this to agree on +// the socket location. +func ResolveSockPath(poolDir string) string { + // Canonicalize so daemon and CLI agree on the path even when one + // uses a relative path and the other uses an absolute one. + if abs, err := filepath.Abs(poolDir); err == nil { + poolDir = abs + } + candidate := filepath.Join(poolDir, "daemon.sock") + if len(candidate) <= 100 { + return candidate + } + h := fnv.New32a() + h.Write([]byte(poolDir)) + return filepath.Join(os.TempDir(), fmt.Sprintf("ap-%x.sock", h.Sum32())) +} diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index c1a1708..55b70d9 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -8,6 +8,7 @@ package daemon import ( "context" "fmt" + "io" "log/slog" "os" "path/filepath" @@ -16,8 +17,6 @@ import ( "sync" "time" - "io" - "github.com/cameronsjo/agent-pool/internal/approval" "github.com/cameronsjo/agent-pool/internal/config" "github.com/cameronsjo/agent-pool/internal/expert" @@ -51,6 +50,12 @@ type Daemon struct { board *taskboard.Board boardPath string draining map[string]bool // re-entrancy guard for expert inbox drains + + wg sync.WaitGroup + drainTimeout time.Duration // max wait for in-flight goroutines on shutdown (default 30s) + startedAt time.Time + sockPathOver string // overrides default socket path (for tests with long TempDir paths) + events *eventBus } // Option configures a Daemon. @@ -71,6 +76,18 @@ func WithStdout(w io.Writer) Option { return func(d *Daemon) { d.stdout = w } } +// WithSocketPath overrides the default socket path ({poolDir}/daemon.sock). +// Used in tests where TempDir paths exceed the macOS Unix socket limit (104 bytes). +func WithSocketPath(path string) Option { + return func(d *Daemon) { d.sockPathOver = path } +} + +// WithDrainTimeout sets the max wait for in-flight goroutines during shutdown. +// Default is 30 seconds. Use a shorter value in tests. +func WithDrainTimeout(d time.Duration) Option { + return func(dm *Daemon) { dm.drainTimeout = d } +} + // New creates a Daemon for the given pool. func New(cfg *config.PoolConfig, poolDir string, logger *slog.Logger, opts ...Option) *Daemon { boardPath := filepath.Join(poolDir, "taskboard.json") @@ -85,13 +102,15 @@ func New(cfg *config.PoolConfig, poolDir string, logger *slog.Logger, opts ...Op } d := &Daemon{ - cfg: cfg, - poolDir: poolDir, - logger: logger, - board: board, - boardPath: boardPath, - draining: make(map[string]bool), - spawner: defaultSpawner{}, + cfg: cfg, + poolDir: poolDir, + logger: logger, + board: board, + boardPath: boardPath, + draining: make(map[string]bool), + spawner: defaultSpawner{}, + drainTimeout: 30 * time.Second, + events: newEventBus(), } for _, opt := range opts { opt(d) @@ -99,12 +118,17 @@ func New(cfg *config.PoolConfig, poolDir string, logger *slog.Logger, opts ...Op return d } -// Run starts the daemon's main loop. It blocks until ctx is cancelled. +// Run starts the daemon's main loop. It blocks until ctx is cancelled (via +// signal) or a stop command is received over the socket. func (d *Daemon) Run(ctx context.Context) error { if err := d.ensureDirs(); err != nil { return fmt.Errorf("ensuring directory structure: %w", err) } + // Create a child context so both signals and socket stop converge here. + childCtx, cancel := context.WithCancel(ctx) + defer cancel() + watcher, err := NewWatcher(d.logger) if err != nil { return fmt.Errorf("creating watcher: %w", err) @@ -137,26 +161,48 @@ func (d *Daemon) Run(ctx context.Context) error { } } + // Start socket server for CLI→daemon communication + sockPath := d.resolveSockPath() + sock, err := newSocketServer(sockPath, d, cancel) + if err != nil { + return fmt.Errorf("starting socket server: %w", err) + } + defer sock.close() + go sock.serve(childCtx) + // Start watcher goroutine - go watcher.Run(ctx) + go watcher.Run(childCtx) + d.startedAt = time.Now() d.logger.Info("Successfully started daemon", "pool", d.cfg.Pool.Name, "pool_dir", d.poolDir, "experts", len(d.cfg.Experts), + "socket", sockPath, ) // Drain pre-existing messages from before the daemon started - d.drainPostoffice(ctx) - d.drainAllInboxes(ctx) + d.drainPostoffice(childCtx) + d.drainAllInboxes(childCtx) // Main event loop for { select { - case <-ctx.Done(): - d.logger.Info("Shutting down daemon", + case <-childCtx.Done(): + d.logger.Info("Preparing to drain in-flight work", "pool", d.cfg.Pool.Name, + "drain_timeout", d.drainTimeout, ) + done := make(chan struct{}) + go func() { d.wg.Wait(); close(done) }() + select { + case <-done: + d.logger.Info("Successfully drained all in-flight work") + case <-time.After(d.drainTimeout): + d.logger.Warn("Skipping drain. Reason: timeout exceeded", + "drain_timeout", d.drainTimeout, + ) + } return nil case event, ok := <-watcher.Events(): @@ -165,9 +211,10 @@ func (d *Daemon) Run(ctx context.Context) error { } if event.Dir == postofficeDir { - d.handlePostoffice(ctx, event.Path) + d.handlePostoffice(childCtx, event.Path) } else if event.Dir == approvalsDir { - go d.handleApprovalRequest(ctx, event.Path) + d.wg.Add(1) + go func() { defer d.wg.Done(); d.handleApprovalRequest(childCtx, event.Path) }() } else { // Determine which expert this inbox belongs to expertName := d.resolveExpertName(event.Dir) @@ -176,7 +223,8 @@ func (d *Daemon) Run(ctx context.Context) error { // block postoffice routing or other experts. // The busy flag inside handleInbox prevents // concurrent spawns for the same expert. - go d.handleInbox(ctx, expertName, event.Path) + d.wg.Add(1) + go func() { defer d.wg.Done(); d.handleInbox(childCtx, expertName, event.Path) }() } else { d.logger.Warn("Received event for unknown inbox", "dir", event.Dir, @@ -221,6 +269,11 @@ func (d *Daemon) handlePostoffice(ctx context.Context, path string) { "id", routed.ID, "to", routed.To, ) + d.events.emit(Event{ + Type: EventTaskRouted, + Timestamp: time.Now(), + Data: TaskRoutedData{ID: routed.ID, From: routed.From, To: routed.To, Type: string(routed.Type)}, + }) if routed.Type == mail.TypeTask || routed.Type == mail.TypeQuestion { d.registerTask(routed) @@ -326,6 +379,11 @@ func (d *Daemon) handleCancel(msg *mail.Message, cancelPath string) { "cancel_id", msg.ID, "target_id", targetID, ) + d.events.emit(Event{ + Type: EventTaskCancelled, + Timestamp: time.Now(), + Data: TaskCancelledData{TaskID: targetID}, + }) case taskboard.StatusActive: task.CancelNote = "cancel requested while active" @@ -492,6 +550,12 @@ func (d *Daemon) processInboxMessage(ctx context.Context, expertName string, pat d.mu.Unlock() model, tools := d.resolveExpertConfig(expertName) + + d.events.emit(Event{ + Type: EventExpertSpawning, + Timestamp: time.Now(), + Data: ExpertSpawningData{Expert: expertName, TaskID: msg.ID, Model: model}, + }) projectDir := d.resolveProjectDir() expertDir := d.resolveExpertDir(expertName) @@ -536,12 +600,18 @@ func (d *Daemon) processInboxMessage(ctx context.Context, expertName string, pat timeout, parseErr := d.resolveSessionTimeout(expertName) if parseErr != nil { - d.logger.Warn("Failed to parse session timeout, using default 10m", + d.logger.Warn("Failed to parse session timeout, running without timeout", "error", parseErr, ) - timeout = 10 * time.Minute } - spawnCtx, spawnCancel := context.WithTimeout(ctx, timeout) + + var spawnCtx context.Context + var spawnCancel context.CancelFunc + if timeout > 0 { + spawnCtx, spawnCancel = context.WithTimeout(ctx, timeout) + } else { + spawnCtx, spawnCancel = context.WithCancel(ctx) + } defer spawnCancel() result, err := d.spawner.Spawn(spawnCtx, d.logger, cfg) @@ -596,6 +666,11 @@ func (d *Daemon) processInboxMessage(ctx context.Context, expertName string, pat "duration", result.Duration, "summary", result.Summary, ) + d.events.emit(Event{ + Type: EventExpertFailed, + Timestamp: time.Now(), + Data: ExpertFailedData{Expert: expertName, TaskID: result.TaskID, ExitCode: result.ExitCode}, + }) d.markTaskFailed(msg.ID, result.ExitCode) return true } @@ -617,6 +692,17 @@ func (d *Daemon) processInboxMessage(ctx context.Context, expertName string, pat "duration", result.Duration, "summary", result.Summary, ) + d.events.emit(Event{ + Type: EventExpertCompleted, + Timestamp: time.Now(), + Data: ExpertCompletedData{ + Expert: expertName, + TaskID: result.TaskID, + Duration: result.Duration.String(), + ExitCode: result.ExitCode, + Summary: result.Summary, + }, + }) return true } @@ -693,6 +779,11 @@ func (d *Daemon) markTaskCompleted(ctx context.Context, taskID string, exitCode if t, ok := d.board.Get(id); ok && !seen[t.Expert] { seen[t.Expert] = true expertsToWake = append(expertsToWake, t.Expert) + d.events.emit(Event{ + Type: EventTaskUnblocked, + Timestamp: time.Now(), + Data: TaskUnblockedData{TaskID: id, Expert: t.Expert}, + }) } } } @@ -701,7 +792,8 @@ func (d *Daemon) markTaskCompleted(ctx context.Context, taskID string, exitCode // Wake experts outside the lock — handleInbox acquires its own lock for _, expert := range expertsToWake { - go d.handleInbox(ctx, expert, "") + d.wg.Add(1) + go func(e string) { defer d.wg.Done(); d.handleInbox(ctx, e, "") }(expert) } } @@ -726,10 +818,12 @@ func (d *Daemon) markTaskFailed(taskID string, exitCode int) { // when the daemon starts. Each drains in its own goroutine via handleInbox. func (d *Daemon) drainAllInboxes(ctx context.Context) { // Drain architect inbox - go d.handleInbox(ctx, "architect", "") + d.wg.Add(1) + go func() { defer d.wg.Done(); d.handleInbox(ctx, "architect", "") }() for name := range d.cfg.Experts { - go d.handleInbox(ctx, name, "") + d.wg.Add(1) + go func(n string) { defer d.wg.Done(); d.handleInbox(ctx, n, "") }(name) } } @@ -932,6 +1026,43 @@ func (d *Daemon) handleApprovalRequest(ctx context.Context, path string) { ) } +// Status returns live daemon state for the socket status method. +func (d *Daemon) Status() map[string]any { + d.mu.Lock() + defer d.mu.Unlock() + + experts := make([]string, 0, len(d.cfg.Experts)) + for name := range d.cfg.Experts { + experts = append(experts, name) + } + sort.Strings(experts) + + counts := make(map[string]int) + var activeTasks []map[string]string + for _, task := range d.board.Tasks { + counts[string(task.Status)]++ + if task.Status == taskboard.StatusActive { + entry := map[string]string{ + "id": task.ID, + "expert": task.Expert, + } + if task.StartedAt != nil { + entry["started"] = time.Since(*task.StartedAt).Truncate(time.Second).String() + } + activeTasks = append(activeTasks, entry) + } + } + + return map[string]any{ + "pool": d.cfg.Pool.Name, + "state": "running", + "uptime": time.Since(d.startedAt).Truncate(time.Second).String(), + "experts": experts, + "task_counts": counts, + "active_tasks": activeTasks, + } +} + // resolveSessionTimeout returns the session timeout for the given role or expert. // Built-in roles with their own timeout config use that; otherwise falls back to defaults. func (d *Daemon) resolveSessionTimeout(name string) (time.Duration, error) { @@ -945,6 +1076,16 @@ func (d *Daemon) resolveSessionTimeout(name string) (time.Duration, error) { return d.cfg.Defaults.ParseSessionTimeout() } +// resolveSockPath returns the unix socket path for CLI→daemon communication. +// Uses the override if set, otherwise delegates to config.ResolveSockPath +// (shared with the CLI to ensure both sides agree on the path). +func (d *Daemon) resolveSockPath() string { + if d.sockPathOver != "" { + return d.sockPathOver + } + return config.ResolveSockPath(d.poolDir) +} + // resolveExpertDir returns the state directory for an expert or built-in role. func (d *Daemon) resolveExpertDir(name string) string { return mail.ResolveExpertDir(d.poolDir, name) diff --git a/internal/daemon/daemon_test.go b/internal/daemon/daemon_test.go index 2185a26..9d96aec 100644 --- a/internal/daemon/daemon_test.go +++ b/internal/daemon/daemon_test.go @@ -74,10 +74,12 @@ func writePoolConfig(t *testing.T, poolDir, toml string) *config.PoolConfig { // startTestDaemon creates a daemon with the given config and spawner, starts it // in a background goroutine, and waits for it to be ready. Returns a cancel // function and error channel. Call shutdownDaemon(t, cancel, errCh) to stop. -func startTestDaemon(t *testing.T, cfg *config.PoolConfig, poolDir string, spawner *fakeSpawner) (context.CancelFunc, <-chan error) { +func startTestDaemon(t *testing.T, cfg *config.PoolConfig, poolDir string, spawner *fakeSpawner, opts ...daemon.Option) (context.CancelFunc, <-chan error) { t.Helper() logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) - d := daemon.New(cfg, poolDir, logger, daemon.WithSpawner(spawner)) + + allOpts := append([]daemon.Option{daemon.WithSpawner(spawner)}, opts...) + d := daemon.New(cfg, poolDir, logger, allOpts...) ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) diff --git a/internal/daemon/events.go b/internal/daemon/events.go new file mode 100644 index 0000000..a284aed --- /dev/null +++ b/internal/daemon/events.go @@ -0,0 +1,128 @@ +package daemon + +import ( + "sync" + "time" +) + +// EventType identifies the kind of daemon event. +type EventType string + +const ( + EventTaskRouted EventType = "task.routed" + EventExpertSpawning EventType = "expert.spawning" + EventExpertCompleted EventType = "expert.completed" + EventExpertFailed EventType = "expert.failed" + EventTaskCancelled EventType = "task.cancelled" + EventTaskUnblocked EventType = "task.unblocked" +) + +// Event is a structured daemon event emitted at state transitions. +type Event struct { + Type EventType `json:"type"` + Timestamp time.Time `json:"timestamp"` + Data any `json:"data"` +} + +// Per-event data types. + +type TaskRoutedData struct { + ID string `json:"id"` + From string `json:"from"` + To string `json:"to"` + Type string `json:"type"` +} + +type ExpertSpawningData struct { + Expert string `json:"expert"` + TaskID string `json:"task_id"` + Model string `json:"model"` +} + +type ExpertCompletedData struct { + Expert string `json:"expert"` + TaskID string `json:"task_id"` + Duration string `json:"duration"` + ExitCode int `json:"exit_code"` + Summary string `json:"summary"` +} + +type ExpertFailedData struct { + Expert string `json:"expert"` + TaskID string `json:"task_id"` + ExitCode int `json:"exit_code"` +} + +type TaskCancelledData struct { + TaskID string `json:"task_id"` + CancelNote string `json:"cancel_note,omitempty"` +} + +type TaskUnblockedData struct { + TaskID string `json:"task_id"` + Expert string `json:"expert"` +} + +// 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 + +// eventBus fans out events to registered subscribers. +type eventBus struct { + mu sync.RWMutex + nextID int + subs map[int]chan Event +} + +func newEventBus() *eventBus { + return &eventBus{ + subs: make(map[int]chan Event), + } +} + +// subscribe returns a subscriber ID and a buffered event channel. +func (b *eventBus) subscribe() (int, <-chan Event) { + b.mu.Lock() + defer b.mu.Unlock() + + id := b.nextID + b.nextID++ + + ch := make(chan Event, EventBufSize) + b.subs[id] = ch + return id, ch +} + +// unsubscribe removes a subscriber and closes its channel. +func (b *eventBus) unsubscribe(id int) { + b.mu.Lock() + defer b.mu.Unlock() + + if ch, ok := b.subs[id]; ok { + close(ch) + delete(b.subs, id) + } +} + +// emit sends an event to all subscribers. Non-blocking: slow subscribers +// that can't keep up will miss events (their channels are buffered at 64). +func (b *eventBus) emit(e Event) { + b.mu.RLock() + defer b.mu.RUnlock() + + for _, ch := range b.subs { + select { + case ch <- e: + default: + // Subscriber too slow, drop event + } + } +} + +// EventBus wraps the internal eventBus for test access. +type EventBus struct{ *eventBus } + +func NewEventBusForTest() *EventBus { return &EventBus{newEventBus()} } +func (b *EventBus) Subscribe() (int, <-chan Event) { return b.eventBus.subscribe() } +func (b *EventBus) Unsubscribe(id int) { b.eventBus.unsubscribe(id) } +func (b *EventBus) Emit(e Event) { b.eventBus.emit(e) } diff --git a/internal/daemon/events_test.go b/internal/daemon/events_test.go new file mode 100644 index 0000000..c8544ba --- /dev/null +++ b/internal/daemon/events_test.go @@ -0,0 +1,101 @@ +// Test plan for events.go: +// +// EventBus: +// - TestEventBus_SubscribeReceives: subscriber gets emitted events +// - TestEventBus_Multiple: multiple subscribers each get events +// - TestEventBus_UnsubscribeCleans: unsubscribed channel is closed +// - TestEventBus_SlowDrops: slow subscriber misses events (buffer overflow) +package daemon_test + +import ( + "testing" + "time" + + "github.com/cameronsjo/agent-pool/internal/daemon" +) + +func TestEventBus_SubscribeReceives(t *testing.T) { + bus := daemon.NewEventBusForTest() + + _, ch := bus.Subscribe() + + bus.Emit(daemon.Event{ + Type: daemon.EventTaskRouted, + Timestamp: time.Now(), + Data: daemon.TaskRoutedData{ID: "t1", From: "a", To: "b", Type: "task"}, + }) + + select { + case e := <-ch: + if e.Type != daemon.EventTaskRouted { + t.Errorf("type = %v, want task.routed", e.Type) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for event") + } +} + +func TestEventBus_Multiple(t *testing.T) { + bus := daemon.NewEventBusForTest() + + _, ch1 := bus.Subscribe() + _, ch2 := bus.Subscribe() + + bus.Emit(daemon.Event{ + Type: daemon.EventExpertCompleted, + Timestamp: time.Now(), + }) + + for _, ch := range []<-chan daemon.Event{ch1, ch2} { + select { + case e := <-ch: + if e.Type != daemon.EventExpertCompleted { + t.Errorf("type = %v, want expert.completed", e.Type) + } + case <-time.After(time.Second): + t.Fatal("subscriber did not receive event") + } + } +} + +func TestEventBus_UnsubscribeCleans(t *testing.T) { + bus := daemon.NewEventBusForTest() + + id, ch := bus.Subscribe() + bus.Unsubscribe(id) + + // Channel should be closed + _, ok := <-ch + if ok { + t.Error("channel should be closed after unsubscribe") + } +} + +func TestEventBus_SlowDrops(t *testing.T) { + bus := daemon.NewEventBusForTest() + + _, ch := bus.Subscribe() + + // Fill the buffer plus extra + for i := 0; i < daemon.EventBufSize+16; i++ { + bus.Emit(daemon.Event{ + Type: daemon.EventExpertSpawning, + Timestamp: time.Now(), + }) + } + + // Should have exactly buffer capacity events + count := 0 + for { + select { + case <-ch: + count++ + default: + goto done + } + } +done: + if count != daemon.EventBufSize { + t.Errorf("received %d events, want %d (buffer capacity)", count, daemon.EventBufSize) + } +} diff --git a/internal/daemon/socket.go b/internal/daemon/socket.go new file mode 100644 index 0000000..b37b62b --- /dev/null +++ b/internal/daemon/socket.go @@ -0,0 +1,183 @@ +package daemon + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "log/slog" + "net" + "os" + "time" +) + +// socketRequest is the NDJSON request format for CLI→daemon communication. +type socketRequest struct { + Method string `json:"method"` +} + +// socketResponse is the NDJSON response format for daemon→CLI communication. +type socketResponse struct { + Status string `json:"status"` // "ok" or "error" + Data any `json:"data,omitempty"` // method-specific payload + Message string `json:"message,omitempty"` // error description +} + +// socketServer listens on a Unix domain socket for CLI commands. +type socketServer struct { + listener net.Listener + logger *slog.Logger + daemon *Daemon + cancel context.CancelFunc // cancels the daemon's child context (for stop) + sockPath string +} + +// newSocketServer creates a socket server at the given path. Removes any stale +// socket file from a previous crash before listening. +func newSocketServer(sockPath string, d *Daemon, cancel context.CancelFunc) (*socketServer, error) { + // Remove stale socket — if the daemon crashed, the file lingers. + if err := os.Remove(sockPath); err != nil && !os.IsNotExist(err) { + return nil, fmt.Errorf("removing stale socket: %w", err) + } + + listener, err := net.Listen("unix", sockPath) + if err != nil { + return nil, fmt.Errorf("listening on %s: %w", sockPath, err) + } + + return &socketServer{ + listener: listener, + logger: d.logger, + daemon: d, + cancel: cancel, + sockPath: sockPath, + }, nil +} + +// serve accepts connections until ctx is cancelled. Each connection is handled +// in its own goroutine. +func (s *socketServer) serve(ctx context.Context) { + go func() { + <-ctx.Done() + s.listener.Close() + }() + + for { + conn, err := s.listener.Accept() + if err != nil { + // Expected when listener is closed during shutdown. + if ctx.Err() != nil { + return + } + s.logger.Warn("Failed to accept socket connection", "error", err) + continue + } + go s.handleConn(ctx, conn) + } +} + +// handleConn reads a single NDJSON request, dispatches by method, writes the +// response, and closes the connection. +func (s *socketServer) handleConn(ctx context.Context, conn net.Conn) { + defer conn.Close() + + conn.SetReadDeadline(time.Now().Add(5 * time.Second)) + + scanner := bufio.NewScanner(conn) + if !scanner.Scan() { + return + } + + var req socketRequest + if err := json.Unmarshal(scanner.Bytes(), &req); err != nil { + s.logger.Warn("Failed to parse socket request", + "error", err, + ) + s.writeResponse(conn, socketResponse{ + Status: "error", + Message: "invalid request: " + err.Error(), + }) + return + } + + switch req.Method { + case "stop": + s.logger.Info("Received stop command via socket") + s.writeResponse(conn, socketResponse{ + Status: "ok", + Data: map[string]string{"message": "shutting down"}, + }) + s.cancel() + + case "status": + s.writeResponse(conn, socketResponse{ + Status: "ok", + Data: s.daemon.Status(), + }) + + case "subscribe": + s.handleSubscribe(ctx, conn) + + default: + s.logger.Warn("Received unknown socket method", + "method", req.Method, + ) + s.writeResponse(conn, socketResponse{ + Status: "error", + Message: fmt.Sprintf("unknown method: %s", req.Method), + }) + } +} + +// handleSubscribe streams events to a client as NDJSON until the client +// disconnects or the context is cancelled. The connection is NOT closed +// by handleConn — this method manages the full lifecycle. +func (s *socketServer) handleSubscribe(ctx context.Context, conn net.Conn) { + id, ch := s.daemon.events.subscribe() + defer s.daemon.events.unsubscribe(id) + + s.logger.Debug("Preparing to stream events to subscriber", + "subscriber_id", id, + ) + + // Send ack + s.writeResponse(conn, socketResponse{ + Status: "ok", + Data: map[string]string{"message": "subscribed"}, + }) + + // Clear deadline for streaming + conn.SetDeadline(time.Time{}) + + enc := json.NewEncoder(conn) + for { + select { + case event, ok := <-ch: + if !ok { + return + } + if err := enc.Encode(event); err != nil { + s.logger.Debug("Subscriber disconnected", + "subscriber_id", id, + ) + return + } + case <-ctx.Done(): + return + } + } +} + +// writeResponse encodes a response as a single JSON line. +func (s *socketServer) writeResponse(conn net.Conn, resp socketResponse) { + conn.SetWriteDeadline(time.Now().Add(5 * time.Second)) + if err := json.NewEncoder(conn).Encode(resp); err != nil { + s.logger.Debug("Failed to write socket response", "error", err) + } +} + +// close shuts down the listener and removes the socket file. +func (s *socketServer) close() error { + s.listener.Close() + return os.Remove(s.sockPath) +} diff --git a/internal/daemon/socket_test.go b/internal/daemon/socket_test.go new file mode 100644 index 0000000..bd2b0b7 --- /dev/null +++ b/internal/daemon/socket_test.go @@ -0,0 +1,395 @@ +// Test plan for socket.go: +// +// Socket lifecycle: +// - TestSocket_Stop: connect, send stop, verify ok response, daemon exits +// - TestSocket_Unknown: send unknown, get error, daemon stays up +// - TestSocket_StaleCleanup: stale socket file doesn't block start +// - TestSocket_RemovedOnShutdown: file cleaned up after exit +// - TestSocket_Status: verify status response has pool name and experts +// +// Graceful drain: +// - TestDaemon_DrainWaits: gated spawn blocks, cancel context, verify daemon +// waits for in-flight work before exiting +package daemon_test + +import ( + "bufio" + "context" + "encoding/json" + "log/slog" + "net" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/cameronsjo/agent-pool/internal/daemon" + "github.com/cameronsjo/agent-pool/internal/expert" +) + +// shortTempDir creates a temp directory with a short path, avoiding macOS Unix +// socket path length limits (104 bytes). t.TempDir() generates paths too long +// for socket files when test names are verbose. +func shortTempDir(t *testing.T) string { + t.Helper() + dir, err := os.MkdirTemp("/tmp", "ap-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + return dir +} + +// connectSocket dials the daemon's unix socket with a short timeout. +// poolDir should be a short path created by shortTempDir when daemon.sock +// is at {poolDir}/daemon.sock. +func connectSocket(t *testing.T, poolDir string) net.Conn { + t.Helper() + sockPath := filepath.Join(poolDir, "daemon.sock") + + // The socket may not be ready immediately after daemon start. + var conn net.Conn + var err error + for i := 0; i < 20; i++ { + conn, err = net.DialTimeout("unix", sockPath, 500*time.Millisecond) + if err == nil { + return conn + } + time.Sleep(50 * time.Millisecond) + } + t.Fatalf("failed to connect to socket after retries: %v", err) + return nil +} + +// sendSocketRequest writes a JSON request and reads the JSON response. +func sendSocketRequest(t *testing.T, conn net.Conn, method string) map[string]any { + t.Helper() + + conn.SetDeadline(time.Now().Add(5 * time.Second)) + + req := map[string]string{"method": method} + if err := json.NewEncoder(conn).Encode(req); err != nil { + t.Fatalf("sending request: %v", err) + } + + scanner := bufio.NewScanner(conn) + if !scanner.Scan() { + t.Fatalf("no response: %v", scanner.Err()) + } + + var resp map[string]any + if err := json.Unmarshal(scanner.Bytes(), &resp); err != nil { + t.Fatalf("parsing response: %v", err) + } + return resp +} + +func TestSocket_Stop(t *testing.T) { + poolDir := shortTempDir(t) + + cfg := writePoolConfig(t, poolDir, `[pool] +name = "socket-test" +project_dir = "PROJECT_DIR" + +[experts.auth] +`) + + _, errCh := startTestDaemon(t, cfg, poolDir, &fakeSpawner{}) + + conn := connectSocket(t, poolDir) + resp := sendSocketRequest(t, conn, "stop") + conn.Close() + + if resp["status"] != "ok" { + t.Errorf("stop status = %v, want ok", resp["status"]) + } + + // Daemon should exit after stop + select { + case err := <-errCh: + if err != nil { + t.Errorf("daemon returned error: %v", err) + } + case <-time.After(3 * time.Second): + t.Error("daemon did not shut down after stop") + } +} + +func TestSocket_Unknown(t *testing.T) { + poolDir := shortTempDir(t) + + cfg := writePoolConfig(t, poolDir, `[pool] +name = "socket-test" +project_dir = "PROJECT_DIR" + +[experts.auth] +`) + + cancel, errCh := startTestDaemon(t, cfg, poolDir, &fakeSpawner{}) + + conn := connectSocket(t, poolDir) + resp := sendSocketRequest(t, conn, "bogus") + conn.Close() + + if resp["status"] != "error" { + t.Errorf("status = %v, want error", resp["status"]) + } + msg, _ := resp["message"].(string) + if msg != "unknown method: bogus" { + t.Errorf("message = %q, want 'unknown method: bogus'", msg) + } + + // Daemon should still be running + shutdownDaemon(t, cancel, errCh) +} + +func TestSocket_StaleCleanup(t *testing.T) { + poolDir := shortTempDir(t) + + // Create a stale socket file before starting the daemon + sockPath := filepath.Join(poolDir, "daemon.sock") + if err := os.WriteFile(sockPath, []byte("stale"), 0o644); err != nil { + t.Fatal(err) + } + + cfg := writePoolConfig(t, poolDir, `[pool] +name = "socket-test" +project_dir = "PROJECT_DIR" + +[experts.auth] +`) + + cancel, errCh := startTestDaemon(t, cfg, poolDir, &fakeSpawner{}) + + // Daemon should start despite stale socket — verify by connecting + conn := connectSocket(t, poolDir) + resp := sendSocketRequest(t, conn, "status") + conn.Close() + + if resp["status"] != "ok" { + t.Errorf("status = %v, want ok", resp["status"]) + } + + shutdownDaemon(t, cancel, errCh) +} + +func TestSocket_RemovedOnShutdown(t *testing.T) { + poolDir := shortTempDir(t) + + cfg := writePoolConfig(t, poolDir, `[pool] +name = "socket-test" +project_dir = "PROJECT_DIR" + +[experts.auth] +`) + + cancel, errCh := startTestDaemon(t, cfg, poolDir, &fakeSpawner{}) + + // Verify socket exists while running + sockPath := filepath.Join(poolDir, "daemon.sock") + if _, err := os.Stat(sockPath); os.IsNotExist(err) { + t.Fatal("socket file should exist while daemon is running") + } + + shutdownDaemon(t, cancel, errCh) + + // Socket file should be removed after shutdown + if _, err := os.Stat(sockPath); !os.IsNotExist(err) { + t.Error("socket file should be removed after shutdown") + } +} + +func TestSocket_Status(t *testing.T) { + poolDir := shortTempDir(t) + + cfg := writePoolConfig(t, poolDir, `[pool] +name = "status-test" +project_dir = "PROJECT_DIR" + +[experts.auth] +[experts.frontend] +`) + + cancel, errCh := startTestDaemon(t, cfg, poolDir, &fakeSpawner{}) + + conn := connectSocket(t, poolDir) + resp := sendSocketRequest(t, conn, "status") + conn.Close() + + if resp["status"] != "ok" { + t.Fatalf("status = %v, want ok", resp["status"]) + } + + data, ok := resp["data"].(map[string]any) + if !ok { + t.Fatalf("data is not a map: %T", resp["data"]) + } + + if data["pool"] != "status-test" { + t.Errorf("pool = %v, want status-test", data["pool"]) + } + + if data["state"] != "running" { + t.Errorf("state = %v, want running", data["state"]) + } + + experts, ok := data["experts"].([]any) + if !ok { + t.Fatalf("experts is not a list: %T", data["experts"]) + } + if len(experts) != 2 { + t.Errorf("experts count = %d, want 2", len(experts)) + } + + shutdownDaemon(t, cancel, errCh) +} + +func TestSocket_Subscribe(t *testing.T) { + poolDir := shortTempDir(t) + + cfg := writePoolConfig(t, poolDir, `[pool] +name = "watch-test" +project_dir = "PROJECT_DIR" + +[experts.auth] +`) + + cancel, errCh := startTestDaemon(t, cfg, poolDir, &fakeSpawner{}) + + // Connect and subscribe + conn := connectSocket(t, poolDir) + defer conn.Close() + + conn.SetDeadline(time.Now().Add(5 * time.Second)) + json.NewEncoder(conn).Encode(map[string]string{"method": "subscribe"}) + + scanner := bufio.NewScanner(conn) + + // Read ack + if !scanner.Scan() { + t.Fatal("no ack") + } + var ack map[string]any + json.Unmarshal(scanner.Bytes(), &ack) + if ack["status"] != "ok" { + t.Fatalf("ack status = %v", ack["status"]) + } + + // Send a task — should produce a task.routed event + writeMessage(t, filepath.Join(poolDir, "postoffice"), + "task-watch-001", "architect", "auth") + + // Read events until we see task.routed + conn.SetDeadline(time.Now().Add(5 * time.Second)) + var foundRouted bool + for scanner.Scan() { + var event map[string]any + json.Unmarshal(scanner.Bytes(), &event) + if event["type"] == "task.routed" { + data, _ := event["data"].(map[string]any) + if data["id"] == "task-watch-001" { + foundRouted = true + break + } + } + } + if !foundRouted { + t.Error("did not receive task.routed event for task-watch-001") + } + + shutdownDaemon(t, cancel, errCh) +} + +// slowSpawner blocks on a channel regardless of context cancellation. +// This tests that the drain waits for in-flight work even when the spawn +// doesn't respond to context cancellation immediately (simulating a real +// claude process that takes time to clean up). +type slowSpawner struct { + mu sync.Mutex + attempts int + blocker chan struct{} +} + +func (s *slowSpawner) Spawn(_ context.Context, _ *slog.Logger, cfg *expert.SpawnConfig) (*expert.Result, error) { + s.mu.Lock() + s.attempts++ + s.mu.Unlock() + + <-s.blocker // blocks until closed, ignores context + + return &expert.Result{ + TaskID: cfg.TaskMessage.ID, + ExitCode: 0, + Output: []byte(`{"type":"result","result":"done"}`), + Summary: "Task completed", + Duration: 100 * time.Millisecond, + }, nil +} + +func (s *slowSpawner) getAttempts() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.attempts +} + +func TestDaemon_DrainWaits(t *testing.T) { + poolDir := shortTempDir(t) + + cfg := writePoolConfig(t, poolDir, `[pool] +name = "drain-test" +project_dir = "PROJECT_DIR" + +[experts.auth] +`) + + blocker := make(chan struct{}) + spawner := &slowSpawner{blocker: blocker} + + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) + d := daemon.New(cfg, poolDir, logger, + daemon.WithSpawner(spawner), + daemon.WithDrainTimeout(5*time.Second)) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + + errCh := make(chan error, 1) + go func() { errCh <- d.Run(ctx) }() + time.Sleep(500 * time.Millisecond) + + // Send a task that will block in the spawner + writeMessage(t, filepath.Join(poolDir, "postoffice"), + "task-drain-001", "architect", "auth") + + // Wait for the spawn to be attempted + deadline := time.Now().Add(3 * time.Second) + for spawner.getAttempts() == 0 && time.Now().Before(deadline) { + time.Sleep(50 * time.Millisecond) + } + if spawner.getAttempts() == 0 { + t.Fatal("spawn was never attempted") + } + + // Cancel context — daemon should start drain but wait for in-flight spawn + cancel() + + // Daemon should NOT have exited yet (spawn is still blocked) + select { + case <-errCh: + t.Fatal("daemon exited before in-flight work completed") + case <-time.After(300 * time.Millisecond): + // Good — daemon is waiting for drain + } + + // Release the blocker — daemon should now complete + close(blocker) + + select { + case err := <-errCh: + if err != nil { + t.Errorf("daemon returned error: %v", err) + } + case <-time.After(5 * time.Second): + t.Error("daemon did not exit after drain completed") + } +} diff --git a/internal/mcp/architect_tools.go b/internal/mcp/architect_tools.go index af818ea..765c6bf 100644 --- a/internal/mcp/architect_tools.go +++ b/internal/mcp/architect_tools.go @@ -27,7 +27,7 @@ func RegisterArchitectTools(srv *server.MCPServer, cfg *ServerConfig) { store := contract.NewStore(cfg.PoolDir).WithLogger(cfg.Logger) srv.AddTool( - mcp.NewTool("pool_define_contract", + mcp.NewTool("define_contract", mcp.WithDescription("Define a new contract between experts. Creates a versioned interface specification."), mcp.WithString("id", mcp.Required(), mcp.Description("Contract ID (must be filename-safe)")), mcp.WithString("between", mcp.Required(), mcp.Description("Comma-separated list of expert names (at least 2)")), @@ -37,7 +37,7 @@ func RegisterArchitectTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_send_task", + mcp.NewTool("send_task", mcp.WithDescription("Delegate a task to an expert via the postoffice. References contracts the expert must follow."), mcp.WithString("to", mcp.Required(), mcp.Description("Recipient expert name")), mcp.WithString("body", mcp.Required(), mcp.Description("Task description (markdown)")), @@ -50,7 +50,7 @@ func RegisterArchitectTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_verify_result", + mcp.NewTool("verify_result", mcp.WithDescription("Log a verification result for a task against a contract specification."), mcp.WithString("task_id", mcp.Required(), mcp.Description("Task ID being verified")), mcp.WithString("contract_id", mcp.Required(), mcp.Description("Contract ID verified against")), @@ -61,7 +61,7 @@ func RegisterArchitectTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_amend_contract", + mcp.NewTool("amend_contract", mcp.WithDescription("Amend an existing contract. Increments version and notifies all parties."), mcp.WithString("id", mcp.Required(), mcp.Description("Contract ID to amend")), mcp.WithString("body", mcp.Required(), mcp.Description("New contract body (markdown)")), diff --git a/internal/mcp/architect_tools_test.go b/internal/mcp/architect_tools_test.go index 581aadf..3ab41da 100644 --- a/internal/mcp/architect_tools_test.go +++ b/internal/mcp/architect_tools_test.go @@ -3,22 +3,22 @@ // RegisterArchitectTools (Classification: INTEGRATION) // [x] Happy: all 4 architect tools + 6 expert tools registered (TestArchitectTools_Registration) // -// pool_define_contract (Classification: FILESYSTEM I/O) +//define_contract (Classification: FILESYSTEM I/O) // [x] Happy: creates contract file and index (TestDefineContract_Happy) // [x] Error: missing params (TestDefineContract_MissingParams) // [x] Error: fewer than 2 between parties (TestDefineContract_TooFewBetween) // -// pool_send_task (Classification: FILESYSTEM I/O) +//send_task (Classification: FILESYSTEM I/O) // [x] Happy: message appears in postoffice (TestSendTask_Happy) // [x] Error: missing params (TestSendTask_MissingParams) // [x] Error: path traversal ID (TestSendTask_PathTraversal) // -// pool_verify_result (Classification: FILESYSTEM I/O) +//verify_result (Classification: FILESYSTEM I/O) // [x] Happy: verification log created (TestVerifyResult_Happy) // [x] Error: invalid status (TestVerifyResult_InvalidStatus) // [x] Error: contract not found (TestVerifyResult_ContractNotFound) // -// pool_amend_contract (Classification: FILESYSTEM I/O) +//amend_contract (Classification: FILESYSTEM I/O) // [x] Happy: version incremented + notify messages (TestAmendContract_Happy) // [x] Error: contract not found (TestAmendContract_NotFound) // @@ -70,12 +70,12 @@ func TestArchitectTools_Registration(t *testing.T) { expected := []string{ // Architect tools - "pool_define_contract", "pool_send_task", - "pool_verify_result", "pool_amend_contract", + "define_contract", "send_task", + "verify_result", "amend_contract", // Expert tools (inherited) - "pool_read_state", "pool_update_state", - "pool_append_error", "pool_send_response", - "pool_recall", "pool_search_index", + "read_state", "update_state", + "append_error", "send_response", + "recall", "search_index", } for _, name := range expected { if !tools[name] { @@ -88,7 +88,7 @@ func TestDefineContract_Happy(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_define_contract", map[string]any{ + result := callTool(t, srv, "define_contract", map[string]any{ "id": "contract-001", "between": "auth, frontend", "body": "## Token Exchange\n\nSpec goes here.", @@ -116,7 +116,7 @@ func TestDefineContract_MissingParams(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_define_contract", map[string]any{ + result := callTool(t, srv, "define_contract", map[string]any{ "between": "auth, frontend", "body": "spec", }) @@ -129,7 +129,7 @@ func TestDefineContract_TooFewBetween(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_define_contract", map[string]any{ + result := callTool(t, srv, "define_contract", map[string]any{ "id": "c1", "between": "only-one", "body": "spec", @@ -144,7 +144,7 @@ func TestSendTask_Happy(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_send_task", map[string]any{ + result := callTool(t, srv, "send_task", map[string]any{ "to": "auth", "body": "Implement the token endpoint", "id": "task-001", @@ -184,7 +184,7 @@ func TestSendTask_MissingParams(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_send_task", map[string]any{ + result := callTool(t, srv, "send_task", map[string]any{ "to": "auth", "body": "do something", }) @@ -197,7 +197,7 @@ func TestSendTask_PathTraversal(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_send_task", map[string]any{ + result := callTool(t, srv, "send_task", map[string]any{ "to": "auth", "body": "do something", "id": "../escape", @@ -213,13 +213,13 @@ func TestVerifyResult_Happy(t *testing.T) { srv := buildArchitectTestServer(t, poolDir) // Create a contract first - callTool(t, srv, "pool_define_contract", map[string]any{ + callTool(t, srv, "define_contract", map[string]any{ "id": "contract-001", "between": "auth, frontend", "body": "spec", }) - result := callTool(t, srv, "pool_verify_result", map[string]any{ + result := callTool(t, srv, "verify_result", map[string]any{ "task_id": "task-001", "contract_id": "contract-001", "status": "pass", @@ -242,7 +242,7 @@ func TestVerifyResult_InvalidStatus(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_verify_result", map[string]any{ + result := callTool(t, srv, "verify_result", map[string]any{ "task_id": "task-001", "contract_id": "contract-001", "status": "unknown", @@ -258,7 +258,7 @@ func TestVerifyResult_ContractNotFound(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_verify_result", map[string]any{ + result := callTool(t, srv, "verify_result", map[string]any{ "task_id": "task-001", "contract_id": "nonexistent", "status": "pass", @@ -275,13 +275,13 @@ func TestAmendContract_Happy(t *testing.T) { srv := buildArchitectTestServer(t, poolDir) // Create initial contract - callTool(t, srv, "pool_define_contract", map[string]any{ + callTool(t, srv, "define_contract", map[string]any{ "id": "contract-001", "between": "auth, frontend", "body": "v1 spec", }) - result := callTool(t, srv, "pool_amend_contract", map[string]any{ + result := callTool(t, srv, "amend_contract", map[string]any{ "id": "contract-001", "body": "## Updated spec v2\n\nNew content.", }) @@ -332,7 +332,7 @@ func TestAmendContract_NotFound(t *testing.T) { poolDir := setupArchitectPool(t) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_amend_contract", map[string]any{ + result := callTool(t, srv, "amend_contract", map[string]any{ "id": "nonexistent", "body": "new body", }) @@ -352,7 +352,7 @@ func TestSendTask_ApprovalNoneMode(t *testing.T) { // approval_mode = "none" (set in setupArchitectPool via buildArchitectTestServer) srv := buildArchitectTestServer(t, poolDir) - result := callTool(t, srv, "pool_send_task", map[string]any{ + result := callTool(t, srv, "send_task", map[string]any{ "to": "auth", "body": "do something", "id": "task-none-001", @@ -402,7 +402,7 @@ func TestSendTask_ApprovalRequired(t *testing.T) { // Call pool_send_task in a goroutine (it blocks on approval) resultCh := make(chan *mcp.CallToolResult, 1) go func() { - r := callTool(t, srv, "pool_send_task", map[string]any{ + r := callTool(t, srv, "send_task", map[string]any{ "to": "auth", "body": "implement token endpoint", "id": "task-approval-001", @@ -477,7 +477,7 @@ func TestSendTask_ApprovalRejected(t *testing.T) { resultCh := make(chan *mcp.CallToolResult, 1) go func() { - r := callTool(t, srv, "pool_send_task", map[string]any{ + r := callTool(t, srv, "send_task", map[string]any{ "to": "auth", "body": "implement auth", "id": "task-rejected-001", diff --git a/internal/mcp/concierge_tools.go b/internal/mcp/concierge_tools.go index 0e3ae45..edef68e 100644 --- a/internal/mcp/concierge_tools.go +++ b/internal/mcp/concierge_tools.go @@ -32,8 +32,8 @@ func RegisterConciergeTools(srv *server.MCPServer, cfg *ServerConfig) { } srv.AddTool( - mcp.NewTool("pool_ask_expert", - mcp.WithDescription("Send a question to an expert and wait for the response. BLOCKS until the expert completes or times out. For non-blocking dispatch, use pool_dispatch instead."), + mcp.NewTool("ask_expert", + mcp.WithDescription("Send a question to an expert and wait for the response. BLOCKS until the expert completes or times out. For non-blocking dispatch, use dispatch instead."), mcp.WithString("expert", mcp.Required(), mcp.Description("Expert name to ask (e.g., 'auth', 'frontend')")), mcp.WithString("question", mcp.Required(), mcp.Description("Question body (markdown)")), ), @@ -41,8 +41,8 @@ func RegisterConciergeTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_dispatch", - mcp.WithDescription("Send a question or task to an expert without waiting. Returns a task ID immediately. Use pool_collect to retrieve results later."), + mcp.NewTool("dispatch", + mcp.WithDescription("Send a question or task to an expert without waiting. Returns a task ID immediately. Use collect to retrieve results later."), mcp.WithString("expert", mcp.Required(), mcp.Description("Expert name (e.g., 'auth', 'frontend')")), mcp.WithString("message", mcp.Required(), mcp.Description("Question or task body (markdown)")), mcp.WithString("type", mcp.Description("Message type: 'question' (default) or 'task'")), @@ -51,7 +51,7 @@ func RegisterConciergeTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_collect", + mcp.NewTool("collect", mcp.WithDescription("Check dispatched tasks and return results for any that have completed. Non-blocking — returns immediately with current status."), mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), mcp.WithString("task_ids", mcp.Required(), mcp.Description("Comma-separated task IDs to check")), @@ -60,8 +60,8 @@ func RegisterConciergeTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_submit_plan", - mcp.WithDescription("Submit a plan to the architect for review and decomposition. Returns immediately with task ID — use pool_check_status to track."), + mcp.NewTool("submit_plan", + mcp.WithDescription("Submit a plan to the architect for review and decomposition. Returns immediately with task ID — use check_status to track."), mcp.WithString("plan", mcp.Required(), mcp.Description("Plan body (markdown)")), mcp.WithString("contracts", mcp.Description("Comma-separated contract IDs to reference (optional)")), ), @@ -69,7 +69,7 @@ func RegisterConciergeTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_check_status", + mcp.NewTool("check_status", mcp.WithDescription("Query the taskboard for task status. Returns all non-terminal tasks if no filters given."), mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), mcp.WithString("task_id", mcp.Description("Specific task ID to look up (optional)")), @@ -80,7 +80,7 @@ func RegisterConciergeTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_list_experts", + mcp.NewTool("list_experts", mcp.WithDescription("List available experts in the pool (pool-scoped and shared)."), mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), ), @@ -209,7 +209,7 @@ func handleDispatch(cfg *ServerConfig) server.ToolHandlerFunc { } } -// collectResult represents one task's status in a pool_collect response. +// collectResult represents one task's status in a collect response. type collectResult struct { ID string `json:"id"` Expert string `json:"expert"` @@ -510,7 +510,8 @@ func handleListExperts(cfg *ServerConfig) server.ToolHandlerFunc { sort.Strings(poolExperts) result := map[string][]string{ - "pool_experts": poolExperts, + "experts": poolExperts, + "pool_experts": poolExperts, // compatibility alias (deprecated) "shared_experts": poolCfg.Shared.Include, } diff --git a/internal/mcp/concierge_tools_test.go b/internal/mcp/concierge_tools_test.go index 7ace779..8296ddc 100644 --- a/internal/mcp/concierge_tools_test.go +++ b/internal/mcp/concierge_tools_test.go @@ -3,26 +3,26 @@ // RegisterConciergeTools (Classification: INTEGRATION) // [x] Happy: all 4 concierge tools + 6 expert tools registered (TestConciergeTools_Registration) // -// pool_ask_expert (Classification: FILESYSTEM I/O + CONCURRENCY) +//ask_expert (Classification: FILESYSTEM I/O + CONCURRENCY) // [x] Happy: question dispatched, polls taskboard, returns result (TestAskExpert_Happy) // [x] Error: missing params (TestAskExpert_MissingParams) // [x] Error: expert task fails (TestAskExpert_ExpertFails) // [x] Error: task cancelled with note (TestAskExpert_TaskCancelled) // [x] Error: context timeout (TestAskExpert_Timeout) // -// pool_submit_plan (Classification: FILESYSTEM I/O) +//submit_plan (Classification: FILESYSTEM I/O) // [x] Happy: plan message in postoffice, returns task ID (TestSubmitPlan_Happy) // [x] Happy: plan with contracts (TestSubmitPlan_WithContracts) // [x] Error: missing plan param (TestSubmitPlan_MissingPlan) // -// pool_check_status (Classification: FILESYSTEM I/O) +//check_status (Classification: FILESYSTEM I/O) // [x] Happy: single task lookup (TestCheckStatus_SingleTask) // [x] Happy: filter by expert (TestCheckStatus_FilterByExpert) // [x] Happy: filter by status (TestCheckStatus_FilterByStatus) // [x] Happy: default excludes terminal (TestCheckStatus_DefaultExcludesTerminal) // [x] Error: task not found (TestCheckStatus_NotFound) // -// pool_list_experts (Classification: FILESYSTEM I/O) +//list_experts (Classification: FILESYSTEM I/O) // [x] Happy: lists pool and shared experts (TestListExperts_Happy) // [x] Error: missing pool.toml (TestListExperts_MissingConfig) @@ -78,12 +78,12 @@ func TestConciergeTools_Registration(t *testing.T) { expected := []string{ // Concierge tools - "pool_ask_expert", "pool_submit_plan", - "pool_check_status", "pool_list_experts", + "ask_expert", "submit_plan", + "check_status", "list_experts", // Expert tools (inherited) - "pool_read_state", "pool_update_state", - "pool_append_error", "pool_send_response", - "pool_recall", "pool_search_index", + "read_state", "update_state", + "append_error", "send_response", + "recall", "search_index", } for _, name := range expected { if !tools[name] { @@ -92,7 +92,7 @@ func TestConciergeTools_Registration(t *testing.T) { } // Should NOT have architect tools - architectOnly := []string{"pool_define_contract", "pool_send_task", "pool_verify_result", "pool_amend_contract"} + architectOnly := []string{"define_contract", "send_task", "verify_result", "amend_contract"} for _, name := range architectOnly { if tools[name] { t.Errorf("unexpected architect tool registered for concierge: %s", name) @@ -156,7 +156,7 @@ func TestAskExpert_Happy(t *testing.T) { board.Save(filepath.Join(poolDir, "taskboard.json")) }() - result := callTool(t, srv, "pool_ask_expert", map[string]any{ + result := callTool(t, srv, "ask_expert", map[string]any{ "expert": "auth", "question": "How does token refresh work?", }) @@ -178,7 +178,7 @@ func TestAskExpert_MissingParams(t *testing.T) { srv := buildConciergeTestServer(t, poolDir) t.Run("missing_expert", func(t *testing.T) { - result := callTool(t, srv, "pool_ask_expert", map[string]any{ + result := callTool(t, srv, "ask_expert", map[string]any{ "question": "How does auth work?", }) if !result.IsError { @@ -187,7 +187,7 @@ func TestAskExpert_MissingParams(t *testing.T) { }) t.Run("missing_question", func(t *testing.T) { - result := callTool(t, srv, "pool_ask_expert", map[string]any{ + result := callTool(t, srv, "ask_expert", map[string]any{ "expert": "auth", }) if !result.IsError { @@ -240,7 +240,7 @@ func TestAskExpert_ExpertFails(t *testing.T) { board.Save(filepath.Join(poolDir, "taskboard.json")) }() - result := callTool(t, srv, "pool_ask_expert", map[string]any{ + result := callTool(t, srv, "ask_expert", map[string]any{ "expert": "auth", "question": "How does auth work?", }) @@ -297,7 +297,7 @@ func TestAskExpert_TaskCancelled(t *testing.T) { board.Save(filepath.Join(poolDir, "taskboard.json")) }() - result := callTool(t, srv, "pool_ask_expert", map[string]any{ + result := callTool(t, srv, "ask_expert", map[string]any{ "expert": "auth", "question": "How does auth work?", }) @@ -323,7 +323,7 @@ func TestAskExpert_Timeout(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond) defer cancel() - result := callToolWithContext(t, ctx, srv, "pool_ask_expert", map[string]any{ + result := callToolWithContext(t, ctx, srv, "ask_expert", map[string]any{ "expert": "auth", "question": "This will time out", }) @@ -343,7 +343,7 @@ func TestSubmitPlan_Happy(t *testing.T) { poolDir := setupConciergePool(t) srv := buildConciergeTestServer(t, poolDir) - result := callTool(t, srv, "pool_submit_plan", map[string]any{ + result := callTool(t, srv, "submit_plan", map[string]any{ "plan": "## OAuth Login Flow\n\nImplement Google OAuth with PKCE.", }) @@ -380,7 +380,7 @@ func TestSubmitPlan_WithContracts(t *testing.T) { poolDir := setupConciergePool(t) srv := buildConciergeTestServer(t, poolDir) - result := callTool(t, srv, "pool_submit_plan", map[string]any{ + result := callTool(t, srv, "submit_plan", map[string]any{ "plan": "Implement the auth flow", "contracts": "auth-api-v1, session-store-v2", }) @@ -412,7 +412,7 @@ func TestSubmitPlan_MissingPlan(t *testing.T) { poolDir := setupConciergePool(t) srv := buildConciergeTestServer(t, poolDir) - result := callTool(t, srv, "pool_submit_plan", map[string]any{}) + result := callTool(t, srv, "submit_plan", map[string]any{}) if !result.IsError { t.Error("expected error for missing plan") } @@ -443,7 +443,7 @@ func TestCheckStatus_SingleTask(t *testing.T) { }, }) - result := callTool(t, srv, "pool_check_status", map[string]any{ + result := callTool(t, srv, "check_status", map[string]any{ "task_id": "task-001", }) @@ -467,7 +467,7 @@ func TestCheckStatus_FilterByExpert(t *testing.T) { "task-003": {ID: "task-003", Status: taskboard.StatusPending, Expert: "auth", CreatedAt: now}, }) - result := callTool(t, srv, "pool_check_status", map[string]any{ + result := callTool(t, srv, "check_status", map[string]any{ "expert": "auth", }) @@ -493,7 +493,7 @@ func TestCheckStatus_FilterByStatus(t *testing.T) { "task-002": {ID: "task-002", Status: taskboard.StatusCompleted, Expert: "auth", CreatedAt: now}, }) - result := callTool(t, srv, "pool_check_status", map[string]any{ + result := callTool(t, srv, "check_status", map[string]any{ "status": "completed", }) @@ -517,7 +517,7 @@ func TestCheckStatus_DefaultExcludesTerminal(t *testing.T) { "task-003": {ID: "task-003", Status: taskboard.StatusFailed, Expert: "frontend", CreatedAt: now}, }) - result := callTool(t, srv, "pool_check_status", map[string]any{}) + result := callTool(t, srv, "check_status", map[string]any{}) text := resultText(t, result) if !strings.Contains(text, "task-001") { @@ -537,7 +537,7 @@ func TestCheckStatus_NotFound(t *testing.T) { setupTaskboard(t, poolDir, map[string]*taskboard.Task{}) - result := callTool(t, srv, "pool_check_status", map[string]any{ + result := callTool(t, srv, "check_status", map[string]any{ "task_id": "nonexistent", }) if !result.IsError { @@ -569,7 +569,7 @@ model = "sonnet" t.Fatalf("writing pool.toml: %v", err) } - result := callTool(t, srv, "pool_list_experts", map[string]any{}) + result := callTool(t, srv, "list_experts", map[string]any{}) text := resultText(t, result) @@ -578,13 +578,13 @@ model = "sonnet" t.Fatalf("parsing result JSON: %v", err) } - poolExperts := parsed["pool_experts"] + poolExperts := parsed["experts"] if len(poolExperts) != 2 { - t.Fatalf("pool_experts count = %d, want 2", len(poolExperts)) + t.Fatalf("experts count = %d, want 2", len(poolExperts)) } // Sorted alphabetically if poolExperts[0] != "auth" || poolExperts[1] != "frontend" { - t.Errorf("pool_experts = %v, want [auth, frontend]", poolExperts) + t.Errorf("experts = %v, want [auth, frontend]", poolExperts) } shared := parsed["shared_experts"] @@ -598,7 +598,7 @@ func TestListExperts_MissingConfig(t *testing.T) { srv := buildConciergeTestServer(t, poolDir) // No pool.toml written — LoadPool should fail - result := callTool(t, srv, "pool_list_experts", map[string]any{}) + result := callTool(t, srv, "list_experts", map[string]any{}) if !result.IsError { t.Error("expected error for missing pool.toml") } diff --git a/internal/mcp/config.go b/internal/mcp/config.go index 7d57604..892bf73 100644 --- a/internal/mcp/config.go +++ b/internal/mcp/config.go @@ -11,12 +11,12 @@ import ( // Claude Code requires MCP tools to be explicitly allowed in headless mode. // Format: mcp____ var ExpertToolNames = []string{ - "mcp__agent-pool__pool_read_state", - "mcp__agent-pool__pool_update_state", - "mcp__agent-pool__pool_append_error", - "mcp__agent-pool__pool_send_response", - "mcp__agent-pool__pool_recall", - "mcp__agent-pool__pool_search_index", + "mcp__agent-pool__read_state", + "mcp__agent-pool__update_state", + "mcp__agent-pool__append_error", + "mcp__agent-pool__send_response", + "mcp__agent-pool__recall", + "mcp__agent-pool__search_index", } // MCPConfig is the JSON structure claude expects for --mcp-config. diff --git a/internal/mcp/tools.go b/internal/mcp/tools.go index 076344d..bfe4cad 100644 --- a/internal/mcp/tools.go +++ b/internal/mcp/tools.go @@ -23,7 +23,7 @@ func RegisterExpertTools(srv *server.MCPServer, cfg *ServerConfig) { expertDir := mail.ResolveExpertDir(cfg.PoolDir, cfg.ExpertName) srv.AddTool( - mcp.NewTool("pool_read_state", + mcp.NewTool("read_state", mcp.WithDescription("Read current expert state files (identity.md, state.md, errors.md)"), mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), ), @@ -31,7 +31,7 @@ func RegisterExpertTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_update_state", + mcp.NewTool("update_state", mcp.WithDescription("Update the expert's working memory (state.md). Content must be non-empty and under 50KB."), mcp.WithString("content", mcp.Required(), mcp.Description("New state.md content")), ), @@ -39,7 +39,7 @@ func RegisterExpertTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_append_error", + mcp.NewTool("append_error", mcp.WithDescription("Append a structured error entry to the expert's error log (errors.md). Each entry is timestamped."), mcp.WithString("entry", mcp.Required(), mcp.Description("Error description to append")), ), @@ -47,7 +47,7 @@ func RegisterExpertTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_send_response", + mcp.NewTool("send_response", mcp.WithDescription("Send a response message to another agent via the postoffice."), mcp.WithString("to", mcp.Required(), mcp.Description("Recipient agent name")), mcp.WithString("body", mcp.Required(), mcp.Description("Response body (markdown)")), @@ -57,7 +57,7 @@ func RegisterExpertTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_recall", + mcp.NewTool("recall", mcp.WithDescription("Read a prior task log by its task ID."), mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), mcp.WithString("task_id", mcp.Required(), mcp.Description("Task ID to recall")), @@ -66,7 +66,7 @@ func RegisterExpertTools(srv *server.MCPServer, cfg *ServerConfig) { ) srv.AddTool( - mcp.NewTool("pool_search_index", + mcp.NewTool("search_index", mcp.WithDescription("Search the task log index for relevant prior tasks. Case-insensitive substring match."), mcp.WithToolAnnotation(mcp.ToolAnnotation{ReadOnlyHint: boolPtr(true)}), mcp.WithString("query", mcp.Required(), mcp.Description("Search query")), diff --git a/internal/mcp/tools_test.go b/internal/mcp/tools_test.go index 1297281..0803154 100644 --- a/internal/mcp/tools_test.go +++ b/internal/mcp/tools_test.go @@ -3,27 +3,27 @@ // Each handler is tested by constructing a JSON-RPC tools/call message, // sending it through HandleMessage, and inspecting the response. // -// pool_read_state: +//read_state: // - All state files present → returns JSON with all three fields // - No state files → returns JSON with empty strings // -// pool_update_state: +//update_state: // - Happy path → state.md written // - Empty content → error result // -// pool_append_error: +//append_error: // - Happy path → errors.md contains entry // - Empty entry → error result // -// pool_send_response: +//send_response: // - Happy path → message file appears in postoffice, round-trips through Parse // - Missing required params → error result // -// pool_recall: +//recall: // - Happy path → returns log content // - Missing log → error result // -// pool_search_index: +//search_index: // - Happy path → returns matching rows // - No matches → returns "no matching tasks found" @@ -56,7 +56,7 @@ func TestReadState_AllPresent(t *testing.T) { os.WriteFile(filepath.Join(expertDir, "errors.md"), []byte("JWT panics"), 0o644) srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_read_state", nil) + result := callTool(t, srv, "read_state", nil) text := resultText(t, result) var data map[string]string @@ -78,7 +78,7 @@ func TestReadState_AllPresent(t *testing.T) { func TestReadState_NoFiles(t *testing.T) { poolDir, _ := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_read_state", nil) + result := callTool(t, srv, "read_state", nil) text := resultText(t, result) var data map[string]string @@ -97,7 +97,7 @@ func TestUpdateState_HappyPath(t *testing.T) { poolDir, expertDir := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_update_state", map[string]any{ + result := callTool(t, srv, "update_state", map[string]any{ "content": "Updated working memory", }) @@ -115,7 +115,7 @@ func TestUpdateState_EmptyContent(t *testing.T) { poolDir, _ := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_update_state", map[string]any{ + result := callTool(t, srv, "update_state", map[string]any{ "content": "", }) @@ -130,7 +130,7 @@ func TestAppendError_HappyPath(t *testing.T) { poolDir, expertDir := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_append_error", map[string]any{ + result := callTool(t, srv, "append_error", map[string]any{ "entry": "Connection timeout to database", }) @@ -148,7 +148,7 @@ func TestAppendError_EmptyEntry(t *testing.T) { poolDir, _ := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_append_error", map[string]any{ + result := callTool(t, srv, "append_error", map[string]any{ "entry": "", }) @@ -163,7 +163,7 @@ func TestSendResponse_HappyPath(t *testing.T) { poolDir, _ := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_send_response", map[string]any{ + result := callTool(t, srv, "send_response", map[string]any{ "to": "architect", "body": "Token endpoint is complete.", "id": "resp-001", @@ -198,7 +198,7 @@ func TestSendResponse_MissingTo(t *testing.T) { poolDir, _ := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_send_response", map[string]any{ + result := callTool(t, srv, "send_response", map[string]any{ "body": "response body", "id": "resp-002", }) @@ -212,7 +212,7 @@ func TestSendResponse_PathTraversalID(t *testing.T) { poolDir, _ := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_send_response", map[string]any{ + result := callTool(t, srv, "send_response", map[string]any{ "to": "architect", "body": "response body", "id": "../../etc/evil", @@ -234,7 +234,7 @@ func TestRecall_HappyPath(t *testing.T) { ) srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_recall", map[string]any{ + result := callTool(t, srv, "recall", map[string]any{ "task_id": "task-042", }) @@ -252,7 +252,7 @@ func TestRecall_MissingLog(t *testing.T) { poolDir, _ := setupExpertPool(t, "auth") srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_recall", map[string]any{ + result := callTool(t, srv, "recall", map[string]any{ "task_id": "nonexistent", }) @@ -273,7 +273,7 @@ func TestSearchIndex_HappyPath(t *testing.T) { os.WriteFile(filepath.Join(expertDir, "logs", "index.md"), []byte(index), 0o644) srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_search_index", map[string]any{ + result := callTool(t, srv, "search_index", map[string]any{ "query": "OAuth", }) @@ -296,7 +296,7 @@ func TestSearchIndex_NoMatches(t *testing.T) { os.WriteFile(filepath.Join(expertDir, "logs", "index.md"), []byte(index), 0o644) srv := buildMCPTestServer(t, poolDir, "auth", "") - result := callTool(t, srv, "pool_search_index", map[string]any{ + result := callTool(t, srv, "search_index", map[string]any{ "query": "nonexistent", }) diff --git a/plugin/concierge-identity.md b/plugin/concierge-identity.md index df19409..087c94f 100644 --- a/plugin/concierge-identity.md +++ b/plugin/concierge-identity.md @@ -13,15 +13,19 @@ You are the concierge — the user-facing coordinator in an expert pool. ## Tools -- `pool_ask_expert` — dispatch a question to an expert and wait for response -- `pool_submit_plan` — send a plan to the architect for decomposition -- `pool_check_status` — query the taskboard for task progress -- `pool_list_experts` — discover available experts +| Tool | Behavior | +|------|----------| +| `dispatch` | Send question/task to expert, return task ID. **Non-blocking.** | +| `collect` | Check task IDs, return results for completed ones. **Non-blocking.** | +| `ask_expert` | Send + wait. **Blocks.** Only for single quick questions. | +| `submit_plan` | Send plan to architect for decomposition | +| `check_status` | Query taskboard for task progress | +| `list_experts` | Discover available experts | ## Principles -1. Know who knows what. Use `pool_list_experts` to understand the pool. +1. Know who knows what. Use `list_experts` to understand the pool. 2. Ask sharp questions. Tailor each question to the expert's domain. -3. Don't bottleneck. Dispatch to multiple experts in parallel when possible. +3. Don't bottleneck. Use `dispatch` + `collect` for multi-expert work. 4. Track everything. The taskboard is your source of truth. 5. Be honest about failure. If an expert fails or times out, say so. diff --git a/plugin/skills/pool-ask.md b/plugin/skills/pool-ask.md index 40675ed..88b0841 100644 --- a/plugin/skills/pool-ask.md +++ b/plugin/skills/pool-ask.md @@ -1,47 +1,51 @@ --- name: pool-ask -description: Ask domain experts a question and get a synthesized answer (read path) +description: Use when the user asks a question that requires domain expertise from one or more experts in the pool (read path) --- # Pool Ask — Read Path -You are the concierge. The user has a question that requires domain expertise. -Your job is to dispatch the question to the right experts, wait for their -responses, and synthesize a unified answer. +You are the concierge. The user has a question that needs expert knowledge. +Dispatch to the right experts, collect their responses, and synthesize. ## Workflow ### 1. Discover experts -Call `pool_list_experts` to see who's available. Present the list to the user -if they haven't specified which experts to ask. +Call `list_experts` to see who's available. Present the list if the user +hasn't specified who to ask. ### 2. Dispatch questions -For each relevant expert, call `pool_ask_expert` with: -- `expert`: the expert's name -- `question`: the question, tailored to that expert's domain +Use `dispatch` (non-blocking) for each relevant expert. Tailor each +question to the expert's specialty — don't send the same generic +question to everyone. Dispatch to multiple experts in parallel when the +question spans domains. -Dispatch to multiple experts in parallel when the question spans domains. -Tailor each question to the expert's specialty — don't send the same generic -question to everyone. +For a single quick question to one expert, `ask_expert` (blocking) is +acceptable. -### 3. Synthesize +### 3. Collect results -Once all experts have responded: +Call `collect` with the returned task IDs to check what's done. +Re-check for pending ones after a short wait. + +### 4. Synthesize + +Once experts have responded: - Identify common themes and agreements -- Surface any contradictions between expert answers -- Combine into a coherent narrative that answers the user's original question +- Surface contradictions between expert answers +- Combine into a coherent narrative answering the original question - Cite which expert provided which insight -If an expert fails or times out, note it and work with the responses you have. +If an expert fails or times out, note it and work with what you have. ## Example User: "How does our auth flow work end-to-end?" -You would: -1. Ask the `auth` expert about token lifecycle and session management -2. Ask the `frontend` expert about login UI and token storage -3. Ask the `backend` expert about middleware and route protection -4. Synthesize into a single end-to-end narrative +1. `dispatch` to `auth` — token lifecycle and session management +2. `dispatch` to `frontend` — login UI and token storage +3. `dispatch` to `backend` — middleware and route protection +4. `collect` all three task IDs +5. Synthesize into a single end-to-end narrative diff --git a/plugin/skills/pool-build.md b/plugin/skills/pool-build.md index 720609a..7ce36c9 100644 --- a/plugin/skills/pool-build.md +++ b/plugin/skills/pool-build.md @@ -1,30 +1,28 @@ --- name: pool-build -description: Build a feature by gathering expert input, planning, and delegating to the architect (write path) +description: Use when the user wants to build a feature, implement a change, or execute a multi-step plan requiring expert coordination (write path) --- # Pool Build — Write Path -You are the concierge. The user wants to build something. Your job is to -gather expert input, synthesize it into a plan, and submit it to the architect -for decomposition and execution. +You are the concierge. The user wants to build something. Gather expert +input, draft a plan, submit to the architect for decomposition. ## Workflow ### 1. Understand the request -Clarify what the user wants to build. Ask questions if the scope is ambiguous. -Identify which domains are involved. +Clarify what the user wants to build. Ask questions if the scope is +ambiguous. Identify which domains are involved. ### 2. Gather expert input (optional) -If the feature spans multiple domains, use `pool_ask_expert` to get domain -input from relevant experts. This is the same read-path flow as pool-ask -but focused on gathering implementation considerations. +If the feature spans multiple domains, use `dispatch` + `collect` to +get implementation considerations from relevant experts. ### 3. Draft the plan -Synthesize the user's requirements and expert input into a plan that includes: +Synthesize requirements and expert input into: - **Goal**: What's being built and why - **Scope**: What's in and out - **Approach**: High-level technical direction @@ -33,30 +31,23 @@ Synthesize the user's requirements and expert input into a plan that includes: ### 4. Submit to architect -Call `pool_submit_plan` with: +Call `submit_plan` with: - `plan`: the plan body (markdown) -- `contracts`: any existing contract IDs that apply (optional) +- `contracts`: existing contract IDs that apply (optional) -This returns a task ID. The architect will review the plan, define contracts, -and dispatch tasks to experts. +The architect will review, define contracts, and dispatch tasks to experts. ### 5. Track progress -Use `pool_check_status` to monitor: -- The plan task itself (is the architect working on it?) -- Sub-tasks dispatched by the architect -- Any blocked or failed tasks - -Report progress to the user at natural milestones. +Use `check_status` to monitor the plan task, sub-tasks dispatched by +the architect, and any blocked or failed tasks. Report progress to the +user at natural milestones. ## Example User: "Build an OAuth login flow" -You would: -1. Ask the `auth` expert about supported providers and token patterns -2. Ask the `frontend` expert about current login UX and routing -3. Ask the `backend` expert about session middleware -4. Draft a plan combining these inputs -5. Submit to architect -6. Track as experts execute their tasks +1. `dispatch` to `auth`, `frontend`, `backend` for domain input +2. `collect` results, draft plan combining insights +3. `submit_plan` to architect +4. `check_status` as experts execute their tasks diff --git a/plugin/skills/pool-status.md b/plugin/skills/pool-status.md index 7a2a9bb..16da698 100644 --- a/plugin/skills/pool-status.md +++ b/plugin/skills/pool-status.md @@ -1,34 +1,32 @@ --- name: pool-status -description: Check task and pool status from the taskboard +description: Use when the user asks about task progress, expert activity, or wants to know what's blocked or failed in the pool --- # Pool Status -You are the concierge. The user wants to know the status of in-flight work. +You are the concierge. The user wants to know the state of in-flight work. ## Workflow ### 1. Query the taskboard -Call `pool_check_status` with appropriate filters: -- No filters: shows all active (non-terminal) tasks -- `task_id`: look up a specific task -- `expert`: show all tasks for one expert -- `status`: filter by status (pending, blocked, active, completed, failed, cancelled) +Call `check_status` with appropriate filters: +- No filters: all active (non-terminal) tasks +- `task_id`: specific task lookup +- `expert`: all tasks for one expert +- `status`: filter by pending, blocked, active, completed, failed, cancelled ### 2. Format for the user -Present the results clearly: - Group by status (active first, then pending/blocked, then completed) - Highlight blocked tasks and what they're waiting on - Highlight failed tasks with exit codes -- Show timing information (created, started, completed) +- Show timing (created, started, completed) ### 3. Suggest next actions -Based on the status: - **All complete**: summarize results, ask if user needs anything else -- **Some blocked**: explain dependencies, suggest checking blocker tasks +- **Some blocked**: explain dependencies, suggest checking blockers - **Some failed**: suggest investigating failed expert logs -- **In progress**: report ETA based on task count and progress +- **In progress**: report based on task count and progress diff --git a/scripts/com.agent-pool.daemon.plist b/scripts/com.agent-pool.daemon.plist new file mode 100644 index 0000000..4e7bb79 --- /dev/null +++ b/scripts/com.agent-pool.daemon.plist @@ -0,0 +1,33 @@ + + + + + Label + com.agent-pool.daemon + + ProgramArguments + + AGENT_POOL_BINARY + start + POOL_DIR + + + RunAtLoad + + + KeepAlive + + + StandardOutPath + POOL_DIR/launchd-stdout.log + + StandardErrorPath + POOL_DIR/launchd-stderr.log + + WorkingDirectory + POOL_DIR + + ExitTimeOut + 35 + +