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 WebSearchPolicy WebSearchPolicy
45
46 // Model and Provider, when non-empty, override the child
47 // subprocess's model resolution. BaseURL and InsecureTLS carry
48 // the effective provider connection settings from the supervisor.
49 // Empty values inherit whatever the child resolves on its own from
50 // config / env / flags. All four settings are persisted in meta.json
51 // so Resume keeps using the same provider configuration across zut
52 // restarts.
53 Model string
54 Provider string
55 BaseURL string
56 InsecureTLS bool
57 Reasoning string
58 FastMode bool
59 Subagent string
60
61 // fastModeOverridesHost records the spawn-time case where an explicit
62 // profile fastMode: true enabled the child while the host was off. It is
63 // intentionally transient: the parent only needs it to report the initial
64 // spawn warning, while the effective FastMode remains durable state.
65 fastModeOverridesHost bool
66
67 // SessionID, when non-empty, scopes the agent to a particular
68 // host zut session: the dashboard only surfaces agents whose
69 // SessionID matches the active session. Empty means "unscoped" and
70 // applies to agents spawned without a session context such as in tests. Set at
71 // Spawn time from Supervisor.activeSession and persisted in
72 // meta.json so the scope survives restarts.
73 SessionID string
74
75 // Resuming is true when this Agent struct was built by Resume
76 // rather than Spawn. The runner consults it to decide whether
77 // to pass the original Task as a positional argv to the child:
78 // on first Spawn we want the child to run the task immediately;
79 // on Resume the task was already run last time and must not replay.
80 // Not persisted — every Resume sets it explicitly.
81 Resuming bool
82
83 // resumePromptText is an optional manager follow-up passed to a resumed
84 // child as its initial prompt. It remains in durable metadata until the
85 // child reports turn.started, so a host exit while the worker is queued
86 // cannot lose it. Access it through ResumePromptInfo.
87 resumePromptText string
88 resumePromptAt time.Time
89
90 // InboxPath is the unix socket the child agent listens on for
91 // follow-up prompts and control messages. The supervisor opens
92 // an Inbox at this path; the child opens a Listener.
93 InboxPath string
94
95 // Attempt increments whenever the supervisor starts a process for this
96 // agent. It is persisted so recovery can distinguish a resumed worker
97 // from a first launch.
98 Attempt int
99
100 // ProcessPID identifies the current worker when known. It is persisted
101 // so a reloaded supervisor can force-kill a detached worker after a
102 // graceful shutdown deadline.
103 ProcessPID int
104
105 // EventLogPath is the durable JSONL event log for this agent.
106 // The runner appends every well-formed event from the child
107 // (plus lifecycle events of its own) here. /subagents open in any
108 // zut process reads from this file to replay the full history.
109 EventLogPath string
110
111 // SessionPath is the child's persistent session file. Surfaced
112 // so the dashboard / /subagents open can resume the agent later.
113 SessionPath string
114
115 // inbox is the supervisor-side socket handle. Populated by
116 // Supervisor.Spawn so callers do not need to manage the socket directly.
117 inbox *Inbox
118
119 mu sync.Mutex
120 status Status
121 activity string
122 transcript []string
123 lastAssistant string
124 streamingAssistantText string
125 streamingAssistantTruncated bool
126 streamingAssistantStart int
127 streamingAssistantLines int
128 finished time.Time
129 lastErr error
130 transcriptLoaded bool
131 legacyEventState bool
132 transcriptMu sync.Mutex
133
134 lifecycleMu sync.Mutex
135 processState ProcessState
136 turnState TurnState
137 currentTurnID string
138 updatedAt time.Time
139 lastActivity time.Time
140 result *TurnResult
141 resultRef string
142 patchRef string
143 changedFiles []string
144 outputBytes int
145 outputLines int
146 outputTruncated bool
147 maxOutputBytes int
148 maxOutputLines int
149 persistFn func(*Agent)
150 stateDir string
151 workspaceCleanup func() error
152 workspaceCapture func() (WorkspaceCapture, error)
153
154 // OnTurnEnd, if set, fires once per prompt-level turn_end event
155 // emitted by the subagent daemon wrapper. Provider/tool-loop
156 // turn_end events (for example stop=tool_use) are ignored by the
157 // runner because they do not mean the sub-agent task finished.
158 // Used by auto-subagents watchers to detect that a sub-agent's first
159 // (or n-th) task has finished without waiting for the long-lived
160 // daemon itself to exit — sub-agents keep running on the inbox
161 // even after the initial task completes, so Wait() never unblocks
162 // for them.
163 OnTurnEnd func(step int, errMsg string)
164 pendingTurnEnds []turnEndNotice
165
166 ctx context.Context
167 cancel context.CancelFunc
168 runner Runner
169 lease *agentLease
170
171 // launchState is an admission gate for the scheduler's goroutine:
172 // 0=not admitted, 1=runner admitted, 2=stopped before admission. It
173 // prevents an immediate Stop from starting a runner after cancellation.
174 launchState int32
175
176 // done closes when the run goroutine finalises the agent's
177 // status (done / failed / killed). Wait blocks on this so
178 // callers don't have to poll.
179 done chan struct{}
180 turnResults chan *TurnResult
181 doneOnce sync.Once
182}
183
184// Inbox exposes the supervisor-side socket handle. Returns nil for
185// agents without inboxes (e.g. tests using a custom Runner that
186// doesn't speak the daemon protocol).
187func (a *Agent) Inbox() *Inbox { return a.inbox }
188
189// FastModeOverridesHost reports whether this child was created with an
190// explicit fast-mode override while the host setting was disabled.
191func (a *Agent) FastModeOverridesHost() bool {
192 return a != nil && a.fastModeOverridesHost
193}
194
195// Status returns the current high-level status. Cheap; safe from any goroutine.
196func (a *Agent) Status() Status {
197 a.mu.Lock()
198 defer a.mu.Unlock()
199 return a.status
200}
201
202// Activity returns the current one-line activity string.
203func (a *Agent) Activity() string {
204 a.mu.Lock()
205 defer a.mu.Unlock()
206 return a.activity
207}
208
209// ProcessState returns the child process lifecycle state.
210func (a *Agent) ProcessState() ProcessState {
211 a.lifecycleMu.Lock()
212 defer a.lifecycleMu.Unlock()
213 return a.processState
214}
215
216// TurnState returns the current delegated-turn state.
217func (a *Agent) TurnState() TurnState {
218 a.lifecycleMu.Lock()
219 defer a.lifecycleMu.Unlock()
220 return a.turnState
221}
222
223// CurrentTurnID returns the stable id of the current or most recent turn.
224func (a *Agent) CurrentTurnID() string {
225 a.lifecycleMu.Lock()
226 defer a.lifecycleMu.Unlock()
227 return a.currentTurnID
228}
229
230// AttemptValue returns the current process attempt number.
231func (a *Agent) AttemptValue() int {
232 a.lifecycleMu.Lock()
233 defer a.lifecycleMu.Unlock()
234 return a.Attempt
235}
236
237// ProcessPIDValue returns the current worker pid, when known.
238func (a *Agent) ProcessPIDValue() int {
239 a.lifecycleMu.Lock()
240 defer a.lifecycleMu.Unlock()
241 return a.ProcessPID
242}
243
244// LastActivity returns the last heartbeat or meaningful child event time.
245func (a *Agent) LastActivity() time.Time {
246 a.lifecycleMu.Lock()
247 defer a.lifecycleMu.Unlock()
248 return a.lastActivity
249}
250
251// UpdatedAt returns the last lifecycle-state update time.
252func (a *Agent) UpdatedAt() time.Time {
253 a.lifecycleMu.Lock()
254 defer a.lifecycleMu.Unlock()
255 return a.updatedAt
256}
257
258// Result returns a copy of the most recent structured turn result.
259func (a *Agent) Result() *TurnResult {
260 a.lifecycleMu.Lock()
261 defer a.lifecycleMu.Unlock()
262 return cloneTurnResult(a.result)
263}
264
265func (a *Agent) setProcessState(state ProcessState) {
266 a.lifecycleMu.Lock()
267 a.processState = state
268 a.updatedAt = time.Now()
269 a.lifecycleMu.Unlock()
270}
271
272func (a *Agent) incrementAttempt() int {
273 a.lifecycleMu.Lock()
274 a.Attempt++
275 a.updatedAt = time.Now()
276 attempt := a.Attempt
277 a.lifecycleMu.Unlock()
278 return attempt
279}
280
281func (a *Agent) setProcessPID(pid int) {
282 a.lifecycleMu.Lock()
283 a.ProcessPID = pid
284 a.updatedAt = time.Now()
285 a.lifecycleMu.Unlock()
286}
287
288func (a *Agent) releaseLease() error {
289 a.lifecycleMu.Lock()
290 lease := a.lease
291 a.lease = nil
292 a.lifecycleMu.Unlock()
293 if lease == nil {
294 return nil
295 }
296 return lease.Close()
297}
298
299func (a *Agent) setTurnState(state TurnState, turnID string) {
300 a.lifecycleMu.Lock()
301 a.turnState = state
302 if turnID != "" {
303 a.currentTurnID = turnID
304 }
305 a.updatedAt = time.Now()
306 a.lifecycleMu.Unlock()
307}
308
309func (a *Agent) markActivity(now time.Time) {
310 if now.IsZero() {
311 now = time.Now()
312 }
313 a.lifecycleMu.Lock()
314 a.lastActivity = now
315 a.updatedAt = now
316 a.lifecycleMu.Unlock()
317}
318
319func (a *Agent) setResult(result *TurnResult) {
320 a.lifecycleMu.Lock()
321 a.result = cloneTurnResult(result)
322 a.updatedAt = time.Now()
323 a.lifecycleMu.Unlock()
324}
325
326// Transcript returns a copy of the running transcript. Detached agents
327// hydrate their durable event history the first time it is requested.
328func (a *Agent) Transcript() []string {
329 _ = a.loadTranscript()
330 a.mu.Lock()
331 defer a.mu.Unlock()
332 out := make([]string, len(a.transcript))
333 copy(out, a.transcript)
334 return out
335}
336
337// Err returns the runner's terminal error, if any.
338func (a *Agent) Err() error {
339 a.mu.Lock()
340 defer a.mu.Unlock()
341 return a.lastErr
342}
343
344// Wait blocks until the agent reaches a terminal state. Used by tests
345// and by /subagents wait <id>.
346func (a *Agent) Wait() { <-a.done }
347
348// WaitContext waits for the agent to finish or for ctx cancellation.
349func (a *Agent) WaitContext(ctx context.Context) error {
350 if ctx == nil {
351 ctx = context.Background()
352 }
353 select {
354 case <-a.done:
355 return nil
356 case <-ctx.Done():
357 return ctx.Err()
358 }
359}
360
361func (a *Agent) waitForTurnResult(ctx context.Context) (*TurnResult, error) {
362 if ctx == nil {
363 ctx = context.Background()
364 }
365 select {
366 case result := <-a.turnResults:
367 if result == nil {
368 return nil, fmt.Errorf("subagents: nil turn result for %s", a.ID)
369 }
370 return cloneTurnResult(result), nil
371 case <-a.done:
372 if result := a.Result(); result != nil {
373 return result, nil
374 }
375 return nil, fmt.Errorf("subagents: agent %s exited without a turn result", a.ID)
376 case <-ctx.Done():
377 return nil, ctx.Err()
378 }
379}
380
381func (a *Agent) closeDone() { a.doneOnce.Do(func() { close(a.done) }) }
382
383func (a *Agent) stateDirectory(defaultRoot string) string {
384 if a.stateDir != "" {
385 return a.stateDir
386 }
387 return filepath.Join(defaultRoot, "agents", a.ID)
388}
389
390// SetOnTurnEnd installs (or clears, with nil) the callback fired
391// from the runner when the child daemon emits a prompt-level
392// turn_end event with a step field. Safe to call from any goroutine:
393// the runner reads the callback under the same mutex.
394func (a *Agent) resumePrompt() string {
395 prompt, _ := a.ResumePromptInfo()
396 return prompt
397}
398
399// ResumePromptInfo returns the queued manager follow-up and its acceptance
400// time. Both values are read as one synchronized snapshot.
401func (a *Agent) ResumePromptInfo() (string, time.Time) {
402 a.mu.Lock()
403 defer a.mu.Unlock()
404 return a.resumePromptText, a.resumePromptAt
405}
406
407func (a *Agent) setResumePrompt(prompt string, acceptedAt time.Time) (string, time.Time) {
408 a.mu.Lock()
409 defer a.mu.Unlock()
410 previousPrompt, previousAcceptedAt := a.resumePromptText, a.resumePromptAt
411 a.resumePromptText = prompt
412 a.resumePromptAt = acceptedAt
413 return previousPrompt, previousAcceptedAt
414}
415
416func (a *Agent) clearResumePrompt() bool {
417 a.mu.Lock()
418 defer a.mu.Unlock()
419 if a.resumePromptText == "" {
420 return false
421 }
422 a.resumePromptText = ""
423 a.resumePromptAt = time.Time{}
424 return true
425}
426
427func (a *Agent) SetOnTurnEnd(fn func(step int, errMsg string)) {
428 a.mu.Lock()
429 a.OnTurnEnd = fn
430 pending := append([]turnEndNotice(nil), a.pendingTurnEnds...)
431 a.pendingTurnEnds = nil
432 a.mu.Unlock()
433 if fn != nil {
434 for _, notice := range pending {
435 fn(notice.step, notice.errMsg)
436 }
437 }
438}
439
440func (a *Agent) setActivity(msg string) {
441 a.mu.Lock()
442 a.activity = strings.TrimSpace(msg)
443 a.mu.Unlock()
444 a.markActivity(time.Now())
445}
446
447func (a *Agent) appendTranscript(chunk string) {
448 a.appendTranscriptChunk(chunk, "", false)
449}
450
451func (a *Agent) appendUserMessage(text string) {
452 a.appendTranscriptChunk(text, "user: ", false)
453}
454
455func (a *Agent) appendAssistantMessage(text string) {
456 if text == "" {
457 return
458 }
459 a.mu.Lock()
460 if a.streamingAssistantText != "" {
461 a.replaceStreamingAssistantLocked(text)
462 a.clearStreamingAssistantLocked()
463 } else {
464 a.appendTranscriptLinesLocked(text, "", true)
465 a.boundTranscriptLocked()
466 }
467 a.mu.Unlock()
468 a.markActivity(time.Now())
469}
470
471// appendAssistantDelta exposes an in-progress assistant response in the
472// dashboard without duplicating it when the final assistant_message arrives.
473// The JSONL event log remains the durable source; this maintains its live and
474// replayed in-memory projection.
475func (a *Agent) appendAssistantDelta(text string) {
476 if text == "" {
477 return
478 }
479 a.mu.Lock()
480 if !a.streamingAssistantTruncated {
481 value := a.streamingAssistantText + text
482 byteCap, lineCap := a.outputLimits()
483 a.streamingAssistantText = boundInlineText(value, byteCap, lineCap)
484 a.streamingAssistantTruncated = a.streamingAssistantText != value
485 }
486 a.replaceStreamingAssistantLocked(a.streamingAssistantText)
487 a.mu.Unlock()
488 a.markActivity(time.Now())
489}
490
491func (a *Agent) appendTranscriptChunk(chunk, linePrefix string, assistant bool) {
492 if strings.TrimRight(chunk, "\n") == "" {
493 return
494 }
495 a.mu.Lock()
496 a.appendTranscriptLinesLocked(chunk, linePrefix, assistant)
497 a.boundTranscriptLocked()
498 a.mu.Unlock()
499 a.markActivity(time.Now())
500}
501
502func (a *Agent) resetStreamingAssistant() {
503 a.mu.Lock()
504 a.clearStreamingAssistantLocked()
505 a.mu.Unlock()
506}
507
508func (a *Agent) replaceStreamingAssistantLocked(text string) {
509 a.removeStreamingAssistantLocked()
510 a.streamingAssistantStart = len(a.transcript)
511 a.streamingAssistantLines = a.appendTranscriptLinesLocked(text, "", true)
512 a.boundTranscriptLocked()
513}
514
515func (a *Agent) removeStreamingAssistantLocked() {
516 if a.streamingAssistantLines == 0 {
517 return
518 }
519 start := a.streamingAssistantStart
520 if start < 0 {
521 start = 0
522 }
523 if start > len(a.transcript) {
524 start = len(a.transcript)
525 }
526 end := start + a.streamingAssistantLines
527 if end > len(a.transcript) {
528 end = len(a.transcript)
529 }
530 for _, line := range a.transcript[start:end] {
531 a.outputBytes -= len(line) + 1
532 a.outputLines--
533 }
534 a.transcript = append(a.transcript[:start], a.transcript[end:]...)
535 a.streamingAssistantLines = 0
536 if a.outputLines == 0 {
537 a.outputBytes = 0
538 }
539}
540
541func (a *Agent) clearStreamingAssistantLocked() {
542 a.streamingAssistantText = ""
543 a.streamingAssistantTruncated = false
544 a.streamingAssistantStart = 0
545 a.streamingAssistantLines = 0
546}
547
548func (a *Agent) appendTranscriptLinesLocked(chunk, linePrefix string, assistant bool) int {
549 message := chunk
550 chunk = strings.TrimRight(chunk, "\n")
551 if chunk == "" {
552 return 0
553 }
554 lines := strings.Split(chunk, "\n")
555 for _, line := range lines {
556 line = linePrefix + line
557 a.transcript = append(a.transcript, line)
558 a.outputBytes += len(line) + 1
559 a.outputLines++
560 }
561 if assistant {
562 a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
563 }
564 return len(lines)
565}
566
567func (a *Agent) outputLimits() (byteCap, lineCap int) {
568 byteCap = a.maxOutputBytes
569 if byteCap <= 0 {
570 byteCap = 500_000
571 }
572 lineCap = a.maxOutputLines
573 if lineCap <= 0 {
574 lineCap = 2_000
575 }
576 return byteCap, lineCap
577}
578
579func (a *Agent) boundTranscriptLocked() {
580 // Bound the inline dashboard copy. The durable event log and session
581 // remain available through the logical history reference.
582 byteCap, lineCap := a.outputLimits()
583 for a.outputLines > lineCap || (a.outputLines > 0 && a.outputBytes > byteCap) {
584 line := a.transcript[0]
585 a.transcript = a.transcript[1:]
586 a.outputBytes -= len(line) + 1
587 a.outputLines--
588 if a.streamingAssistantLines > 0 {
589 if a.streamingAssistantStart > 0 {
590 a.streamingAssistantStart--
591 } else {
592 a.streamingAssistantLines--
593 }
594 }
595 a.outputTruncated = true
596 }
597 if a.outputLines == 0 {
598 a.outputBytes = 0
599 }
600}
601
602// newAgentID returns a short, mostly-collision-free identifier of the
603// form "<slug>-<nano>". The slug is derived from the task text so
604// dashboards stay readable; the nano suffix guarantees uniqueness
605// even when two agents are spawned in the same millisecond.
606func newAgentID(task string, now time.Time) string {
607 slug := taskSlug(task)
608 return fmt.Sprintf("%s-%d", slug, now.UnixNano()%1_000_000)
609}
610
611func taskSlug(task string) string {
612 task = strings.ToLower(task)
613 var b strings.Builder
614 dash := false
615 for _, r := range task {
616 switch {
617 case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
618 b.WriteRune(r)
619 dash = false
620 default:
621 if !dash && b.Len() > 0 {
622 b.WriteByte('-')
623 dash = true
624 }
625 }
626 if b.Len() >= 24 {
627 break
628 }
629 }
630 out := strings.Trim(b.String(), "-")
631 if out == "" {
632 out = "agent"
633 }
634 return out
635}