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 // Large detached logs are loaded in full only when their transcript is viewed or resumed.
73 needsFullReplay bool
74 finished time.Time
75 lastErr error
76
77 // OnTurnEnd, if set, fires once per prompt-level turn_end event
78 // emitted by the swarm daemon wrapper. Provider/tool-loop
79 // turn_end events (for example stop=tool_use) are ignored by the
80 // runner because they do not mean the sub-agent task finished.
81 // Used by auto-swarm watchers to detect that a sub-agent's first
82 // (or n-th) task has finished without waiting for the long-lived
83 // daemon itself to exit — sub-agents keep running on the inbox
84 // even after the initial task completes, so Wait() never unblocks
85 // for them.
86 OnTurnEnd func(step int, errMsg string)
87
88 ctx context.Context
89 cancel context.CancelFunc
90 runner Runner
91
92 // done closes when the run goroutine finalises the agent's
93 // status (done / failed / killed). Wait blocks on this so
94 // callers don't have to poll.
95 done chan struct{}
96}
97
98// Inbox exposes the supervisor-side socket handle. Returns nil for
99// agents without inboxes (e.g. tests using a custom Runner that
100// doesn't speak the daemon protocol).
101func (a *Agent) Inbox() *Inbox { return a.inbox }
102
103// Status returns the current high-level status. Cheap; safe from any goroutine.
104func (a *Agent) Status() Status {
105 a.mu.Lock()
106 defer a.mu.Unlock()
107 return a.status
108}
109
110// Activity returns the current one-line activity string.
111func (a *Agent) Activity() string {
112 a.mu.Lock()
113 defer a.mu.Unlock()
114 return a.activity
115}
116
117// Transcript returns a copy of the running transcript.
118func (a *Agent) Transcript() []string {
119 a.mu.Lock()
120 defer a.mu.Unlock()
121 out := make([]string, len(a.transcript))
122 copy(out, a.transcript)
123 return out
124}
125
126// Err returns the runner's terminal error, if any.
127func (a *Agent) Err() error {
128 a.mu.Lock()
129 defer a.mu.Unlock()
130 return a.lastErr
131}
132
133// Wait blocks until the agent reaches a terminal state. Used by tests
134// and by /swarm wait <id>.
135func (a *Agent) Wait() { <-a.done }
136
137// SetOnTurnEnd installs (or clears, with nil) the callback fired
138// from the runner when the child daemon emits a prompt-level
139// turn_end event with a step field. Safe to call from any goroutine:
140// the runner reads the callback under the same mutex.
141func (a *Agent) SetOnTurnEnd(fn func(step int, errMsg string)) {
142 a.mu.Lock()
143 a.OnTurnEnd = fn
144 a.mu.Unlock()
145}
146
147func (a *Agent) setStatus(s Status) {
148 a.mu.Lock()
149 a.status = s
150 a.mu.Unlock()
151}
152
153func (a *Agent) setActivity(msg string) {
154 a.mu.Lock()
155 a.activity = strings.TrimSpace(msg)
156 a.mu.Unlock()
157}
158
159func (a *Agent) appendTranscript(chunk string) {
160 a.appendTranscriptLocked(chunk, "", false)
161}
162
163func (a *Agent) appendUserMessage(text string) {
164 a.appendTranscriptLocked(text, "user: ", false)
165}
166
167func (a *Agent) appendAssistantMessage(text string) {
168 a.appendTranscriptLocked(text, "", true)
169}
170
171func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {
172 message := chunk
173 chunk = strings.TrimRight(chunk, "\n")
174 if chunk == "" {
175 return
176 }
177 a.mu.Lock()
178 for _, line := range strings.Split(chunk, "\n") {
179 a.transcript = append(a.transcript, linePrefix+line)
180 }
181 if assistant {
182 a.lastAssistant = message
183 }
184 // Bound the transcript so dashboards don't hold gigabytes for
185 // long-running agents. 2000 lines is plenty for inspection;
186 // the durable record lives in the agent's session file.
187 const cap = 2000
188 if len(a.transcript) > cap {
189 a.transcript = a.transcript[len(a.transcript)-cap:]
190 }
191 a.mu.Unlock()
192}
193
194// newAgentID returns a short, mostly-collision-free identifier of the
195// form "<slug>-<nano>". The slug is derived from the task text so
196// dashboards stay readable; the nano suffix guarantees uniqueness
197// even when two agents are spawned in the same millisecond.
198func newAgentID(task string, now time.Time) string {
199 slug := taskSlug(task)
200 return fmt.Sprintf("%s-%d", slug, now.UnixNano()%1_000_000)
201}
202
203func taskSlug(task string) string {
204 task = strings.ToLower(task)
205 var b strings.Builder
206 dash := false
207 for _, r := range task {
208 switch {
209 case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
210 b.WriteRune(r)
211 dash = false
212 default:
213 if !dash && b.Len() > 0 {
214 b.WriteByte('-')
215 dash = true
216 }
217 }
218 if b.Len() >= 24 {
219 break
220 }
221 }
222 out := strings.Trim(b.String(), "-")
223 if out == "" {
224 out = "agent"
225 }
226 return out
227}