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// UpdatedAt returns the last lifecycle-state update time.
227func (a *Agent) UpdatedAt() time.Time {
228 a.lifecycleMu.Lock()
229 defer a.lifecycleMu.Unlock()
230 return a.updatedAt
231}
232
233// Result returns a copy of the most recent structured turn result.
234func (a *Agent) Result() *TurnResult {
235 a.lifecycleMu.Lock()
236 defer a.lifecycleMu.Unlock()
237 return cloneTurnResult(a.result)
238}
239
240func (a *Agent) setProcessState(state ProcessState) {
241 a.lifecycleMu.Lock()
242 a.processState = state
243 a.updatedAt = time.Now()
244 a.lifecycleMu.Unlock()
245}
246
247func (a *Agent) incrementAttempt() int {
248 a.lifecycleMu.Lock()
249 a.Attempt++
250 a.updatedAt = time.Now()
251 attempt := a.Attempt
252 a.lifecycleMu.Unlock()
253 return attempt
254}
255
256func (a *Agent) setProcessPID(pid int) {
257 a.lifecycleMu.Lock()
258 a.ProcessPID = pid
259 a.updatedAt = time.Now()
260 a.lifecycleMu.Unlock()
261}
262
263func (a *Agent) releaseLease() error {
264 a.lifecycleMu.Lock()
265 lease := a.lease
266 a.lease = nil
267 a.lifecycleMu.Unlock()
268 if lease == nil {
269 return nil
270 }
271 return lease.Close()
272}
273
274func (a *Agent) setTurnState(state TurnState, turnID string) {
275 a.lifecycleMu.Lock()
276 a.turnState = state
277 if turnID != "" {
278 a.currentTurnID = turnID
279 }
280 a.updatedAt = time.Now()
281 a.lifecycleMu.Unlock()
282}
283
284func (a *Agent) markActivity(now time.Time) {
285 if now.IsZero() {
286 now = time.Now()
287 }
288 a.lifecycleMu.Lock()
289 a.lastActivity = now
290 a.updatedAt = now
291 a.lifecycleMu.Unlock()
292}
293
294func (a *Agent) setResult(result *TurnResult) {
295 a.lifecycleMu.Lock()
296 a.result = cloneTurnResult(result)
297 a.updatedAt = time.Now()
298 a.lifecycleMu.Unlock()
299}
300
301// Transcript returns a copy of the running transcript.
302func (a *Agent) Transcript() []string {
303 a.mu.Lock()
304 defer a.mu.Unlock()
305 out := make([]string, len(a.transcript))
306 copy(out, a.transcript)
307 return out
308}
309
310// Err returns the runner's terminal error, if any.
311func (a *Agent) Err() error {
312 a.mu.Lock()
313 defer a.mu.Unlock()
314 return a.lastErr
315}
316
317// Wait blocks until the agent reaches a terminal state. Used by tests
318// and by /subagents wait <id>.
319func (a *Agent) Wait() { <-a.done }
320
321// WaitContext waits for the agent to finish or for ctx cancellation.
322func (a *Agent) WaitContext(ctx context.Context) error {
323 if ctx == nil {
324 ctx = context.Background()
325 }
326 select {
327 case <-a.done:
328 return nil
329 case <-ctx.Done():
330 return ctx.Err()
331 }
332}
333
334func (a *Agent) waitForTurnResult(ctx context.Context) (*TurnResult, error) {
335 if ctx == nil {
336 ctx = context.Background()
337 }
338 select {
339 case result := <-a.turnResults:
340 if result == nil {
341 return nil, fmt.Errorf("subagents: nil turn result for %s", a.ID)
342 }
343 return cloneTurnResult(result), nil
344 case <-a.done:
345 if result := a.Result(); result != nil {
346 return result, nil
347 }
348 return nil, fmt.Errorf("subagents: agent %s exited without a turn result", a.ID)
349 case <-ctx.Done():
350 return nil, ctx.Err()
351 }
352}
353
354func (a *Agent) closeDone() { a.doneOnce.Do(func() { close(a.done) }) }
355
356func (a *Agent) stateDirectory(defaultRoot string) string {
357 if a.stateDir != "" {
358 return a.stateDir
359 }
360 return filepath.Join(defaultRoot, "agents", a.ID)
361}
362
363// SetOnTurnEnd installs (or clears, with nil) the callback fired
364// from the runner when the child daemon emits a prompt-level
365// turn_end event with a step field. Safe to call from any goroutine:
366// the runner reads the callback under the same mutex.
367func (a *Agent) SetOnTurnEnd(fn func(step int, errMsg string)) {
368 a.mu.Lock()
369 a.OnTurnEnd = fn
370 pending := append([]turnEndNotice(nil), a.pendingTurnEnds...)
371 a.pendingTurnEnds = nil
372 a.mu.Unlock()
373 if fn != nil {
374 for _, notice := range pending {
375 fn(notice.step, notice.errMsg)
376 }
377 }
378}
379
380func (a *Agent) setActivity(msg string) {
381 a.mu.Lock()
382 a.activity = strings.TrimSpace(msg)
383 a.mu.Unlock()
384 a.markActivity(time.Now())
385}
386
387func (a *Agent) appendTranscript(chunk string) {
388 a.appendTranscriptLocked(chunk, "", false)
389}
390
391func (a *Agent) appendUserMessage(text string) {
392 a.appendTranscriptLocked(text, "user: ", false)
393}
394
395func (a *Agent) appendAssistantMessage(text string) {
396 a.appendTranscriptLocked(text, "", true)
397}
398
399func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) {
400 message := chunk
401 chunk = strings.TrimRight(chunk, "\n")
402 if chunk == "" {
403 return
404 }
405 a.mu.Lock()
406 for _, line := range strings.Split(chunk, "\n") {
407 line = linePrefix + line
408 a.transcript = append(a.transcript, line)
409 a.outputBytes += len(line) + 1
410 a.outputLines++
411 }
412 if assistant {
413 a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
414 }
415 // Bound the inline dashboard copy. The durable event log and session
416 // remain available through the logical history reference.
417 lineCap := a.maxOutputLines
418 if lineCap <= 0 {
419 lineCap = 2_000
420 }
421 byteCap := a.maxOutputBytes
422 if byteCap <= 0 {
423 byteCap = 500_000
424 }
425 for a.outputLines > lineCap {
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 for a.outputLines > 0 && a.outputBytes > byteCap {
433 line := a.transcript[0]
434 a.transcript = a.transcript[1:]
435 a.outputBytes -= len(line) + 1
436 a.outputLines--
437 a.outputTruncated = true
438 }
439 if a.outputLines == 0 {
440 a.outputBytes = 0
441 }
442 a.mu.Unlock()
443 a.markActivity(time.Now())
444}
445
446// newAgentID returns a short, mostly-collision-free identifier of the
447// form "<slug>-<nano>". The slug is derived from the task text so
448// dashboards stay readable; the nano suffix guarantees uniqueness
449// even when two agents are spawned in the same millisecond.
450func newAgentID(task string, now time.Time) string {
451 slug := taskSlug(task)
452 return fmt.Sprintf("%s-%d", slug, now.UnixNano()%1_000_000)
453}
454
455func taskSlug(task string) string {
456 task = strings.ToLower(task)
457 var b strings.Builder
458 dash := false
459 for _, r := range task {
460 switch {
461 case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
462 b.WriteRune(r)
463 dash = false
464 default:
465 if !dash && b.Len() > 0 {
466 b.WriteByte('-')
467 dash = true
468 }
469 }
470 if b.Len() >= 24 {
471 break
472 }
473 }
474 out := strings.Trim(b.String(), "-")
475 if out == "" {
476 out = "agent"
477 }
478 return out
479}