| 1 | package agents |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "crypto/rand" |
| 6 | "encoding/hex" |
| 7 | "encoding/json" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "os" |
| 11 | "os/exec" |
| 12 | "path/filepath" |
| 13 | "regexp" |
| 14 | "strings" |
| 15 | "sync" |
| 16 | "time" |
| 17 | ) |
| 18 | |
| 19 | // session is one live headless Claude Code process. The bridge keeps exactly |
| 20 | // one process per session for its whole life — steering rides stdin |
| 21 | // (stream-json input), so a turn's context never leaves the process; if the |
| 22 | // bridge dies, the CLI's own on-disk session survives and a human can |
| 23 | // `claude --resume <claudeSessionID>` from a real terminal (the honest |
| 24 | // handoff the spec settles on). |
| 25 | type session struct { |
| 26 | id string // our id — the one session.* events carry |
| 27 | name string // the human tag ("fix-login") |
| 28 | flavor string // "scoped" | "plain" |
| 29 | threadID string |
| 30 | root string // scoped: the configured root it was spawned under |
| 31 | worktree string // scoped in a git root: the fresh worktree (else "") |
| 32 | branch string // the worktree's branch (survives worktree removal) |
| 33 | |
| 34 | cmd *exec.Cmd |
| 35 | stdin io.WriteCloser |
| 36 | sink actSink // set before spawn; translated stream acts flow up through it |
| 37 | |
| 38 | mu sync.Mutex |
| 39 | claudeSessionID string // from the init event; what --resume takes |
| 40 | ended bool |
| 41 | } |
| 42 | |
| 43 | // actSink is the bridge's receiver for a session's translated stream acts. |
| 44 | type actSink func(*session, Act) |
| 45 | |
| 46 | func newSessionID() string { |
| 47 | b := make([]byte, 6) |
| 48 | rand.Read(b) |
| 49 | return "s-" + hex.EncodeToString(b) |
| 50 | } |
| 51 | |
| 52 | var slugRe = regexp.MustCompile(`[^a-z0-9]+`) |
| 53 | |
| 54 | // slugify derives a session tag from the first words of a prompt — |
| 55 | // "fix the flaky reminder test" → "fix-the-flaky". |
| 56 | func slugify(text string) string { |
| 57 | words := strings.Fields(strings.ToLower(text)) |
| 58 | if len(words) > 3 { |
| 59 | words = words[:3] |
| 60 | } |
| 61 | s := slugRe.ReplaceAllString(strings.Join(words, "-"), "-") |
| 62 | s = strings.Trim(s, "-") |
| 63 | if len(s) > 40 { |
| 64 | s = s[:40] |
| 65 | } |
| 66 | if s == "" { |
| 67 | s = "session" |
| 68 | } |
| 69 | return s |
| 70 | } |
| 71 | |
| 72 | // spawn starts the claude process for a session. Scoped sessions in a git |
| 73 | // root get a fresh worktree beside the root (parallel sessions never trample |
| 74 | // each other; the repo's own .claude settings, CLAUDE.md and permission |
| 75 | // config govern the session exactly as they would a human dev); a non-git |
| 76 | // root runs in the root itself. Plain sessions run from the operator's home. |
| 77 | func (s *session) spawn(command string, extraArgs []string, logf func(string, ...any)) error { |
| 78 | dir := s.root |
| 79 | if s.flavor == "plain" { |
| 80 | home, err := os.UserHomeDir() |
| 81 | if err != nil { |
| 82 | return err |
| 83 | } |
| 84 | dir = home |
| 85 | } else if isGitRepo(s.root) { |
| 86 | wt := filepath.Join(filepath.Dir(s.root), "agents-"+s.name+"-"+strings.TrimPrefix(s.id, "s-")[:4]) |
| 87 | branch := "agents/" + s.name + "-" + strings.TrimPrefix(s.id, "s-")[:4] |
| 88 | out, err := exec.Command("git", "-C", s.root, "worktree", "add", "-b", branch, wt).CombinedOutput() |
| 89 | if err != nil { |
| 90 | return fmt.Errorf("worktree add: %v: %s", err, strings.TrimSpace(string(out))) |
| 91 | } |
| 92 | s.worktree, s.branch, dir = wt, branch, wt |
| 93 | } |
| 94 | |
| 95 | args := append([]string{ |
| 96 | "-p", |
| 97 | "--input-format", "stream-json", |
| 98 | "--output-format", "stream-json", |
| 99 | "--verbose", |
| 100 | "--append-system-prompt", "You are driven from a shared group chat; each prompt is prefixed with the speaker's name, and several people may steer you.", |
| 101 | }, extraArgs...) |
| 102 | cmd := exec.Command(command, args...) |
| 103 | cmd.Dir = dir |
| 104 | stdin, err := cmd.StdinPipe() |
| 105 | if err != nil { |
| 106 | return err |
| 107 | } |
| 108 | stdout, err := cmd.StdoutPipe() |
| 109 | if err != nil { |
| 110 | return err |
| 111 | } |
| 112 | cmd.Stderr = &logWriter{prefix: s.name, logf: logf} |
| 113 | if err := cmd.Start(); err != nil { |
| 114 | s.removeWorktree(logf) |
| 115 | return err |
| 116 | } |
| 117 | s.cmd, s.stdin = cmd, stdin |
| 118 | go s.readLoop(stdout, logf) |
| 119 | return nil |
| 120 | } |
| 121 | |
| 122 | // steer writes one attributed user turn onto the session's stdin. The CLI |
| 123 | // processes stdin messages in order, so mid-turn prompts queue naturally. |
| 124 | func (s *session) steer(speaker, text string) error { |
| 125 | if speaker != "" { |
| 126 | text = speaker + ": " + text |
| 127 | } |
| 128 | msg := map[string]any{ |
| 129 | "type": "user", |
| 130 | "message": map[string]any{ |
| 131 | "role": "user", |
| 132 | "content": []map[string]any{{"type": "text", "text": text}}, |
| 133 | }, |
| 134 | } |
| 135 | data, err := json.Marshal(msg) |
| 136 | if err != nil { |
| 137 | return err |
| 138 | } |
| 139 | s.mu.Lock() |
| 140 | defer s.mu.Unlock() |
| 141 | if s.ended || s.stdin == nil { |
| 142 | return fmt.Errorf("session %s is not running", s.name) |
| 143 | } |
| 144 | _, err = s.stdin.Write(append(data, '\n')) |
| 145 | return err |
| 146 | } |
| 147 | |
| 148 | // readLoop consumes the process's stream-json stdout until EOF, translating |
| 149 | // each line. Lines can be large (a whole assistant message); the scanner |
| 150 | // buffer allows well past the server's 64K blob cap. |
| 151 | func (s *session) readLoop(stdout io.Reader, logf func(string, ...any)) { |
| 152 | sc := bufio.NewScanner(stdout) |
| 153 | sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024) |
| 154 | for sc.Scan() { |
| 155 | for _, act := range Translate(sc.Bytes()) { |
| 156 | if act.Kind == "init" { |
| 157 | s.mu.Lock() |
| 158 | s.claudeSessionID = act.ClaudeSessionID |
| 159 | s.mu.Unlock() |
| 160 | } |
| 161 | if s.sink != nil { |
| 162 | s.sink(s, act) |
| 163 | } |
| 164 | } |
| 165 | } |
| 166 | if err := sc.Err(); err != nil { |
| 167 | logf("agents: %s: read: %v", s.name, err) |
| 168 | } |
| 169 | // EOF: the process is going (or gone). A synthetic act lets the bridge |
| 170 | // distinguish a deliberate close from a session dying under it. |
| 171 | if s.sink != nil { |
| 172 | s.sink(s, Act{Kind: "eof"}) |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | // close ends the process (stdin close lets an idle -p session exit cleanly; |
| 177 | // SIGTERM then kill covers a stuck one) and removes the worktree — the |
| 178 | // accepted design's close-session-closes-worktree: the checkout goes, the |
| 179 | // branch (any committed work) survives, and the CLI's own session file stays |
| 180 | // resumable from the ROOT checkout via --resume. |
| 181 | func (s *session) close(logf func(string, ...any)) { |
| 182 | s.mu.Lock() |
| 183 | if s.ended { |
| 184 | s.mu.Unlock() |
| 185 | return |
| 186 | } |
| 187 | s.ended = true |
| 188 | stdin, cmd := s.stdin, s.cmd |
| 189 | s.mu.Unlock() |
| 190 | if stdin != nil { |
| 191 | stdin.Close() |
| 192 | } |
| 193 | if cmd != nil && cmd.Process != nil { |
| 194 | done := make(chan struct{}) |
| 195 | go func() { cmd.Wait(); close(done) }() |
| 196 | select { |
| 197 | case <-done: |
| 198 | case <-time.After(5 * time.Second): |
| 199 | cmd.Process.Kill() |
| 200 | <-done |
| 201 | } |
| 202 | } |
| 203 | s.removeWorktree(logf) |
| 204 | } |
| 205 | |
| 206 | func (s *session) removeWorktree(logf func(string, ...any)) { |
| 207 | if s.worktree == "" { |
| 208 | return |
| 209 | } |
| 210 | // --force: closing the session closes the worktree, uncommitted changes |
| 211 | // included — that's what the board's confirm says. Committed work lives on |
| 212 | // the branch. |
| 213 | out, err := exec.Command("git", "-C", s.root, "worktree", "remove", "--force", s.worktree).CombinedOutput() |
| 214 | if err != nil { |
| 215 | logf("agents: %s: worktree remove: %v: %s", s.name, err, strings.TrimSpace(string(out))) |
| 216 | } |
| 217 | s.worktree = "" |
| 218 | } |
| 219 | |
| 220 | func isGitRepo(dir string) bool { |
| 221 | return exec.Command("git", "-C", dir, "rev-parse", "--git-dir").Run() == nil |
| 222 | } |
| 223 | |
| 224 | // logWriter forwards a process's stderr to the bridge log, line-tagged. |
| 225 | type logWriter struct { |
| 226 | prefix string |
| 227 | logf func(string, ...any) |
| 228 | buf []byte |
| 229 | } |
| 230 | |
| 231 | func (w *logWriter) Write(p []byte) (int, error) { |
| 232 | w.buf = append(w.buf, p...) |
| 233 | for { |
| 234 | i := strings.IndexByte(string(w.buf), '\n') |
| 235 | if i < 0 { |
| 236 | break |
| 237 | } |
| 238 | if line := strings.TrimSpace(string(w.buf[:i])); line != "" { |
| 239 | w.logf("agents: %s: %s", w.prefix, line) |
| 240 | } |
| 241 | w.buf = w.buf[i+1:] |
| 242 | } |
| 243 | return len(p), nil |
| 244 | } |
| 245 | |
amadan