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