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 shutdownOrigin ShutdownOrigin
156 transcriptLoaded bool
157 legacyEventState bool
158 transcriptMu sync.Mutex
159
160 lifecycleMu sync.Mutex
161 processState ProcessState
162 turnState TurnState
163 currentTurnID string
164 updatedAt time.Time
165 lastActivity time.Time
166 result *TurnResult
167 resultRef string
168 requirement RequirementSnapshot
169 requirementPersisting bool
170 requirementChanged chan struct{}
171 patchRef string
172 changedFiles []string
173 outputBytes int
174 outputLines int
175 outputTruncated bool
176 maxOutputBytes int
177 maxOutputLines int
178 persistFn func(*Agent) error
179 persistMu sync.Mutex
180 stateDir string
181 workspaceCleanup func() error
182 workspaceCapture func() (WorkspaceCapture, error)
183 onTurnIdle func()
184
185 // OnTurnEnd, if set, fires once per prompt-level turn_end event
186 // emitted by the subagent daemon wrapper. Provider/tool-loop
187 // turn_end events (for example stop=tool_use) are ignored by the
188 // runner because they do not mean the sub-agent task finished.
189 // Used by auto-subagents watchers to detect that a sub-agent's first
190 // (or n-th) task has finished without waiting for the long-lived
191 // daemon itself to exit — sub-agents keep running on the inbox
192 // even after the initial task completes, so Wait() never unblocks
193 // for them.
194 OnTurnEnd func(step int, errMsg string)
195 pendingTurnEnds []turnEndNotice
196
197 ctx context.Context
198 cancel context.CancelFunc
199 runner Runner
200 lease *agentLease
201
202 // launchState is an admission gate for the scheduler's goroutine:
203 // 0=not admitted, 1=runner admitted, 2=stopped before admission. It
204 // prevents an immediate Stop from starting a runner after cancellation.
205 launchState int32
206
207 // done closes when the run goroutine finalises the agent's
208 // status (done / failed / killed). Wait blocks on this so
209 // callers don't have to poll.
210 done chan struct{}
211 turnResults chan *TurnResult
212 doneOnce sync.Once
213}
214
215// Inbox exposes the supervisor-side socket handle. Returns nil for
216// agents without inboxes (e.g. tests using a custom Runner that
217// doesn't speak the daemon protocol).
218func (a *Agent) Inbox() *Inbox { return a.inbox }
219
220// FastModeOverridesHost reports whether this child was created with an
221// explicit fast-mode override while the host setting was disabled.
222func (a *Agent) FastModeOverridesHost() bool {
223 return a != nil && a.fastModeOverridesHost
224}
225
226// Status returns the current high-level status. Cheap; safe from any goroutine.
227func (a *Agent) Status() Status {
228 a.mu.Lock()
229 defer a.mu.Unlock()
230 return a.status
231}
232
233// Activity returns the current one-line activity string.
234func (a *Agent) Activity() string {
235 a.mu.Lock()
236 defer a.mu.Unlock()
237 return a.activity
238}
239
240// ProcessState returns the child process lifecycle state.
241func (a *Agent) ProcessState() ProcessState {
242 a.lifecycleMu.Lock()
243 defer a.lifecycleMu.Unlock()
244 return a.processState
245}
246
247// TurnState returns the current delegated-turn state.
248func (a *Agent) TurnState() TurnState {
249 a.lifecycleMu.Lock()
250 defer a.lifecycleMu.Unlock()
251 return a.turnState
252}
253
254// CurrentTurnID returns the stable id of the current or most recent turn.
255func (a *Agent) CurrentTurnID() string {
256 a.lifecycleMu.Lock()
257 defer a.lifecycleMu.Unlock()
258 return a.currentTurnID
259}
260
261// AttemptValue returns the current process attempt number.
262func (a *Agent) AttemptValue() int {
263 a.lifecycleMu.Lock()
264 defer a.lifecycleMu.Unlock()
265 return a.Attempt
266}
267
268// LifetimeTurnsValue returns the number of accepted message turns across all
269// explicit runs of this worker.
270func (a *Agent) LifetimeTurnsValue() int {
271 a.lifecycleMu.Lock()
272 defer a.lifecycleMu.Unlock()
273 return a.LifetimeTurns
274}
275
276// CurrentRunTurnsValue returns the number of accepted message turns in the
277// current explicit run.
278func (a *Agent) CurrentRunTurnsValue() int {
279 a.lifecycleMu.Lock()
280 defer a.lifecycleMu.Unlock()
281 return a.CurrentRunTurns
282}
283
284func (a *Agent) setTurnCounts(lifetime, currentRun int) {
285 if lifetime < 0 {
286 lifetime = 0
287 }
288 if currentRun < 0 {
289 currentRun = 0
290 }
291 a.lifecycleMu.Lock()
292 a.LifetimeTurns = lifetime
293 a.CurrentRunTurns = currentRun
294 a.updatedAt = time.Now()
295 a.lifecycleMu.Unlock()
296}
297
298func (a *Agent) incrementTurnCounts() (lifetime, currentRun int) {
299 a.lifecycleMu.Lock()
300 a.LifetimeTurns++
301 a.CurrentRunTurns++
302 a.updatedAt = time.Now()
303 lifetime = a.LifetimeTurns
304 currentRun = a.CurrentRunTurns
305 a.lifecycleMu.Unlock()
306 return lifetime, currentRun
307}
308
309func (a *Agent) resetCurrentRunTurns() (previous int) {
310 a.lifecycleMu.Lock()
311 previous = a.CurrentRunTurns
312 a.CurrentRunTurns = 0
313 a.updatedAt = time.Now()
314 a.lifecycleMu.Unlock()
315 return previous
316}
317
318func (a *Agent) setCurrentRunTurns(currentRun int) {
319 if currentRun < 0 {
320 currentRun = 0
321 }
322 a.lifecycleMu.Lock()
323 a.CurrentRunTurns = currentRun
324 a.updatedAt = time.Now()
325 a.lifecycleMu.Unlock()
326}
327
328// ProcessPIDValue returns the current worker pid, when known.
329func (a *Agent) ProcessPIDValue() int {
330 a.lifecycleMu.Lock()
331 defer a.lifecycleMu.Unlock()
332 return a.ProcessPID
333}
334
335// LastActivity returns the last heartbeat or meaningful child event time.
336func (a *Agent) LastActivity() time.Time {
337 a.lifecycleMu.Lock()
338 defer a.lifecycleMu.Unlock()
339 return a.lastActivity
340}
341
342// UpdatedAt returns the last lifecycle-state update time.
343func (a *Agent) UpdatedAt() time.Time {
344 a.lifecycleMu.Lock()
345 defer a.lifecycleMu.Unlock()
346 return a.updatedAt
347}
348
349// Result returns a copy of the most recent structured turn result.
350func (a *Agent) Result() *TurnResult {
351 a.lifecycleMu.Lock()
352 defer a.lifecycleMu.Unlock()
353 return cloneTurnResult(a.result)
354}
355
356func (a *Agent) setProcessState(state ProcessState) {
357 a.lifecycleMu.Lock()
358 a.processState = state
359 a.updatedAt = time.Now()
360 a.lifecycleMu.Unlock()
361}
362
363func (a *Agent) incrementAttempt() int {
364 a.lifecycleMu.Lock()
365 a.Attempt++
366 a.updatedAt = time.Now()
367 attempt := a.Attempt
368 a.lifecycleMu.Unlock()
369 return attempt
370}
371
372func (a *Agent) setProcessPID(pid int) {
373 a.lifecycleMu.Lock()
374 a.ProcessPID = pid
375 a.updatedAt = time.Now()
376 a.lifecycleMu.Unlock()
377}
378
379func (a *Agent) releaseLease() error {
380 a.lifecycleMu.Lock()
381 lease := a.lease
382 a.lease = nil
383 a.lifecycleMu.Unlock()
384 if lease == nil {
385 return nil
386 }
387 return lease.Close()
388}
389
390func (a *Agent) setTurnState(state TurnState, turnID string) {
391 a.lifecycleMu.Lock()
392 a.turnState = state
393 if turnID != "" {
394 a.currentTurnID = turnID
395 }
396 a.updatedAt = time.Now()
397 a.lifecycleMu.Unlock()
398}
399
400func (a *Agent) markActivity(now time.Time) {
401 if now.IsZero() {
402 now = time.Now()
403 }
404 a.lifecycleMu.Lock()
405 a.lastActivity = now
406 a.updatedAt = now
407 a.lifecycleMu.Unlock()
408}
409
410func (a *Agent) setResult(result *TurnResult) {
411 a.lifecycleMu.Lock()
412 a.result = cloneTurnResult(result)
413 a.updatedAt = time.Now()
414 a.lifecycleMu.Unlock()
415}
416
417func (a *Agent) shutdownOriginValue() ShutdownOrigin {
418 a.mu.Lock()
419 defer a.mu.Unlock()
420 return a.shutdownOrigin
421}
422
423// prepareShutdown publishes the shutdown origin in the same critical section
424// as the killed status. Finalization can therefore observe either the previous
425// terminal state or a fully attributed cancellation, never an intermediate one.
426func (a *Agent) prepareShutdown(origin ShutdownOrigin) (Status, bool) {
427 a.mu.Lock()
428 defer a.mu.Unlock()
429 status := a.status
430 if status == StatusDone || status == StatusFailed || status == StatusKilled {
431 return status, false
432 }
433 a.shutdownOrigin = origin.Sanitized()
434 if status != StatusDetached {
435 a.status = StatusKilled
436 a.activity = "canceling"
437 }
438 return status, true
439}
440
441// Transcript returns a copy of the running transcript. Detached agents
442// hydrate their durable event history the first time it is requested.
443func (a *Agent) Transcript() []string {
444 _ = a.loadTranscript()
445 a.mu.Lock()
446 defer a.mu.Unlock()
447 out := make([]string, len(a.transcript))
448 copy(out, a.transcript)
449 return out
450}
451
452// Err returns the runner's terminal error, if any.
453func (a *Agent) Err() error {
454 a.mu.Lock()
455 defer a.mu.Unlock()
456 return a.lastErr
457}
458
459// Wait blocks until the agent reaches a terminal state. Used by tests
460// and by /subagents wait <id>.
461func (a *Agent) Wait() { <-a.done }
462
463// WaitContext waits for the agent to finish or for ctx cancellation.
464func (a *Agent) WaitContext(ctx context.Context) error {
465 if ctx == nil {
466 ctx = context.Background()
467 }
468 select {
469 case <-a.done:
470 return nil
471 case <-ctx.Done():
472 return ctx.Err()
473 }
474}
475
476func (a *Agent) waitForTurnResult(ctx context.Context) (*TurnResult, error) {
477 if ctx == nil {
478 ctx = context.Background()
479 }
480 select {
481 case result := <-a.turnResults:
482 if result == nil {
483 return nil, fmt.Errorf("subagents: nil turn result for %s", a.ID)
484 }
485 return cloneTurnResult(result), nil
486 case <-a.done:
487 if result := a.Result(); result != nil {
488 return result, nil
489 }
490 return nil, fmt.Errorf("subagents: agent %s exited without a turn result", a.ID)
491 case <-ctx.Done():
492 return nil, ctx.Err()
493 }
494}
495
496func (a *Agent) closeDone() { a.doneOnce.Do(func() { close(a.done) }) }
497
498func (a *Agent) stateDirectory(defaultRoot string) string {
499 if a.stateDir != "" {
500 return a.stateDir
501 }
502 return filepath.Join(defaultRoot, "agents", a.ID)
503}
504
505// SetOnTurnEnd installs (or clears, with nil) the callback fired
506// from the runner when the child daemon emits a prompt-level
507// turn_end event with a step field. Safe to call from any goroutine:
508// the runner reads the callback under the same mutex.
509func (a *Agent) resumePrompt() string {
510 prompt, _ := a.ResumePromptInfo()
511 return prompt
512}
513
514// ResumePromptInfo returns the queued manager follow-up and its acceptance
515// time. Both values are read as one synchronized snapshot.
516func (a *Agent) ResumePromptInfo() (string, time.Time) {
517 a.mu.Lock()
518 defer a.mu.Unlock()
519 return a.resumePromptText, a.resumePromptAt
520}
521
522func (a *Agent) setResumePrompt(prompt string, acceptedAt time.Time) resumePromptSnapshot {
523 a.mu.Lock()
524 defer a.mu.Unlock()
525 previous := resumePromptSnapshot{
526 Prompt: a.resumePromptText,
527 AcceptedAt: a.resumePromptAt,
528 CommandID: a.resumePromptID,
529 }
530 a.resumePromptText = prompt
531 a.resumePromptAt = acceptedAt
532 a.resumePromptID = uuid.NewString()
533 return previous
534}
535
536func (a *Agent) restoreResumePrompt(previous resumePromptSnapshot) {
537 a.mu.Lock()
538 defer a.mu.Unlock()
539 a.resumePromptText = previous.Prompt
540 a.resumePromptAt = previous.AcceptedAt
541 a.resumePromptID = previous.CommandID
542 if a.resumePromptText != "" && a.resumePromptID == "" {
543 a.resumePromptID = uuid.NewString()
544 }
545}
546
547func (a *Agent) resumePromptState() resumePromptSnapshot {
548 a.mu.Lock()
549 defer a.mu.Unlock()
550 return resumePromptSnapshot{
551 Prompt: a.resumePromptText,
552 AcceptedAt: a.resumePromptAt,
553 CommandID: a.resumePromptID,
554 }
555}
556
557func (a *Agent) resumePromptCommandID() string {
558 a.mu.Lock()
559 defer a.mu.Unlock()
560 return a.resumePromptID
561}
562
563func (a *Agent) clearResumePrompt() bool {
564 a.mu.Lock()
565 defer a.mu.Unlock()
566 if a.resumePromptText == "" {
567 return false
568 }
569 a.resumePromptText = ""
570 a.resumePromptAt = time.Time{}
571 a.resumePromptID = ""
572 return true
573}
574
575func (a *Agent) queueResumePrompt(prompt string, acceptedAt time.Time) string {
576 commandID := uuid.NewString()
577 a.mu.Lock()
578 a.resumePromptQueue = append(a.resumePromptQueue, queuedResumePrompt{Prompt: prompt, AcceptedAt: acceptedAt, CommandID: commandID})
579 a.mu.Unlock()
580 return commandID
581}
582
583func (a *Agent) removeQueuedResumePrompt(commandID string) bool {
584 if commandID == "" {
585 return false
586 }
587 a.mu.Lock()
588 defer a.mu.Unlock()
589 for i := len(a.resumePromptQueue) - 1; i >= 0; i-- {
590 if a.resumePromptQueue[i].CommandID == commandID {
591 a.resumePromptQueue = append(a.resumePromptQueue[:i], a.resumePromptQueue[i+1:]...)
592 return true
593 }
594 }
595 return false
596}
597
598func (a *Agent) resumePromptQueueSnapshot() []queuedResumePrompt {
599 a.mu.Lock()
600 defer a.mu.Unlock()
601 return append([]queuedResumePrompt(nil), a.resumePromptQueue...)
602}
603
604func (a *Agent) hasQueuedResumePrompt() bool {
605 a.mu.Lock()
606 defer a.mu.Unlock()
607 return len(a.resumePromptQueue) != 0
608}
609
610// startQueuedResumePrompt moves the oldest active follow-up into the
611// in-flight prompt slot. The caller must persist the returned state before
612// delivering the command so a host exit cannot lose the prompt.
613func (a *Agent) startQueuedResumePrompt() (string, time.Time, string, bool) {
614 a.mu.Lock()
615 defer a.mu.Unlock()
616 if a.resumePromptText != "" || len(a.resumePromptQueue) == 0 {
617 return "", time.Time{}, "", false
618 }
619 queued := a.resumePromptQueue[0]
620 a.resumePromptQueue = a.resumePromptQueue[1:]
621 a.resumePromptText = queued.Prompt
622 a.resumePromptAt = queued.AcceptedAt
623 a.resumePromptID = queued.CommandID
624 if a.resumePromptID == "" {
625 a.resumePromptID = uuid.NewString()
626 }
627 return queued.Prompt, queued.AcceptedAt, a.resumePromptID, true
628}
629
630func (a *Agent) restoreResumePromptToQueue(prompt string, acceptedAt time.Time) bool {
631 a.mu.Lock()
632 defer a.mu.Unlock()
633 if a.resumePromptText != prompt || !a.resumePromptAt.Equal(acceptedAt) {
634 return false
635 }
636 commandID := a.resumePromptID
637 if commandID == "" {
638 commandID = uuid.NewString()
639 }
640 a.resumePromptText = ""
641 a.resumePromptAt = time.Time{}
642 a.resumePromptID = ""
643 a.resumePromptQueue = append([]queuedResumePrompt{{Prompt: prompt, AcceptedAt: acceptedAt, CommandID: commandID}}, a.resumePromptQueue...)
644 return true
645}
646
647func (a *Agent) rejectActiveResumePrompt() bool {
648 a.mu.Lock()
649 defer a.mu.Unlock()
650 if a.resumePromptText == "" {
651 return false
652 }
653 a.resumePromptQueue = append([]queuedResumePrompt{{Prompt: a.resumePromptText, AcceptedAt: a.resumePromptAt, CommandID: a.resumePromptID}}, a.resumePromptQueue...)
654 a.resumePromptText = ""
655 a.resumePromptAt = time.Time{}
656 a.resumePromptID = ""
657 return true
658}
659
660func (a *Agent) setOnTurnIdle(fn func()) {
661 a.mu.Lock()
662 a.onTurnIdle = fn
663 a.mu.Unlock()
664}
665
666func (a *Agent) notifyTurnIdle() {
667 a.mu.Lock()
668 fn := a.onTurnIdle
669 a.mu.Unlock()
670 if fn != nil {
671 fn()
672 }
673}
674
675func (a *Agent) SetOnTurnEnd(fn func(step int, errMsg string)) {
676 a.mu.Lock()
677 a.OnTurnEnd = fn
678 pending := append([]turnEndNotice(nil), a.pendingTurnEnds...)
679 a.pendingTurnEnds = nil
680 a.mu.Unlock()
681 if fn != nil {
682 for _, notice := range pending {
683 fn(notice.step, notice.errMsg)
684 }
685 }
686}
687
688func (a *Agent) recordPersistenceError(err error) {
689 if a == nil || err == nil {
690 return
691 }
692 a.mu.Lock()
693 if a.lastErr == nil {
694 a.lastErr = fmt.Errorf("subagent metadata persistence: %w", err)
695 }
696 if a.status != StatusDone && a.status != StatusFailed && a.status != StatusKilled {
697 a.activity = "metadata error: " + truncate(err.Error(), 120)
698 }
699 a.mu.Unlock()
700}
701
702func (a *Agent) setActivity(msg string) {
703 a.mu.Lock()
704 a.activity = strings.TrimSpace(msg)
705 a.mu.Unlock()
706 a.markActivity(time.Now())
707}
708
709func (a *Agent) appendTranscript(chunk string) {
710 a.appendTranscriptChunk(chunk, "", false)
711}
712
713func (a *Agent) appendUserMessage(text string) {
714 a.appendTranscriptChunk(text, "user: ", false)
715}
716
717func (a *Agent) appendAssistantMessage(text string) {
718 if text == "" {
719 return
720 }
721 a.mu.Lock()
722 if a.streamingAssistantText != "" {
723 a.replaceStreamingAssistantLocked(text)
724 a.clearStreamingAssistantLocked()
725 } else {
726 a.appendTranscriptLinesLocked(text, "", true)
727 a.boundTranscriptLocked()
728 }
729 a.mu.Unlock()
730 a.markActivity(time.Now())
731}
732
733// appendAssistantDelta exposes an in-progress assistant response in the
734// dashboard without duplicating it when the final assistant_message arrives.
735// The JSONL event log remains the durable source; this maintains its live and
736// replayed in-memory projection.
737func (a *Agent) appendAssistantDelta(text string) {
738 if text == "" {
739 return
740 }
741 a.mu.Lock()
742 if !a.streamingAssistantTruncated {
743 value := a.streamingAssistantText + text
744 byteCap, lineCap := a.outputLimits()
745 a.streamingAssistantText = boundInlineText(value, byteCap, lineCap)
746 a.streamingAssistantTruncated = a.streamingAssistantText != value
747 }
748 a.replaceStreamingAssistantLocked(a.streamingAssistantText)
749 a.mu.Unlock()
750 a.markActivity(time.Now())
751}
752
753func (a *Agent) appendTranscriptChunk(chunk, linePrefix string, assistant bool) {
754 if strings.TrimRight(chunk, "\n") == "" {
755 return
756 }
757 a.mu.Lock()
758 a.appendTranscriptLinesLocked(chunk, linePrefix, assistant)
759 a.boundTranscriptLocked()
760 a.mu.Unlock()
761 a.markActivity(time.Now())
762}
763
764func (a *Agent) resetStreamingAssistant() {
765 a.mu.Lock()
766 a.clearStreamingAssistantLocked()
767 a.mu.Unlock()
768}
769
770func (a *Agent) replaceStreamingAssistantLocked(text string) {
771 a.removeStreamingAssistantLocked()
772 a.streamingAssistantStart = len(a.transcript)
773 a.streamingAssistantLines = a.appendTranscriptLinesLocked(text, "", true)
774 a.boundTranscriptLocked()
775}
776
777func (a *Agent) removeStreamingAssistantLocked() {
778 if a.streamingAssistantLines == 0 {
779 return
780 }
781 start := a.streamingAssistantStart
782 if start < 0 {
783 start = 0
784 }
785 if start > len(a.transcript) {
786 start = len(a.transcript)
787 }
788 end := start + a.streamingAssistantLines
789 if end > len(a.transcript) {
790 end = len(a.transcript)
791 }
792 for _, line := range a.transcript[start:end] {
793 a.outputBytes -= len(line) + 1
794 a.outputLines--
795 }
796 a.transcript = append(a.transcript[:start], a.transcript[end:]...)
797 a.streamingAssistantLines = 0
798 if a.outputLines == 0 {
799 a.outputBytes = 0
800 }
801}
802
803func (a *Agent) clearStreamingAssistantLocked() {
804 a.streamingAssistantText = ""
805 a.streamingAssistantTruncated = false
806 a.streamingAssistantStart = 0
807 a.streamingAssistantLines = 0
808}
809
810func (a *Agent) appendTranscriptLinesLocked(chunk, linePrefix string, assistant bool) int {
811 message := chunk
812 chunk = strings.TrimRight(chunk, "\n")
813 if chunk == "" {
814 return 0
815 }
816 lines := strings.Split(chunk, "\n")
817 for _, line := range lines {
818 line = linePrefix + line
819 a.transcript = append(a.transcript, line)
820 a.outputBytes += len(line) + 1
821 a.outputLines++
822 }
823 if assistant {
824 a.lastAssistant = boundInlineText(message, a.maxOutputBytes, a.maxOutputLines)
825 }
826 return len(lines)
827}
828
829func (a *Agent) outputLimits() (byteCap, lineCap int) {
830 byteCap = a.maxOutputBytes
831 if byteCap <= 0 {
832 byteCap = 500_000
833 }
834 lineCap = a.maxOutputLines
835 if lineCap <= 0 {
836 lineCap = 2_000
837 }
838 return byteCap, lineCap
839}
840
841func (a *Agent) boundTranscriptLocked() {
842 // Bound the inline dashboard copy. The durable event log and session
843 // remain available through the logical history reference.
844 byteCap, lineCap := a.outputLimits()
845 for a.outputLines > lineCap || (a.outputLines > 0 && a.outputBytes > byteCap) {
846 line := a.transcript[0]
847 a.transcript = a.transcript[1:]
848 a.outputBytes -= len(line) + 1
849 a.outputLines--
850 if a.streamingAssistantLines > 0 {
851 if a.streamingAssistantStart > 0 {
852 a.streamingAssistantStart--
853 } else {
854 a.streamingAssistantLines--
855 }
856 }
857 a.outputTruncated = true
858 }
859 if a.outputLines == 0 {
860 a.outputBytes = 0
861 }
862}
863
864// newAgentID returns a short, mostly-collision-free identifier of the
865// form "<slug>-<nano>". The slug is derived from the task text so
866// dashboards stay readable; the nano suffix guarantees uniqueness
867// even when two agents are spawned in the same millisecond.
868func newAgentID(task string, now time.Time) string {
869 slug := taskSlug(task)
870 return fmt.Sprintf("%s-%d", slug, now.UnixNano()%1_000_000)
871}
872
873func taskSlug(task string) string {
874 task = strings.ToLower(task)
875 var b strings.Builder
876 dash := false
877 for _, r := range task {
878 switch {
879 case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
880 b.WriteRune(r)
881 dash = false
882 default:
883 if !dash && b.Len() > 0 {
884 b.WriteByte('-')
885 dash = true
886 }
887 }
888 if b.Len() >= 24 {
889 break
890 }
891 }
892 out := strings.Trim(b.String(), "-")
893 if out == "" {
894 out = "agent"
895 }
896 return out
897}