1package subagents
2
3import (
4 "context"
5 "fmt"
6 "path/filepath"
7 "strings"
8 "sync"
9 "time"
10
11 "github.com/google/uuid"
12)
13
14// Agent is one supervised task. Public fields are immutable after
15// Spawn; mutable state (status, activity, transcript) lives behind
16// the embedded mutex.
17type turnEndNotice struct {
18 step int
19 errMsg string
20}
21
22type queuedResumePrompt struct {
23 Prompt string `json:"prompt"`
24 AcceptedAt time.Time `json:"accepted_at"`
25 CommandID string `json:"command_id,omitempty"`
26}
27
28type resumePromptSnapshot struct {
29 Prompt string
30 AcceptedAt time.Time
31 CommandID string
32}
33
34type Agent struct {
35 ID string
36 Task string
37 Dir string // shared checkout or isolated worktree used by the child.
38 Started time.Time
39
40 // ParentID and RootSessionID identify the supervisor/session that owns
41 // this worker. Empty ParentID means the main supervisor created it;
42 // child-originated spawning is rejected in v1.
43 ParentID string
44 BatchID string
45 RootSessionID string
46 OriginalTask string
47 RepositoryRoot string
48 WorkspaceMode WorkspaceMode
49 WorkspacePath string
50 WorkspaceBase string
51 WorkspaceCapture CaptureMode
52 MaxTurns int
53
54 // LifetimeTurns counts every accepted message turn for this worker across
55 // all explicit runs. CurrentRunTurns resets only when an explicit
56 // subagent_resume follow-up starts a new run; retries, model loops,
57 // compaction, and cancellation do not reset it.
58 LifetimeTurns int
59 CurrentRunTurns int
60
61 // Timeout is the effective per-agent lifetime, persisted so a resumed
62 // worker keeps the timeout selected at its spawn boundary.
63 Timeout time.Duration
64 HeartbeatInterval time.Duration
65 Tools []string
66 WebSearchPolicy WebSearchPolicy
67
68 // Model and Provider, when non-empty, override the child
69 // subprocess's model resolution. BaseURL and InsecureTLS carry
70 // the effective provider connection settings from the supervisor.
71 // Empty values inherit whatever the child resolves on its own from
72 // config / env / flags. All four settings are persisted in meta.json
73 // so Resume keeps using the same provider configuration across zut
74 // restarts.
75 Model string
76 Provider string
77 BaseURL string
78 InsecureTLS bool
79 Reasoning string
80 FastMode bool
81 Subagent string
82
83 // fastModeOverridesHost records the spawn-time case where an explicit
84 // profile fastMode: true enabled the child while the host was off. It is
85 // intentionally transient: the parent only needs it to report the initial
86 // spawn warning, while the effective FastMode remains durable state.
87 fastModeOverridesHost bool
88
89 // SessionID, when non-empty, scopes the agent to a particular
90 // host zut session: the dashboard only surfaces agents whose
91 // SessionID matches the active session. Empty means "unscoped" and
92 // applies to agents spawned without a session context such as in tests. Set at
93 // Spawn time from Supervisor.activeSession and persisted in
94 // meta.json so the scope survives restarts.
95 SessionID string
96
97 // Resuming is true when this Agent struct was built by Resume
98 // rather than Spawn. The runner consults it to decide whether
99 // to pass the original Task as a positional argv to the child:
100 // on first Spawn we want the child to run the task immediately;
101 // on Resume the task was already run last time and must not replay.
102 // Not persisted — every Resume sets it explicitly.
103 Resuming bool
104
105 // resumePromptText is the manager follow-up currently handed to the
106 // child. It remains in durable metadata until the child reports
107 // turn.started, so a host exit while the worker is queued cannot lose it.
108 // Additional follow-ups accepted while the child is busy stay in
109 // resumePromptQueue until the next agent idle turn.
110 resumePromptText string
111 resumePromptAt time.Time
112 resumePromptID string
113 resumePromptQueue []queuedResumePrompt
114
115 // InboxPath is the unix socket the child agent listens on for
116 // follow-up prompts and control messages. The supervisor opens
117 // an Inbox at this path; the child opens a Listener.
118 InboxPath string
119
120 // Attempt increments whenever the supervisor starts a process for this
121 // agent. It is persisted so recovery can distinguish a resumed worker
122 // from a first launch.
123 Attempt int
124
125 // ProcessPID identifies the current worker when known. It is persisted
126 // so a reloaded supervisor can force-kill a detached worker after a
127 // graceful shutdown deadline.
128 ProcessPID int
129
130 // EventLogPath is the durable JSONL event log for this agent.
131 // The runner appends every well-formed event from the child
132 // (plus lifecycle events of its own) here. /subagents open in any
133 // zut process reads from this file to replay the full history.
134 EventLogPath string
135
136 // SessionPath is the child's persistent session file. Surfaced
137 // so the dashboard / /subagents open can resume the agent later.
138 SessionPath string
139
140 // inbox is the supervisor-side socket handle. Populated by
141 // Supervisor.Spawn so callers do not need to manage the socket directly.
142 inbox *Inbox
143
144 mu sync.Mutex
145 status Status
146 activity string
147 transcript []string
148 lastAssistant string
149 streamingAssistantText string
150 streamingAssistantTruncated bool
151 streamingAssistantStart int
152 streamingAssistantLines int
153 finished time.Time
154 lastErr error
155 transcriptLoaded bool
156 legacyEventState bool
157 transcriptMu sync.Mutex
158
159 lifecycleMu sync.Mutex
160 processState ProcessState
161 turnState TurnState
162 currentTurnID string
163 updatedAt time.Time
164 lastActivity time.Time
165 result *TurnResult
166 resultRef string
167 patchRef string
168 changedFiles []string
169 outputBytes int
170 outputLines int
171 outputTruncated bool
172 maxOutputBytes int
173 maxOutputLines int
174 persistFn func(*Agent) error
175 persistMu sync.Mutex
176 stateDir string
177 workspaceCleanup func() error
178 workspaceCapture func() (WorkspaceCapture, error)
179 onTurnIdle func()
180
181 // OnTurnEnd, if set, fires once per prompt-level turn_end event
182 // emitted by the subagent daemon wrapper. Provider/tool-loop
183 // turn_end events (for example stop=tool_use) are ignored by the
184 // runner because they do not mean the sub-agent task finished.
185 // Used by auto-subagents watchers to detect that a sub-agent's first
186 // (or n-th) task has finished without waiting for the long-lived
187 // daemon itself to exit — sub-agents keep running on the inbox
188 // even after the initial task completes, so Wait() never unblocks
189 // for them.
190 OnTurnEnd func(step int, errMsg string)
191 pendingTurnEnds []turnEndNotice
192
193 ctx context.Context
194 cancel context.CancelFunc
195 runner Runner
196 lease *agentLease
197
198 // launchState is an admission gate for the scheduler's goroutine:
199 // 0=not admitted, 1=runner admitted, 2=stopped before admission. It
200 // prevents an immediate Stop from starting a runner after cancellation.
201 launchState int32
202
203 // done closes when the run goroutine finalises the agent's
204 // status (done / failed / killed). Wait blocks on this so
205 // callers don't have to poll.
206 done chan struct{}
207 turnResults chan *TurnResult
208 doneOnce sync.Once
209}
210
211// Inbox exposes the supervisor-side socket handle. Returns nil for
212// agents without inboxes (e.g. tests using a custom Runner that
213// doesn't speak the daemon protocol).
214func (a *Agent) Inbox() *Inbox { return a.inbox }
215
216// FastModeOverridesHost reports whether this child was created with an
217// explicit fast-mode override while the host setting was disabled.
218func (a *Agent) FastModeOverridesHost() bool {
219 return a != nil && a.fastModeOverridesHost
220}
221
222// Status returns the current high-level status. Cheap; safe from any goroutine.
223func (a *Agent) Status() Status {
224 a.mu.Lock()
225 defer a.mu.Unlock()
226 return a.status
227}
228
229// Activity returns the current one-line activity string.
230func (a *Agent) Activity() string {
231 a.mu.Lock()
232 defer a.mu.Unlock()
233 return a.activity
234}
235
236// ProcessState returns the child process lifecycle state.
237func (a *Agent) ProcessState() ProcessState {
238 a.lifecycleMu.Lock()
239 defer a.lifecycleMu.Unlock()
240 return a.processState
241}
242
243// TurnState returns the current delegated-turn state.
244func (a *Agent) TurnState() TurnState {
245 a.lifecycleMu.Lock()
246 defer a.lifecycleMu.Unlock()
247 return a.turnState
248}
249
250// CurrentTurnID returns the stable id of the current or most recent turn.
251func (a *Agent) CurrentTurnID() string {
252 a.lifecycleMu.Lock()
253 defer a.lifecycleMu.Unlock()
254 return a.currentTurnID
255}
256
257// AttemptValue returns the current process attempt number.
258func (a *Agent) AttemptValue() int {
259 a.lifecycleMu.Lock()
260 defer a.lifecycleMu.Unlock()
261 return a.Attempt
262}
263
264// LifetimeTurnsValue returns the number of accepted message turns across all
265// explicit runs of this worker.
266func (a *Agent) LifetimeTurnsValue() int {
267 a.lifecycleMu.Lock()
268 defer a.lifecycleMu.Unlock()
269 return a.LifetimeTurns
270}
271
272// CurrentRunTurnsValue returns the number of accepted message turns in the
273// current explicit run.
274func (a *Agent) CurrentRunTurnsValue() int {
275 a.lifecycleMu.Lock()
276 defer a.lifecycleMu.Unlock()
277 return a.CurrentRunTurns
278}
279
280func (a *Agent) setTurnCounts(lifetime, currentRun int) {
281 if lifetime < 0 {
282 lifetime = 0
283 }
284 if currentRun < 0 {
285 currentRun = 0
286 }
287 a.lifecycleMu.Lock()
288 a.LifetimeTurns = lifetime
289 a.CurrentRunTurns = currentRun
290 a.updatedAt = time.Now()
291 a.lifecycleMu.Unlock()
292}
293
294func (a *Agent) incrementTurnCounts() (lifetime, currentRun int) {
295 a.lifecycleMu.Lock()
296 a.LifetimeTurns++
297 a.CurrentRunTurns++
298 a.updatedAt = time.Now()
299 lifetime = a.LifetimeTurns
300 currentRun = a.CurrentRunTurns
301 a.lifecycleMu.Unlock()
302 return lifetime, currentRun
303}
304
305func (a *Agent) resetCurrentRunTurns() (previous int) {
306 a.lifecycleMu.Lock()
307 previous = a.CurrentRunTurns
308 a.CurrentRunTurns = 0
309 a.updatedAt = time.Now()
310 a.lifecycleMu.Unlock()
311 return previous
312}
313
314func (a *Agent) setCurrentRunTurns(currentRun int) {
315 if currentRun < 0 {
316 currentRun = 0
317 }
318 a.lifecycleMu.Lock()
319 a.CurrentRunTurns = currentRun
320 a.updatedAt = time.Now()
321 a.lifecycleMu.Unlock()
322}
323
324// ProcessPIDValue returns the current worker pid, when known.
325func (a *Agent) ProcessPIDValue() int {
326 a.lifecycleMu.Lock()
327 defer a.lifecycleMu.Unlock()
328 return a.ProcessPID
329}
330
331// LastActivity returns the last heartbeat or meaningful child event time.
332func (a *Agent) LastActivity() time.Time {
333 a.lifecycleMu.Lock()
334 defer a.lifecycleMu.Unlock()
335 return a.lastActivity
336}
337
338// UpdatedAt returns the last lifecycle-state update time.
339func (a *Agent) UpdatedAt() time.Time {
340 a.lifecycleMu.Lock()
341 defer a.lifecycleMu.Unlock()
342 return a.updatedAt
343}
344
345// Result returns a copy of the most recent structured turn result.
346func (a *Agent) Result() *TurnResult {
347 a.lifecycleMu.Lock()
348 defer a.lifecycleMu.Unlock()
349 return cloneTurnResult(a.result)
350}
351
352func (a *Agent) setProcessState(state ProcessState) {
353 a.lifecycleMu.Lock()
354 a.processState = state
355 a.updatedAt = time.Now()
356 a.lifecycleMu.Unlock()
357}
358
359func (a *Agent) incrementAttempt() int {
360 a.lifecycleMu.Lock()
361 a.Attempt++
362 a.updatedAt = time.Now()
363 attempt := a.Attempt
364 a.lifecycleMu.Unlock()
365 return attempt
366}
367
368func (a *Agent) setProcessPID(pid int) {
369 a.lifecycleMu.Lock()
370 a.ProcessPID = pid
371 a.updatedAt = time.Now()
372 a.lifecycleMu.Unlock()
373}
374
375func (a *Agent) releaseLease() error {
376 a.lifecycleMu.Lock()
377 lease := a.lease
378 a.lease = nil
379 a.lifecycleMu.Unlock()
380 if lease == nil {
381 return nil
382 }
383 return lease.Close()
384}
385
386func (a *Agent) setTurnState(state TurnState, turnID string) {
387 a.lifecycleMu.Lock()
388 a.turnState = state
389 if turnID != "" {
390 a.currentTurnID = turnID
391 }
392 a.updatedAt = time.Now()
393 a.lifecycleMu.Unlock()
394}
395
396func (a *Agent) markActivity(now time.Time) {
397 if now.IsZero() {
398 now = time.Now()
399 }
400 a.lifecycleMu.Lock()
401 a.lastActivity = now
402 a.updatedAt = now
403 a.lifecycleMu.Unlock()
404}
405
406func (a *Agent) setResult(result *TurnResult) {
407 a.lifecycleMu.Lock()
408 a.result = cloneTurnResult(result)
409 a.updatedAt = time.Now()
410 a.lifecycleMu.Unlock()
411}
412
413// Transcript returns a copy of the running transcript. Detached agents
414// hydrate their durable event history the first time it is requested.
415func (a *Agent) Transcript() []string {
416 _ = a.loadTranscript()
417 a.mu.Lock()
418 defer a.mu.Unlock()
419 out := make([]string, len(a.transcript))
420 copy(out, a.transcript)
421 return out
422}
423
424// Err returns the runner's terminal error, if any.
425func (a *Agent) Err() error {
426 a.mu.Lock()
427 defer a.mu.Unlock()
428 return a.lastErr
429}
430
431// Wait blocks until the agent reaches a terminal state. Used by tests
432// and by /subagents wait <id>.
433func (a *Agent) Wait() { <-a.done }
434
435// WaitContext waits for the agent to finish or for ctx cancellation.
436func (a *Agent) WaitContext(ctx context.Context) error {
437 if ctx == nil {
438 ctx = context.Background()
439 }
440 select {
441 case <-a.done:
442 return nil
443 case <-ctx.Done():
444 return ctx.Err()
445 }
446}
447
448func (a *Agent) waitForTurnResult(ctx context.Context) (*TurnResult, error) {
449 if ctx == nil {
450 ctx = context.Background()
451 }
452 select {
453 case result := <-a.turnResults:
454 if result == nil {
455 return nil, fmt.Errorf("subagents: nil turn result for %s", a.ID)
456 }
457 return cloneTurnResult(result), nil
458 case <-a.done:
459 if result := a.Result(); result != nil {
460 return result, nil
461 }
462 return nil, fmt.Errorf("subagents: agent %s exited without a turn result", a.ID)
463 case <-ctx.Done():
464 return nil, ctx.Err()
465 }
466}
467
468func (a *Agent) closeDone() { a.doneOnce.Do(func() { close(a.done) }) }
469
470func (a *Agent) stateDirectory(defaultRoot string) string {
471 if a.stateDir != "" {
472 return a.stateDir
473 }
474 return filepath.Join(defaultRoot, "agents", a.ID)
475}
476
477// SetOnTurnEnd installs (or clears, with nil) the callback fired
478// from the runner when the child daemon emits a prompt-level
479// turn_end event with a step field. Safe to call from any goroutine:
480// the runner reads the callback under the same mutex.
481func (a *Agent) resumePrompt() string {
482 prompt, _ := a.ResumePromptInfo()
483 return prompt
484}
485
486// ResumePromptInfo returns the queued manager follow-up and its acceptance
487// time. Both values are read as one synchronized snapshot.
488func (a *Agent) ResumePromptInfo() (string, time.Time) {
489 a.mu.Lock()
490 defer a.mu.Unlock()
491 return a.resumePromptText, a.resumePromptAt
492}
493
494func (a *Agent) setResumePrompt(prompt string, acceptedAt time.Time) resumePromptSnapshot {
495 a.mu.Lock()
496 defer a.mu.Unlock()
497 previous := resumePromptSnapshot{
498 Prompt: a.resumePromptText,
499 AcceptedAt: a.resumePromptAt,
500 CommandID: a.resumePromptID,
501 }
502 a.resumePromptText = prompt
503 a.resumePromptAt = acceptedAt
504 a.resumePromptID = uuid.NewString()
505 return previous
506}
507
508func (a *Agent) restoreResumePrompt(previous resumePromptSnapshot) {
509 a.mu.Lock()
510 defer a.mu.Unlock()
511 a.resumePromptText = previous.Prompt
512 a.resumePromptAt = previous.AcceptedAt
513 a.resumePromptID = previous.CommandID
514 if a.resumePromptText != "" && a.resumePromptID == "" {
515 a.resumePromptID = uuid.NewString()
516 }
517}
518
519func (a *Agent) resumePromptState() resumePromptSnapshot {
520 a.mu.Lock()
521 defer a.mu.Unlock()
522 return resumePromptSnapshot{
523 Prompt: a.resumePromptText,
524 AcceptedAt: a.resumePromptAt,
525 CommandID: a.resumePromptID,
526 }
527}
528
529func (a *Agent) resumePromptCommandID() string {
530 a.mu.Lock()
531 defer a.mu.Unlock()
532 return a.resumePromptID
533}
534
535func (a *Agent) clearResumePrompt() bool {
536 a.mu.Lock()
537 defer a.mu.Unlock()
538 if a.resumePromptText == "" {
539 return false
540 }
541 a.resumePromptText = ""
542 a.resumePromptAt = time.Time{}
543 a.resumePromptID = ""
544 return true
545}
546
547func (a *Agent) queueResumePrompt(prompt string, acceptedAt time.Time) string {
548 commandID := uuid.NewString()
549 a.mu.Lock()
550 a.resumePromptQueue = append(a.resumePromptQueue, queuedResumePrompt{Prompt: prompt, AcceptedAt: acceptedAt, CommandID: commandID})
551 a.mu.Unlock()
552 return commandID
553}
554
555func (a *Agent) removeQueuedResumePrompt(commandID string) bool {
556 if commandID == "" {
557 return false
558 }
559 a.mu.Lock()
560 defer a.mu.Unlock()
561 for i := len(a.resumePromptQueue) - 1; i >= 0; i-- {
562 if a.resumePromptQueue[i].CommandID == commandID {
563 a.resumePromptQueue = append(a.resumePromptQueue[:i], a.resumePromptQueue[i+1:]...)
564 return true
565 }
566 }
567 return false
568}
569
570func (a *Agent) resumePromptQueueSnapshot() []queuedResumePrompt {
571 a.mu.Lock()
572 defer a.mu.Unlock()
573 return append([]queuedResumePrompt(nil), a.resumePromptQueue...)
574}
575
576func (a *Agent) hasQueuedResumePrompt() bool {
577 a.mu.Lock()
578 defer a.mu.Unlock()
579 return len(a.resumePromptQueue) != 0
580}
581
582// startQueuedResumePrompt moves the oldest active follow-up into the
583// in-flight prompt slot. The caller must persist the returned state before
584// delivering the command so a host exit cannot lose the prompt.
585func (a *Agent) startQueuedResumePrompt() (string, time.Time, string, bool) {
586 a.mu.Lock()
587 defer a.mu.Unlock()
588 if a.resumePromptText != "" || len(a.resumePromptQueue) == 0 {
589 return "", time.Time{}, "", false
590 }
591 queued := a.resumePromptQueue[0]
592 a.resumePromptQueue = a.resumePromptQueue[1:]
593 a.resumePromptText = queued.Prompt
594 a.resumePromptAt = queued.AcceptedAt
595 a.resumePromptID = queued.CommandID
596 if a.resumePromptID == "" {
597 a.resumePromptID = uuid.NewString()
598 }
599 return queued.Prompt, queued.AcceptedAt, a.resumePromptID, true
600}
601
602func (a *Agent) restoreResumePromptToQueue(prompt string, acceptedAt time.Time) bool {
603 a.mu.Lock()
604 defer a.mu.Unlock()
605 if a.resumePromptText != prompt || !a.resumePromptAt.Equal(acceptedAt) {
606 return false
607 }
608 commandID := a.resumePromptID
609 if commandID == "" {
610 commandID = uuid.NewString()
611 }
612 a.resumePromptText = ""
613 a.resumePromptAt = time.Time{}
614 a.resumePromptID = ""
615 a.resumePromptQueue = append([]queuedResumePrompt{{Prompt: prompt, AcceptedAt: acceptedAt, CommandID: commandID}}, a.resumePromptQueue...)
616 return true
617}
618
619func (a *Agent) rejectActiveResumePrompt() bool {
620 a.mu.Lock()
621 defer a.mu.Unlock()
622 if a.resumePromptText == "" {
623 return false
624 }
625 a.resumePromptQueue = append([]queuedResumePrompt{{Prompt: a.resumePromptText, AcceptedAt: a.resumePromptAt, CommandID: a.resumePromptID}}, a.resumePromptQueue...)
626 a.resumePromptText = ""
627 a.resumePromptAt = time.Time{}
628 a.resumePromptID = ""
629 return true
630}
631
632func (a *Agent) setOnTurnIdle(fn func()) {
633 a.mu.Lock()
634 a.onTurnIdle = fn
635 a.mu.Unlock()
636}
637
638func (a *Agent) notifyTurnIdle() {
639 a.mu.Lock()
640 fn := a.onTurnIdle
641 a.mu.Unlock()
642 if fn != nil {
643 fn()
644 }
645}
646
647func (a *Agent) SetOnTurnEnd(fn func(step int, errMsg string)) {
648 a.mu.Lock()
649 a.OnTurnEnd = fn
650 pending := append([]turnEndNotice(nil), a.pendingTurnEnds...)
651 a.pendingTurnEnds = nil
652 a.mu.Unlock()
653 if fn != nil {
654 for _, notice := range pending {
655 fn(notice.step, notice.errMsg)
656 }
657 }
658}
659
660func (a *Agent) recordPersistenceError(err error) {
661 if a == nil || err == nil {
662 return
663 }
664 a.mu.Lock()
665 if a.lastErr == nil {
666 a.lastErr = fmt.Errorf("subagent metadata persistence: %w", err)
667 }
668 if a.status != StatusDone && a.status != StatusFailed && a.status != StatusKilled {
669 a.activity = "metadata error: " + truncate(err.Error(), 120)
670 }
671 a.mu.Unlock()
672}
673
674func (a *Agent) setActivity(msg string) {
675 a.mu.Lock()
676 a.activity = strings.TrimSpace(msg)
677 a.mu.Unlock()
678 a.markActivity(time.Now())
679}
680
681func (a *Agent) appendTranscript(chunk string) {
682 a.appendTranscriptChunk(chunk, "", false)
683}
684
685func (a *Agent) appendUserMessage(text string) {
686 a.appendTranscriptChunk(text, "user: ", false)
687}
688
689func (a *Agent) appendAssistantMessage(text string) {
690 if text == "" {
691 return
692 }
693 a.mu.Lock()
694 if a.streamingAssistantText != "" {
695 a.replaceStreamingAssistantLocked(text)
696 a.clearStreamingAssistantLocked()
697 } else {
698 a.appendTranscriptLinesLocked(text, "", true)
699 a.boundTranscriptLocked()
700 }
701 a.mu.Unlock()
702 a.markActivity(time.Now())
703}
704
705// appendAssistantDelta exposes an in-progress assistant response in the
706// dashboard without duplicating it when the final assistant_message arrives.
707// The JSONL event log remains the durable source; this maintains its live and
708// replayed in-memory projection.
709func (a *Agent) appendAssistantDelta(text string) {
710 if text == "" {
711 return
712 }
713 a.mu.Lock()
714 if !a.streamingAssistantTruncated {
715 value := a.streamingAssistantText + text
716 byteCap, lineCap := a.outputLimits()
717 a.streamingAssistantText = boundInlineText(value, byteCap, lineCap)
718 a.streamingAssistantTruncated = a.streamingAssistantText != value
719 }
720 a.replaceStreamingAssistantLocked(a.streamingAssistantText)
721 a.mu.Unlock()
722 a.markActivity(time.Now())
723}
724
725func (a *Agent) appendTranscriptChunk(chunk, linePrefix string, assistant bool) {
726 if strings.TrimRight(chunk, "\n") == "" {
727 return
728 }
729 a.mu.Lock()
730 a.appendTranscriptLinesLocked(chunk, linePrefix, assistant)
731 a.boundTranscriptLocked()
732 a.mu.Unlock()
733 a.markActivity(time.Now())
734}
735
736func (a *Agent) resetStreamingAssistant() {
737 a.mu.Lock()
738 a.clearStreamingAssistantLocked()
739 a.mu.Unlock()
740}
741
742func (a *Agent) replaceStreamingAssistantLocked(text string) {
743 a.removeStreamingAssistantLocked()
744 a.streamingAssistantStart = len(a.transcript)
745 a.streamingAssistantLines = a.appendTranscriptLinesLocked(text, "", true)
746 a.boundTranscriptLocked()
747}
748
749func (a *Agent) removeStreamingAssistantLocked() {
750 if a.streamingAssistantLines == 0 {
751 return
752 }
753 start := a.streamingAssistantStart
754 if start < 0 {
755 start = 0
756 }
757 if start > len(a.transcript) {
758 start = len(a.transcript)
759 }
760 end := start + a.streamingAssistantLines
761 if end > len(a.transcript) {
762 end = len(a.transcript)
763 }
764 for _, line := range a.transcript[start:end] {
765 a.outputBytes -= len(line) + 1
766 a.outputLines--
767 }
768 a.transcript = append(a.transcript[:start], a.transcript[end:]...)
769 a.streamingAssistantLines = 0
770 if a.outputLines == 0 {
771 a.outputBytes = 0
772 }
773}
774
775func (a *Agent) clearStreamingAssistantLocked() {
776 a.streamingAssistantText = ""
777 a.streamingAssistantTruncated = false
778 a.streamingAssistantStart = 0
779 a.streamingAssistantLines = 0
780}
781
782func (a *Agent) appendTranscriptLinesLocked(chunk, linePrefix string, assistant bool) int {
783 message := chunk
784 chunk = strings.TrimRight(chunk, "\n")
785 if chunk == "" {
786 return 0
787 }
788 lines := strings.Split(chunk, "\n")
789 for _, line := range lines {
790 line = linePrefix + line
791 a.transcript = append(a.transcript, line)
792 a.outputBytes += len(line) + 1
793 a.outputLines++
794 }
795 if assistant {
796 a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
797 }
798 return len(lines)
799}
800
801func (a *Agent) outputLimits() (byteCap, lineCap int) {
802 byteCap = a.maxOutputBytes
803 if byteCap <= 0 {
804 byteCap = 500_000
805 }
806 lineCap = a.maxOutputLines
807 if lineCap <= 0 {
808 lineCap = 2_000
809 }
810 return byteCap, lineCap
811}
812
813func (a *Agent) boundTranscriptLocked() {
814 // Bound the inline dashboard copy. The durable event log and session
815 // remain available through the logical history reference.
816 byteCap, lineCap := a.outputLimits()
817 for a.outputLines > lineCap || (a.outputLines > 0 && a.outputBytes > byteCap) {
818 line := a.transcript[0]
819 a.transcript = a.transcript[1:]
820 a.outputBytes -= len(line) + 1
821 a.outputLines--
822 if a.streamingAssistantLines > 0 {
823 if a.streamingAssistantStart > 0 {
824 a.streamingAssistantStart--
825 } else {
826 a.streamingAssistantLines--
827 }
828 }
829 a.outputTruncated = true
830 }
831 if a.outputLines == 0 {
832 a.outputBytes = 0
833 }
834}
835
836// newAgentID returns a short, mostly-collision-free identifier of the
837// form "<slug>-<nano>". The slug is derived from the task text so
838// dashboards stay readable; the nano suffix guarantees uniqueness
839// even when two agents are spawned in the same millisecond.
840func newAgentID(task string, now time.Time) string {
841 slug := taskSlug(task)
842 return fmt.Sprintf("%s-%d", slug, now.UnixNano()%1_000_000)
843}
844
845func taskSlug(task string) string {
846 task = strings.ToLower(task)
847 var b strings.Builder
848 dash := false
849 for _, r := range task {
850 switch {
851 case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
852 b.WriteRune(r)
853 dash = false
854 default:
855 if !dash && b.Len() > 0 {
856 b.WriteByte('-')
857 dash = true
858 }
859 }
860 if b.Len() >= 24 {
861 break
862 }
863 }
864 out := strings.Trim(b.String(), "-")
865 if out == "" {
866 out = "agent"
867 }
868 return out
869}