1package swarm
2
3import (
4 "context"
5 "fmt"
6 "strings"
7 "sync"
8 "time"
9)
10
11// Agent is one supervised task. Public fields are immutable after
12// Spawn; mutable state (status, activity, transcript) lives behind
13// the embedded mutex.
14type Agent struct {
15 ID string
16 Task string
17 Dir string // always the host's RepoRoot; agents share its cwd.
18 Started time.Time
19
20 // Model and Provider, when non-empty, override the child
21 // subprocess's model resolution. Empty means "inherit whatever
22 // the child resolves on its own from config / env / flags" —
23 // the historical behaviour. Persisted in meta.json so Resume
24 // keeps using the same model across zot restarts.
25 Model string
26 Provider string
27
28 // SessionID, when non-empty, scopes the agent to a particular
29 // host zot session: the dashboard only surfaces agents whose
30 // SessionID matches the active session. Empty means "unscoped"
31 // (legacy meta files from before the field existed, or agents
32 // spawned without a session context such as in tests). Set at
33 // Spawn time from Swarm.activeSession and persisted in
34 // meta.json so the scope survives restarts.
35 SessionID string
36
37 // Resuming is true when this Agent struct was built by Resume
38 // rather than Spawn. The runner consults it to decide whether
39 // to pass the original Task as a positional argv to the child:
40 // on first Spawn we want the child to run the task immediately;
41 // on Resume the task was already run last time and replaying it
42 // would produce a second turn that collides ("agent busy; send
43 // 'cancel' first") with whatever the user types next. Not
44 // persisted — every Resume sets it explicitly.
45 Resuming bool
46
47 // InboxPath is the unix socket the child agent listens on for
48 // follow-up prompts and control messages. The supervisor opens
49 // an Inbox at this path; the child opens a Listener.
50 InboxPath string
51
52 // EventLogPath is the durable JSONL event log for this agent.
53 // The runner appends every well-formed event from the child
54 // (plus lifecycle events of its own) here. /swarm open in any
55 // zot process reads from this file to replay the full history.
56 EventLogPath string
57
58 // SessionPath is the child's persistent session file. Surfaced
59 // so the dashboard / /swarm open can resume the agent later.
60 SessionPath string
61
62 // inbox is the supervisor-side socket handle. Populated by
63 // Swarm.Spawn so SendInput can route messages without each
64 // caller redialing.
65 inbox *Inbox
66
67 mu sync.Mutex
68 status Status
69 activity string
70 transcript []string
71 lastAssistant string
72 finished time.Time
73 lastErr error
74
75 // OnTurnEnd, if set, fires once per prompt-level turn_end event
76 // emitted by the swarm daemon wrapper. Provider/tool-loop
77 // turn_end events (for example stop=tool_use) are ignored by the
78 // runner because they do not mean the sub-agent task finished.
79 // Used by auto-swarm watchers to detect that a sub-agent's first
80 // (or n-th) task has finished without waiting for the long-lived
81 // daemon itself to exit — sub-agents keep running on the inbox
82 // even after the initial task completes, so Wait() never unblocks
83 // for them.
84 OnTurnEnd func(step int, errMsg string)
85
86 ctx context.Context
87 cancel context.CancelFunc
88 runner Runner
89
90 // done closes when the run goroutine finalises the agent's
91 // status (done / failed / killed). Wait blocks on this so
92 // callers don't have to poll.
93 done chan struct{}
94}
95
96// Inbox exposes the supervisor-side socket handle. Returns nil for
97// agents without inboxes (e.g. tests using a custom Runner that
98// doesn't speak the daemon protocol).
99func (a *Agent) Inbox() *Inbox { return a.inbox }
100
101// Status returns the current high-level status. Cheap; safe from any goroutine.
102func (a *Agent) Status() Status {
103 a.mu.Lock()
104 defer a.mu.Unlock()
105 return a.status
106}
107
108// Activity returns the current one-line activity string.
109func (a *Agent) Activity() string {
110 a.mu.Lock()
111 defer a.mu.Unlock()
112 return a.activity
113}
114
115// Transcript returns a copy of the running transcript.
116func (a *Agent) Transcript() []string {
117 a.mu.Lock()
118 defer a.mu.Unlock()
119 out := make([]string, len(a.transcript))
120 copy(out, a.transcript)
121 return out
122}
123
124// Err returns the runner's terminal error, if any.
125func (a *Agent) Err() error {
126 a.mu.Lock()
127 defer a.mu.Unlock()
128 return a.lastErr
129}
130
131// Wait blocks until the agent reaches a terminal state. Used by tests
132// and by /swarm wait <id>.
133func (a *Agent) Wait() { <-a.done }
134
135// SetOnTurnEnd installs (or clears, with nil) the callback fired
136// from the runner when the child daemon emits a prompt-level
137// turn_end event with a step field. Safe to call from any goroutine:
138// the runner reads the callback under the same mutex.
139func (a *Agent) SetOnTurnEnd(fn func(step int, errMsg string)) {
140 a.mu.Lock()
141 a.OnTurnEnd = fn
142 a.mu.Unlock()
143}
144
145func (a *Agent) setStatus(s Status) {
146 a.mu.Lock()
147 a.status = s
148 a.mu.Unlock()
149}
150
151func (a *Agent) setActivity(msg string) {
152 a.mu.Lock()
153 a.activity = strings.TrimSpace(msg)
154 a.mu.Unlock()
155}
156
157func (a *Agent) appendTranscript(chunk string) {
158 a.appendTranscriptLocked(chunk, "", false)
159}
160
161func (a *Agent) appendUserMessage(text string) {
162 a.appendTranscriptLocked(text, "user: ", false)
163}
164
165func (a *Agent) appendAssistantMessage(text string) {
166 a.appendTranscriptLocked(text, "", true)
167}
168
169func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {
170 message := chunk
171 chunk = strings.TrimRight(chunk, "\n")
172 if chunk == "" {
173 return
174 }
175 a.mu.Lock()
176 for _, line := range strings.Split(chunk, "\n") {
177 a.transcript = append(a.transcript, linePrefix+line)
178 }
179 if assistant {
180 a.lastAssistant = message
181 }
182 // Bound the transcript so dashboards don't hold gigabytes for
183 // long-running agents. 2000 lines is plenty for inspection;
184 // the durable record lives in the agent's session file.
185 const cap = 2000
186 if len(a.transcript) > cap {
187 a.transcript = a.transcript[len(a.transcript)-cap:]
188 }
189 a.mu.Unlock()
190}
191
192// newAgentID returns a short, mostly-collision-free identifier of the
193// form "<slug>-<nano>". The slug is derived from the task text so
194// dashboards stay readable; the nano suffix guarantees uniqueness
195// even when two agents are spawned in the same millisecond.
196func newAgentID(task string, now time.Time) string {
197 slug := taskSlug(task)
198 return fmt.Sprintf("%s-%d", slug, now.UnixNano()%1_000_000)
199}
200
201func taskSlug(task string) string {
202 task = strings.ToLower(task)
203 var b strings.Builder
204 dash := false
205 for _, r := range task {
206 switch {
207 case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
208 b.WriteRune(r)
209 dash = false
210 default:
211 if !dash && b.Len() > 0 {
212 b.WriteByte('-')
213 dash = true
214 }
215 }
216 if b.Len() >= 24 {
217 break
218 }
219 }
220 out := strings.Trim(b.String(), "-")
221 if out == "" {
222 out = "agent"
223 }
224 return out
225}