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