1package subagents
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "encoding/json"
8 "fmt"
9 "os"
10 "os/exec"
11 "path/filepath"
12 "strings"
13 "sync"
14 "time"
15)
16
17// execRunner spawns `zut --subagent-worker <inbox> --session <path>` in
18// Agent.Dir (the shared repository root or an isolated worktree) and consumes
19// its JSONL event stream on stdout.
20//
21// Why a long-lived daemon and not `zut --print`: the supervisor and
22// the user expect agents to keep accepting follow-up prompts. A
23// one-shot subprocess can't do that; this design gives each subagent
24// agent a persistent session file plus an inbox socket the parent
25// writes to, mirroring Claude Code's "Agents view" model.
26//
27// Events flow:
28//
29// child stdout --> decoder --> EventLog (events.jsonl)
30// --> Sink (Activity/Transcript)
31//
32// The on-disk log is the durable record. The Sink updates are an
33// in-memory mirror so the dashboard doesn't have to tail the file
34// for the parent's own agents. /subagents open in a separate zut would
35// read the log directly.
36type execRunner struct {
37 agent *Agent
38 resolveCredential func(context.Context, string) (Credential, error)
39
40 // Command overrides the default `zut --subagent-worker ...`
41 // invocation. Tests set this to a fake binary (or `go run`
42 // against a tiny stub program) so the supervisor logic can be
43 // tested without a real child. Production code leaves it nil.
44 Command []string
45
46 // SessionPath is the agent's session file. Empty means "defer
47 // to r.agent.SessionPath", which Supervisor.Spawn always populates
48 // with <subagent-root>/agents/<id>/session.json. Tests that
49 // hand-build an Agent without going through Spawn must set
50 // one of the two; the runner refuses to invent a fallback
51 // because the only plausible one (<Dir>/.zut/session.json)
52 // would litter the user's repo — every agent's Dir points
53 // at it directly.
54 SessionPath string
55}
56
57// subagentWorkerArgsOpts captures every dynamic input to subagentWorkerArgs.
58// The fields map 1:1 onto child CLI flags; empty values omit the flag
59// entirely and let the child resolve a default the same way a normal
60// `zut` invocation does.
61type subagentWorkerArgsOpts struct {
62 Exe string
63 Dir string
64 SessionPath string
65 InboxPath string
66 Task string
67 Model string
68 Provider string
69 BaseURL string
70 InsecureTLS bool
71 Reasoning string
72 FastMode bool
73 FastModeSet bool
74 Subagent string
75 MaxTurns int
76 Tools []string
77 WebSearchPolicy WebSearchPolicy
78}
79
80// defaultChildArgs builds the argv execRunner uses when its Command
81// override is empty. Centralised so the spawn-vs-resume decision
82// (whether to pass the original Task as a positional) lives in one
83// place that tests can hit directly without going through Run's
84// side effects.
85//
86// On Spawn (Resuming==false) we pass the task so the child's first
87// turn runs immediately. On Resume (Resuming==true) we omit it: the
88// child reopens the existing session file, loads the prior
89// conversation, and just waits on the inbox for the next prompt.
90// Re-firing the task on every resume produces a duplicate turn that
91// collides with whatever the user types next, surfacing the
92// "agent busy; send 'cancel' first" error between two assistant
93// messages — which is exactly the bug this helper fixes.
94func defaultChildArgs(exe string, a *Agent, sessionPath, inboxPath string) []string {
95 task := a.Task
96 if a.Resuming {
97 task = ""
98 }
99 return subagentWorkerArgs(subagentWorkerArgsOpts{
100 Exe: exe,
101 Dir: a.Dir,
102 SessionPath: sessionPath,
103 InboxPath: inboxPath,
104 Task: task,
105 Model: a.Model,
106 Provider: a.Provider,
107 BaseURL: a.BaseURL,
108 InsecureTLS: a.InsecureTLS,
109 Reasoning: a.Reasoning,
110 FastMode: a.FastMode,
111 FastModeSet: true,
112 Subagent: a.Subagent,
113 MaxTurns: a.MaxTurns,
114 Tools: a.Tools,
115 WebSearchPolicy: a.WebSearchPolicy,
116 })
117}
118
119// subagentWorkerArgs builds the argv used when execRunner.Command is
120// empty. Pulled out so tests can lock in the flag set without
121// actually spawning a subprocess.
122func subagentWorkerArgs(opts subagentWorkerArgsOpts) []string {
123 args := []string{
124 opts.Exe,
125 "--subagent-worker", opts.InboxPath,
126 "--session", opts.SessionPath,
127 "--cwd", opts.Dir,
128 }
129 if opts.Model != "" {
130 args = append(args, "--model", opts.Model)
131 }
132 if opts.Provider != "" {
133 args = append(args, "--provider", opts.Provider)
134 }
135 if opts.BaseURL != "" {
136 args = append(args, "--base-url", opts.BaseURL)
137 }
138 if opts.InsecureTLS {
139 args = append(args, "--insecure")
140 }
141 if opts.Reasoning != "" {
142 args = append(args, "--reasoning", opts.Reasoning)
143 }
144 if opts.FastModeSet {
145 if opts.FastMode {
146 args = append(args, "--fast-mode")
147 } else {
148 args = append(args, "--no-fast-mode")
149 }
150 }
151 if opts.Subagent != "" {
152 args = append(args, "--subagent", opts.Subagent)
153 }
154 if opts.MaxTurns > 0 {
155 args = append(args, "--max-turns", fmt.Sprint(opts.MaxTurns))
156 }
157 webSearchPolicy := childWebSearchPolicy(opts.WebSearchPolicy, opts.Subagent, opts.Tools)
158 // Always propagate the final capability decision, including deny, so a
159 // child with no ordinary --tools list cannot re-enable web search from its
160 // own persisted configuration.
161 args = append(args, "--web-search-policy", webSearchPolicy.String())
162 if len(opts.Tools) > 0 {
163 args = append(args, "--tools", strings.Join(opts.Tools, ","))
164 }
165 if opts.Task != "" {
166 // First task is positional so the child treats it as the
167 // initial user turn; the terminator keeps flag-like task text
168 // from being parsed as a child CLI option.
169 args = append(args, "--", opts.Task)
170 }
171 return args
172}
173
174// resolveSubagentExecutable finds a runnable zut binary without assuming a
175// particular installation directory. A parent process can outlive the file
176// it was started from (for example after a reinstall), so all known
177// candidates are checked concurrently and the first successful lookup wins.
178// The cancellation signal stops candidates that have not finished yet. PATH
179// is one of the candidates because it covers GOBIN, user-local bin
180// directories, package-manager locations, and system installs without
181// hard-coding any of them.
182func resolveSubagentExecutable(ctx context.Context, self, argv0 string, lookPath func(context.Context, string) (string, error)) (string, error) {
183 if ctx == nil {
184 ctx = context.Background()
185 }
186 if lookPath == nil {
187 lookPath = func(ctx context.Context, candidate string) (string, error) {
188 select {
189 case <-ctx.Done():
190 return "", ctx.Err()
191 default:
192 }
193 return exec.LookPath(candidate)
194 }
195 }
196
197 candidates := make([]string, 0, 5)
198 addCandidate := func(candidate string) {
199 if candidate == "" || candidate == "." {
200 return
201 }
202 for _, existing := range candidates {
203 if existing == candidate {
204 return
205 }
206 }
207 candidates = append(candidates, candidate)
208 }
209 addCandidate(self)
210 addCandidate(argv0)
211 if self != "" {
212 addCandidate(filepath.Base(self))
213 }
214 if argv0 != "" {
215 addCandidate(filepath.Base(argv0))
216 }
217 addCandidate("zut")
218
219 if len(candidates) == 0 {
220 return "", fmt.Errorf("locate zut executable: no executable candidates")
221 }
222
223 type lookupResult struct {
224 candidate string
225 path string
226 err error
227 }
228 childCtx, cancel := context.WithCancel(ctx)
229 defer cancel()
230 results := make(chan lookupResult, len(candidates))
231 for _, candidate := range candidates {
232 go func(candidate string) {
233 path, err := lookPath(childCtx, candidate)
234 if err == nil && path == "" {
235 err = fmt.Errorf("empty executable path")
236 }
237 results <- lookupResult{candidate: candidate, path: path, err: err}
238 }(candidate)
239 }
240
241 lookupErrors := make([]string, 0, len(candidates))
242 for range candidates {
243 result := <-results
244 if result.err == nil {
245 cancel()
246 return result.path, nil
247 }
248 lookupErrors = append(lookupErrors, fmt.Sprintf("%q: %v", result.candidate, result.err))
249 }
250 return "", fmt.Errorf("locate zut executable: %s", strings.Join(lookupErrors, "; "))
251}
252
253func (r *execRunner) Run(ctx context.Context, sink Sink) error {
254 // SessionPath resolution order:
255 // 1. explicit r.SessionPath set by the test / caller
256 // 2. r.agent.SessionPath baked in by Supervisor.Spawn — the
257 // production path. Always lives under
258 // <subagent-root>/agents/<id>/session.json so the per-
259 // agent state is entirely outside the working tree.
260 // Crucial because Agent.Dir points at the user's repo;
261 // any .zut/ scratch directory under Dir would litter
262 // their source tree.
263 //
264 // There is no third fallback. If neither path is set we
265 // refuse to start instead of inventing a directory; that
266 // way a misconfigured caller fails loudly the first time
267 // instead of silently dumping session data into someone's
268 // repo.
269 sessionPath := r.SessionPath
270 if sessionPath == "" {
271 sessionPath = r.agent.SessionPath
272 }
273 if sessionPath == "" {
274 return fmt.Errorf("subagent: agent missing session path (set SpawnRequest via Supervisor.SpawnReq, or hand-build Agent with SessionPath populated)")
275 }
276 if err := os.MkdirAll(filepath.Dir(sessionPath), 0o755); err != nil {
277 return fmt.Errorf("session dir: %w", err)
278 }
279
280 inboxPath := r.agent.InboxPath
281 logPath := r.agent.EventLogPath
282 if logPath == "" {
283 return fmt.Errorf("subagent: agent missing event log path")
284 }
285 log, err := OpenEventLog(logPath)
286 if err != nil {
287 return err
288 }
289 defer log.Close()
290
291 args := r.Command
292 if len(args) == 0 {
293 self, selfErr := os.Executable()
294 argv0 := ""
295 if len(os.Args) > 0 {
296 argv0 = os.Args[0]
297 }
298 exe, err := resolveSubagentExecutable(ctx, self, argv0, nil)
299 if err != nil {
300 if selfErr != nil {
301 return fmt.Errorf("locate zut executable (os.Executable: %v): %w", selfErr, err)
302 }
303 return err
304 }
305 args = defaultChildArgs(exe, r.agent, sessionPath, inboxPath)
306 }
307
308 cmd := exec.CommandContext(ctx, args[0], args[1:]...)
309 cmd.Dir = r.agent.Dir
310 cmd.Env = append(workerEnvironment(r.agent.Provider),
311 "ZUT_SUBAGENT_AGENT_ID="+r.agent.ID,
312 "ZUT_SUBAGENT_EVENT_LOG="+logPath,
313 )
314 if r.agent.HeartbeatInterval > 0 {
315 cmd.Env = append(cmd.Env, "ZUT_SUBAGENT_HEARTBEAT_INTERVAL="+r.agent.HeartbeatInterval.String())
316 }
317 if r.agent.maxOutputBytes > 0 {
318 cmd.Env = append(cmd.Env, fmt.Sprintf("ZUT_SUBAGENT_MAX_OUTPUT_BYTES=%d", r.agent.maxOutputBytes))
319 }
320 if r.agent.maxOutputLines > 0 {
321 cmd.Env = append(cmd.Env, fmt.Sprintf("ZUT_SUBAGENT_MAX_OUTPUT_LINES=%d", r.agent.maxOutputLines))
322 }
323 if r.resolveCredential != nil {
324 credential, resolveErr := r.resolveCredential(ctx, r.agent.Provider)
325 if resolveErr != nil {
326 return fmt.Errorf("resolve subagent credential for %s: %w", r.agent.Provider, resolveErr)
327 }
328 if credential.Value != "" {
329 encoded, encodeErr := json.Marshal(credential)
330 if encodeErr != nil {
331 return fmt.Errorf("encode subagent credential: %w", encodeErr)
332 }
333 cmd.Stdin = bytes.NewReader(encoded)
334 cmd.Env = append(cmd.Env, "ZUT_SUBAGENT_CREDENTIAL_STDIN=1")
335 }
336 }
337
338 stdout, err := cmd.StdoutPipe()
339 if err != nil {
340 return err
341 }
342 stderr, err := cmd.StderrPipe()
343 if err != nil {
344 return err
345 }
346 configureWorkerProcess(cmd, stdout, stderr)
347
348 // "spawning" is briefly shown until the first event arrives;
349 // the child's "spawned" lifecycle event then overwrites it.
350 sink.Activity("starting")
351 if err := cmd.Start(); err != nil {
352 return err
353 }
354 r.agent.setProcessPID(cmd.Process.Pid)
355 if r.agent.persistFn != nil {
356 r.agent.persistFn(r.agent)
357 }
358
359 var logErrMu sync.Mutex
360 var logErr error
361 var stopOnLogErr sync.Once
362 appendLog := func(ev Event) {
363 if err := log.Append(ev); err != nil {
364 logErrMu.Lock()
365 if logErr == nil {
366 logErr = err
367 }
368 logErrMu.Unlock()
369 // A durable event log is part of the runner's recovery contract.
370 // Stop the worker rather than letting it continue in a state the
371 // supervisor cannot reconstruct after restart.
372 stopOnLogErr.Do(func() { _ = killProcessGroup(cmd) })
373 }
374 }
375 firstLogErr := func() error {
376 logErrMu.Lock()
377 defer logErrMu.Unlock()
378 return logErr
379 }
380
381 // stdout: parsed as JSONL. Every well-formed event is appended
382 // to the durable log AND forwarded to the in-memory sink so the
383 // dashboard updates without having to tail the file. Malformed
384 // lines are surfaced as plain transcript so they don't vanish.
385 done := make(chan struct{}, 2)
386 go func() {
387 defer func() { done <- struct{}{} }()
388 dec := bufio.NewReader(stdout)
389 lineLimit := r.agent.maxOutputBytes + 64*1024
390 if lineLimit <= 0 {
391 lineLimit = 512 * 1024
392 }
393 if lineLimit > maxEventLogLineBytes {
394 lineLimit = maxEventLogLineBytes
395 }
396 for {
397 line, err, truncated := readBoundedLine(dec, lineLimit)
398 if len(line) > 0 {
399 trimmed := strings.TrimRight(string(line), "\r\n")
400 if trimmed != "" && !truncated {
401 if ev, ok := parseEventLine(trimmed); ok {
402 appendLog(ev)
403 updateAgentFromEvent(r.agent, ev)
404 applyEventToSink(ev, sink)
405 // Fan prompt-level task completions up to any
406 // subscriber on the supervised Agent. The child
407 // also forwards provider/tool-loop turn_end
408 // events (for example stop=tool_use); those do
409 // not contain step and must not be treated as
410 // subagent task completion.
411 notifyPromptTurnEnd(r.agent, ev)
412 } else {
413 sink.Transcript(trimmed)
414 appendLog(NewEvent("stdout", map[string]any{"text": trimmed}))
415 }
416 } else if trimmed != "" {
417 trimmed += "\n...[line truncated]"
418 sink.Transcript(trimmed)
419 appendLog(NewEvent("stdout", map[string]any{"text": trimmed, "truncated": true}))
420 }
421 }
422 if err != nil {
423 return
424 }
425 }
426 }()
427
428 // stderr: lifecycle/error chatter from the child. Every line
429 // is mirrored as a stderr event in the durable log AND surfaced
430 // in the transcript so users can diagnose a failing agent
431 // without leaving the dashboard.
432 go func() {
433 defer func() { done <- struct{}{} }()
434 br := bufio.NewReader(stderr)
435 lineLimit := r.agent.maxOutputBytes + 64*1024
436 if lineLimit <= 0 {
437 lineLimit = 512 * 1024
438 }
439 if lineLimit > maxEventLogLineBytes {
440 lineLimit = maxEventLogLineBytes
441 }
442 for {
443 line, err, truncated := readBoundedLine(br, lineLimit)
444 if len(line) > 0 {
445 txt := strings.TrimRight(string(line), "\r\n")
446 if truncated {
447 txt += "\n...[line truncated]"
448 }
449 sink.Transcript("stderr: " + txt)
450 appendLog(NewEvent("stderr", map[string]any{"text": txt, "truncated": truncated}))
451 }
452 if err != nil {
453 return
454 }
455 }
456 }()
457
458 <-done
459 <-done
460
461 err = cmd.Wait()
462 exit := 0
463 if ee, ok := err.(*exec.ExitError); ok {
464 exit = ee.ExitCode()
465 }
466 if err != nil && ctx.Err() != nil {
467 appendLog(NewEvent("agent_stopped", map[string]any{"reason": "cancelled"}))
468 if logErr := firstLogErr(); logErr != nil {
469 return fmt.Errorf("subagent event log: %w", logErr)
470 }
471 return ctx.Err()
472 }
473 if err != nil {
474 appendLog(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": exit, "error": err.Error()}))
475 if logErr := firstLogErr(); logErr != nil {
476 return fmt.Errorf("subagent event log: %v; worker: %w", logErr, err)
477 }
478 return err
479 }
480 appendLog(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": 0}))
481 if logErr := firstLogErr(); logErr != nil {
482 return fmt.Errorf("subagent event log: %w", logErr)
483 }
484 sink.Activity("done")
485 return nil
486}
487
488func workerEnvironment(provider string) []string {
489 // Provider credentials are resolved by the supervisor and transferred
490 // over the child's stdin. Do not inherit API-key/token variables into the
491 // worker environment, where child tools, crash reports, or diagnostics
492 // could expose them. Bedrock's current client still resolves bearer/IAM
493 // credentials directly from AWS environment variables, so preserve that
494 // provider's ambient AWS chain until it has a structured stdin handoff.
495 provider = strings.ToLower(strings.TrimSpace(provider))
496 var env []string
497 for _, entry := range os.Environ() {
498 name, _, ok := strings.Cut(entry, "=")
499 if ok && workerSecretEnvName(name) && (provider != "amazon-bedrock" || !isBedrockCredentialEnv(name)) {
500 continue
501 }
502 env = append(env, entry)
503 }
504 return env
505}
506
507func isBedrockCredentialEnv(name string) bool {
508 switch strings.ToUpper(strings.TrimSpace(name)) {
509 case "AWS_BEARER_TOKEN_BEDROCK", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN":
510 return true
511 default:
512 return false
513 }
514}
515
516func workerSecretEnvName(name string) bool {
517 name = strings.ToUpper(strings.TrimSpace(name))
518 for _, suffix := range []string{"_KEY", "_API_KEY", "_OAUTH_TOKEN", "_TOKEN", "_SECRET", "_PASSWORD", "_CREDENTIAL", "_CREDENTIALS"} {
519 if strings.HasSuffix(name, suffix) {
520 return true
521 }
522 }
523 switch name {
524 case "AWS_ACCESS_KEY_ID", "HF_TOKEN", "AWS_BEARER_TOKEN_BEDROCK", "COPILOT_GITHUB_TOKEN", "GITHUB_COPILOT_TOKEN", "GOOGLE_APPLICATION_CREDENTIALS":
525 return true
526 default:
527 return false
528 }
529}
530
531func readBoundedLine(r *bufio.Reader, limit int) ([]byte, error, bool) {
532 line, truncated, _, _, err := readBoundedLineStats(r, limit)
533 return line, err, truncated
534}
535
536// readBoundedLineStats reads through one newline or EOF while retaining the
537// number of source bytes consumed and whether a newline terminated the record.
538// The follower uses those extra values to avoid advancing past a partial final
539// JSONL record that may be completed by a concurrent append.
540func readBoundedLineStats(r *bufio.Reader, limit int) ([]byte, bool, int64, bool, error) {
541 if limit <= 0 {
542 limit = 512 * 1024
543 }
544 var line []byte
545 var consumed int64
546 truncated := false
547 for {
548 chunk, err := r.ReadSlice('\n')
549 consumed += int64(len(chunk))
550 if len(chunk) > 0 {
551 if len(line) < limit {
552 keep := len(chunk)
553 if remaining := limit - len(line); keep > remaining {
554 keep = remaining
555 truncated = true
556 }
557 line = append(line, chunk[:keep]...)
558 } else {
559 truncated = true
560 }
561 }
562 if err != bufio.ErrBufferFull {
563 return line, truncated, consumed, len(chunk) > 0 && chunk[len(chunk)-1] == '\n', err
564 }
565 }
566}
567
568// parseEventLine attempts to decode one current-protocol JSONL line as an Event.
569func parseEventLine(line string) (Event, bool) {
570 if len(line) == 0 || line[0] != '{' {
571 return Event{}, false
572 }
573 env, err := ParseEvent([]byte(line))
574 if err != nil {
575 return Event{}, false
576 }
577 fields, err := env.PayloadFields()
578 if err != nil {
579 return Event{}, false
580 }
581 data := make(map[string]any, len(fields))
582 for key, raw := range fields {
583 var value any
584 if err := json.Unmarshal(raw, &value); err != nil {
585 return Event{}, false
586 }
587 data[key] = value
588 }
589 when := env.Timestamp
590 if when.IsZero() {
591 when = time.Now()
592 }
593 return Event{
594 Time: when, Type: env.Type, Version: env.Version,
595 MessageID: env.MessageID, AgentID: env.AgentID, TurnID: env.TurnID,
596 Data: data,
597 }, true
598}
599
600func updateAgentFromEvent(a *Agent, ev Event) {
601 if a == nil {
602 return
603 }
604 now := ev.Time
605 if now.IsZero() {
606 now = time.Now()
607 }
608 a.markActivity(now)
609 persist := false
610 switch ev.Type {
611 case EventAgentReady, "agent_ready":
612 a.setProcessState(ProcessAlive)
613 a.setTurnState(TurnIdle, ev.TurnID)
614 persist = true
615 case EventAgentHeartbeat, "agent_heartbeat":
616 a.setProcessState(ProcessAlive)
617 persist = true
618 case EventTurnStarted, "turn_start":
619 turnID := ev.TurnID
620 if turnID == "" {
621 if step, ok := ev.Data["step"].(float64); ok {
622 turnID = fmt.Sprintf("turn-%d", int(step))
623 }
624 }
625 a.setProcessState(ProcessAlive)
626 a.setTurnState(TurnRunning, turnID)
627 persist = true
628 case EventTurnResult, "turn_result":
629 if result, err := decodeTurnResultEvent(ev, a.ID, a.maxOutputBytes, a.maxOutputLines); err == nil {
630 a.setResult(result)
631 if a.turnResults != nil {
632 select {
633 case a.turnResults <- result:
634 default:
635 }
636 }
637 if a.stateDir != "" {
638 if err := writeTurnResult(a.stateDir, result); err == nil {
639 a.lifecycleMu.Lock()
640 a.resultRef = ResultRef(a.ID)
641 a.lifecycleMu.Unlock()
642 }
643 }
644 switch result.Status {
645 case ResultCanceled:
646 a.setTurnState(TurnCanceled, result.TurnID)
647 case ResultFailed:
648 a.setTurnState(TurnFailed, result.TurnID)
649 default:
650 a.setTurnState(TurnSucceeded, result.TurnID)
651 }
652 }
653 persist = true
654 case EventTurnFailed, "turn_failed":
655 a.setTurnState(TurnFailed, ev.TurnID)
656 persist = true
657 case EventAgentIdle, "agent_idle":
658 a.setTurnState(TurnIdle, ev.TurnID)
659 persist = true
660 case EventAgentExited, "agent_exited", "agent_stopped":
661 a.setProcessState(ProcessExited)
662 persist = true
663 case "turn_end":
664 // Provider/tool-loop turn_end events (for example stop=tool_use)
665 // do not carry the daemon's prompt step and are not terminal for
666 // the delegated task. They remain in the event log, but must not
667 // overwrite the delegated turn state or trigger persistence.
668 if _, ok := ev.Data["step"].(float64); !ok {
669 break
670 }
671 if message, _ := ev.Data["error"].(string); message != "" {
672 a.setTurnState(TurnFailed, ev.TurnID)
673 } else {
674 a.setTurnState(TurnSucceeded, ev.TurnID)
675 }
676 persist = true
677 }
678 if persist && a.persistFn != nil {
679 a.persistFn(a)
680 }
681}
682
683// notifyPromptTurnEnd calls Agent.OnTurnEnd only for the subagent
684// daemon's prompt-level completion event. Provider/tool-loop
685// turn_end events (such as stop=tool_use) do not include step and
686// are not terminal for the delegated task.
687func notifyPromptTurnEnd(a *Agent, ev Event) {
688 if a == nil || ev.Type != "turn_end" {
689 return
690 }
691 step, ok := ev.Data["step"].(float64)
692 if !ok {
693 return
694 }
695
696 errMsg, _ := ev.Data["error"].(string)
697 a.mu.Lock()
698 fn := a.OnTurnEnd
699 if fn == nil {
700 a.pendingTurnEnds = append(a.pendingTurnEnds, turnEndNotice{step: int(step), errMsg: errMsg})
701 a.mu.Unlock()
702 return
703 }
704 a.mu.Unlock()
705 go fn(int(step), errMsg)
706}
707
708// applyEventToSink translates an Event into Sink updates. Only a
709// few event types are interpreted; the rest still land in the
710// durable log via the caller.
711func applyEventToSink(ev Event, sink Sink) {
712 type roleSink interface {
713 userMessage(string)
714 assistantMessage(string)
715 }
716 appendMessage := func(text string, assistant bool) {
717 if text == "" {
718 return
719 }
720 if roles, ok := sink.(roleSink); ok {
721 if assistant {
722 roles.assistantMessage(text)
723 } else {
724 roles.userMessage(text)
725 }
726 return
727 }
728 if assistant {
729 sink.Transcript(text)
730 } else {
731 sink.Transcript("user: " + text)
732 }
733 }
734
735 switch ev.Type {
736 case "assistant_message", "user_message":
737 var text []string
738 if c, ok := ev.Data["content"].([]any); ok {
739 for _, blk := range c {
740 m, _ := blk.(map[string]any)
741 if t, _ := m["type"].(string); t == "text" {
742 if txt, _ := m["text"].(string); txt != "" {
743 text = append(text, txt)
744 }
745 }
746 }
747 }
748 appendMessage(strings.Join(text, "\n"), ev.Type == "assistant_message")
749 if ev.Type == "assistant_message" {
750 sink.Activity("idle")
751 }
752 case "turn_start", EventTurnStarted:
753 sink.Activity("thinking")
754 case "tool_call", EventToolStarted:
755 if name, _ := ev.Data["name"].(string); name != "" {
756 sink.Activity("tool: " + truncate(name, 60))
757 }
758 case "tool_result", EventToolFinished:
759 sink.Activity("idle")
760 case "turn_end", EventTurnResult, EventAgentIdle:
761 sink.Activity("idle")
762 case "agent_ready", EventAgentReady, EventAgentHeartbeat:
763 sink.Activity("idle")
764 case "agent_stopped":
765 // terminal status is decided by Supervisor.run from the runner's
766 // return value, not from this event. Don't overwrite the
767 // activity here.
768 case "error":
769 if msg, _ := ev.Data["message"].(string); msg != "" {
770 sink.Transcript("error: " + msg)
771 }
772 }
773}
774
775// RunnerFunc adapts a plain function into a Runner. Useful for tests
776// and for callers who don't need their own type.
777type RunnerFunc func(ctx context.Context, sink Sink) error
778
779func (f RunnerFunc) Run(ctx context.Context, sink Sink) error { return f(ctx, sink) }