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