| 1 | package agents |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "strings" |
| 6 | "sync" |
| 7 | |
| 8 | "lchat/internal/client" |
| 9 | ) |
| 10 | |
| 11 | // maxTurnText keeps a session.turn's payload safely under the server's 64K |
| 12 | // blob cap (ciphertext is base64 of the JSON payload plus overhead). A turn |
| 13 | // this long is pathological anyway; the tail carries the detail. |
| 14 | const maxTurnText = 40000 |
| 15 | |
| 16 | // Bridge is the running sidecar: one E2EE member (@claude) fronting this |
| 17 | // machine's Claude Code sessions to every group the config grants. |
| 18 | type Bridge struct { |
| 19 | // OnReady, if set, fires each time the bridge finishes connecting (tests |
| 20 | // wait on it; live frames sent before the connect are only seen via |
| 21 | // catch-up on the NEXT connect, so callers who race sends need it). |
| 22 | OnReady func() |
| 23 | |
| 24 | cfg *Config |
| 25 | cl *client.Client |
| 26 | logf func(string, ...any) |
| 27 | uid string |
| 28 | |
| 29 | mu sync.Mutex |
| 30 | sessions map[string]*session // by session id |
| 31 | } |
| 32 | |
| 33 | // New wires a bridge onto a member client. The caller owns the session file |
| 34 | // (bot.EnsureSession) exactly as the other sidecars do. |
| 35 | func New(base string, sess *client.Session, cfg *Config, logf func(string, ...any)) *Bridge { |
| 36 | b := &Bridge{cfg: cfg, logf: logf, sessions: map[string]*session{}} |
| 37 | b.cl = &client.Client{Base: base, Session: sess, Logf: logf} |
| 38 | b.cl.OnMessage = b.onMessage |
| 39 | b.cl.OnEvent = b.onEvent |
| 40 | return b |
| 41 | } |
| 42 | |
| 43 | // Run connects and serves until ctx is cancelled. Sessions the bridge spawned |
| 44 | // are closed on the way out — their worktrees are only removed by an explicit |
| 45 | // session.end, so a bridge restart leaves work in place and the CLI's own |
| 46 | // session files resumable. |
| 47 | func (b *Bridge) Run(ctx context.Context) error { |
| 48 | if b.uid == "" { |
| 49 | if uid, err := b.cl.Me(); err == nil { |
| 50 | b.uid = uid |
| 51 | } |
| 52 | } |
| 53 | b.cl.OnReady = b.OnReady |
| 54 | err := b.cl.Run(ctx) |
| 55 | b.mu.Lock() |
| 56 | open := make([]*session, 0, len(b.sessions)) |
| 57 | for _, s := range b.sessions { |
| 58 | open = append(open, s) |
| 59 | } |
| 60 | b.mu.Unlock() |
| 61 | for _, s := range open { |
| 62 | s.mu.Lock() |
| 63 | ended := s.ended |
| 64 | s.mu.Unlock() |
| 65 | if !ended { |
| 66 | // Terminate the process but keep the worktree: only session.end |
| 67 | // means close-and-clean; a dying bridge must not eat work. |
| 68 | wt := s.worktree |
| 69 | s.worktree = "" |
| 70 | s.close(b.logf) |
| 71 | s.worktree = wt |
| 72 | } |
| 73 | } |
| 74 | return err |
| 75 | } |
| 76 | |
| 77 | // onMessage is the conversational trigger: an "@claude …" message in a |
| 78 | // configured group steers the live session (or births one — the spec's |
| 79 | // conversational birth). The bridge only ever acts on members' typed text and |
| 80 | // events; session OUTPUT lands in chat as content, never as commands. |
| 81 | func (b *Bridge) onMessage(m client.Message) { |
| 82 | if m.Mine { |
| 83 | return |
| 84 | } |
| 85 | group, ok := b.cfg.Groups[m.ThreadID] |
| 86 | if !ok { |
| 87 | return |
| 88 | } |
| 89 | text := strings.TrimSpace(m.Text) |
| 90 | lower := strings.ToLower(text) |
| 91 | if !strings.HasPrefix(lower, "@claude") { |
| 92 | return |
| 93 | } |
| 94 | prompt := strings.TrimSpace(text[len("@claude"):]) |
| 95 | if prompt == "" { |
| 96 | return |
| 97 | } |
| 98 | live := b.liveSessions(m.ThreadID) |
| 99 | switch len(live) { |
| 100 | case 0: |
| 101 | if len(group.Roots) == 0 { |
| 102 | b.reply(m.ThreadID, "This group has no project roots configured on the workstation.") |
| 103 | return |
| 104 | } |
| 105 | s, err := b.spawn(m.ThreadID, group.Roots[0], slugify(prompt), m.Name) |
| 106 | if err != nil { |
| 107 | b.logf("agents: spawn: %v", err) |
| 108 | b.reply(m.ThreadID, "Couldn't start a session on the workstation — check the bridge log.") |
| 109 | return |
| 110 | } |
| 111 | b.steer(s, m.Name, prompt) |
| 112 | case 1: |
| 113 | b.steer(live[0], m.Name, prompt) |
| 114 | default: |
| 115 | // Several desks: chat-only routing (reply-pinning, composer chips) is |
| 116 | // client work still to come, so be honest rather than guess a target. |
| 117 | b.reply(m.ThreadID, "Several sessions are live — steer one from the Agents board for now.") |
| 118 | } |
| 119 | } |
| 120 | |
| 121 | // onEvent handles members' typed steering events (the board's emits, and — |
| 122 | // once composer routing ships — the prompt-is-the-message path). |
| 123 | func (b *Bridge) onEvent(e client.Event) { |
| 124 | if uid, _ := e.Payload["uid"].(string); uid != "" && uid == b.uid { |
| 125 | return // our own echo |
| 126 | } |
| 127 | if _, ok := b.cfg.Groups[e.ThreadID]; !ok { |
| 128 | return |
| 129 | } |
| 130 | id, _ := e.Payload["id"].(string) |
| 131 | switch e.Type { |
| 132 | case "session.prompt": |
| 133 | text, _ := e.Payload["text"].(string) |
| 134 | name, _ := e.Payload["name"].(string) |
| 135 | if s := b.session(id); s != nil && text != "" { |
| 136 | b.steer(s, name, text) |
| 137 | } |
| 138 | case "session.end": |
| 139 | if s := b.session(id); s != nil { |
| 140 | // The member's event is the durable record (the fold marks the |
| 141 | // session ended); the bridge's job is the machine side — end the |
| 142 | // process and remove the worktree (close-session-closes-worktree). |
| 143 | s.close(b.logf) |
| 144 | b.mu.Lock() |
| 145 | delete(b.sessions, id) |
| 146 | b.mu.Unlock() |
| 147 | } |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | func (b *Bridge) session(id string) *session { |
| 152 | b.mu.Lock() |
| 153 | defer b.mu.Unlock() |
| 154 | return b.sessions[id] |
| 155 | } |
| 156 | |
| 157 | func (b *Bridge) liveSessions(threadID string) []*session { |
| 158 | b.mu.Lock() |
| 159 | defer b.mu.Unlock() |
| 160 | var out []*session |
| 161 | for _, s := range b.sessions { |
| 162 | s.mu.Lock() |
| 163 | ended := s.ended |
| 164 | s.mu.Unlock() |
| 165 | if s.threadID == threadID && !ended { |
| 166 | out = append(out, s) |
| 167 | } |
| 168 | } |
| 169 | return out |
| 170 | } |
| 171 | |
| 172 | // spawn births a scoped session under a configured root and announces it. |
| 173 | func (b *Bridge) spawn(threadID, root, name, by string) (*session, error) { |
| 174 | s := &session{ |
| 175 | id: newSessionID(), |
| 176 | name: name, |
| 177 | flavor: "scoped", |
| 178 | threadID: threadID, |
| 179 | root: root, |
| 180 | sink: b.onAct, |
| 181 | } |
| 182 | if err := s.spawn(b.cfg.Command, nil, b.logf); err != nil { |
| 183 | return nil, err |
| 184 | } |
| 185 | b.mu.Lock() |
| 186 | b.sessions[s.id] = s |
| 187 | b.mu.Unlock() |
| 188 | b.post(threadID, map[string]any{ |
| 189 | "type": "session.spawn", "id": s.id, "session": s.name, "flavor": s.flavor, |
| 190 | "root": root, "worktree": s.worktree, "by": by, |
| 191 | }) |
| 192 | return s, nil |
| 193 | } |
| 194 | |
| 195 | func (b *Bridge) steer(s *session, speaker, text string) { |
| 196 | if err := s.steer(speaker, text); err != nil { |
| 197 | b.logf("agents: steer %s: %v", s.name, err) |
| 198 | return |
| 199 | } |
| 200 | b.post(s.threadID, map[string]any{"type": "session.status", "id": s.id, "state": "running"}) |
| 201 | } |
| 202 | |
| 203 | // onAct receives a session's translated stream acts and turns them into |
| 204 | // thread events: turns and status transitions durably, tool chatter as |
| 205 | // relay-only frames (the durable/ephemeral split is the design's |
| 206 | // load-bearing wall — a busy session must not bloat every member's log). |
| 207 | func (b *Bridge) onAct(s *session, act Act) { |
| 208 | switch act.Kind { |
| 209 | case "tool": |
| 210 | if err := b.cl.SendEphemeral(s.threadID, map[string]any{ |
| 211 | "type": "session.tool", "id": s.id, "line": act.Tool, |
| 212 | }); err != nil { |
| 213 | // Best-effort by design: an ephemeral frame needs a live socket, |
| 214 | // and losing one loses nothing durable. |
| 215 | b.logf("agents: %s: tool frame: %v", s.name, err) |
| 216 | } |
| 217 | case "turn": |
| 218 | text := act.Text |
| 219 | if len(text) > maxTurnText { |
| 220 | text = text[:maxTurnText] + "…" |
| 221 | } |
| 222 | b.post(s.threadID, map[string]any{"type": "session.turn", "id": s.id, "session": s.name, "text": text}) |
| 223 | b.post(s.threadID, map[string]any{"type": "session.status", "id": s.id, "state": "idle"}) |
| 224 | case "error": |
| 225 | b.post(s.threadID, map[string]any{"type": "session.turn", "id": s.id, "session": s.name, "text": "⚠️ " + act.Text}) |
| 226 | b.post(s.threadID, map[string]any{"type": "session.status", "id": s.id, "state": "idle"}) |
| 227 | case "eof": |
| 228 | s.mu.Lock() |
| 229 | ended := s.ended |
| 230 | s.mu.Unlock() |
| 231 | if !ended { |
| 232 | // The process died under us — the board must say so, not linger |
| 233 | // "running" (a dead bridge silently eating prompts is the worst |
| 234 | // failure mode; a dead session is its little sibling). |
| 235 | if s.cmd != nil { |
| 236 | s.cmd.Wait() |
| 237 | } |
| 238 | b.post(s.threadID, map[string]any{"type": "session.status", "id": s.id, "state": "dead"}) |
| 239 | } |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | func (b *Bridge) post(threadID string, event map[string]any) { |
| 244 | if err := b.cl.SendEvent(threadID, event); err != nil { |
| 245 | b.logf("agents: post %s: %v", event["type"], err) |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | func (b *Bridge) reply(threadID, text string) { |
| 250 | if err := b.cl.Send(threadID, text); err != nil { |
| 251 | b.logf("agents: reply: %v", err) |
| 252 | } |
| 253 | } |
| 254 | |
amadan