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