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