Skip to content

feat: v0.6 daemon lifecycle + observability #17

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cameronsjo merged 13 commits into from
Apr 5, 2026
Merged

feat: v0.6 daemon lifecycle + observability #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
adc5847
feat: add unix domain socket and stop command (v0.6 phase 1)
cameronsjo Apr 5, 2026
7624495
feat: add graceful drain with WaitGroup and double-signal (v0.6 phase 2)
cameronsjo Apr 5, 2026
031530f
feat: add status command with pretty-print output (v0.6 phase 3)
cameronsjo Apr 5, 2026
1966bea
feat: add event bus and watch command (v0.6 phase 4)
cameronsjo Apr 5, 2026
f176139
feat: make session_timeout optional (v0.6 phase 5a)
cameronsjo Apr 5, 2026
651bf93
refactor: drop pool_ prefix from MCP tool names (v0.6 phase 5b)
cameronsjo Apr 5, 2026
dd5de61
docs: add launchd plist template and install guide (v0.6 phase 6)
cameronsjo Apr 5, 2026
aae7592
refactor: add narrative logging to socket server
cameronsjo Apr 5, 2026
4b408a2
docs: update test audit report for v0.6
cameronsjo Apr 5, 2026
2a07d2b
refactor: improve plugin skills per writing-skills guidelines
cameronsjo Apr 5, 2026
9132934
fix: address review findings (round 1)
cameronsjo Apr 5, 2026
85f7ea0
fix: address review findings (round 2)
cameronsjo Apr 5, 2026
3afd8bb
fix: address review findings (round 3)
cameronsjo Apr 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
335 changes: 334 additions & 1 deletion cmd/agent-pool/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,19 @@
package main

import (
"bufio"
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"net"
"os"
"os/signal"
"path/filepath"
"strings"
"syscall"
"time"

"github.com/cameronsjo/agent-pool/internal/config"
"github.com/cameronsjo/agent-pool/internal/daemon"
Expand All @@ -25,14 +30,20 @@ func main() {
switch os.Args[1] {
case "start":
cmdStart()
case "stop":
cmdStop()
case "status":
cmdStatus()
case "watch":
cmdWatch()
case "mcp":
cmdMCP()
case "flush":
cmdFlush()
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:
Expand Down Expand Up @@ -86,13 +97,330 @@ 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)
os.Exit(1)
}
}

func cmdStop() {
explicit := ""
if len(os.Args) > 2 {
explicit = os.Args[2]
}

poolDir, err := config.DiscoverPoolDir(explicit)
if err != nil {
fmt.Fprintf(os.Stderr, "error: %v\n", err)
os.Exit(1)
}

resp, err := connectAndSend(config.ResolveSockPath(poolDir), "stop")
Comment thread
coderabbitai[bot] marked this conversation as resolved.
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.")
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

func cmdStatus() {
explicit := ""
if len(os.Args) > 2 {
explicit = os.Args[2]
}

poolDir, err := config.DiscoverPoolDir(explicit)
if err != nil {
fmt.Fprintf(os.Stderr, "error: %v\n", err)
os.Exit(1)
}

resp, err := connectAndSend(config.ResolveSockPath(poolDir), "status")
if err != nil {
fmt.Fprintf(os.Stderr, "error: %v\n", err)
os.Exit(1)
}

if resp.Status != "ok" {
fmt.Fprintf(os.Stderr, "error: %s\n", resp.Message)
os.Exit(1)
}

var data map[string]json.RawMessage
if err := json.Unmarshal(resp.Data, &data); err != nil {
// Fallback to raw JSON
fmt.Println(string(resp.Data))
return
}

printStatusField := func(label, key string) {
if v, ok := data[key]; ok {
var s string
if err := json.Unmarshal(v, &s); err != nil {
fmt.Printf("%-10s %s\n", label+":", string(v))
return
}
fmt.Printf("%-10s %s\n", label+":", s)
}
}
Comment on lines +174 to +183

@coderabbitai coderabbitai Bot Apr 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial

Consider handling unmarshal errors in printStatusField.

The helper ignores json.Unmarshal errors, which is acceptable for string fields since they'll default to empty. However, for robustness, you might want to skip printing the field if unmarshal fails for non-string types.

💡 Optional: Handle unmarshal errors 
 	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)
+			if err := json.Unmarshal(v, &s); err == nil {
+				fmt.Printf("%-10s %s\n", label+":", s)
+			}
 		}
 	}
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@cmd/agent-pool/main.go` around lines 174 - 180, The printStatusField helper
currently ignores the error returned by json.Unmarshal which can lead to
printing incorrect or empty values; update printStatusField to capture the error
(e.g., err := json.Unmarshal(v, &s)) and if err != nil simply skip printing the
field (return) so only successfully unmarshalled string values are printed;
reference the printStatusField closure and its use of label, key, data,
json.Unmarshal and variable s when making the change.


printStatusField("Pool", "pool")
printStatusField("State", "state")
printStatusField("Uptime", "uptime")

// Experts
if v, ok := data["experts"]; ok {
var experts []string
json.Unmarshal(v, &experts)
fmt.Printf("%-10s %s\n", "Experts:", strings.Join(experts, ", "))
}

// Task counts
if v, ok := data["task_counts"]; ok {
var counts map[string]int
json.Unmarshal(v, &counts)
if len(counts) > 0 {
fmt.Println("\nTasks:")
for _, status := range []string{"pending", "blocked", "active", "completed", "failed", "cancelled"} {
if n, ok := counts[status]; ok && n > 0 {
fmt.Printf(" %-12s %d\n", status+":", n)
}
}
}
}

// Active tasks
if v, ok := data["active_tasks"]; ok {
var tasks []map[string]string
json.Unmarshal(v, &tasks)
if len(tasks) > 0 {
fmt.Println("\nActive Tasks:")
for _, t := range tasks {
started := t["started"]
if started != "" {
started = " (" + started + " ago)"
}
fmt.Printf(" %-20s %s%s\n", t["id"], t["expert"], started)
}
}
}
}

func cmdWatch() {
explicit := ""
if len(os.Args) > 2 {
explicit = os.Args[2]
}

poolDir, err := config.DiscoverPoolDir(explicit)
if err != nil {
fmt.Fprintf(os.Stderr, "error: %v\n", err)
os.Exit(1)
}

sockPath := config.ResolveSockPath(poolDir)
conn, err := net.DialTimeout("unix", sockPath, 5*time.Second)
if err != nil {
fmt.Fprintf(os.Stderr, "error: connecting to daemon (is it running?): %v\n", err)
os.Exit(1)
}
defer conn.Close()

// Send subscribe request
conn.SetDeadline(time.Now().Add(5 * time.Second))
if err := json.NewEncoder(conn).Encode(map[string]string{"method": "subscribe"}); err != nil {
fmt.Fprintf(os.Stderr, "error: %v\n", err)
os.Exit(1)
}

// Read ack
scanner := bufio.NewScanner(conn)
if !scanner.Scan() {
if err := scanner.Err(); err != nil {
fmt.Fprintf(os.Stderr, "error: reading ack: %v\n", err)
} else {
fmt.Fprintf(os.Stderr, "error: no ack from daemon\n")
}
os.Exit(1)
}
var ack socketResponse
if err := json.Unmarshal(scanner.Bytes(), &ack); err != nil || ack.Status != "ok" {
fmt.Fprintf(os.Stderr, "error: subscribe failed: %s\n", ack.Message)
os.Exit(1)
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

// 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()
}()
Comment on lines +273 to +279

@coderabbitai coderabbitai Bot Apr 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

cd /repo && wc -l cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 120

🏁 Script executed:

cd /repo && sed -n '273,279p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 120

🏁 Script executed:

cd /repo && sed -n '380,383p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 120

🏁 Script executed:

cd /repo && sed -n '260,300p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 120

🏁 Script executed:

cd /repo && sed -n '370,400p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 120

🏁 Script executed:

find . -name "main.go" -path "*/agent-pool/*" -type f

Repository: cameronsjo/agent-pool

Length of output: 89

🏁 Script executed:

wc -l cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 91

🏁 Script executed:

sed -n '273,279p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 222

🏁 Script executed:

sed -n '380,383p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 184

🏁 Script executed:

sed -n '260,300p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 967

🏁 Script executed:

sed -n '370,410p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 1294

🏁 Script executed:

sed -n '240,280p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 1179

🏁 Script executed:

sed -n '330,410p' cmd/agent-pool/main.go

Repository: cameronsjo/agent-pool

Length of output: 2292

🏁 Script executed:

rg "scanner.Err\(\)" cmd/agent-pool/main.go -A 2 -B 2

Repository: cameronsjo/agent-pool

Length of output: 514

🏁 Script executed:

rg "net.ErrClosed" . -t go

Repository: cameronsjo/agent-pool

Length of output: 47

Track intentional shutdown to suppress error message on Ctrl-C.

The signal handler closes conn when Ctrl-C is pressed, causing scanner.Scan() to fail. The subsequent scanner.Err() check will return that error and print "error: stream interrupted" before exiting 1, even though this is a normal, graceful shutdown. Set a flag when the signal handler triggers and suppress the error message for intentional closes.

Also applies to: 380-383

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@cmd/agent-pool/main.go` around lines 273 - 279, The signal handler currently
closes conn which makes scanner.Scan() return an error that is then logged as
"error: stream interrupted" even on intentional Ctrl-C; add an
intentional-shutdown flag (e.g., a package-level uint32 used with sync/atomic or
an atomic.Bool) and set it inside the goroutine before calling conn.Close()
(where sigCh, conn.Close() are used), then when checking scanner.Err() (and the
similar check around lines 380-383), suppress logging/exit if the
intentional-shutdown flag is set; apply the same pattern to the other signal
handler block so intentional closes don’t produce error messages.


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)
}
Comment on lines +281 to +378

@coderabbitai coderabbitai Bot Apr 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

watch is still event-only.

Issue #10 calls for a live task table plus the event stream, but this implementation never fetches an initial status snapshot or maintains task state for redraws. Operators only get an append-only log.

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@cmd/agent-pool/main.go` around lines 277 - 374, The current watcher only
prints events (the scanner loop handling event struct and switch on e.Type) but
never loads or maintains task state for a live table; update the logic to fetch
an initial task snapshot (via the same backend API used for events) into an
in-memory map keyed by task ID, implement a redraw function that clears the
terminal and prints a compact task table (showing status, expert, model,
duration, summary) and call it initially and whenever events arrive, and in the
scanner loop update the in-memory task entries based on event types
("task.routed", "expert.spawning", "expert.completed", "expert.failed",
"task.cancelled", "task.unblocked") before calling redraw; keep the existing
event log output but use fmt.Printf and the ANSI constants (reset, green, red,
yellow, cyan) to colorize both the table rows and the appended event line.


if err := scanner.Err(); err != nil {
fmt.Fprintf(os.Stderr, "error: stream interrupted: %v\n", err)
os.Exit(1)
}
}

// socketResponse mirrors the daemon's response type for CLI deserialization.
type socketResponse struct {
Status string `json:"status"`
Data json.RawMessage `json:"data,omitempty"`
Message string `json:"message,omitempty"`
}

// connectAndSend dials the daemon socket, sends a method request, and reads the response.
func connectAndSend(sockPath, method string) (*socketResponse, error) {
conn, err := net.DialTimeout("unix", sockPath, 5*time.Second)
if err != nil {
return nil, fmt.Errorf("connecting to daemon (is it running?): %w", err)
}
defer conn.Close()

conn.SetDeadline(time.Now().Add(5 * time.Second))

req := map[string]string{"method": method}
if err := json.NewEncoder(conn).Encode(req); err != nil {
return nil, fmt.Errorf("sending request: %w", err)
}

scanner := bufio.NewScanner(conn)
if !scanner.Scan() {
if err := scanner.Err(); err != nil {
return nil, fmt.Errorf("reading response: %w", err)
}
return nil, fmt.Errorf("no response from daemon")
}

var resp socketResponse
if err := json.Unmarshal(scanner.Bytes(), &resp); err != nil {
return nil, fmt.Errorf("parsing response: %w", err)
}

return &resp, nil
}

// cmdMCP starts the stdio MCP server. Stdout is the MCP transport; logs go to stderr.
//
// Two invocation modes:
Expand Down Expand Up @@ -229,6 +557,9 @@ func printUsage() {

Usage:
agent-pool start [pool-dir] Start the daemon for a pool
agent-pool stop [pool-dir] Stop a running daemon
agent-pool status [pool-dir] Show daemon status
agent-pool watch [pool-dir] Stream daemon events
agent-pool mcp --pool <dir> --expert <name> Start expert MCP server (stdio)
agent-pool mcp --pool <dir> --role <role> Start built-in role MCP server
agent-pool flush --pool <dir> --expert <name> --task <id> Stop hook: verify state
Expand All @@ -243,5 +574,7 @@ Roles:

Examples:
agent-pool start ~/.agent-pool/pools/api-gateway
agent-pool stop
agent-pool status
agent-pool mcp --pool ./my-pool --role concierge`)
}
Loading
Loading