1package subagents
2
3import (
4 "context"
5 "fmt"
6 "path/filepath"
7 "strings"
8 "sync"
9 "time"
10)
11
12// Agent is one supervised task. Public fields are immutable after
13// Spawn; mutable state (status, activity, transcript) lives behind
14// the embedded mutex.
15type turnEndNotice struct {
16 step int
17 errMsg string
18}
19
20type Agent struct {
21 ID string
22 Task string
23 Dir string // shared checkout or isolated worktree used by the child.
24 Started time.Time
25
26 // ParentID and RootSessionID identify the supervisor/session that owns
27 // this worker. Empty ParentID means the main supervisor created it;
28 // child-originated spawning is rejected in v1.
29 ParentID string
30 BatchID string
31 RootSessionID string
32 OriginalTask string
33 RepositoryRoot string
34 WorkspaceMode WorkspaceMode
35 WorkspacePath string
36 WorkspaceBase string
37 WorkspaceCapture CaptureMode
38 MaxTurns int
39 // Timeout is the effective per-agent lifetime, persisted so a resumed
40 // worker keeps the timeout selected at its spawn boundary.
41 Timeout time.Duration
42 HeartbeatInterval time.Duration
43 Tools []string
44
45 // Model and Provider, when non-empty, override the child
46 // subprocess's model resolution. BaseURL and InsecureTLS carry
47 // the effective provider connection settings from the supervisor.
48 // Empty values inherit whatever the child resolves on its own from
49 // config / env / flags. All four settings are persisted in meta.json
50 // so Resume keeps using the same provider configuration across zut
51 // restarts.
52 Model string
53 Provider string
54 BaseURL string
55 InsecureTLS bool
56 Reasoning string
57 FastMode bool
58 Subagent string
59
60 // SessionID, when non-empty, scopes the agent to a particular
61 // host zut session: the dashboard only surfaces agents whose
62 // SessionID matches the active session. Empty means "unscoped" and
63 // applies to agents spawned without a session context such as in tests. Set at
64 // Spawn time from Supervisor.activeSession and persisted in
65 // meta.json so the scope survives restarts.
66 SessionID string
67
68 // Resuming is true when this Agent struct was built by Resume
69 // rather than Spawn. The runner consults it to decide whether
70 // to pass the original Task as a positional argv to the child:
71 // on first Spawn we want the child to run the task immediately;
72 // on Resume the task was already run last time and replaying it
73 // would produce a second turn that collides ("agent busy; send
74 // 'cancel' first") with whatever the user types next. Not
75 // persisted — every Resume sets it explicitly.
76 Resuming bool
77
78 // InboxPath is the unix socket the child agent listens on for
79 // follow-up prompts and control messages. The supervisor opens
80 // an Inbox at this path; the child opens a Listener.
81 InboxPath string
82
83 // Attempt increments whenever the supervisor starts a process for this
84 // agent. It is persisted so recovery can distinguish a resumed worker
85 // from a first launch.
86 Attempt int
87
88 // ProcessPID identifies the current worker when known. It is persisted
89 // so a reloaded supervisor can force-kill a detached worker after a
90 // graceful shutdown deadline.
91 ProcessPID int
92
93 // EventLogPath is the durable JSONL event log for this agent.
94 // The runner appends every well-formed event from the child
95 // (plus lifecycle events of its own) here. /subagents open in any
96 // zut process reads from this file to replay the full history.
97 EventLogPath string
98
99 // SessionPath is the child's persistent session file. Surfaced
100 // so the dashboard / /subagents open can resume the agent later.
101 SessionPath string
102
103 // inbox is the supervisor-side socket handle. Populated by
104 // Supervisor.Spawn so callers do not need to manage the socket directly.
105 inbox *Inbox
106
107 mu sync.Mutex
108 status Status
109 activity string
110 transcript []string
111 lastAssistant string
112 finished time.Time
113 lastErr error
114
115 lifecycleMu sync.Mutex
116 processState ProcessState
117 turnState TurnState
118 currentTurnID string
119 updatedAt time.Time
120 lastActivity time.Time
121 result *TurnResult
122 resultRef string
123 patchRef string
124 changedFiles []string
125 outputBytes int
126 outputLines int
127 outputTruncated bool
128 maxOutputBytes int
129 maxOutputLines int
130 persistFn func(*Agent)
131 stateDir string
132 workspaceCleanup func() error
133 workspaceCapture func() (WorkspaceCapture, error)
134
135 // OnTurnEnd, if set, fires once per prompt-level turn_end event
136 // emitted by the subagent daemon wrapper. Provider/tool-loop
137 // turn_end events (for example stop=tool_use) are ignored by the
138 // runner because they do not mean the sub-agent task finished.
139 // Used by auto-subagents watchers to detect that a sub-agent's first
140 // (or n-th) task has finished without waiting for the long-lived
141 // daemon itself to exit — sub-agents keep running on the inbox
142 // even after the initial task completes, so Wait() never unblocks
143 // for them.
144 OnTurnEnd func(step int, errMsg string)
145 pendingTurnEnds []turnEndNotice
146
147 ctx context.Context
148 cancel context.CancelFunc
149 runner Runner
150 lease *agentLease
151
152 // launchState is an admission gate for the scheduler's goroutine:
153 // 0=not admitted, 1=runner admitted, 2=stopped before admission. It
154 // prevents an immediate Stop from starting a runner after cancellation.
155 launchState int32
156
157 // done closes when the run goroutine finalises the agent's
158 // status (done / failed / killed). Wait blocks on this so
159 // callers don't have to poll.
160 done chan struct{}
161 turnResults chan *TurnResult
162 doneOnce sync.Once
163}
164
165// Inbox exposes the supervisor-side socket handle. Returns nil for
166// agents without inboxes (e.g. tests using a custom Runner that
167// doesn't speak the daemon protocol).
168func (a *Agent) Inbox() *Inbox { return a.inbox }
169
170// Status returns the current high-level status. Cheap; safe from any goroutine.
171func (a *Agent) Status() Status {
172 a.mu.Lock()
173 defer a.mu.Unlock()
174 return a.status
175}
176
177// Activity returns the current one-line activity string.
178func (a *Agent) Activity() string {
179 a.mu.Lock()
180 defer a.mu.Unlock()
181 return a.activity
182}
183
184// ProcessState returns the child process lifecycle state.
185func (a *Agent) ProcessState() ProcessState {
186 a.lifecycleMu.Lock()
187 defer a.lifecycleMu.Unlock()
188 return a.processState
189}
190
191// TurnState returns the current delegated-turn state.
192func (a *Agent) TurnState() TurnState {
193 a.lifecycleMu.Lock()
194 defer a.lifecycleMu.Unlock()
195 return a.turnState
196}
197
198// CurrentTurnID returns the stable id of the current or most recent turn.
199func (a *Agent) CurrentTurnID() string {
200 a.lifecycleMu.Lock()
201 defer a.lifecycleMu.Unlock()
202 return a.currentTurnID
203}
204
205// AttemptValue returns the current process attempt number.
206func (a *Agent) AttemptValue() int {
207 a.lifecycleMu.Lock()
208 defer a.lifecycleMu.Unlock()
209 return a.Attempt
210}
211
212// ProcessPIDValue returns the current worker pid, when known.
213func (a *Agent) ProcessPIDValue() int {
214 a.lifecycleMu.Lock()
215 defer a.lifecycleMu.Unlock()
216 return a.ProcessPID
217}
218
219// LastActivity returns the last heartbeat or meaningful child event time.
220func (a *Agent) LastActivity() time.Time {
221 a.lifecycleMu.Lock()
222 defer a.lifecycleMu.Unlock()
223 return a.lastActivity
224}
225
226// Result returns a copy of the most recent structured turn result.
227func (a *Agent) Result() *TurnResult {
228 a.lifecycleMu.Lock()
229 defer a.lifecycleMu.Unlock()
230 return cloneTurnResult(a.result)
231}
232
233func (a *Agent) setProcessState(state ProcessState) {
234 a.lifecycleMu.Lock()
235 a.processState = state
236 a.updatedAt = time.Now()
237 a.lifecycleMu.Unlock()
238}
239
240func (a *Agent) incrementAttempt() int {
241 a.lifecycleMu.Lock()
242 a.Attempt++
243 a.updatedAt = time.Now()
244 attempt := a.Attempt
245 a.lifecycleMu.Unlock()
246 return attempt
247}
248
249func (a *Agent) setProcessPID(pid int) {
250 a.lifecycleMu.Lock()
251 a.ProcessPID = pid
252 a.updatedAt = time.Now()
253 a.lifecycleMu.Unlock()
254}
255
256func (a *Agent) releaseLease() error {
257 a.lifecycleMu.Lock()
258 lease := a.lease
259 a.lease = nil
260 a.lifecycleMu.Unlock()
261 if lease == nil {
262 return nil
263 }
264 return lease.Close()
265}
266
267func (a *Agent) setTurnState(state TurnState, turnID string) {
268 a.lifecycleMu.Lock()
269 a.turnState = state
270 if turnID != "" {
271 a.currentTurnID = turnID
272 }
273 a.updatedAt = time.Now()
274 a.lifecycleMu.Unlock()
275}
276
277func (a *Agent) markActivity(now time.Time) {
278 if now.IsZero() {
279 now = time.Now()
280 }
281 a.lifecycleMu.Lock()
282 a.lastActivity = now
283 a.updatedAt = now
284 a.lifecycleMu.Unlock()
285}
286
287func (a *Agent) setResult(result *TurnResult) {
288 a.lifecycleMu.Lock()
289 a.result = cloneTurnResult(result)
290 a.updatedAt = time.Now()
291 a.lifecycleMu.Unlock()
292}
293
294// Transcript returns a copy of the running transcript.
295func (a *Agent) Transcript() []string {
296 a.mu.Lock()
297 defer a.mu.Unlock()
298 out := make([]string, len(a.transcript))
299 copy(out, a.transcript)
300 return out
301}
302
303// Err returns the runner's terminal error, if any.
304func (a *Agent) Err() error {
305 a.mu.Lock()
306 defer a.mu.Unlock()
307 return a.lastErr
308}
309
310// Wait blocks until the agent reaches a terminal state. Used by tests
311// and by /subagents wait <id>.
312func (a *Agent) Wait() { <-a.done }
313
314// WaitContext waits for the agent to finish or for ctx cancellation.
315func (a *Agent) WaitContext(ctx context.Context) error {
316 if ctx == nil {
317 ctx = context.Background()
318 }
319 select {
320 case <-a.done:
321 return nil
322 case <-ctx.Done():
323 return ctx.Err()
324 }
325}
326
327func (a *Agent) waitForTurnResult(ctx context.Context) (*TurnResult, error) {
328 if ctx == nil {
329 ctx = context.Background()
330 }
331 select {
332 case result := <-a.turnResults:
333 if result == nil {
334 return nil, fmt.Errorf("subagents: nil turn result for %s", a.ID)
335 }
336 return cloneTurnResult(result), nil
337 case <-a.done:
338 if result := a.Result(); result != nil {
339 return result, nil
340 }
341 return nil, fmt.Errorf("subagents: agent %s exited without a turn result", a.ID)
342 case <-ctx.Done():
343 return nil, ctx.Err()
344 }
345}
346
347func (a *Agent) closeDone() { a.doneOnce.Do(func() { close(a.done) }) }
348
349func (a *Agent) stateDirectory(defaultRoot string) string {
350 if a.stateDir != "" {
351 return a.stateDir
352 }
353 return filepath.Join(defaultRoot, "agents", a.ID)
354}
355
356// SetOnTurnEnd installs (or clears, with nil) the callback fired
357// from the runner when the child daemon emits a prompt-level
358// turn_end event with a step field. Safe to call from any goroutine:
359// the runner reads the callback under the same mutex.
360func (a *Agent) SetOnTurnEnd(fn func(step int, errMsg string)) {
361 a.mu.Lock()
362 a.OnTurnEnd = fn
363 pending := append([]turnEndNotice(nil), a.pendingTurnEnds...)
364 a.pendingTurnEnds = nil
365 a.mu.Unlock()
366 if fn != nil {
367 for _, notice := range pending {
368 fn(notice.step, notice.errMsg)
369 }
370 }
371}
372
373func (a *Agent) setActivity(msg string) {
374 a.mu.Lock()
375 a.activity = strings.TrimSpace(msg)
376 a.mu.Unlock()
377 a.markActivity(time.Now())
378}
379
380func (a *Agent) appendTranscript(chunk string) {
381 a.appendTranscriptLocked(chunk, "", false)
382}
383
384func (a *Agent) appendUserMessage(text string) {
385 a.appendTranscriptLocked(text, "user: ", false)
386}
387
388func (a *Agent) appendAssistantMessage(text string) {
389 a.appendTranscriptLocked(text, "", true)
390}
391
392func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {
393 message := chunk
394 chunk = strings.TrimRight(chunk, "\n")
395 if chunk == "" {
396 return
397 }
398 a.mu.Lock()
399 for _, line := range strings.Split(chunk, "\n") {
400 line = linePrefix + line
401 a.transcript = append(a.transcript, line)
402 a.outputBytes += len(line) + 1
403 a.outputLines++
404 }
405 if assistant {
406 a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
407 }
408 // Bound the inline dashboard copy. The durable event log and session
409 // remain available through the logical history reference.
410 lineCap := a.maxOutputLines
411 if lineCap <= 0 {
412 lineCap = 2_000
413 }
414 byteCap := a.maxOutputBytes
415 if byteCap <= 0 {
416 byteCap = 500_000
417 }
418 for a.outputLines > lineCap {
419 line := a.transcript[0]
420 a.transcript = a.transcript[1:]
421 a.outputBytes -= len(line) + 1
422 a.outputLines--
423 a.outputTruncated = true
424 }
425 for a.outputLines > 0 && a.outputBytes > byteCap {
426 line := a.transcript[0]
427 a.transcript = a.transcript[1:]
428 a.outputBytes -= len(line) + 1
429 a.outputLines--
430 a.outputTruncated = true
431 }
432 if a.outputLines == 0 {
433 a.outputBytes = 0
434 }
435 a.mu.Unlock()
436 a.markActivity(time.Now())
437}
438
439// newAgentID returns a short, mostly-collision-free identifier of the
440// form "<slug>-<nano>". The slug is derived from the task text so
441// dashboards stay readable; the nano suffix guarantees uniqueness
442// even when two agents are spawned in the same millisecond.
443func newAgentID(task string, now time.Time) string {
444 slug := taskSlug(task)
445 return fmt.Sprintf("%s-%d", slug, now.UnixNano()%1_000_000)
446}
447
448func taskSlug(task string) string {
449 task = strings.ToLower(task)
450 var b strings.Builder
451 dash := false
452 for _, r := range task {
453 switch {
454 case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
455 b.WriteRune(r)
456 dash = false
457 default:
458 if !dash && b.Len() > 0 {
459 b.WriteByte('-')
460 dash = true
461 }
462 }
463 if b.Len() >= 24 {
464 break
465 }
466 }
467 out := strings.Trim(b.String(), "-")
468 if out == "" {
469 out = "agent"
470 }
471 return out
472}