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 finished time.Time
125 lastErr error
126 transcriptLoaded bool
127 legacyEventState bool
128 transcriptMu sync.Mutex
129
130 lifecycleMu sync.Mutex
131 processState ProcessState
132 turnState TurnState
133 currentTurnID string
134 updatedAt time.Time
135 lastActivity time.Time
136 result *TurnResult
137 resultRef string
138 patchRef string
139 changedFiles []string
140 outputBytes int
141 outputLines int
142 outputTruncated bool
143 maxOutputBytes int
144 maxOutputLines int
145 persistFn func(*Agent)
146 stateDir string
147 workspaceCleanup func() error
148 workspaceCapture func() (WorkspaceCapture, error)
149
150 // OnTurnEnd, if set, fires once per prompt-level turn_end event
151 // emitted by the subagent daemon wrapper. Provider/tool-loop
152 // turn_end events (for example stop=tool_use) are ignored by the
153 // runner because they do not mean the sub-agent task finished.
154 // Used by auto-subagents watchers to detect that a sub-agent's first
155 // (or n-th) task has finished without waiting for the long-lived
156 // daemon itself to exit — sub-agents keep running on the inbox
157 // even after the initial task completes, so Wait() never unblocks
158 // for them.
159 OnTurnEnd func(step int, errMsg string)
160 pendingTurnEnds []turnEndNotice
161
162 ctx context.Context
163 cancel context.CancelFunc
164 runner Runner
165 lease *agentLease
166
167 // launchState is an admission gate for the scheduler's goroutine:
168 // 0=not admitted, 1=runner admitted, 2=stopped before admission. It
169 // prevents an immediate Stop from starting a runner after cancellation.
170 launchState int32
171
172 // done closes when the run goroutine finalises the agent's
173 // status (done / failed / killed). Wait blocks on this so
174 // callers don't have to poll.
175 done chan struct{}
176 turnResults chan *TurnResult
177 doneOnce sync.Once
178}
179
180// Inbox exposes the supervisor-side socket handle. Returns nil for
181// agents without inboxes (e.g. tests using a custom Runner that
182// doesn't speak the daemon protocol).
183func (a *Agent) Inbox() *Inbox { return a.inbox }
184
185// FastModeOverridesHost reports whether this child was created with an
186// explicit fast-mode override while the host setting was disabled.
187func (a *Agent) FastModeOverridesHost() bool {
188 return a != nil && a.fastModeOverridesHost
189}
190
191// Status returns the current high-level status. Cheap; safe from any goroutine.
192func (a *Agent) Status() Status {
193 a.mu.Lock()
194 defer a.mu.Unlock()
195 return a.status
196}
197
198// Activity returns the current one-line activity string.
199func (a *Agent) Activity() string {
200 a.mu.Lock()
201 defer a.mu.Unlock()
202 return a.activity
203}
204
205// ProcessState returns the child process lifecycle state.
206func (a *Agent) ProcessState() ProcessState {
207 a.lifecycleMu.Lock()
208 defer a.lifecycleMu.Unlock()
209 return a.processState
210}
211
212// TurnState returns the current delegated-turn state.
213func (a *Agent) TurnState() TurnState {
214 a.lifecycleMu.Lock()
215 defer a.lifecycleMu.Unlock()
216 return a.turnState
217}
218
219// CurrentTurnID returns the stable id of the current or most recent turn.
220func (a *Agent) CurrentTurnID() string {
221 a.lifecycleMu.Lock()
222 defer a.lifecycleMu.Unlock()
223 return a.currentTurnID
224}
225
226// AttemptValue returns the current process attempt number.
227func (a *Agent) AttemptValue() int {
228 a.lifecycleMu.Lock()
229 defer a.lifecycleMu.Unlock()
230 return a.Attempt
231}
232
233// ProcessPIDValue returns the current worker pid, when known.
234func (a *Agent) ProcessPIDValue() int {
235 a.lifecycleMu.Lock()
236 defer a.lifecycleMu.Unlock()
237 return a.ProcessPID
238}
239
240// LastActivity returns the last heartbeat or meaningful child event time.
241func (a *Agent) LastActivity() time.Time {
242 a.lifecycleMu.Lock()
243 defer a.lifecycleMu.Unlock()
244 return a.lastActivity
245}
246
247// UpdatedAt returns the last lifecycle-state update time.
248func (a *Agent) UpdatedAt() time.Time {
249 a.lifecycleMu.Lock()
250 defer a.lifecycleMu.Unlock()
251 return a.updatedAt
252}
253
254// Result returns a copy of the most recent structured turn result.
255func (a *Agent) Result() *TurnResult {
256 a.lifecycleMu.Lock()
257 defer a.lifecycleMu.Unlock()
258 return cloneTurnResult(a.result)
259}
260
261func (a *Agent) setProcessState(state ProcessState) {
262 a.lifecycleMu.Lock()
263 a.processState = state
264 a.updatedAt = time.Now()
265 a.lifecycleMu.Unlock()
266}
267
268func (a *Agent) incrementAttempt() int {
269 a.lifecycleMu.Lock()
270 a.Attempt++
271 a.updatedAt = time.Now()
272 attempt := a.Attempt
273 a.lifecycleMu.Unlock()
274 return attempt
275}
276
277func (a *Agent) setProcessPID(pid int) {
278 a.lifecycleMu.Lock()
279 a.ProcessPID = pid
280 a.updatedAt = time.Now()
281 a.lifecycleMu.Unlock()
282}
283
284func (a *Agent) releaseLease() error {
285 a.lifecycleMu.Lock()
286 lease := a.lease
287 a.lease = nil
288 a.lifecycleMu.Unlock()
289 if lease == nil {
290 return nil
291 }
292 return lease.Close()
293}
294
295func (a *Agent) setTurnState(state TurnState, turnID string) {
296 a.lifecycleMu.Lock()
297 a.turnState = state
298 if turnID != "" {
299 a.currentTurnID = turnID
300 }
301 a.updatedAt = time.Now()
302 a.lifecycleMu.Unlock()
303}
304
305func (a *Agent) markActivity(now time.Time) {
306 if now.IsZero() {
307 now = time.Now()
308 }
309 a.lifecycleMu.Lock()
310 a.lastActivity = now
311 a.updatedAt = now
312 a.lifecycleMu.Unlock()
313}
314
315func (a *Agent) setResult(result *TurnResult) {
316 a.lifecycleMu.Lock()
317 a.result = cloneTurnResult(result)
318 a.updatedAt = time.Now()
319 a.lifecycleMu.Unlock()
320}
321
322// Transcript returns a copy of the running transcript. Detached agents
323// hydrate their durable event history the first time it is requested.
324func (a *Agent) Transcript() []string {
325 _ = a.loadTranscript()
326 a.mu.Lock()
327 defer a.mu.Unlock()
328 out := make([]string, len(a.transcript))
329 copy(out, a.transcript)
330 return out
331}
332
333// Err returns the runner's terminal error, if any.
334func (a *Agent) Err() error {
335 a.mu.Lock()
336 defer a.mu.Unlock()
337 return a.lastErr
338}
339
340// Wait blocks until the agent reaches a terminal state. Used by tests
341// and by /subagents wait <id>.
342func (a *Agent) Wait() { <-a.done }
343
344// WaitContext waits for the agent to finish or for ctx cancellation.
345func (a *Agent) WaitContext(ctx context.Context) error {
346 if ctx == nil {
347 ctx = context.Background()
348 }
349 select {
350 case <-a.done:
351 return nil
352 case <-ctx.Done():
353 return ctx.Err()
354 }
355}
356
357func (a *Agent) waitForTurnResult(ctx context.Context) (*TurnResult, error) {
358 if ctx == nil {
359 ctx = context.Background()
360 }
361 select {
362 case result := <-a.turnResults:
363 if result == nil {
364 return nil, fmt.Errorf("subagents: nil turn result for %s", a.ID)
365 }
366 return cloneTurnResult(result), nil
367 case <-a.done:
368 if result := a.Result(); result != nil {
369 return result, nil
370 }
371 return nil, fmt.Errorf("subagents: agent %s exited without a turn result", a.ID)
372 case <-ctx.Done():
373 return nil, ctx.Err()
374 }
375}
376
377func (a *Agent) closeDone() { a.doneOnce.Do(func() { close(a.done) }) }
378
379func (a *Agent) stateDirectory(defaultRoot string) string {
380 if a.stateDir != "" {
381 return a.stateDir
382 }
383 return filepath.Join(defaultRoot, "agents", a.ID)
384}
385
386// SetOnTurnEnd installs (or clears, with nil) the callback fired
387// from the runner when the child daemon emits a prompt-level
388// turn_end event with a step field. Safe to call from any goroutine:
389// the runner reads the callback under the same mutex.
390func (a *Agent) resumePrompt() string {
391 prompt, _ := a.ResumePromptInfo()
392 return prompt
393}
394
395// ResumePromptInfo returns the queued manager follow-up and its acceptance
396// time. Both values are read as one synchronized snapshot.
397func (a *Agent) ResumePromptInfo() (string, time.Time) {
398 a.mu.Lock()
399 defer a.mu.Unlock()
400 return a.resumePromptText, a.resumePromptAt
401}
402
403func (a *Agent) setResumePrompt(prompt string, acceptedAt time.Time) (string, time.Time) {
404 a.mu.Lock()
405 defer a.mu.Unlock()
406 previousPrompt, previousAcceptedAt := a.resumePromptText, a.resumePromptAt
407 a.resumePromptText = prompt
408 a.resumePromptAt = acceptedAt
409 return previousPrompt, previousAcceptedAt
410}
411
412func (a *Agent) clearResumePrompt() bool {
413 a.mu.Lock()
414 defer a.mu.Unlock()
415 if a.resumePromptText == "" {
416 return false
417 }
418 a.resumePromptText = ""
419 a.resumePromptAt = time.Time{}
420 return true
421}
422
423func (a *Agent) SetOnTurnEnd(fn func(step int, errMsg string)) {
424 a.mu.Lock()
425 a.OnTurnEnd = fn
426 pending := append([]turnEndNotice(nil), a.pendingTurnEnds...)
427 a.pendingTurnEnds = nil
428 a.mu.Unlock()
429 if fn != nil {
430 for _, notice := range pending {
431 fn(notice.step, notice.errMsg)
432 }
433 }
434}
435
436func (a *Agent) setActivity(msg string) {
437 a.mu.Lock()
438 a.activity = strings.TrimSpace(msg)
439 a.mu.Unlock()
440 a.markActivity(time.Now())
441}
442
443func (a *Agent) appendTranscript(chunk string) {
444 a.appendTranscriptLocked(chunk, "", false)
445}
446
447func (a *Agent) appendUserMessage(text string) {
448 a.appendTranscriptLocked(text, "user: ", false)
449}
450
451func (a *Agent) appendAssistantMessage(text string) {
452 a.appendTranscriptLocked(text, "", true)
453}
454
455func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {
456 message := chunk
457 chunk = strings.TrimRight(chunk, "\n")
458 if chunk == "" {
459 return
460 }
461 a.mu.Lock()
462 for _, line := range strings.Split(chunk, "\n") {
463 line = linePrefix + line
464 a.transcript = append(a.transcript, line)
465 a.outputBytes += len(line) + 1
466 a.outputLines++
467 }
468 if assistant {
469 a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
470 }
471 // Bound the inline dashboard copy. The durable event log and session
472 // remain available through the logical history reference.
473 lineCap := a.maxOutputLines
474 if lineCap <= 0 {
475 lineCap = 2_000
476 }
477 byteCap := a.maxOutputBytes
478 if byteCap <= 0 {
479 byteCap = 500_000
480 }
481 for a.outputLines > lineCap {
482 line := a.transcript[0]
483 a.transcript = a.transcript[1:]
484 a.outputBytes -= len(line) + 1
485 a.outputLines--
486 a.outputTruncated = true
487 }
488 for a.outputLines > 0 && a.outputBytes > byteCap {
489 line := a.transcript[0]
490 a.transcript = a.transcript[1:]
491 a.outputBytes -= len(line) + 1
492 a.outputLines--
493 a.outputTruncated = true
494 }
495 if a.outputLines == 0 {
496 a.outputBytes = 0
497 }
498 a.mu.Unlock()
499 a.markActivity(time.Now())
500}
501
502// newAgentID returns a short, mostly-collision-free identifier of the
503// form "<slug>-<nano>". The slug is derived from the task text so
504// dashboards stay readable; the nano suffix guarantees uniqueness
505// even when two agents are spawned in the same millisecond.
506func newAgentID(task string, now time.Time) string {
507 slug := taskSlug(task)
508 return fmt.Sprintf("%s-%d", slug, now.UnixNano()%1_000_000)
509}
510
511func taskSlug(task string) string {
512 task = strings.ToLower(task)
513 var b strings.Builder
514 dash := false
515 for _, r := range task {
516 switch {
517 case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
518 b.WriteRune(r)
519 dash = false
520 default:
521 if !dash && b.Len() > 0 {
522 b.WriteByte('-')
523 dash = true
524 }
525 }
526 if b.Len() >= 24 {
527 break
528 }
529 }
530 out := strings.Trim(b.String(), "-")
531 if out == "" {
532 out = "agent"
533 }
534 return out
535}