diff --git a/cmd/berthd/main.go b/cmd/berthd/main.go index d1e1bf9c..faa3bffd 100644 --- a/cmd/berthd/main.go +++ b/cmd/berthd/main.go @@ -20,6 +20,7 @@ import ( "text/tabwriter" "time" + "github.com/cosscom/shipyard/internal/agenthost" "github.com/cosscom/shipyard/internal/box" "github.com/cosscom/shipyard/internal/boxcmd" "github.com/cosscom/shipyard/internal/debugserver" @@ -205,6 +206,12 @@ func run(args []string) error { return nil case "headless": return box.RunHeadless(args[1:], os.Stdout) + case "agent-host": + code, err := agenthost.Main(args[1:], os.Stdout) + if err != nil { + return err + } + os.Exit(code) case "mcp": return mcpserver.Serve(context.Background(), b.socket(), os.Stdin, os.Stdout) case "agents": diff --git a/design/structured-agents/plan.md b/design/structured-agents/plan.md index bec3d619..0206e364 100644 --- a/design/structured-agents/plan.md +++ b/design/structured-agents/plan.md @@ -1,6 +1,6 @@ # Structured agents: plan -Status: 2026-10-13. Phase 0 (spike) done: go. Phase 1 (driver interface) in progress. The spike's recordings join the repo as test fixtures in phase 3, once reviewed. +Status: 2026-10-13. Phase 0 (spike) done: go. Phase 1 (driver interface) merged. Phase 2 (agent-host + log) in review. The spike's recordings join the repo as test fixtures in phase 3, once reviewed. ## Why @@ -90,8 +90,9 @@ Exit: a go/no-go note per item; any no-go changes the phases below. - Keep streamed partial text in the log (a killed agent loses it). Stop = interrupt, then SIGTERM, then kill the process tree (SIGKILL leaves Bash children running). -- SQLite index: conversation id, location/worktree, agent, driver, agent session - id, title, status, cost, timestamps. +- The SQLite index (conversation id, location/worktree, agent, driver, agent + session id, title, status, cost, timestamps) comes with the Claude driver + (phase 3), the first thing that has those to record. - Push to the app via the events bus (items in the event, not a "changed" ping); no stat-polling for driver agents. - Attaching to the tmux session shows a readable live render, so `berth attach` keeps working. diff --git a/internal/agenthost/client.go b/internal/agenthost/client.go new file mode 100644 index 00000000..18f7f557 --- /dev/null +++ b/internal/agenthost/client.go @@ -0,0 +1,93 @@ +package agenthost + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "net" + "time" +) + +// ErrNoHost is a socket no host answers on: the agent has ended (its log +// is still there to read), or never started. +var ErrNoHost = errors.New("agent-host: no host is running") + +func dial(ctx context.Context, sock string) (net.Conn, error) { + var d net.Dialer + c, err := d.DialContext(ctx, "unix", sock) + if err != nil { + return nil, ErrNoHost + } + if dl, ok := ctx.Deadline(); ok { + c.SetDeadline(dl) + } + return c, nil +} + +func call(ctx context.Context, sock string, req request) (reply, error) { + c, err := dial(ctx, sock) + if err != nil { + return reply{}, err + } + defer c.Close() + if err := json.NewEncoder(c).Encode(req); err != nil { + return reply{}, err + } + var rep reply + if err := json.NewDecoder(bufio.NewReader(c)).Decode(&rep); err != nil { + return reply{}, err + } + if rep.Error != "" { + return rep, errors.New(rep.Error) + } + return rep, nil +} + +// Write gives the agent one JSON line on its stdin, and returns the log +// entry it became. +func Write(ctx context.Context, sock string, data json.RawMessage) (int64, error) { + rep, err := call(ctx, sock, request{Op: "write", Data: data}) + return rep.Seq, err +} + +// Stop ends the agent and waits for it to be gone. +func Stop(ctx context.Context, sock string) error { + _, err := call(ctx, sock, request{Op: "stop"}) + return err +} + +// Attach calls fn with each log entry after from, then each new one as the +// agent works, until the stream ends (nil), ctx ends or fn fails. The +// stream ends after the agent's "exited" entry; one that ends before it is +// a host that let a reader fall behind go, or went away: attach again from +// the last Seq seen. With no host running it is ErrNoHost: read the log +// file instead. +func Attach(ctx context.Context, sock string, from int64, fn func(Entry) error) error { + c, err := dial(ctx, sock) + if err != nil { + return err + } + defer c.Close() + stop := context.AfterFunc(ctx, func() { c.SetReadDeadline(time.Now()) }) + defer stop() + if err := json.NewEncoder(c).Encode(request{Op: "attach", From: from}); err != nil { + return err + } + dec := json.NewDecoder(bufio.NewReaderSize(c, 64<<10)) + for { + var e Entry + if err := dec.Decode(&e); err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return nil + } + if e.Seq == 0 { + return errors.New("agent-host: the host refused to attach") + } + if err := fn(e); err != nil { + return err + } + } +} diff --git a/internal/agenthost/cmd.go b/internal/agenthost/cmd.go new file mode 100644 index 00000000..7b2b9807 --- /dev/null +++ b/internal/agenthost/cmd.go @@ -0,0 +1,31 @@ +package agenthost + +import ( + "context" + "errors" + "flag" + "io" + "os" + "os/signal" + "syscall" +) + +// Main is `berthd agent-host --dir DIR [--socket S] [--cwd W] -- COMMAND...`, +// run by berthd as a tmux session's program: it hosts the agent until it +// ends, and returns its exit code. Closing the session (SIGHUP) or SIGTERM +// stops the agent first. +func Main(args []string, stdout io.Writer) (int, error) { + fs := flag.NewFlagSet("agent-host", flag.ContinueOnError) + dir := fs.String("dir", "", "the conversation's folder (its log, and socket)") + sock := fs.String("socket", "", "where to listen (default DIR/"+SocketName+")") + cwd := fs.String("cwd", "", "where the agent runs (default here)") + if err := fs.Parse(args); err != nil { + return 2, err + } + if *dir == "" || fs.NArg() == 0 { + return 2, errors.New("usage: berthd agent-host --dir DIR [--socket S] [--cwd W] -- COMMAND...") + } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGHUP, syscall.SIGINT) + defer stop() + return Run(ctx, Options{Dir: *dir, Socket: *sock, WorkDir: *cwd, Command: fs.Args(), Render: stdout, Env: os.Environ()}) +} diff --git a/internal/agenthost/host.go b/internal/agenthost/host.go new file mode 100644 index 00000000..79dc53e4 --- /dev/null +++ b/internal/agenthost/host.go @@ -0,0 +1,386 @@ +package agenthost + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "syscall" + "time" +) + +// Options say what a host runs and where it keeps it. +type Options struct { + // Dir is the conversation's folder: its log and, by default, socket. + Dir string + // Socket is where the host listens; empty is Dir/host.sock. A unix + // socket's path is limited to about 100 bytes. + Socket string + // Command is the agent and its arguments; WorkDir where it runs (empty + // is the host's own); Env its environment (nil is the host's own). + Command []string + WorkDir string + Env []string + // Render, when set, gets a line for each entry: the tmux pane, so + // someone attached sees the agent work. + Render io.Writer + // StopGrace is how long the agent has to end after its stdin closes, + // and again after SIGTERM; 0 is 5 seconds. + StopGrace time.Duration +} + +// request is one call on the socket, a JSON line: "attach" streams the log +// from From (one Entry per line, until the agent ends); "write" gives the +// agent Data as one line of its stdin; "stop" ends it. +type request struct { + Op string `json:"op"` + From int64 `json:"from,omitempty"` + Data json.RawMessage `json:"data,omitempty"` +} + +// reply answers a write or a stop. +type reply struct { + Seq int64 `json:"seq,omitempty"` + Error string `json:"error,omitempty"` +} + +// Run starts the agent, keeps its log and serves its socket until it ends, +// and returns its exit code (-1 for a signal). Cancelling ctx stops the +// agent as Stop does. +func Run(ctx context.Context, o Options) (int, error) { + if len(o.Command) == 0 { + return 0, errors.New("agent-host: no command to run") + } + if o.StopGrace <= 0 { + o.StopGrace = 5 * time.Second + } + sock := o.Socket + if sock == "" { + sock = filepath.Join(o.Dir, SocketName) + } + log, err := OpenLog(o.Dir) + if err != nil { + return 0, err + } + defer log.Close() + ln, err := listen(sock) + if err != nil { + return 0, err + } + defer os.Remove(sock) + defer ln.Close() + + cmd := exec.Command(o.Command[0], o.Command[1:]...) + cmd.Dir, cmd.Env = o.WorkDir, o.Env + // Its own group, so a stop reaches what it starts. + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + stdin, err := cmd.StdinPipe() + if err != nil { + return 0, err + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return 0, err + } + stderr, err := cmd.StderrPipe() + if err != nil { + return 0, err + } + if err := cmd.Start(); err != nil { + e := Entry{Dir: "host", Event: "exited", Text: err.Error(), Code: intp(127)} + log.Append(e) + return 127, err + } + h := &host{log: log, stdin: stdin, pid: cmd.Process.Pid, grace: o.StopGrace, render: o.Render, exited: make(chan struct{})} + h.note(Entry{Dir: "host", Event: "started", PID: h.pid, Text: strings.Join(o.Command, " ")}) + + var reading sync.WaitGroup + reading.Add(2) + go func() { defer reading.Done(); h.read(stdout, "out") }() + go func() { defer reading.Done(); h.read(stderr, "err") }() + go h.serve(ln) + go func() { + select { + case <-ctx.Done(): + h.stop() + case <-h.exited: + } + }() + + st, werr := cmd.Process.Wait() + close(h.exited) + h.inMu.Lock() + stdin.Close() + h.inMu.Unlock() + // What it wrote is read to the end before its exit is logged, so the + // exit is the last entry. Something it started may still hold its + // stdout open: that is cut off after a moment. + drained := make(chan struct{}) + go func() { reading.Wait(); close(drained) }() + select { + case <-drained: + case <-time.After(2 * time.Second): + stdout.Close() + stderr.Close() + <-drained + } + code := -1 + if werr == nil && st != nil { + code = st.ExitCode() + } + h.note(Entry{Dir: "host", Event: "exited", Code: intp(code)}) + // Readers get the exit before the host goes: the log's close ends + // their streams once they have it. + ln.Close() + h.connMu.Lock() + h.done = true + h.connMu.Unlock() + log.Close() + flushed := make(chan struct{}) + go func() { h.conns.Wait(); close(flushed) }() + select { + case <-flushed: + case <-time.After(2 * time.Second): + } + return code, nil +} + +func intp(i int) *int { return &i } + +// listen makes the socket, refusing one a live host still answers on. +func listen(sock string) (net.Listener, error) { + if c, err := net.DialTimeout("unix", sock, time.Second); err == nil { + c.Close() + return nil, fmt.Errorf("agent-host: another host is running on %s", sock) + } + os.Remove(sock) + if err := os.MkdirAll(filepath.Dir(sock), 0o700); err != nil { + return nil, err + } + ln, err := net.Listen("unix", sock) + if err != nil { + return nil, err + } + os.Chmod(sock, 0o600) + return ln, nil +} + +type host struct { + log *Log + pid int + grace time.Duration + render io.Writer + + inMu sync.Mutex + stdin io.WriteCloser + + stopOnce sync.Once + exited chan struct{} + // conns are the calls being answered; once done is set no more start. + connMu sync.Mutex + done bool + conns sync.WaitGroup +} + +// note logs an entry and draws it. +func (h *host) note(e Entry) (Entry, error) { + e, err := h.log.Append(e) + if err == nil && h.render != nil { + fmt.Fprintln(h.render, renderLine(e)) + } + return e, err +} + +// read logs each line from the agent's stdout or stderr. Lines can be long +// (a tool's whole output): they are read whole. +func (h *host) read(r io.Reader, dir string) { + br := bufio.NewReaderSize(r, 64<<10) + for { + line, err := br.ReadBytes('\n') + if s := strings.TrimRight(string(line), "\r\n"); s != "" { + e := Entry{Dir: dir} + if dir == "out" && json.Valid([]byte(s)) { + e.Data = json.RawMessage(s) + } else { + e.Text = s + } + h.note(e) + } + if err != nil { + return + } + } +} + +// write gives the agent one JSON line, logging it first. +func (h *host) write(data json.RawMessage) (int64, error) { + if !json.Valid(data) { + return 0, errors.New("data must be JSON") + } + var one bytes.Buffer + if err := json.Compact(&one, data); err != nil { + return 0, err + } + h.inMu.Lock() + defer h.inMu.Unlock() + select { + case <-h.exited: + return 0, errors.New("the agent has ended") + default: + } + e, err := h.note(Entry{Dir: "in", Data: json.RawMessage(one.String())}) + if err != nil { + return 0, err + } + if _, err := io.WriteString(h.stdin, one.String()+"\n"); err != nil { + return 0, err + } + return e.Seq, nil +} + +// stop ends the agent: its stdin closes (structured agents end there), then +// SIGTERM, then SIGKILL to everything it started. The processes are found +// before the first signal, since once the agent goes they are no longer +// its children. +func (h *host) stop() { + h.stopOnce.Do(func() { + h.inMu.Lock() + h.stdin.Close() + h.inMu.Unlock() + if h.waitExit(h.grace) { + return + } + kids := descendants(h.pid) + signalTree(h.pid, syscall.SIGTERM) + if h.waitExit(h.grace) && !anyAlive(kids) { + return + } + for _, p := range kids { + syscall.Kill(p, syscall.SIGKILL) + } + signalTree(h.pid, syscall.SIGKILL) + }) +} + +func (h *host) waitExit(d time.Duration) bool { + select { + case <-h.exited: + return true + case <-time.After(d): + return false + } +} + +func anyAlive(pids []int) bool { + for _, p := range pids { + if syscall.Kill(p, 0) == nil { + return true + } + } + return false +} + +func (h *host) serve(ln net.Listener) { + for { + c, err := ln.Accept() + if err != nil { + return + } + h.connMu.Lock() + if h.done { + h.connMu.Unlock() + c.Close() + continue + } + h.conns.Add(1) + h.connMu.Unlock() + go func() { + defer h.conns.Done() + h.conn(c) + }() + } +} + +func (h *host) conn(c net.Conn) { + defer c.Close() + var req request + if err := json.NewDecoder(bufio.NewReader(c)).Decode(&req); err != nil { + return + } + enc := json.NewEncoder(c) + switch req.Op { + case "attach": + ch, stop, err := h.log.Follow(req.From) + if err != nil { + enc.Encode(reply{Error: err.Error()}) + return + } + defer stop() + // A reader that goes away ends its stream. + gone := make(chan struct{}) + go func() { io.Copy(io.Discard, c); close(gone) }() + for { + select { + case e, ok := <-ch: + if !ok { + return + } + if enc.Encode(e) != nil { + return + } + case <-gone: + return + } + } + case "write": + seq, err := h.write(req.Data) + if err != nil { + enc.Encode(reply{Error: err.Error()}) + return + } + enc.Encode(reply{Seq: seq}) + case "stop": + go h.stop() + <-h.exited + enc.Encode(reply{}) + default: + enc.Encode(reply{Error: "unknown op " + req.Op}) + } +} + +// renderLine is an entry as one line for the pane. +func renderLine(e Entry) string { + switch { + case e.Dir == "host" && e.Event == "exited" && e.Code != nil: + return fmt.Sprintf("— the agent ended (exit %d)", *e.Code) + case e.Dir == "host": + return "— " + e.Event + " " + e.Text + case e.Data != nil: + var head struct{ Type, Subtype, Method string } + json.Unmarshal(e.Data, &head) + kind := strings.TrimSpace(strings.Join([]string{head.Type, head.Subtype, head.Method}, " ")) + return e.Dir + " " + firstNonEmpty(kind, "json") + default: + t := e.Text + if len(t) > 200 { + t = t[:200] + "…" + } + return e.Dir + " " + t + } +} + +func firstNonEmpty(a, b string) string { + if a != "" { + return a + } + return b +} diff --git a/internal/agenthost/host_test.go b/internal/agenthost/host_test.go new file mode 100644 index 00000000..fdcf845a --- /dev/null +++ b/internal/agenthost/host_test.go @@ -0,0 +1,274 @@ +package agenthost + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" +) + +// echoAgent says it is ready, warns once on stderr, then answers each line +// it is sent with the same line, as a structured agent answers JSON. +const echoAgent = `echo ready; echo warming up >&2; while IFS= read -r l; do printf '%s\n' "$l"; done` + +// started runs a host for script in its own folder, and returns the +// folder, its socket and the host's result once it ends. +func started(t *testing.T, dir, script string) (sock string, result <-chan int) { + t.Helper() + // A unix socket's path is short: never under the test's long temp dir. + sd, err := os.MkdirTemp("/tmp", "ah") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(sd) }) + sock = filepath.Join(sd, "s") + ctx, cancel := context.WithCancel(context.Background()) + res, done := make(chan int, 1), make(chan struct{}) + go func() { + defer close(done) + code, err := Run(ctx, Options{Dir: dir, Socket: sock, Command: []string{"sh", "-c", script}, StopGrace: 500 * time.Millisecond}) + if err != nil { + t.Error(err) + } + res <- code + }() + t.Cleanup(func() { cancel(); <-done }) + deadline := time.Now().Add(5 * time.Second) + for { + if c, err := dial(context.Background(), sock); err == nil { + c.Close() + break + } + if time.Now().After(deadline) { + t.Fatal("the host never listened") + } + time.Sleep(20 * time.Millisecond) + } + return sock, res +} + +// collect attaches from from in the background, gathering entries. +type collected struct { + mu sync.Mutex + all []Entry + done chan error +} + +func collect(sock string, from int64) *collected { + c := &collected{done: make(chan error, 1)} + go func() { + c.done <- Attach(context.Background(), sock, from, func(e Entry) error { + c.mu.Lock() + c.all = append(c.all, e) + c.mu.Unlock() + return nil + }) + }() + return c +} + +func (c *collected) waitFor(t *testing.T, ok func([]Entry) bool) []Entry { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + c.mu.Lock() + got := append([]Entry(nil), c.all...) + c.mu.Unlock() + if ok(got) { + return got + } + time.Sleep(20 * time.Millisecond) + } + c.mu.Lock() + defer c.mu.Unlock() + t.Fatalf("never saw it; entries: %+v", c.all) + return nil +} + +func has(es []Entry, f func(Entry) bool) bool { + for _, e := range es { + if f(e) { + return true + } + } + return false +} + +func TestTheHostLogsWhatPassesBetweenItAndTheAgent(t *testing.T) { + dir := t.TempDir() + sock, result := started(t, dir, echoAgent) + live := collect(sock, 0) + + seq, err := Write(context.Background(), sock, json.RawMessage(`{"type": "user", "text": "hi"}`)) + if err != nil || seq == 0 { + t.Fatalf("write: %d, %v", seq, err) + } + got := live.waitFor(t, func(es []Entry) bool { + return has(es, func(e Entry) bool { return e.Dir == "out" && e.Data != nil }) + }) + for i, e := range got { + if e.Seq != int64(i+1) { + t.Fatalf("entries out of order or missing: %+v", got) + } + } + if got[0].Dir != "host" || got[0].Event != "started" || got[0].PID == 0 { + t.Fatalf("the first entry isn't the start: %+v", got[0]) + } + if !has(got, func(e Entry) bool { return e.Dir == "out" && e.Text == "ready" }) || + !has(got, func(e Entry) bool { return e.Dir == "err" && e.Text == "warming up" }) { + t.Fatalf("a line that isn't JSON wasn't kept as text: %+v", got) + } + var in, out Entry + for _, e := range got { + if e.Dir == "in" { + in = e + } + if e.Dir == "out" && e.Data != nil { + out = e + } + } + if in.Seq != seq || string(in.Data) != `{"type":"user","text":"hi"}` || string(out.Data) != string(in.Data) || out.Seq <= in.Seq { + t.Fatalf("in = %+v, out = %+v", in, out) + } + + if _, err := Write(context.Background(), sock, json.RawMessage(`not json`)); err == nil { + t.Fatal("a line that isn't JSON was written to the agent") + } + + // A reader attaching later from where another left off misses nothing. + late := collect(sock, in.Seq) + lateGot := late.waitFor(t, func(es []Entry) bool { return len(es) >= len(got)-int(in.Seq) }) + if lateGot[0].Seq != in.Seq+1 { + t.Fatalf("attaching from %d began at %d", in.Seq, lateGot[0].Seq) + } + + if err := Stop(context.Background(), sock); err != nil { + t.Fatal(err) + } + if code := <-result; code != 0 { + // The agent ends at its stdin's end, as structured agents do. + t.Fatalf("exit code %d", code) + } + if err := <-live.done; err != nil { + t.Fatal(err) + } + final := live.waitFor(t, func(es []Entry) bool { return len(es) > 0 && es[len(es)-1].Event == "exited" }) + if c := final[len(final)-1].Code; c == nil || *c != 0 { + t.Fatalf("the exit wasn't logged last: %+v", final[len(final)-1]) + } + if err := Attach(context.Background(), sock, 0, func(Entry) error { return nil }); !errors.Is(err, ErrNoHost) { + t.Fatalf("attach after the end: %v", err) + } + + // The log outlives the host, and a host started again carries it on. + var fromFile []Entry + ReadLog(filepath.Join(dir, LogName), 0, func(e Entry) error { fromFile = append(fromFile, e); return nil }) + if len(fromFile) != len(final) { + t.Fatalf("the log file has %d entries, the stream had %d", len(fromFile), len(final)) + } + sock2, result2 := started(t, dir, `sleep 0.5; exit 3`) + _ = sock2 + if code := <-result2; code != 3 { + t.Fatalf("exit code %d", code) + } + var again []Entry + ReadLog(filepath.Join(dir, LogName), int64(len(final)), func(e Entry) error { again = append(again, e); return nil }) + if len(again) != 2 || again[0].Seq != int64(len(final)+1) || again[0].Event != "started" || *again[1].Code != 3 { + t.Fatalf("a second host didn't carry the log on: %+v", again) + } +} + +func TestStoppingKillsWhatTheAgentStartedToo(t *testing.T) { + // An agent that ignores its stdin's end and SIGTERM, with a child that + // left its group, as a tool's process can. + script := `trap '' TERM; perl -e 'use POSIX; setsid(); exec "sleep", "300"' & echo "child $!"; while :; do sleep 1; done` + sock, result := started(t, t.TempDir(), script) + live := collect(sock, 0) + got := live.waitFor(t, func(es []Entry) bool { + return has(es, func(e Entry) bool { return strings.HasPrefix(e.Text, "child ") }) + }) + var child int + for _, e := range got { + if strings.HasPrefix(e.Text, "child ") { + child, _ = strconv.Atoi(strings.TrimPrefix(e.Text, "child ")) + } + } + if child == 0 || syscall.Kill(child, 0) != nil { + t.Fatalf("the child isn't running: %d", child) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := Stop(ctx, sock); err != nil { + t.Fatal(err) + } + if code := <-result; code != -1 { + t.Fatalf("a killed agent's exit code is %d", code) + } + deadline := time.Now().Add(3 * time.Second) + for syscall.Kill(child, 0) == nil { + if time.Now().After(deadline) { + syscall.Kill(child, syscall.SIGKILL) + t.Fatal("the agent's child outlived the stop") + } + time.Sleep(50 * time.Millisecond) + } +} + +func TestALogCutShortCarriesOnFromItsLastWholeEntry(t *testing.T) { + dir := t.TempDir() + l, err := OpenLog(dir) + if err != nil { + t.Fatal(err) + } + l.Append(Entry{Dir: "out", Text: "one"}) + l.Append(Entry{Dir: "out", Text: "two"}) + l.Close() + f, _ := os.OpenFile(filepath.Join(dir, LogName), os.O_APPEND|os.O_WRONLY, 0) + f.WriteString(`{"seq":3,"dir":"out","te`) + f.Close() + + l, err = OpenLog(dir) + if err != nil { + t.Fatal(err) + } + e, _ := l.Append(Entry{Dir: "out", Text: "three"}) + l.Close() + if e.Seq != 3 { + t.Fatalf("seq %d after a cut line", e.Seq) + } + b, _ := os.ReadFile(filepath.Join(dir, LogName)) + if n := bytes.Count(b, []byte("\n")); n != 3 || bytes.Contains(b, []byte(`"te`+"\n")) { + t.Fatalf("the log:\n%s", b) + } +} + +func TestAReaderThatFallsBehindIsLetGo(t *testing.T) { + l, err := OpenLog(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer l.Close() + ch, stop, err := l.Follow(0) + if err != nil { + t.Fatal(err) + } + defer stop() + for range subBuffer + 1 { + l.Append(Entry{Dir: "out", Text: "x"}) + } + n := 0 + for range ch { + n++ + } + if n != subBuffer { + t.Fatalf("read %d entries before being let go", n) + } +} diff --git a/internal/agenthost/log.go b/internal/agenthost/log.go new file mode 100644 index 00000000..7cf724d9 --- /dev/null +++ b/internal/agenthost/log.go @@ -0,0 +1,210 @@ +// Package agenthost keeps a structured agent (Claude Code's stream-json, +// Codex's app-server, an ACP agent) alive apart from berthd. `berthd +// agent-host` runs in the agent's tmux session, owns the agent's stdin and +// stdout, writes everything that passes between them to the conversation's +// log, and serves that log and the agent's stdin on a unix socket. berthd +// can restart or upgrade, or the laptop's link drop, and the agent keeps +// working: berthd attaches again from the last entry it saw. +// +// The log is berthd's own record of the conversation, so nothing the agent +// streamed is lost when it is killed mid-turn (the agents' own transcripts +// keep only finished messages). +package agenthost + +import ( + "bufio" + "bytes" + "encoding/json" + "errors" + "io" + "os" + "path/filepath" + "sync" + "time" +) + +// Entry is one line of a conversation's log. +type Entry struct { + // Seq numbers the entries from 1, across every host that ran in the + // conversation's folder (a resumed conversation carries on its log). + Seq int64 `json:"seq"` + // At is when the entry was written, in Unix milliseconds. + At int64 `json:"at"` + // Dir is where it came from: "out" the agent's stdout, "in" what was + // written to its stdin, "err" its stderr, "host" the host itself. + Dir string `json:"dir"` + // Data is a JSON line as the agent wrote or was sent it. + Data json.RawMessage `json:"data,omitempty"` + // Text is a line that isn't JSON, or what a host entry says. + Text string `json:"text,omitempty"` + // Event is a host entry's kind: "started" (PID set) or "exited" (Code + // set; -1 when the agent was killed by a signal). + Event string `json:"event,omitempty"` + PID int `json:"pid,omitempty"` + Code *int `json:"code,omitempty"` +} + +// LogName and SocketName are the files a host keeps in its folder. +const ( + LogName = "events.jsonl" + SocketName = "host.sock" +) + +// subBuffer is how many entries an attached reader may fall behind before +// it is let go (it attaches again from the last entry it saw). +const subBuffer = 1024 + +// Log is a conversation's log, appended to by one host, and the readers +// following it live. +type Log struct { + mu sync.Mutex + f *os.File + last int64 + subs map[chan Entry]struct{} +} + +// OpenLog opens the log in dir, carrying on its numbering. A line cut short +// (a host that died mid-write) is dropped. +func OpenLog(dir string) (*Log, error) { + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, err + } + path := filepath.Join(dir, LogName) + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, err + } + var last, good int64 + r := bufio.NewReader(f) + for { + line, err := r.ReadBytes('\n') + if err != nil { + break // a line without its newline is cut short + } + var e Entry + if json.Unmarshal(line, &e) == nil { + last = e.Seq + } + good += int64(len(line)) + } + if err := f.Truncate(good); err != nil { + f.Close() + return nil, err + } + if _, err := f.Seek(good, io.SeekStart); err != nil { + f.Close() + return nil, err + } + return &Log{f: f, last: last, subs: map[chan Entry]struct{}{}}, nil +} + +// Append numbers e, writes it, and hands it to the readers following. +func (l *Log) Append(e Entry) (Entry, error) { + l.mu.Lock() + defer l.mu.Unlock() + if l.f == nil { + return e, errors.New("the log is closed") + } + e.Seq = l.last + 1 + if e.At == 0 { + e.At = time.Now().UnixMilli() + } + b, err := json.Marshal(e) + if err != nil { + return e, err + } + if _, err := l.f.Write(append(b, '\n')); err != nil { + return e, err + } + l.last = e.Seq + for ch := range l.subs { + select { + case ch <- e: + default: + // Too far behind: let it go, so one slow reader never holds + // the agent up. + delete(l.subs, ch) + close(ch) + } + } + return e, nil +} + +// Follow sends the entries after from, then every new one, to the channel +// it returns, with none missed or repeated between the two. The channel +// closes when the log does, or when the reader falls too far behind; stop +// lets it go. +func (l *Log) Follow(from int64) (ch <-chan Entry, stop func(), err error) { + l.mu.Lock() + defer l.mu.Unlock() + if l.f == nil { + return nil, nil, errors.New("the log is closed") + } + var past []Entry + if from < l.last { + err := ReadLog(l.f.Name(), from, func(e Entry) error { + past = append(past, e) + return nil + }) + if err != nil { + return nil, nil, err + } + } + c := make(chan Entry, len(past)+subBuffer) + for _, e := range past { + c <- e + } + l.subs[c] = struct{}{} + stop = func() { + l.mu.Lock() + defer l.mu.Unlock() + if _, ok := l.subs[c]; ok { + delete(l.subs, c) + close(c) + } + } + return c, stop, nil +} + +// Close ends the log and every reader's channel. +func (l *Log) Close() error { + l.mu.Lock() + defer l.mu.Unlock() + for ch := range l.subs { + close(ch) + } + l.subs = map[chan Entry]struct{}{} + if l.f == nil { + return nil + } + err := l.f.Close() + l.f = nil + return err +} + +// ReadLog calls fn with each entry in the log at path after from, in +// order. A missing log has none. +func ReadLog(path string, from int64, fn func(Entry) error) error { + f, err := os.Open(path) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return err + } + defer f.Close() + r := bufio.NewReader(f) + for { + line, err := r.ReadBytes('\n') + if err != nil { + return nil // the end, or a line still being written + } + var e Entry + if json.Unmarshal(bytes.TrimSpace(line), &e) != nil || e.Seq <= from { + continue + } + if err := fn(e); err != nil { + return err + } + } +} diff --git a/internal/agenthost/procs.go b/internal/agenthost/procs.go new file mode 100644 index 00000000..b82d6428 --- /dev/null +++ b/internal/agenthost/procs.go @@ -0,0 +1,56 @@ +package agenthost + +import ( + "context" + "os/exec" + "strconv" + "strings" + "syscall" + "time" +) + +// descendants are pid's children, theirs, and so on, deepest last. An +// agent's tools can start processes in groups and sessions of their own, +// which a signal to the agent's group doesn't reach. +func descendants(pid int) []int { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + out, err := exec.CommandContext(ctx, "ps", "-A", "-o", "pid=,ppid=").Output() + if err != nil { + return nil + } + children := map[int][]int{} + for _, line := range strings.Split(string(out), "\n") { + f := strings.Fields(line) + if len(f) != 2 { + continue + } + p, err1 := strconv.Atoi(f[0]) + pp, err2 := strconv.Atoi(f[1]) + if err1 == nil && err2 == nil { + children[pp] = append(children[pp], p) + } + } + var all []int + queue := []int{pid} + for len(queue) > 0 { + p := queue[0] + queue = queue[1:] + for _, c := range children[p] { + all = append(all, c) + queue = append(queue, c) + } + } + return all +} + +// signalTree sends sig to the process tree under pid (its descendants as +// they are now, then its group, then pid itself). +func signalTree(pid int, sig syscall.Signal) { + kids := descendants(pid) + for i := len(kids) - 1; i >= 0; i-- { + syscall.Kill(kids[i], sig) + } + syscall.Kill(-pid, sig) + syscall.Kill(pid, sig) +}