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