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