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