1// Package subagents implements zut's subagent supervisor.
2//
3// A Supervisor manages headless, long-lived zut subprocesses and keeps their
4// process/turn state, event logs, sessions, results, and optional worktree
5// artifacts durable across supervisor restarts.
6//
7// Shared workspaces preserve the historical behavior, while an explicit
8// worktree request gives a child an isolated checkout whose patch is captured
9// before cleanup. The Runner abstraction lets tests inject a fake process;
10// production uses `zut --subagent-worker ...`.
11package subagents
12
13import (
14 "context"
15 "errors"
16 "fmt"
17 "os"
18 "path/filepath"
19 "sort"
20 "strings"
21 "sync"
22 "sync/atomic"
23 "time"
24
25 "github.com/google/uuid"
26)
27
28// Status is the high-level lifecycle state of an Agent.
29type Status string
30
31const (
32 StatusPending Status = "pending" // created, not started yet
33 StatusRunning Status = "running" // Runner.Run is in flight
34 StatusDone Status = "done" // Runner.Run returned nil
35 StatusFailed Status = "failed" // Runner.Run returned an error
36 StatusKilled Status = "killed" // Stop() called before completion
37 StatusDetached Status = "detached" // reloaded from disk; no live runner
38)
39
40// Config configures a Supervisor.
41type Config struct {
42 // Context is the supervisor lifetime. Child workers use it as their
43 // parent context rather than inheriting an individual tool/model-turn
44 // context. A nil value means context.Background().
45 Context context.Context
46
47 // Root is the directory under which per-agent state files live.
48 // Typically <ZutHome>/subagents, but tests pass a tempdir.
49 Root string
50
51 // Policy owns concurrency, timeout, output, and path safety limits.
52 Policy SubagentPolicy
53
54 // RepoRoot is the repository used by spawned agents. Shared-mode children
55 // run here; worktree-mode children use an isolated checkout created from
56 // this root.
57 RepoRoot string
58
59 // Provider is the active host provider. It is used only to decide whether
60 // an explicit child provider may inherit the host endpoint settings.
61 Provider string
62
63 // FastMode is copied to every newly spawned child and persisted so
64 // child subprocesses use the same OpenAI service-tier setting.
65 FastMode bool
66
67 // WebSearchPolicy is the already-resolved capability of the parent
68 // session. Generic children may inherit it only when the subagent policy
69 // also permits web_search; named profiles resolve their own explicit
70 // tools opt-in at SpawnReq.
71 WebSearchPolicy WebSearchPolicy
72
73 // BaseURL and InsecureTLS are the effective provider connection
74 // settings inherited by newly spawned children. They are copied to
75 // each Agent and persisted so a reload/resume does not silently fall
76 // back to the child's independent provider configuration.
77 BaseURL string
78 InsecureTLS bool
79
80 // NewRunner produces the Runner for an Agent. If nil, the default
81 // `zut --subagent-worker ...` exec runner is used. Tests inject a fake
82 // here.
83 NewRunner func(a *Agent) Runner
84
85 // ResolveCredential resolves the credential inherited by a child
86 // process. The default runner transfers it over the child's stdin,
87 // keeping command-backed credentials out of argv and the environment.
88 ResolveCredential func(ctx context.Context, provider string) (Credential, error)
89
90 // TraceDir enables local execution tracing. When non-empty, the supervisor
91 // creates one private bundle beneath this directory and is its sole writer.
92 // An empty directory leaves tracing disabled.
93 TraceDir string
94 TraceMode TraceMode
95
96 // Now is a clock seam for tests; defaults to time.Now.
97 Now func() time.Time
98}
99
100// Credential is the resolved authentication state inherited by a subagent child.
101type Credential struct {
102 Value string `json:"value"`
103 Method string `json:"method"`
104 AccountID string `json:"account_id,omitempty"`
105}
106
107// Runner executes one agent task. Run blocks until the task finishes,
108// is cancelled via ctx, or hits an unrecoverable error.
109//
110// Run should report progress by writing short human-readable strings
111// to the activity channel and final transcript text to transcript.
112// Both channels are non-blocking sinks owned by the Supervisor; if the
113// dashboard isn't reading, sends are dropped.
114type Runner interface {
115 Run(ctx context.Context, sink Sink) error
116}
117
118// Sink is how a Runner reports activity and transcript back to the
119// supervisor. All methods are safe to call from any goroutine and
120// never block.
121type Sink interface {
122 // Activity sets the one-line "what is this agent doing right now"
123 // string shown in the dashboard.
124 Activity(msg string)
125 // Transcript appends a chunk of agent output (typically a final
126 // assistant message) to the agent's running transcript.
127 Transcript(chunk string)
128}
129
130// Supervisor supervises a set of Agents.
131type Supervisor struct {
132 cfg Config
133
134 mu sync.Mutex
135 operationMu sync.Mutex
136 lifetimeCtx context.Context
137 lifetimeCancel context.CancelFunc
138 agents map[string]*Agent
139 order []string // creation order for stable listing
140 queue []*Agent
141 active int
142 activeByParent map[string]int
143 activeByBatch map[string]int
144 batches map[string]*Batch
145 trace *TraceWriter
146
147 // activeSession is the host session id the dashboard is
148 // currently scoped to. When non-empty, SnapshotAll filters out
149 // agents whose SessionID doesn't match (unscoped
150 // agents with SessionID == "" are always shown). Spawn stamps
151 // new agents with this value so they appear only in the session
152 // that created them. When empty (the default), the historical
153 // "show everything" behaviour is preserved — important for
154 // tests and any scripted use of the Supervisor that doesn't bother
155 // with sessions.
156 activeSession string
157}
158
159// New constructs a Supervisor from cfg. Missing config fields are filled
160// with defaults.
161func New(cfg Config) *Supervisor {
162 if cfg.Now == nil {
163 cfg.Now = time.Now
164 }
165 if cfg.NewRunner == nil {
166 cfg.NewRunner = func(a *Agent) Runner {
167 return &execRunner{
168 agent: a,
169 resolveCredential: cfg.ResolveCredential,
170 GracePeriod: cfg.Policy.CancelGracePeriod,
171 }
172 }
173 }
174 cfg.Policy.normalize()
175 trace := NewMemoryTraceWriter()
176 if cfg.TraceDir != "" {
177 // Tracing is observational only: inability to create its optional local
178 // bundle must not prevent the supervisor from running workers. The
179 // in-memory trace remains the sole status projection in that case.
180 if bundle, err := NewTraceWriter(filepath.Join(cfg.TraceDir, uuid.NewString()), cfg.TraceMode); err == nil {
181 trace = bundle
182 } else {
183 trace.Record(TraceEvent{Type: "trace.bundle.failed", Data: map[string]any{"error_kind": "create"}})
184 }
185 }
186 lifetimeParent := cfg.Context
187 if lifetimeParent == nil {
188 lifetimeParent = context.Background()
189 }
190 lifetimeCtx, lifetimeCancel := context.WithCancel(lifetimeParent)
191 return &Supervisor{
192 cfg: cfg,
193 lifetimeCtx: lifetimeCtx,
194 lifetimeCancel: lifetimeCancel,
195 agents: map[string]*Agent{},
196 activeByParent: map[string]int{},
197 activeByBatch: map[string]int{},
198 batches: map[string]*Batch{},
199 trace: trace,
200 }
201}
202
203// workerContext derives a child worker context from the supervisor lifetime.
204// Turn timeouts are enforced by the worker while a delegated turn is active;
205// the process remains cancellable while idle so it can accept follow-ups.
206func (f *Supervisor) workerContext() (context.Context, context.CancelFunc) {
207 base := f.lifetimeCtx
208 if base == nil {
209 base = context.Background()
210 }
211 return context.WithCancel(base)
212}
213
214// SetRepoRoot updates the shared working directory for subsequently
215// spawned agents. Existing agents keep the directory they were started
216// with; callers use this when the host session changes cwd.
217func (f *Supervisor) SetRepoRoot(root string) {
218 f.mu.Lock()
219 defer f.mu.Unlock()
220 f.cfg.RepoRoot = root
221}
222
223// RepoRoot returns the working directory used by subsequently spawned
224// agents.
225func (f *Supervisor) RepoRoot() string {
226 f.mu.Lock()
227 defer f.mu.Unlock()
228 return f.cfg.RepoRoot
229}
230
231// SetFastMode updates the fast-mode setting for subsequently spawned
232// children. Existing agents keep the setting captured at their spawn
233// boundary so resumes remain deterministic.
234func (f *Supervisor) SetFastMode(enabled bool) {
235 f.mu.Lock()
236 f.cfg.FastMode = enabled
237 f.mu.Unlock()
238}
239
240// FastMode reports the setting used for subsequently spawned children.
241func (f *Supervisor) FastMode() bool {
242 f.mu.Lock()
243 defer f.mu.Unlock()
244 return f.cfg.FastMode
245}
246
247// SetWebSearchPolicy updates the parent capability used for subsequently
248// spawned generic children. Existing agents retain their persisted decision.
249func (f *Supervisor) SetWebSearchPolicy(policy WebSearchPolicy) {
250 f.mu.Lock()
251 f.cfg.WebSearchPolicy = policy
252 f.mu.Unlock()
253}
254
255// WebSearchPolicy reports the parent capability used for new generic
256// children.
257func (f *Supervisor) WebSearchPolicy() WebSearchPolicy {
258 f.mu.Lock()
259 defer f.mu.Unlock()
260 return f.cfg.WebSearchPolicy
261}
262
263// SetProvider updates the provider used by subsequently spawned children.
264// Existing agents keep their persisted provider selection.
265func (f *Supervisor) SetProvider(provider string) {
266 f.mu.Lock()
267 f.cfg.Provider = strings.TrimSpace(provider)
268 f.mu.Unlock()
269}
270
271// SetProviderSettings updates the effective connection settings used by
272// subsequently spawned children. Existing agents keep their persisted values
273// so a model/provider switch cannot silently change a running or resumed
274// worker's endpoint.
275func (f *Supervisor) SetProviderSettings(baseURL string, insecureTLS bool) {
276 f.mu.Lock()
277 f.cfg.BaseURL = strings.TrimSpace(baseURL)
278 f.cfg.InsecureTLS = insecureTLS
279 f.mu.Unlock()
280}
281
282// ProviderSettings returns the settings used by subsequently spawned
283// children. It is primarily useful to host integrations that need to verify
284// the manager follows the active provider selection.
285func (f *Supervisor) ProviderSettings() (string, bool) {
286 f.mu.Lock()
287 defer f.mu.Unlock()
288 return f.cfg.BaseURL, f.cfg.InsecureTLS
289}
290
291// SetActiveSession scopes the dashboard view (and Spawn stamping)
292// to a particular host zut session id. Pass empty to clear the
293// scope and revert to "show every agent" (the original behaviour).
294//
295// Existing in-memory agents keep their SessionID; only the filter
296// applied at snapshot time changes. So swapping the active session
297// with /sessions instantly re-narrows the dashboard without
298// touching any agent state.
299func (f *Supervisor) SetActiveSession(id string) {
300 f.mu.Lock()
301 defer f.mu.Unlock()
302 f.activeSession = id
303}
304
305// ActiveSession returns the current scope, mostly for tests and
306// diagnostics. Empty means "no scope; show everything".
307func (f *Supervisor) ActiveSession() string {
308 f.mu.Lock()
309 defer f.mu.Unlock()
310 return f.activeSession
311}
312
313// agentStateDir is the per-agent state directory laid out as:
314//
315// <root>/agents/<id>/
316// events.jsonl durable event log (runner-owned)
317// session.json persistent agent session (child-owned)
318// meta.json static metadata (id, task)
319//
320// The transient unix socket inbox lives in a probed runtime directory,
321// which may be on a different filesystem from this durable state.
322func (f *Supervisor) agentStateDir(id string) string {
323 return filepath.Join(f.cfg.Root, "agents", id)
324}
325
326func (f *Supervisor) uniqueAgentIDLocked(base string) string {
327 id := base
328 for suffix := 2; ; suffix++ {
329 _, inMemory := f.agents[id]
330 _, onDiskErr := os.Stat(f.agentStateDir(id))
331 if !inMemory && onDiskErr != nil {
332 return id
333 }
334 id = fmt.Sprintf("%s-%d", base, suffix)
335 }
336}
337
338// SpawnRequest configures a Spawn. Only Task is required; the rest
339// are optional. Model + Provider, when set, get baked into the
340// child argv as --model / --provider so the agent runs against the
341// chosen model regardless of the parent's current selection. A named
342// Subagent is passed to the child as --subagent, where it resolves the
343// markdown profile and applies its instructions. Reasoning is passed
344// as --reasoning when set. FastMode overrides the host setting when a
345// selected profile declares a fastMode preference.
346type SpawnRequest struct {
347 Task string
348 Model string // optional override; child resolves default if empty
349 Provider string // optional override; usually paired with Model
350 BaseURL string // optional provider endpoint override
351 // InsecureTLS enables the child to skip TLS verification for its
352 // effective custom provider endpoint.
353 InsecureTLS bool
354 Reasoning string // optional reasoning/thinking level override
355 // FastMode is an optional profile override. A non-nil value enables or
356 // disables fast mode for this child; nil inherits Config.FastMode.
357 FastMode *bool
358 Subagent string // optional named markdown profile
359
360 // WebSearchPolicy is an internal child capability override. Generic
361 // requests normally inherit the supervisor's parent decision; named
362 // requests use Allow only when their profile explicitly lists web_search.
363 WebSearchPolicy WebSearchPolicy
364
365 // ParentID is metadata used for per-parent scheduling limits. A
366 // non-empty RequesterAgentID identifies an untrusted child request;
367 // recursive spawning is rejected in v1.
368 ParentID string
369 BatchID string
370 RootSessionID string
371 RequesterAgentID string
372
373 // Required keeps this delegated turn on the parent's completion path. The
374 // supervisor persists its outcome while the manager remains asynchronous.
375 Required bool
376 Timeout time.Duration
377 Tools []string
378 WorkspaceMode WorkspaceMode
379 WorkspaceBase string
380 WorkspaceCapture CaptureMode
381}
382
383// Spawn creates a new Agent for the given task, allocates its on-disk
384// state directory (events log, inbox socket path, session file path), and
385// queues it for the manager-owned scheduler. The returned Agent is pending
386// when the concurrency budget is full and transitions to running when a slot
387// opens. This is the historical signature; callers that want to override the
388// child's model or workspace use SpawnReq instead.
389func (f *Supervisor) Spawn(ctx context.Context, task string) (*Agent, error) {
390 return f.SpawnReq(ctx, SpawnRequest{Task: task})
391}
392
393// unqualifiedModelID removes a redundant provider prefix from a model ID.
394// Model IDs themselves may contain slashes, so only the explicitly selected
395// provider's exact prefix is removed.
396func unqualifiedModelID(provider, model string) string {
397 provider = strings.TrimSpace(provider)
398 model = strings.TrimSpace(model)
399 if provider == "" {
400 return model
401 }
402 return strings.TrimPrefix(model, provider+"/")
403}
404
405// SpawnReq is the full-fat variant of Spawn that accepts a SpawnRequest.
406// Existing callers can keep using Spawn; new code that wants to pin the
407// child's model, policy, or workspace uses this.
408func (f *Supervisor) SpawnReq(ctx context.Context, req SpawnRequest) (*Agent, error) {
409 task := strings.TrimSpace(req.Task)
410 if task == "" {
411 return nil, errors.New("subagents: empty task")
412 }
413 if req.RequesterAgentID != "" {
414 return nil, errors.New("subagents: child-originated spawning is disabled")
415 }
416 if err := f.validateSpawnRequest(req); err != nil {
417 return nil, err
418 }
419 req.Model = unqualifiedModelID(req.Provider, req.Model)
420 if ctx == nil {
421 ctx = context.Background()
422 }
423
424 now := f.cfg.Now()
425 baseID := newAgentID(task, now)
426
427 f.mu.Lock()
428 id := f.uniqueAgentIDLocked(baseID)
429 dir := f.cfg.RepoRoot
430 fastMode := f.cfg.FastMode
431 configProvider := f.cfg.Provider
432 configBaseURL := f.cfg.BaseURL
433 configInsecureTLS := f.cfg.InsecureTLS
434 sessionID := f.activeSession
435 f.mu.Unlock()
436 fastModeOverridesHost := false
437 if req.FastMode != nil {
438 fastModeOverridesHost = !fastMode && *req.FastMode
439 fastMode = *req.FastMode
440 }
441 childProvider := strings.TrimSpace(req.Provider)
442 inheritProviderSettings := childProvider == "" || configProvider == "" || strings.EqualFold(childProvider, strings.TrimSpace(configProvider))
443 baseURL := strings.TrimSpace(req.BaseURL)
444 if baseURL == "" && inheritProviderSettings {
445 baseURL = strings.TrimSpace(configBaseURL)
446 }
447 insecureTLS := req.InsecureTLS || (inheritProviderSettings && configInsecureTLS)
448
449 stateDir := f.agentStateDir(id)
450 if err := os.MkdirAll(stateDir, 0o700); err != nil {
451 return nil, fmt.Errorf("subagents state dir: %w", err)
452 }
453 lease, err := acquireAgentLease(stateDir)
454 if err != nil {
455 return nil, fmt.Errorf("subagents agent lease: %w", err)
456 }
457 leaseOwned := true
458 defer func() {
459 if leaseOwned {
460 _ = lease.Close()
461 }
462 }()
463 logPath := filepath.Join(stateDir, "events.jsonl")
464 sessionPath := filepath.Join(stateDir, "session.json")
465 // Keep the transient unix socket outside the durable state root.
466 // ZUT_HOME may live on a shared or network filesystem that supports
467 // regular state files but cannot host unix socket nodes.
468 inboxPath, err := inboxSocketPath(f.cfg.Root, id)
469 if err != nil {
470 _ = lease.Close()
471 leaseOwned = false
472 _ = os.RemoveAll(stateDir)
473 return nil, fmt.Errorf("subagents inbox path: %w", err)
474 }
475
476 workspaceMode := req.WorkspaceMode
477 if workspaceMode == "" {
478 workspaceMode = WorkspaceShared
479 }
480 workspace, err := PrepareWorkspace(ctx, WorkspaceRequest{
481 Mode: workspaceMode,
482 RepositoryRoot: dir,
483 StateDir: stateDir,
484 AgentID: id,
485 Base: req.WorkspaceBase,
486 Capture: req.WorkspaceCapture,
487 AllowedRoots: append([]string(nil), f.cfg.Policy.AllowedRoots...),
488 })
489 if err != nil {
490 _ = lease.Close()
491 leaseOwned = false
492 _ = os.RemoveAll(stateDir)
493 return nil, err
494 }
495 dir = workspace.Dir()
496 timeout := req.Timeout
497 if timeout <= 0 {
498 timeout = f.cfg.Policy.DefaultTimeout
499 }
500 runCtx, cancel := f.workerContext()
501 tools := append([]string(nil), req.Tools...)
502 if len(tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 {
503 tools = append([]string(nil), f.cfg.Policy.AllowedTools...)
504 }
505 webSearchPolicy := f.resolveWebSearchPolicy(req)
506
507 a := &Agent{
508 ID: id,
509 Task: task,
510 OriginalTask: task,
511 Dir: dir,
512 RepositoryRoot: workspace.RepositoryRoot(),
513 WorkspacePath: workspace.Dir(),
514 Started: now,
515 ParentID: strings.TrimSpace(req.ParentID),
516 BatchID: strings.TrimSpace(req.BatchID),
517 RootSessionID: strings.TrimSpace(req.RootSessionID),
518 Model: strings.TrimSpace(req.Model),
519 Provider: strings.TrimSpace(req.Provider),
520 BaseURL: baseURL,
521 InsecureTLS: insecureTLS,
522 Reasoning: strings.TrimSpace(req.Reasoning),
523 FastMode: fastMode,
524 Subagent: strings.TrimSpace(req.Subagent),
525 SessionID: sessionID,
526 WorkspaceMode: workspaceMode,
527 WorkspaceBase: strings.TrimSpace(req.WorkspaceBase),
528 WorkspaceCapture: req.WorkspaceCapture,
529 Timeout: timeout,
530 HeartbeatInterval: f.cfg.Policy.HeartbeatInterval,
531 Tools: tools,
532 WebSearchPolicy: webSearchPolicy,
533 InboxPath: inboxPath,
534 EventLogPath: logPath,
535 SessionPath: sessionPath,
536 inbox: NewInbox(inboxPath),
537 status: StatusPending,
538 activity: "queued",
539 transcriptLoaded: true,
540 processState: ProcessPending,
541 turnState: TurnQueued,
542 updatedAt: now,
543 lastActivity: now,
544 requirement: newRequirement(req.Required, 1, now),
545 stateDir: stateDir,
546 lease: lease,
547 maxOutputBytes: f.cfg.Policy.MaxOutputBytes,
548 maxOutputLines: f.cfg.Policy.MaxOutputLines,
549 done: make(chan struct{}),
550 }
551 a.fastModeOverridesHost = fastModeOverridesHost
552 if a.RootSessionID == "" {
553 a.RootSessionID = sessionID
554 }
555 a.ctx, a.cancel = runCtx, cancel
556 a.persistFn = f.persistAgent
557 a.setOnTurnIdle(func() { f.dispatchQueuedResumeWithTimeout(a) })
558 a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
559 a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
560 a.trace = f.trace
561 a.runner = f.cfg.NewRunner(a)
562 if err := writeAgentMeta(stateDir, a); err != nil {
563 if a.cancel != nil {
564 a.cancel()
565 }
566 _ = workspace.Cleanup(context.Background())
567 _ = lease.Close()
568 leaseOwned = false
569 _ = os.RemoveAll(stateDir)
570 return nil, fmt.Errorf("subagents initial metadata: %w", err)
571 }
572
573 f.mu.Lock()
574 f.agents[id] = a
575 f.order = append(f.order, id)
576 f.queue = append(f.queue, a)
577 f.mu.Unlock()
578 leaseOwned = false
579 f.recordTrace(TraceEvent{Type: "agent.spawned", AgentID: a.ID, Data: map[string]any{
580 "task_length": len(task), "workspace_mode": string(workspaceMode),
581 }})
582
583 // Persisted metadata was written before queue admission, so a later
584 // supervisor can reconstruct work that never reached a process.
585 f.armQueueTimeout(a)
586 f.schedule()
587 return a, nil
588}
589
590// SendUserTurn is sugar for the common "send the next user turn"
591// case. It quotes nothing and forwards verbatim; callers are
592// expected to have already trimmed and expanded the text.
593// CancelTurn requests cancellation of the current turn while keeping the
594// worker process alive for a later follow-up.
595func traceErrorKind(err error) string {
596 if err == nil {
597 return ""
598 }
599 switch {
600 case errors.Is(err, context.Canceled):
601 return "canceled"
602 case errors.Is(err, context.DeadlineExceeded):
603 return "deadline_exceeded"
604 default:
605 return "error"
606 }
607}
608
609func (f *Supervisor) recordTrace(event TraceEvent) {
610 if f == nil || f.trace == nil {
611 return
612 }
613 f.trace.Record(event)
614}
615
616// TraceDir returns the opt-in bundle directory, or an empty string when tracing
617// is disabled.
618func (f *Supervisor) TraceDir() string {
619 if f == nil || f.trace == nil {
620 return ""
621 }
622 return f.trace.Dir()
623}
624
625// TraceViews projects the supervisor's factual trace into the diagnostic state
626// for each observed agent. It intentionally does not expose lifecycle labels.
627func (f *Supervisor) TraceViews() map[string]AgentTraceView {
628 if f == nil || f.trace == nil {
629 return map[string]AgentTraceView{}
630 }
631 return f.trace.Views()
632}
633
634// Close flushes and closes the optional trace bundle. It does not stop agents;
635// their lifetime remains controlled by the supervisor context and Stop.
636func (f *Supervisor) Close() error { return f.CloseContext(context.Background()) }
637
638// CloseContext flushes and closes the optional trace bundle within ctx.
639func (f *Supervisor) CloseContext(ctx context.Context) error {
640 if f == nil || f.trace == nil {
641 return nil
642 }
643 return f.trace.CloseContext(ctx)
644}
645
646func (f *Supervisor) CancelTurn(id string) error {
647 a := f.Get(id)
648 if a == nil {
649 return fmt.Errorf("subagents: no such agent %q", id)
650 }
651 if a.inbox == nil {
652 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
653 }
654 a.setTurnState(TurnCanceling, a.CurrentTurnID())
655 f.recordTrace(TraceEvent{Type: "turn.cancel.requested", AgentID: a.ID, TurnID: a.CurrentTurnID()})
656 return a.inbox.SendCommand(NewCommand(CommandTurnCancel, a.ID, a.CurrentTurnID(), TurnCancelPayload{Reason: "user"}))
657}
658
659func (f *Supervisor) SendUserTurn(id, text string) error {
660 a := f.Get(id)
661 if a == nil {
662 return fmt.Errorf("subagents: no such agent %q", id)
663 }
664 if a.inbox == nil {
665 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
666 }
667 f.recordTrace(TraceEvent{Type: "turn.requested", AgentID: a.ID, Data: map[string]any{"prompt_length": len(text)}})
668 return a.inbox.SendCommand(NewCommand(CommandTurnStart, a.ID, a.CurrentTurnID(), TurnStartPayload{Prompt: text}))
669}
670
671func (f *Supervisor) run(a *Agent) {
672 a.mu.Lock()
673 killedBeforeStart := a.status == StatusKilled
674 ctxErr := error(nil)
675 if a.ctx != nil {
676 ctxErr = a.ctx.Err()
677 }
678 a.mu.Unlock()
679 var err error
680 if killedBeforeStart || ctxErr != nil || !atomic.CompareAndSwapInt32(&a.launchState, 0, 1) {
681 if ctxErr != nil {
682 err = ctxErr
683 } else {
684 err = context.Canceled
685 }
686 } else {
687 f.recordTrace(TraceEvent{Type: "agent.started", AgentID: a.ID})
688 a.setActivity("starting")
689 if a.runner == nil {
690 err = errors.New("subagents: agent has no runner")
691 } else {
692 err = a.runner.Run(a.ctx, agentSink{a: a})
693 }
694 }
695
696 a.mu.Lock()
697 a.finished = f.cfg.Now()
698 startupErr := a.startupErr
699 terminalErr := err
700 switch {
701 case a.status == StatusKilled:
702 // Stop requested a terminal state; preserve it.
703 case startupErr != nil:
704 a.status = StatusFailed
705 a.activity = "startup timeout: worker did not become ready"
706 a.lastErr = startupErr
707 terminalErr = startupErr
708 case errors.Is(err, context.Canceled):
709 a.status = StatusKilled
710 a.activity = "cancelled"
711 a.lastErr = err
712 case errors.Is(err, context.DeadlineExceeded):
713 a.status = StatusFailed
714 a.activity = "timeout"
715 a.lastErr = err
716 case err != nil:
717 a.status = StatusFailed
718 a.activity = "error: " + truncate(err.Error(), 120)
719 a.lastErr = err
720 default:
721 a.status = StatusDone
722 a.activity = "done"
723 }
724 finalStatus := a.status
725 a.mu.Unlock()
726 terminalType := "agent.finished"
727 switch {
728 case finalStatus == StatusKilled:
729 terminalType = "agent.cancelled"
730 case terminalErr != nil:
731 terminalType = "agent.failed"
732 }
733 f.recordTrace(TraceEvent{Type: terminalType, AgentID: a.ID, Data: map[string]any{"status": string(finalStatus), "error": traceErrorKind(terminalErr)}})
734
735 switch finalStatus {
736 case StatusKilled:
737 a.setProcessState(ProcessKilled)
738 a.setTurnState(TurnCanceled, a.CurrentTurnID())
739 case StatusFailed:
740 a.setProcessState(ProcessExited)
741 a.setTurnState(TurnFailed, a.CurrentTurnID())
742 default:
743 a.setProcessState(ProcessExited)
744 if a.TurnState() == TurnRunning || a.TurnState() == TurnIdle {
745 a.setTurnState(TurnSucceeded, a.CurrentTurnID())
746 }
747 }
748
749 // The process slot is no longer occupied once Runner.Run has returned.
750 // Release it before potentially slow artifact capture and workspace
751 // cleanup so queued work can start without waiting on git/filesystem I/O.
752 f.releaseCapacity(a)
753
754 // Capture isolated output and write the structured result before the
755 // temporary workspace is removed. This ordering is the recovery
756 // invariant: a user can always inspect a durable result after failure.
757 f.captureWorkspace(a)
758 f.ensureResult(a, finalStatus, terminalErr)
759 if a.cancel != nil {
760 a.cancel()
761 }
762 if a.inbox != nil {
763 _ = a.inbox.Close()
764 }
765 a.setProcessPID(0)
766 f.persistAgent(a)
767 if a.workspaceCleanup != nil {
768 _ = a.workspaceCleanup()
769 }
770 _ = a.releaseLease()
771 a.closeDone()
772}
773
774// List returns a snapshot of every agent in creation order. The
775// returned slice is a copy; callers may iterate without holding the
776// subagent lock. Agent fields are read under their own mutex during
777// formatting in Snapshot.
778func (f *Supervisor) List() []*Agent {
779 f.mu.Lock()
780 defer f.mu.Unlock()
781 out := make([]*Agent, 0, len(f.order))
782 for _, id := range f.order {
783 out = append(out, f.agents[id])
784 }
785 return out
786}
787
788// Get returns the agent with the given (possibly truncated) id, or
789// nil. Matching is prefix-based so the user can type the first few
790// characters of a long id.
791func (f *Supervisor) Get(id string) *Agent {
792 f.mu.Lock()
793 defer f.mu.Unlock()
794 if a, ok := f.agents[id]; ok {
795 return a
796 }
797 // Prefix match.
798 var hits []*Agent
799 for k, a := range f.agents {
800 if strings.HasPrefix(k, id) {
801 hits = append(hits, a)
802 }
803 }
804 if len(hits) == 1 {
805 return hits[0]
806 }
807 return nil
808}
809
810var errDetachedStopTimeout = errors.New("detached worker stop timeout")
811
812const detachedStopPollInterval = 20 * time.Millisecond
813
814// waitForDetachedWorker waits for the worker inbox to disappear, returning a
815// sentinel when the grace period expires. Polling is interruptible so a
816// caller's context is not made to wait for the next fixed sleep interval.
817func waitForDetachedWorker(ctx context.Context, path string, timeout time.Duration) error {
818 if ctx == nil {
819 ctx = context.Background()
820 }
821 timer := time.NewTimer(timeout)
822 defer timer.Stop()
823 ticker := time.NewTicker(detachedStopPollInterval)
824 defer ticker.Stop()
825 for {
826 if err := ctx.Err(); err != nil {
827 return err
828 }
829 if !inboxLive(path) {
830 return nil
831 }
832 select {
833 case <-ctx.Done():
834 return ctx.Err()
835 case <-timer.C:
836 if !inboxLive(path) {
837 return nil
838 }
839 return errDetachedStopTimeout
840 case <-ticker.C:
841 }
842 }
843}
844
845// Stop cancels the agent's context. The Runner should observe the
846// cancellation and return promptly; the goroutine then finalises the
847// agent in StatusKilled. Also closes the supervisor-side inbox handle
848// so any pending send retries fail fast instead of dialing a
849// socket about to be unlinked. Stop uses a background context; callers
850// that need cancellation while waiting for a detached worker should use
851// StopContext.
852//
853// Stop is a no-op for any agent that's not in a live runnable state
854// — Done / Failed / Killed (already finalised). Reloaded Detached agents
855// are probed first: if their worker survived the supervisor, Stop sends the
856// shutdown command through the durable inbox instead of pretending no
857// process exists.
858func (f *Supervisor) Stop(id string) error {
859 return f.stop(context.Background(), id, ShutdownOriginTargeted)
860}
861
862// StopContext is Stop with a context that can interrupt detached-worker
863// shutdown and its graceful and force-stop waits. The context is not held
864// under the supervisor operation lock while those waits run.
865func (f *Supervisor) StopContext(ctx context.Context, id string) error {
866 return f.stop(ctx, id, ShutdownOriginTargeted)
867}
868
869func (f *Supervisor) stop(ctx context.Context, id string, origin ShutdownOrigin) error {
870 if ctx == nil {
871 ctx = context.Background()
872 }
873 f.operationMu.Lock()
874 a := f.Get(id)
875 if a == nil {
876 f.operationMu.Unlock()
877 return fmt.Errorf("no such agent %q", id)
878 }
879
880 // Coordinate the status transition with schedule. Holding f.mu while
881 // checking and changing a.status closes the small window where Stop could
882 // observe pending, schedule could promote the same agent to running, and
883 // the pending cleanup below could then finalize an agent that was already
884 // admitted to the runner.
885 f.mu.Lock()
886 status, shutdownStarted := a.prepareShutdown(origin)
887 if !shutdownStarted {
888 f.mu.Unlock()
889 f.operationMu.Unlock()
890 return nil
891 }
892 if status == StatusDetached {
893 path := a.InboxPath
894 inbox := a.inbox
895 pid := a.ProcessPIDValue()
896 f.mu.Unlock()
897 // The detached worker may take the full grace period to exit. Do not
898 // serialize unrelated supervisor operations for that entire wait.
899 f.operationMu.Unlock()
900 return f.stopDetached(ctx, a, path, inbox, pid, origin)
901 }
902 atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
903 if status == StatusPending {
904 for i, queued := range f.queue {
905 if queued == a {
906 f.queue = append(f.queue[:i], f.queue[i+1:]...)
907 break
908 }
909 }
910 }
911 f.mu.Unlock()
912 f.operationMu.Unlock()
913 a.setTurnState(TurnCanceling, a.CurrentTurnID())
914
915 if status == StatusPending {
916 a.setProcessState(ProcessKilled)
917 a.setTurnState(TurnCanceled, a.CurrentTurnID())
918 f.captureWorkspace(a)
919 f.ensureResult(a, StatusKilled, context.Canceled)
920 if a.cancel != nil {
921 a.cancel()
922 }
923 f.persistAgent(a)
924 if a.workspaceCleanup != nil {
925 _ = a.workspaceCleanup()
926 }
927 _ = a.releaseLease()
928 a.closeDone()
929 f.schedule()
930 if err := ctx.Err(); err != nil {
931 return err
932 }
933 return nil
934 }
935
936 // Ask a live worker to shut down cleanly first. If the socket is not
937 // ready, fall back immediately; a forceful context cancellation is
938 // safer than leaving a supervisor entry running forever.
939 if err := ctx.Err(); err != nil {
940 if a.cancel != nil {
941 a.cancel()
942 }
943 return err
944 }
945 graceful := false
946 if a.inbox != nil {
947 err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()}))
948 if err == nil {
949 graceful = true
950 }
951 }
952 if !graceful && a.cancel != nil {
953 a.cancel()
954 }
955 if graceful {
956 grace := f.cfg.Policy.CancelGracePeriod
957 if grace <= 0 {
958 grace = 10 * time.Second
959 }
960 go func() {
961 timer := time.NewTimer(grace)
962 defer timer.Stop()
963 select {
964 case <-a.done:
965 case <-timer.C:
966 if a.cancel != nil {
967 a.cancel()
968 }
969 if a.inbox != nil {
970 _ = a.inbox.Close()
971 }
972 }
973 }()
974 }
975 if err := ctx.Err(); err != nil {
976 if a.cancel != nil {
977 a.cancel()
978 }
979 if a.inbox != nil {
980 _ = a.inbox.Close()
981 }
982 return err
983 }
984 return nil
985}
986
987func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int, origin ShutdownOrigin) error {
988 if inbox == nil {
989 return nil
990 }
991 if err := ctx.Err(); err != nil {
992 return err
993 }
994 if !inboxLive(path) {
995 return nil
996 }
997 if err := ctx.Err(); err != nil {
998 return err
999 }
1000 if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()})); err != nil {
1001 if ctxErr := ctx.Err(); ctxErr != nil {
1002 return ctxErr
1003 }
1004 return err
1005 }
1006 a.setActivity("shutdown requested")
1007 grace := f.cfg.Policy.CancelGracePeriod
1008 if grace <= 0 {
1009 grace = 10 * time.Second
1010 }
1011 waitErr := waitForDetachedWorker(ctx, path, grace)
1012 if waitErr != nil && !errors.Is(waitErr, errDetachedStopTimeout) {
1013 return waitErr
1014 }
1015 if waitErr == nil {
1016 f.markDetachedKilled(a)
1017 return nil
1018 }
1019 if err := ctx.Err(); err != nil {
1020 return err
1021 }
1022 if pid <= 0 {
1023 return fmt.Errorf("agent %s live worker did not stop within %s and has no recorded pid", a.ID, grace)
1024 }
1025 if killErr := forceKillProcess(pid); killErr != nil {
1026 return fmt.Errorf("agent %s force-stop: %w", a.ID, killErr)
1027 }
1028 if waitErr := waitForDetachedWorker(ctx, path, time.Second); waitErr != nil {
1029 if errors.Is(waitErr, errDetachedStopTimeout) {
1030 return fmt.Errorf("agent %s live worker did not stop after force-stop", a.ID)
1031 }
1032 return waitErr
1033 }
1034 f.markDetachedKilled(a)
1035 return nil
1036}
1037
1038func (f *Supervisor) markDetachedKilled(a *Agent) {
1039 f.operationMu.Lock()
1040 defer f.operationMu.Unlock()
1041 a.mu.Lock()
1042 a.status = StatusKilled
1043 a.activity = "killed"
1044 a.mu.Unlock()
1045 a.setProcessState(ProcessKilled)
1046 a.setTurnState(TurnCanceled, a.CurrentTurnID())
1047 a.setProcessPID(0)
1048 f.ensureResult(a, StatusKilled, context.Canceled)
1049 f.persistAgent(a)
1050 a.closeDone()
1051}
1052
1053// StopByRootSession stops and waits for every agent owned by rootSessionID.
1054// It leaves the supervisor lifetime and agents from other sessions running.
1055func (f *Supervisor) StopByRootSession(rootSessionID string) error {
1056 return f.StopByRootSessionContext(context.Background(), rootSessionID)
1057}
1058
1059// StopByRootSessionContext is the context-aware form of StopByRootSession.
1060func (f *Supervisor) StopByRootSessionContext(ctx context.Context, rootSessionID string) error {
1061 if ctx == nil {
1062 ctx = context.Background()
1063 }
1064 rootSessionID = strings.TrimSpace(rootSessionID)
1065 if rootSessionID == "" {
1066 return errors.New("subagents: root session id is required")
1067 }
1068 agents := f.List()
1069 owned := make([]*Agent, 0, len(agents))
1070 for _, a := range agents {
1071 if a.RootSessionID == rootSessionID {
1072 owned = append(owned, a)
1073 }
1074 }
1075 return f.stopAgents(ctx, owned, ShutdownOriginSession)
1076}
1077
1078func (f *Supervisor) stopAgents(ctx context.Context, agents []*Agent, origin ShutdownOrigin) error {
1079 var firstErr error
1080 for _, a := range agents {
1081 if err := f.stop(ctx, a.ID, origin); err != nil && firstErr == nil {
1082 firstErr = err
1083 }
1084 }
1085 for _, a := range agents {
1086 select {
1087 case <-a.done:
1088 case <-ctx.Done():
1089 if firstErr == nil {
1090 firstErr = ctx.Err()
1091 }
1092 return firstErr
1093 }
1094 }
1095 return firstErr
1096}
1097
1098// StopAll cancels every running agent. Used on shutdown.
1099func (f *Supervisor) StopAll() {
1100 _ = f.stopAll(context.Background())
1101 _ = f.Close()
1102}
1103
1104// StopAllContext is the context-aware shutdown variant. It returns the
1105// context error when cancellation interrupts a detached-worker wait.
1106func (f *Supervisor) StopAllContext(ctx context.Context) error {
1107 return f.stopAll(ctx)
1108}
1109
1110func (f *Supervisor) stopAll(ctx context.Context) error {
1111 if ctx == nil {
1112 ctx = context.Background()
1113 }
1114 defer func() {
1115 if f.lifetimeCancel != nil {
1116 f.lifetimeCancel()
1117 }
1118 }()
1119 return f.stopAgents(ctx, f.List(), ShutdownOriginProcess)
1120}
1121
1122// Remove tears down the per-agent state for a terminated agent. It
1123// is an error to remove an agent that's still running; call Stop
1124// first and wait for the status to settle. Detached agents
1125// (reloaded from disk) remove cleanly because they have no live
1126// runner racing for the same files.
1127//
1128// Shared-workspace agents never touch source files during Remove; worktree
1129// agents have already captured their output and removed the isolated checkout
1130// during finalization. Remove then deletes the agent's durable state directory
1131// under <root>/agents/<id>/.
1132func (f *Supervisor) Remove(id string) error {
1133 f.operationMu.Lock()
1134 defer f.operationMu.Unlock()
1135 a := f.Get(id)
1136 if a == nil {
1137 return fmt.Errorf("no such agent %q", id)
1138 }
1139 a.mu.Lock()
1140 st := a.status
1141 inboxPath := a.InboxPath
1142 a.mu.Unlock()
1143 if st == StatusRunning || st == StatusPending {
1144 return fmt.Errorf("agent %s still %s", a.ID, st)
1145 }
1146 if inboxLive(inboxPath) {
1147 return fmt.Errorf("agent %s still has a live worker; stop it first", a.ID)
1148 }
1149 select {
1150 case <-a.done:
1151 default:
1152 return fmt.Errorf("agent %s is still finalizing", a.ID)
1153 }
1154 lease, err := acquireAgentLease(a.stateDirectory(f.cfg.Root))
1155 if err != nil {
1156 return fmt.Errorf("agent %s state is owned by another supervisor: %w", a.ID, err)
1157 }
1158 _ = lease.Close()
1159 // Best-effort cleanup of the per-agent state directory
1160 // (meta.json, events.jsonl, session.json). Failing here would leave
1161 // the user with no recourse,
1162 // so swallow the error.
1163 _ = os.RemoveAll(a.stateDirectory(f.cfg.Root))
1164 f.mu.Lock()
1165 delete(f.agents, a.ID)
1166 for i, k := range f.order {
1167 if k == a.ID {
1168 f.order = append(f.order[:i], f.order[i+1:]...)
1169 break
1170 }
1171 }
1172 f.mu.Unlock()
1173 return nil
1174}
1175
1176// Snapshot returns a read-only view of one agent. Safe for the TUI
1177// goroutine to call repeatedly; never blocks on the Runner.
1178type AgentSnapshot struct {
1179 ID string
1180 Task string
1181 Dir string
1182 Status Status
1183 ProcessState ProcessState
1184 TurnState TurnState
1185 CurrentTurnID string
1186 LifetimeTurns int
1187 CurrentRunTurns int
1188 Attempt int
1189 Activity string
1190 Started time.Time
1191 Finished time.Time
1192 CreatedAt time.Time
1193 UpdatedAt time.Time
1194 LastActivity time.Time
1195 Err string
1196 Tail string // last few transcript lines, joined with "\n"
1197 Lines []string // full transcript (already capped by Agent.appendTranscript)
1198 LastAssistant string // complete final assistant message, without role formatting
1199 Result *TurnResult
1200 ResultRef string
1201 Requirement RequirementSnapshot
1202 PatchRef string
1203 ChangedFiles []string
1204 OutputTruncated bool
1205
1206 // Model, Provider, Reasoning, and the provider connection settings
1207 // expose the per-agent configuration captured at Spawn time. Empty
1208 // values inherit the child's defaults. Subagent is the selected named
1209 // markdown profile. The dashboard surfaces these so the user can
1210 // confirm the child configuration.
1211 Model string
1212 Provider string
1213 BaseURL string
1214 InsecureTLS bool
1215 Reasoning string
1216 FastMode bool
1217 Subagent string
1218 WorkspaceMode WorkspaceMode
1219 WorkspacePath string
1220
1221 // Paths to the agent's durable state. Surface them in the
1222 // snapshot so the dashboard /subagents view can
1223 // read events.jsonl or resume the session without going back through the Agent.
1224 InboxPath string
1225 EventLogPath string
1226 SessionPath string
1227}
1228
1229// Snapshot copies the live agent state into a value the caller can
1230// inspect at leisure.
1231func (a *Agent) Snapshot() AgentSnapshot {
1232 a.mu.Lock()
1233 tail := strings.Join(lastN(a.transcript, 6), "\n")
1234 lines := make([]string, len(a.transcript))
1235 copy(lines, a.transcript)
1236 errStr := ""
1237 if a.lastErr != nil {
1238 errStr = a.lastErr.Error()
1239 }
1240 status := a.status
1241 activity := a.activity
1242 lastAssistant := a.lastAssistant
1243 outputTruncated := a.outputTruncated || a.streamingAssistantTruncated
1244 finished := a.finished
1245 a.mu.Unlock()
1246
1247 a.lifecycleMu.Lock()
1248 processState := a.processState
1249 turnState := a.turnState
1250 currentTurnID := a.currentTurnID
1251 lifetimeTurns := a.LifetimeTurns
1252 currentRunTurns := a.CurrentRunTurns
1253 attempt := a.Attempt
1254 lastActivity := a.lastActivity
1255 updatedAt := a.updatedAt
1256 result := cloneTurnResult(a.result)
1257 resultRef := a.resultRef
1258 requirement := a.visibleRequirementLocked()
1259 patchRef := a.patchRef
1260 changedFiles := append([]string(nil), a.changedFiles...)
1261 a.lifecycleMu.Unlock()
1262 return AgentSnapshot{
1263 ID: a.ID, Task: a.Task, Dir: a.Dir,
1264 Status: status, ProcessState: processState, TurnState: turnState,
1265 CurrentTurnID: currentTurnID, LifetimeTurns: lifetimeTurns, CurrentRunTurns: currentRunTurns, Attempt: attempt, Activity: activity,
1266 Started: a.Started, Finished: finished, CreatedAt: a.Started, UpdatedAt: updatedAt, LastActivity: lastActivity,
1267 Err: errStr, Tail: tail, Lines: lines,
1268 LastAssistant: lastAssistant, Result: result, ResultRef: resultRef,
1269 Requirement: requirement, PatchRef: patchRef, ChangedFiles: changedFiles, OutputTruncated: outputTruncated,
1270 Model: a.Model, Provider: a.Provider, BaseURL: a.BaseURL,
1271 InsecureTLS: a.InsecureTLS, Reasoning: a.Reasoning,
1272 FastMode: a.FastMode, Subagent: a.Subagent, WorkspaceMode: a.WorkspaceMode,
1273 WorkspacePath: a.WorkspacePath, InboxPath: a.InboxPath,
1274 EventLogPath: a.EventLogPath, SessionPath: a.SessionPath,
1275 }
1276}
1277
1278// SnapshotAll returns the supervisor's established status view: live agents
1279// are scoped to the active session, while durable terminal entries remain
1280// discoverable for recovery callers.
1281func (f *Supervisor) SnapshotAll() []AgentSnapshot {
1282 f.mu.Lock()
1283 active := f.activeSession
1284 f.mu.Unlock()
1285 return f.snapshotMatching(func(a *Agent) bool {
1286 if active == "" || a.SessionID == "" || a.SessionID == active {
1287 return true
1288 }
1289 status := a.Status()
1290 return status != StatusRunning && status != StatusPending
1291 })
1292}
1293
1294// SnapshotAllSessions returns every agent, regardless of host session. It is
1295// the explicit cross-session view used by the dashboard's All filter.
1296func (f *Supervisor) SnapshotAllSessions() []AgentSnapshot {
1297 return f.snapshotMatching(func(*Agent) bool { return true })
1298}
1299
1300// SnapshotCurrentSession returns only agents owned by the active host
1301// session. When no active session is set, it includes every agent for
1302// command-line and test callers that do not establish a session boundary.
1303func (f *Supervisor) SnapshotCurrentSession() []AgentSnapshot {
1304 f.mu.Lock()
1305 active := f.activeSession
1306 f.mu.Unlock()
1307 return f.snapshotMatching(func(a *Agent) bool {
1308 return active == "" || a.SessionID == active
1309 })
1310}
1311
1312func (f *Supervisor) snapshotMatching(include func(*Agent) bool) []AgentSnapshot {
1313 agents := f.List()
1314 out := make([]AgentSnapshot, 0, len(agents))
1315 for _, a := range agents {
1316 if include(a) {
1317 out = append(out, a.Snapshot())
1318 }
1319 }
1320 // Sort by start time for a stable, deterministic listing.
1321 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
1322 return out
1323}
1324
1325// agentSink is the Sink the Supervisor hands to each Runner.
1326type agentSink struct{ a *Agent }
1327
1328func (s agentSink) Activity(msg string) { s.a.setActivity(msg) }
1329func (s agentSink) Transcript(chunk string) {
1330 if strings.HasPrefix(chunk, "stderr: ") || strings.HasPrefix(chunk, "error: ") ||
1331 strings.HasPrefix(chunk, "tool: ") || strings.HasPrefix(chunk, "tool result: ") {
1332 s.a.appendTranscript(chunk)
1333 return
1334 }
1335 s.a.appendAssistantMessage(chunk)
1336}
1337func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) }
1338func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
1339func (s agentSink) assistantDelta(text string) { s.a.appendAssistantDelta(text) }
1340func (s agentSink) resetStreamingAssistant() { s.a.resetStreamingAssistant() }
1341
1342func truncate(s string, n int) string {
1343 runes := []rune(s)
1344 if len(runes) <= n {
1345 return s
1346 }
1347 if n <= 3 {
1348 return strings.Repeat(".", n)
1349 }
1350 return string(runes[:n-3]) + "..."
1351}
1352
1353func lastN(lines []string, n int) []string {
1354 if len(lines) <= n {
1355 return lines
1356 }
1357 return lines[len(lines)-n:]
1358}