From adc584746600f6f2d36ef21c789562a620b3e513 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 13:50:41 -0500 Subject: [PATCH 01/13] feat: add unix domain socket and stop command (v0.6 phase 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add CLI→daemon communication via Unix domain socket at {poolDir}/daemon.sock. The socket uses NDJSON protocol with method dispatch (stop, status, subscribe). - New socketServer in internal/daemon/socket.go with accept loop and per-connection handling - daemon.Run() creates child context shared with socket — stop RPC cancels it - `agent-pool stop` CLI command connects to socket and sends shutdown request - `agent-pool status` CLI command queries live daemon state (pool, experts, task counts) - Auto-fallback to /tmp/ hashed path when poolDir exceeds macOS 104-byte socket limit - 5 integration tests covering stop, unknown method, stale cleanup, shutdown cleanup, status - Includes v0.6 implementation plan in docs/plans/ Co-Authored-By: Claude Opus 4.6 (1M context) --- cmd/agent-pool/main.go | 105 +++++++++- docs/plans/2026-04-05-daemon-lifecycle.md | 213 +++++++++++++++++++ internal/daemon/daemon.go | 87 +++++++- internal/daemon/daemon_test.go | 6 +- internal/daemon/socket.go | 144 +++++++++++++ internal/daemon/socket_test.go | 236 ++++++++++++++++++++++ 6 files changed, 778 insertions(+), 13 deletions(-) create mode 100644 docs/plans/2026-04-05-daemon-lifecycle.md create mode 100644 internal/daemon/socket.go create mode 100644 internal/daemon/socket_test.go diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index 638ea54..7cbf1ac 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -1,14 +1,18 @@ package main import ( + "bufio" "context" + "encoding/json" "fmt" "io" "log/slog" + "net" "os" "os/signal" "path/filepath" "syscall" + "time" "github.com/cameronsjo/agent-pool/internal/config" "github.com/cameronsjo/agent-pool/internal/daemon" @@ -25,6 +29,10 @@ func main() { switch os.Args[1] { case "start": cmdStart() + case "stop": + cmdStop() + case "status": + cmdStatus() case "mcp": cmdMCP() case "flush": @@ -32,7 +40,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: @@ -93,6 +101,97 @@ 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(filepath.Join(poolDir, "daemon.sock"), "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(filepath.Join(poolDir, "daemon.sock"), "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) + } + + data, _ := json.MarshalIndent(resp.Data, "", " ") + fmt.Println(string(data)) +} + +// 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 +328,8 @@ 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 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 +344,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/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/internal/daemon/daemon.go b/internal/daemon/daemon.go index c1a1708..09c3163 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -8,6 +8,8 @@ package daemon import ( "context" "fmt" + "hash/fnv" + "io" "log/slog" "os" "path/filepath" @@ -16,8 +18,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 +51,9 @@ type Daemon struct { board *taskboard.Board boardPath string draining map[string]bool // re-entrancy guard for expert inbox drains + + startedAt time.Time + sockPathOver string // overrides default socket path (for tests with long TempDir paths) } // Option configures a Daemon. @@ -71,6 +74,12 @@ 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 } +} + // 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") @@ -99,12 +108,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,23 +151,34 @@ 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(): + case <-childCtx.Done(): d.logger.Info("Shutting down daemon", "pool", d.cfg.Pool.Name, ) @@ -165,9 +190,9 @@ 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) + go d.handleApprovalRequest(childCtx, event.Path) } else { // Determine which expert this inbox belongs to expertName := d.resolveExpertName(event.Dir) @@ -176,7 +201,7 @@ 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) + go d.handleInbox(childCtx, expertName, event.Path) } else { d.logger.Warn("Received event for unknown inbox", "dir", event.Dir, @@ -932,6 +957,31 @@ 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) + for _, task := range d.board.Tasks { + counts[string(task.Status)]++ + } + + 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, + } +} + // 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 +995,23 @@ 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 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). +func (d *Daemon) resolveSockPath() string { + if d.sockPathOver != "" { + return d.sockPathOver + } + candidate := filepath.Join(d.poolDir, "daemon.sock") + if len(candidate) <= 100 { + return candidate + } + h := fnv.New32a() + h.Write([]byte(d.poolDir)) + return filepath.Join(os.TempDir(), fmt.Sprintf("ap-%x.sock", h.Sum32())) +} + // 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/socket.go b/internal/daemon/socket.go new file mode 100644 index 0000000..1f2abd4 --- /dev/null +++ b/internal/daemon/socket.go @@ -0,0 +1,144 @@ +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("Socket accept error", "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.writeResponse(conn, socketResponse{ + Status: "error", + Message: "invalid request: " + err.Error(), + }) + return + } + + switch req.Method { + case "stop": + s.writeResponse(conn, socketResponse{ + Status: "ok", + Data: map[string]string{"message": "shutting down"}, + }) + s.cancel() + + case "status": + data := s.daemon.Status() + s.writeResponse(conn, socketResponse{ + Status: "ok", + Data: data, + }) + + case "subscribe": + s.handleSubscribe(ctx, conn) + + default: + s.writeResponse(conn, socketResponse{ + Status: "error", + Message: fmt.Sprintf("unknown method: %s", req.Method), + }) + } +} + +// handleSubscribe is a placeholder for Phase 4 event streaming. +func (s *socketServer) handleSubscribe(_ context.Context, conn net.Conn) { + s.writeResponse(conn, socketResponse{ + Status: "error", + Message: "subscribe not implemented", + }) +} + +// 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)) + json.NewEncoder(conn).Encode(resp) +} + +// 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..72d466b --- /dev/null +++ b/internal/daemon/socket_test.go @@ -0,0 +1,236 @@ +// 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 +package daemon_test + +import ( + "bufio" + "encoding/json" + "net" + "os" + "path/filepath" + "testing" + "time" +) + +// 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) +} From 7624495820562dffcc80c4e3b125d2cf1406c0a6 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 13:55:22 -0500 Subject: [PATCH 02/13] feat: add graceful drain with WaitGroup and double-signal (v0.6 phase 2) Track all goroutine dispatch sites with sync.WaitGroup so the daemon waits for in-flight work before exiting. 5 dispatch sites wrapped: handleApprovalRequest, handleInbox (event loop), wake experts, drainAllInboxes (architect + per-expert). - 30s drain timeout with configurable WithDrainTimeout option - Double-signal handler: first SIGTERM/SIGINT triggers graceful drain, second forces immediate exit via os.Exit(1) - Integration test with slowSpawner verifies daemon waits for blocked spawn before completing shutdown Co-Authored-By: Claude Opus 4.6 (1M context) --- cmd/agent-pool/main.go | 11 ++++ internal/daemon/daemon.go | 49 +++++++++++----- internal/daemon/socket_test.go | 103 +++++++++++++++++++++++++++++++++ 3 files changed, 150 insertions(+), 13 deletions(-) diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index 7cbf1ac..70e98be 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -94,6 +94,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) diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 09c3163..f4fd1c8 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -52,6 +52,8 @@ type Daemon struct { 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) } @@ -80,6 +82,12 @@ 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") @@ -94,13 +102,14 @@ 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, } for _, opt := range opts { opt(d) @@ -179,9 +188,18 @@ func (d *Daemon) Run(ctx context.Context) error { for { select { case <-childCtx.Done(): - d.logger.Info("Shutting down daemon", + d.logger.Info("Shutting down daemon, draining 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("All in-flight work completed") + case <-time.After(d.drainTimeout): + d.logger.Warn("Drain timeout exceeded, forcing shutdown") + } return nil case event, ok := <-watcher.Events(): @@ -192,7 +210,8 @@ func (d *Daemon) Run(ctx context.Context) error { if event.Dir == postofficeDir { d.handlePostoffice(childCtx, event.Path) } else if event.Dir == approvalsDir { - go d.handleApprovalRequest(childCtx, 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) @@ -201,7 +220,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(childCtx, 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, @@ -726,7 +746,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) } } @@ -751,10 +772,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) } } diff --git a/internal/daemon/socket_test.go b/internal/daemon/socket_test.go index 72d466b..328b0c5 100644 --- a/internal/daemon/socket_test.go +++ b/internal/daemon/socket_test.go @@ -6,16 +6,26 @@ // - 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 @@ -234,3 +244,96 @@ project_dir = "PROJECT_DIR" 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") + } +} From 031530f686c196d18a04ed767ac304e6756084e3 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 13:57:01 -0500 Subject: [PATCH 03/13] feat: add status command with pretty-print output (v0.6 phase 3) Enhance the Status() method to include active task details (ID, expert, duration). The CLI status command renders pool name, state, uptime, expert list, task count table, and active task list in a human-readable format instead of raw JSON. Co-Authored-By: Claude Opus 4.6 (1M context) --- cmd/agent-pool/main.go | 58 +++++++++++++++++++++++++++++++++++++-- internal/daemon/daemon.go | 22 +++++++++++---- 2 files changed, 73 insertions(+), 7 deletions(-) diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index 70e98be..2f9f055 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -11,6 +11,7 @@ import ( "os" "os/signal" "path/filepath" + "strings" "syscall" "time" @@ -161,8 +162,61 @@ func cmdStatus() { os.Exit(1) } - data, _ := json.MarshalIndent(resp.Data, "", " ") - fmt.Println(string(data)) + 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 + json.Unmarshal(v, &s) + 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) + } + } + } } // socketResponse mirrors the daemon's response type for CLI deserialization. diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index f4fd1c8..88ae131 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -992,16 +992,28 @@ func (d *Daemon) Status() map[string]any { 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, + "pool": d.cfg.Pool.Name, + "state": "running", + "uptime": time.Since(d.startedAt).Truncate(time.Second).String(), + "experts": experts, + "task_counts": counts, + "active_tasks": activeTasks, } } From 1966bea981e3a10096500b631f51bf756e979276 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:01:17 -0500 Subject: [PATCH 04/13] feat: add event bus and watch command (v0.6 phase 4) Add channel-based event bus for daemon state transitions. Events are emitted at 6 existing log points: task.routed, expert.spawning, expert.completed, expert.failed, task.cancelled, task.unblocked. - eventBus with fan-out to subscribers via buffered channels (cap 64) - Non-blocking emit prevents slow subscribers from blocking the daemon - Socket subscribe method streams events as NDJSON - `agent-pool watch` CLI renders events with ANSI colors - 4 unit tests for event bus, 1 integration test for subscribe streaming Co-Authored-By: Claude Opus 4.6 (1M context) --- cmd/agent-pool/main.go | 153 +++++++++++++++++++++++++++++++++ internal/daemon/daemon.go | 39 +++++++++ internal/daemon/events.go | 124 ++++++++++++++++++++++++++ internal/daemon/events_test.go | 101 ++++++++++++++++++++++ internal/daemon/socket.go | 32 ++++++- internal/daemon/socket_test.go | 56 ++++++++++++ 6 files changed, 501 insertions(+), 4 deletions(-) create mode 100644 internal/daemon/events.go create mode 100644 internal/daemon/events_test.go diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index 2f9f055..c6509e8 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -34,6 +34,8 @@ func main() { cmdStop() case "status": cmdStatus() + case "watch": + cmdWatch() case "mcp": cmdMCP() case "flush": @@ -219,6 +221,156 @@ func cmdStatus() { } } +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 := filepath.Join(poolDir, "daemon.sock") + 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() { + 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) + } +} + // socketResponse mirrors the daemon's response type for CLI deserialization. type socketResponse struct { Status string `json:"status"` @@ -395,6 +547,7 @@ 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 diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 88ae131..e1030f6 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -56,6 +56,7 @@ type Daemon struct { 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. @@ -110,6 +111,7 @@ func New(cfg *config.PoolConfig, poolDir string, logger *slog.Logger, opts ...Op draining: make(map[string]bool), spawner: defaultSpawner{}, drainTimeout: 30 * time.Second, + events: newEventBus(), } for _, opt := range opts { opt(d) @@ -266,6 +268,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) @@ -371,6 +378,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" @@ -537,6 +549,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) @@ -641,6 +659,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 } @@ -662,6 +685,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 } @@ -738,6 +772,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}, + }) } } } diff --git a/internal/daemon/events.go b/internal/daemon/events.go new file mode 100644 index 0000000..0c05c20 --- /dev/null +++ b/internal/daemon/events.go @@ -0,0 +1,124 @@ +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"` +} + +// 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, 64) + 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..1f4f8a8 --- /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 (capacity 64) plus extra + for i := 0; i < 80; i++ { + bus.Emit(daemon.Event{ + Type: daemon.EventExpertSpawning, + Timestamp: time.Now(), + }) + } + + // Should have 64 events (buffer capacity), not 80 + count := 0 + for { + select { + case <-ch: + count++ + default: + goto done + } + } +done: + if count != 64 { + t.Errorf("received %d events, want 64 (buffer capacity)", count) + } +} diff --git a/internal/daemon/socket.go b/internal/daemon/socket.go index 1f2abd4..2656ca2 100644 --- a/internal/daemon/socket.go +++ b/internal/daemon/socket.go @@ -123,12 +123,36 @@ func (s *socketServer) handleConn(ctx context.Context, conn net.Conn) { } } -// handleSubscribe is a placeholder for Phase 4 event streaming. -func (s *socketServer) handleSubscribe(_ context.Context, conn net.Conn) { +// 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) + + // Send ack s.writeResponse(conn, socketResponse{ - Status: "error", - Message: "subscribe not implemented", + 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 { + return // client disconnected + } + case <-ctx.Done(): + return + } + } } // writeResponse encodes a response as a single JSON line. diff --git a/internal/daemon/socket_test.go b/internal/daemon/socket_test.go index 328b0c5..bd2b0b7 100644 --- a/internal/daemon/socket_test.go +++ b/internal/daemon/socket_test.go @@ -245,6 +245,62 @@ project_dir = "PROJECT_DIR" 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 From f17613912604c832734e9798b27f0532449d5f79 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:04:25 -0500 Subject: [PATCH 05/13] feat: make session_timeout optional (v0.6 phase 5a) Remove the default 10m session timeout. When session_timeout is empty or unset, expert sessions run to completion without a time limit. - ParseSessionTimeout returns (0, nil) for empty strings - Daemon uses context.WithCancel instead of context.WithTimeout when 0 - Set session_timeout in pool.toml to impose a limit Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/config/config.go | 7 ++++++- internal/config/config_test.go | 6 +++--- internal/daemon/daemon.go | 12 +++++++++--- 3 files changed, 18 insertions(+), 7 deletions(-) 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/daemon/daemon.go b/internal/daemon/daemon.go index e1030f6..f9812cf 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -599,12 +599,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) From 651bf93ea338f8a810257d787d4ed7b44a438365 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:08:06 -0500 Subject: [PATCH 06/13] refactor: drop pool_ prefix from MCP tool names (v0.6 phase 5b) Clean break: all tool names simplified from pool_X to X. The prefix only existed for ~1 day in production (bosun dogfood). Updated across code, tests, plugin skills, and bosun concierge identity. Tool renames: pool_read_state -> read_state, pool_update_state -> update_state, pool_append_error -> append_error, pool_send_response -> send_response, pool_recall -> recall, pool_search_index -> search_index, pool_define_contract -> define_contract, pool_send_task -> send_task, pool_verify_result -> verify_result, pool_amend_contract -> amend_contract, pool_ask_expert -> ask_expert, pool_dispatch -> dispatch, pool_collect -> collect, pool_submit_plan -> submit_plan, pool_check_status -> check_status, pool_list_experts -> list_experts Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/mcp/architect_tools.go | 8 ++-- internal/mcp/architect_tools_test.go | 50 ++++++++++++------------ internal/mcp/concierge_tools.go | 22 +++++------ internal/mcp/concierge_tools_test.go | 58 ++++++++++++++-------------- internal/mcp/config.go | 12 +++--- internal/mcp/tools.go | 12 +++--- internal/mcp/tools_test.go | 38 +++++++++--------- plugin/concierge-identity.md | 10 ++--- plugin/skills/pool-ask.md | 4 +- plugin/skills/pool-build.md | 6 +-- plugin/skills/pool-status.md | 2 +- 11 files changed, 111 insertions(+), 111 deletions(-) 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..8e2597f 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,7 @@ func handleListExperts(cfg *ServerConfig) server.ToolHandlerFunc { sort.Strings(poolExperts) result := map[string][]string{ - "pool_experts": poolExperts, + "experts": poolExperts, "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..dfaf138 100644 --- a/plugin/concierge-identity.md +++ b/plugin/concierge-identity.md @@ -13,14 +13,14 @@ 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 +- `ask_expert` — dispatch a question to an expert and wait for response +- `submit_plan` — send a plan to the architect for decomposition +- `check_status` — query the 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. 4. Track everything. The taskboard is your source of truth. diff --git a/plugin/skills/pool-ask.md b/plugin/skills/pool-ask.md index 40675ed..5301be6 100644 --- a/plugin/skills/pool-ask.md +++ b/plugin/skills/pool-ask.md @@ -13,12 +13,12 @@ responses, and synthesize a unified answer. ### 1. Discover experts -Call `pool_list_experts` to see who's available. Present the list to the user +Call `list_experts` to see who's available. Present the list to the user if they haven't specified which experts to ask. ### 2. Dispatch questions -For each relevant expert, call `pool_ask_expert` with: +For each relevant expert, call `ask_expert` with: - `expert`: the expert's name - `question`: the question, tailored to that expert's domain diff --git a/plugin/skills/pool-build.md b/plugin/skills/pool-build.md index 720609a..38aea16 100644 --- a/plugin/skills/pool-build.md +++ b/plugin/skills/pool-build.md @@ -18,7 +18,7 @@ Identify which domains are involved. ### 2. Gather expert input (optional) -If the feature spans multiple domains, use `pool_ask_expert` to get domain +If the feature spans multiple domains, use `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. @@ -33,7 +33,7 @@ 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) @@ -42,7 +42,7 @@ and dispatch tasks to experts. ### 5. Track progress -Use `pool_check_status` to monitor: +Use `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 diff --git a/plugin/skills/pool-status.md b/plugin/skills/pool-status.md index 7a2a9bb..710236a 100644 --- a/plugin/skills/pool-status.md +++ b/plugin/skills/pool-status.md @@ -11,7 +11,7 @@ You are the concierge. The user wants to know the status of in-flight work. ### 1. Query the taskboard -Call `pool_check_status` with appropriate filters: +Call `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 From dd5de617199b9c141c528abeccb9e5ac0adc4840 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:08:40 -0500 Subject: [PATCH 07/13] docs: add launchd plist template and install guide (v0.6 phase 6) Ship a macOS launchd plist template with placeholder markers for binary path and pool directory. Includes install/stop/unload documentation and notes on KeepAlive and double-signal behavior. Co-Authored-By: Claude Opus 4.6 (1M context) --- docs/launchd.md | 64 +++++++++++++++++++++++++++++ scripts/com.agent-pool.daemon.plist | 30 ++++++++++++++ 2 files changed, 94 insertions(+) create mode 100644 docs/launchd.md create mode 100644 scripts/com.agent-pool.daemon.plist 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/scripts/com.agent-pool.daemon.plist b/scripts/com.agent-pool.daemon.plist new file mode 100644 index 0000000..376b8a3 --- /dev/null +++ b/scripts/com.agent-pool.daemon.plist @@ -0,0 +1,30 @@ + + + + + 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 + + From aae7592e5bbf48a6cf9008ea3216ca679ac313e4 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:12:17 -0500 Subject: [PATCH 08/13] refactor: add narrative logging to socket server Apply action-oriented logging pattern to socket infrastructure: - Info: stop command received - Warn: invalid request, unknown method, accept failures - Debug: subscriber connect/disconnect Also align daemon drain messages with the narrative pattern (Preparing to / Successfully / Skipping. Reason:). Co-Authored-By: Claude Opus 4.6 (1M context) --- internal/daemon/daemon.go | 8 +++++--- internal/daemon/socket.go | 21 +++++++++++++++++---- 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index f9812cf..4208563 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -190,7 +190,7 @@ func (d *Daemon) Run(ctx context.Context) error { for { select { case <-childCtx.Done(): - d.logger.Info("Shutting down daemon, draining in-flight work", + d.logger.Info("Preparing to drain in-flight work", "pool", d.cfg.Pool.Name, "drain_timeout", d.drainTimeout, ) @@ -198,9 +198,11 @@ func (d *Daemon) Run(ctx context.Context) error { go func() { d.wg.Wait(); close(done) }() select { case <-done: - d.logger.Info("All in-flight work completed") + d.logger.Info("Successfully drained all in-flight work") case <-time.After(d.drainTimeout): - d.logger.Warn("Drain timeout exceeded, forcing shutdown") + d.logger.Warn("Skipping drain. Reason: timeout exceeded", + "drain_timeout", d.drainTimeout, + ) } return nil diff --git a/internal/daemon/socket.go b/internal/daemon/socket.go index 2656ca2..714b9fb 100644 --- a/internal/daemon/socket.go +++ b/internal/daemon/socket.go @@ -69,7 +69,7 @@ func (s *socketServer) serve(ctx context.Context) { if ctx.Err() != nil { return } - s.logger.Warn("Socket accept error", "error", err) + s.logger.Warn("Failed to accept socket connection", "error", err) continue } go s.handleConn(ctx, conn) @@ -90,6 +90,9 @@ func (s *socketServer) handleConn(ctx context.Context, conn net.Conn) { 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(), @@ -99,6 +102,7 @@ func (s *socketServer) handleConn(ctx context.Context, conn net.Conn) { 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"}, @@ -106,16 +110,18 @@ func (s *socketServer) handleConn(ctx context.Context, conn net.Conn) { s.cancel() case "status": - data := s.daemon.Status() s.writeResponse(conn, socketResponse{ Status: "ok", - Data: data, + 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), @@ -130,6 +136,10 @@ 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", @@ -147,7 +157,10 @@ func (s *socketServer) handleSubscribe(ctx context.Context, conn net.Conn) { return } if err := enc.Encode(event); err != nil { - return // client disconnected + s.logger.Debug("Subscriber disconnected", + "subscriber_id", id, + ) + return } case <-ctx.Done(): return From 4b408a2ed0393327beadacb62f10e194fcffd5d7 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:15:44 -0500 Subject: [PATCH 09/13] docs: update test audit report for v0.6 Full-scope audit after v0.6 implementation. 69.9% coverage, 25 test files, no P0 issues. Key gaps: DiscoverPoolDir (high risk), connectAndSend (medium risk). Main quality concern is time.Sleep-based synchronization in integration tests (30+ occurrences). [skip-ci] Co-Authored-By: Claude Opus 4.6 (1M context) --- docs/test-audit-report.md | 137 +++++++++++++++++++++----------------- 1 file changed, 77 insertions(+), 60 deletions(-) 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 From 2a07d2b2fcdb588f60e729c3281bfb42161a8a71 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:28:30 -0500 Subject: [PATCH 10/13] refactor: improve plugin skills per writing-skills guidelines Apply CSO (Claude Search Optimization) to skill descriptions: - Change from workflow summaries to triggering conditions ("Use when...") - This prevents Claude from shortcutting the skill body Content improvements: - pool-ask: Use dispatch+collect (non-blocking) instead of ask_expert - pool-build: Same dispatch+collect pattern for expert input gathering - concierge-identity: Add dispatch and collect to tools table, use table format for tool reference [skip-ci] Co-Authored-By: Claude Opus 4.6 (1M context) --- plugin/concierge-identity.md | 14 +++++++---- plugin/skills/pool-ask.md | 48 +++++++++++++++++++----------------- plugin/skills/pool-build.md | 46 +++++++++++++--------------------- plugin/skills/pool-status.md | 20 +++++++-------- 4 files changed, 61 insertions(+), 67 deletions(-) diff --git a/plugin/concierge-identity.md b/plugin/concierge-identity.md index dfaf138..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 -- `ask_expert` — dispatch a question to an expert and wait for response -- `submit_plan` — send a plan to the architect for decomposition -- `check_status` — query the taskboard for task progress -- `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 `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 5301be6..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 `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 `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 38aea16..d309d7e 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 `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,20 @@ Synthesize the user's requirements and expert input into a plan that includes: ### 4. Submit to architect -Call `submit_plan` with: -- `plan`: the plan body (markdown) -- `contracts`: any existing contract IDs that apply (optional) - -This returns a task ID. The architect will review the plan, define contracts, -and dispatch tasks to experts. +Call `submit_plan` with the plan body. The architect will review, +define contracts, and dispatch tasks to experts. ### 5. Track progress -Use `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 710236a..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 `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) +- 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 From 91329349aa1197ec8a7a3d9a471395062772147e Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:44:53 -0500 Subject: [PATCH 11/13] fix: address review findings (round 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit findings: 1. [Critical] Socket path mismatch — CLI used hardcoded path while daemon had fnv fallback for long paths. Extracted shared config.ResolveSockPath used by both sides. 2. [Nitpick] Magic number 64 in event bus test — exported EventBufSize constant, test uses it instead of hardcoded value. 3. [Nitpick] writeResponse silently ignored errors — added debug log. 4. [Major] list_experts response renamed pool_experts to experts without compatibility — added pool_experts as deprecated alias alongside new experts key. 5. [Major] launchd plist missing ExitTimeOut — added 35s to exceed the 30s drain timeout, preventing SIGKILL before graceful shutdown completes. Co-Authored-By: Claude Opus 4.6 (1M context) --- cmd/agent-pool/main.go | 6 +++--- internal/config/socket.go | 23 +++++++++++++++++++++++ internal/daemon/daemon.go | 14 +++----------- internal/daemon/events.go | 6 +++++- internal/daemon/events_test.go | 10 +++++----- internal/daemon/socket.go | 4 +++- internal/mcp/concierge_tools.go | 3 ++- scripts/com.agent-pool.daemon.plist | 3 +++ 8 files changed, 47 insertions(+), 22 deletions(-) create mode 100644 internal/config/socket.go diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index c6509e8..59bb26e 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -127,7 +127,7 @@ func cmdStop() { os.Exit(1) } - resp, err := connectAndSend(filepath.Join(poolDir, "daemon.sock"), "stop") + resp, err := connectAndSend(config.ResolveSockPath(poolDir), "stop") if err != nil { fmt.Fprintf(os.Stderr, "error: %v\n", err) os.Exit(1) @@ -153,7 +153,7 @@ func cmdStatus() { os.Exit(1) } - resp, err := connectAndSend(filepath.Join(poolDir, "daemon.sock"), "status") + resp, err := connectAndSend(config.ResolveSockPath(poolDir), "status") if err != nil { fmt.Fprintf(os.Stderr, "error: %v\n", err) os.Exit(1) @@ -233,7 +233,7 @@ func cmdWatch() { os.Exit(1) } - sockPath := filepath.Join(poolDir, "daemon.sock") + 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) diff --git a/internal/config/socket.go b/internal/config/socket.go new file mode 100644 index 0000000..02edc0b --- /dev/null +++ b/internal/config/socket.go @@ -0,0 +1,23 @@ +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 { + 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 4208563..55b70d9 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -8,7 +8,6 @@ package daemon import ( "context" "fmt" - "hash/fnv" "io" "log/slog" "os" @@ -1078,20 +1077,13 @@ func (d *Daemon) resolveSessionTimeout(name string) (time.Duration, error) { } // resolveSockPath returns the unix socket path for CLI→daemon communication. -// Uses the override if set, otherwise 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). +// 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 } - candidate := filepath.Join(d.poolDir, "daemon.sock") - if len(candidate) <= 100 { - return candidate - } - h := fnv.New32a() - h.Write([]byte(d.poolDir)) - return filepath.Join(os.TempDir(), fmt.Sprintf("ap-%x.sock", h.Sum32())) + return config.ResolveSockPath(d.poolDir) } // resolveExpertDir returns the state directory for an expert or built-in role. diff --git a/internal/daemon/events.go b/internal/daemon/events.go index 0c05c20..a284aed 100644 --- a/internal/daemon/events.go +++ b/internal/daemon/events.go @@ -63,6 +63,10 @@ type TaskUnblockedData struct { 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 @@ -84,7 +88,7 @@ func (b *eventBus) subscribe() (int, <-chan Event) { id := b.nextID b.nextID++ - ch := make(chan Event, 64) + ch := make(chan Event, EventBufSize) b.subs[id] = ch return id, ch } diff --git a/internal/daemon/events_test.go b/internal/daemon/events_test.go index 1f4f8a8..c8544ba 100644 --- a/internal/daemon/events_test.go +++ b/internal/daemon/events_test.go @@ -76,15 +76,15 @@ func TestEventBus_SlowDrops(t *testing.T) { _, ch := bus.Subscribe() - // Fill the buffer (capacity 64) plus extra - for i := 0; i < 80; i++ { + // 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 64 events (buffer capacity), not 80 + // Should have exactly buffer capacity events count := 0 for { select { @@ -95,7 +95,7 @@ func TestEventBus_SlowDrops(t *testing.T) { } } done: - if count != 64 { - t.Errorf("received %d events, want 64 (buffer capacity)", count) + 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 index 714b9fb..b37b62b 100644 --- a/internal/daemon/socket.go +++ b/internal/daemon/socket.go @@ -171,7 +171,9 @@ func (s *socketServer) handleSubscribe(ctx context.Context, conn net.Conn) { // 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)) - json.NewEncoder(conn).Encode(resp) + 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. diff --git a/internal/mcp/concierge_tools.go b/internal/mcp/concierge_tools.go index 8e2597f..edef68e 100644 --- a/internal/mcp/concierge_tools.go +++ b/internal/mcp/concierge_tools.go @@ -510,7 +510,8 @@ func handleListExperts(cfg *ServerConfig) server.ToolHandlerFunc { sort.Strings(poolExperts) result := map[string][]string{ - "experts": poolExperts, + "experts": poolExperts, + "pool_experts": poolExperts, // compatibility alias (deprecated) "shared_experts": poolCfg.Shared.Include, } diff --git a/scripts/com.agent-pool.daemon.plist b/scripts/com.agent-pool.daemon.plist index 376b8a3..4e7bb79 100644 --- a/scripts/com.agent-pool.daemon.plist +++ b/scripts/com.agent-pool.daemon.plist @@ -26,5 +26,8 @@ WorkingDirectory POOL_DIR + + ExitTimeOut + 35 From 85f7ea01a505170cc065227272655957dd7b26d8 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 14:53:04 -0500 Subject: [PATCH 12/13] fix: address review findings (round 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. [Minor] pool-build skill omitted contracts parameter from submit_plan call — restored parameter list matching implementation. 2. [Nitpick] printStatusField ignored unmarshal errors — added fallback to print raw JSON when field isn't a string. Co-Authored-By: Claude Opus 4.6 (1M context) --- cmd/agent-pool/main.go | 5 ++++- plugin/skills/pool-build.md | 7 +++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index 59bb26e..ee1036a 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -174,7 +174,10 @@ func cmdStatus() { printStatusField := func(label, key string) { if v, ok := data[key]; ok { var s string - json.Unmarshal(v, &s) + if err := json.Unmarshal(v, &s); err != nil { + fmt.Printf("%-10s %s\n", label+":", string(v)) + return + } fmt.Printf("%-10s %s\n", label+":", s) } } diff --git a/plugin/skills/pool-build.md b/plugin/skills/pool-build.md index d309d7e..7ce36c9 100644 --- a/plugin/skills/pool-build.md +++ b/plugin/skills/pool-build.md @@ -31,8 +31,11 @@ Synthesize requirements and expert input into: ### 4. Submit to architect -Call `submit_plan` with the plan body. The architect will review, -define contracts, and dispatch tasks to experts. +Call `submit_plan` with: +- `plan`: the plan body (markdown) +- `contracts`: existing contract IDs that apply (optional) + +The architect will review, define contracts, and dispatch tasks to experts. ### 5. Track progress From 3afd8bb126f97d650e874ba9b746f9ff5b146380 Mon Sep 17 00:00:00 2001 From: Cameron Sjo Date: Sun, 5 Apr 2026 15:05:18 -0500 Subject: [PATCH 13/13] fix: address review findings (round 3) 1. [Critical] ResolveSockPath now canonicalizes poolDir with filepath.Abs internally, ensuring daemon and CLI always compute the same socket path regardless of relative vs absolute input. 2. [Major] cmdWatch now checks scanner.Err() after ack read and after the event stream loop, surfacing frame-size and I/O errors instead of silently dropping out. Co-Authored-By: Claude Opus 4.6 (1M context) --- cmd/agent-pool/main.go | 11 ++++++++++- internal/config/socket.go | 5 +++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/cmd/agent-pool/main.go b/cmd/agent-pool/main.go index ee1036a..0989432 100644 --- a/cmd/agent-pool/main.go +++ b/cmd/agent-pool/main.go @@ -254,7 +254,11 @@ func cmdWatch() { // Read ack scanner := bufio.NewScanner(conn) if !scanner.Scan() { - fmt.Fprintf(os.Stderr, "error: no ack from daemon\n") + 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 @@ -372,6 +376,11 @@ func cmdWatch() { 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. diff --git a/internal/config/socket.go b/internal/config/socket.go index 02edc0b..991278e 100644 --- a/internal/config/socket.go +++ b/internal/config/socket.go @@ -13,6 +13,11 @@ import ( // 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