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