1package agent
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "os"
10 "strconv"
11 "strings"
12 "sync"
13 "time"
14 "unicode/utf8"
15
16 "github.com/bnema/zut/packages/agent/modes"
17 "github.com/bnema/zut/packages/agent/subagents"
18 "github.com/bnema/zut/packages/core"
19 "github.com/bnema/zut/packages/provider"
20)
21
22// runSubagentWorkerMode is the daemon-mode entry point used by every
23// subagent-spawned zut subprocess. It's intentionally close in shape to
24// runJSONMode but with two key differences:
25//
26// - Lifetime: the process stays alive across many user turns. The
27// initial positional task (if any) is the first turn; subsequent
28// turns arrive through the inbox unix socket at args.SubagentWorker.
29//
30// - Output: every emitted JSON line is also mirrored verbatim into
31// events.jsonl (see ZUT_SUBAGENT_EVENT_LOG) so a separate zut
32// process can /subagents open this agent and replay its full history
33// even after the parent that spawned us is long gone.
34//
35// The runner in packages/agent/subagents/runner.go is the only caller in
36// production; tests use the stubchild binary under
37// packages/agent/subagents/testdata/cmd/stubchild instead of the real model
38// loop.
39// workerTurnBudget is deliberately independent of the model loop: one
40// admitted message turn consumes one current-run and lifetime turn, while
41// retries, compaction, and provider/tool loops remain inside that turn.
42type workerTurnBudget struct {
43 sequence int
44 lifetime int
45 current int
46}
47
48type sessionPersistenceState struct {
49 mu sync.Mutex
50 errValue error
51}
52
53func (s *sessionPersistenceState) record(err error) {
54 if err == nil {
55 return
56 }
57 s.mu.Lock()
58 if s.errValue == nil {
59 s.errValue = err
60 }
61 s.mu.Unlock()
62}
63
64func (s *sessionPersistenceState) failed() bool {
65 s.mu.Lock()
66 defer s.mu.Unlock()
67 return s.errValue != nil
68}
69
70func (s *sessionPersistenceState) err() error {
71 s.mu.Lock()
72 defer s.mu.Unlock()
73 return s.errValue
74}
75
76func (b *workerTurnBudget) start(maxTurns int, newRun bool) (step, lifetime, current int, admitted bool) {
77 if newRun {
78 b.current = 0
79 }
80 b.sequence++
81 if maxTurns > 0 && b.current >= maxTurns {
82 return b.sequence, b.lifetime, b.current, false
83 }
84 b.lifetime++
85 b.current++
86 return b.sequence, b.lifetime, b.current, true
87}
88
89func subagentTurnContext(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) {
90 if timeout > 0 {
91 return context.WithTimeout(parent, timeout)
92 }
93 return context.WithCancel(parent)
94}
95
96func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runErr error) {
97 if args.SubagentWorker == "" {
98 return fmt.Errorf("--subagent-worker requires a socket path")
99 }
100 if os.Getenv("ZUT_SUBAGENT_CREDENTIAL_STDIN") == "1" {
101 var inherited subagents.Credential
102 dec := json.NewDecoder(io.LimitReader(os.Stdin, 1<<20))
103 if err := dec.Decode(&inherited); err != nil {
104 return fmt.Errorf("read inherited subagent credential: %w", err)
105 }
106 if inherited.Value == "" || inherited.Method == "" {
107 return fmt.Errorf("read inherited subagent credential: missing value or method")
108 }
109 args.inheritedCredential = inherited.Value
110 args.inheritedAuthMethod = inherited.Method
111 args.inheritedAccountID = inherited.AccountID
112 }
113
114 cfg, err := loadSubagentWorkerConfig()
115 if err != nil {
116 return err
117 }
118 r, err := Resolve(args, true)
119 if err != nil {
120 return err
121 }
122 extMgr, stopExt := setupNonInteractiveExtensions(ctx, args, &r, version)
123 defer stopExt()
124
125 ag := r.NewAgent()
126 initialAg := ag
127 defer func() {
128 closeAgentLSP(ag)
129 if ag != initialAg {
130 closeAgentLSP(initialAg)
131 }
132 }()
133 wireNonInteractiveAgentExtHooks(ctx, ag, extMgr)
134 sess, err := openOrCreateSession(ctx, args, r, ag, version)
135 if err != nil {
136 return err
137 }
138 var sessionPersistence sessionPersistenceState
139 if sess != nil {
140 var providerName, model string
141 sess, ag, providerName, model, err = applyInitialSessionResume(ctx, args, r, extMgr, sess, ag)
142 if err != nil {
143 return err
144 }
145 r.Provider, r.Model = providerName, model
146 ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) }
147 // A prompt is acknowledged by the supervisor only after this callback
148 // has appended and synced the matching user message. This closes the
149 // crash window between accepting a follow-up and writing session.json.
150 ag.OnMessageAppended = func(message provider.Message) {
151 if err := sess.AppendMessage(message); err != nil {
152 sessionPersistence.record(err)
153 return
154 }
155 if err := sess.Sync(); err != nil {
156 sessionPersistence.record(err)
157 }
158 }
159 ag.OnUsage = func(usage provider.Usage) {
160 if err := sess.AppendUsage(usage, usage); err != nil {
161 sessionPersistence.record(err)
162 }
163 }
164 ag.OnRetryLifecycle = sess.RecordRetryLifecycle
165 defer func() {
166 runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost()))
167 runErr = joinSessionCloseError(runErr, sess)
168 }()
169 }
170 announceSession(extMgr, sess)
171
172 // Resolve retains metadata for catalog models and for valid synthesized
173 // local/routed/open-catalog models. The active agent also survives a session
174 // resume that rebuilt it for a different provider/model pair.
175 contextWindow := ag.ContextWindow
176
177 // Open the inbox listener BEFORE emitting agent_ready so the
178 // supervisor can dial through on the very first send. The
179 // subagent supervisor's Inbox retries dialing for a
180 // short window, but emitting ready first and then listening
181 // would still race in tight loops.
182 ln, err := subagents.Listen(args.SubagentWorker)
183 if err != nil {
184 return fmt.Errorf("subagent-worker listen: %w", err)
185 }
186 defer ln.Close()
187
188 // Event log is owned by the supervisor's runner via stdout, but
189 // the daemon also writes a redundant copy here when the runner's
190 // pipe is closed (e.g. parent zut exited but the agent is still
191 // running headless). The env var is set by the runner; if it's
192 // empty we silently skip the second mirror.
193 var logMirror *subagents.EventLog
194 if path := os.Getenv("ZUT_SUBAGENT_EVENT_LOG"); path != "" {
195 logMirror, _ = subagents.OpenEventLog(path)
196 }
197 if logMirror != nil {
198 defer logMirror.Close()
199 }
200
201 em := newSubagentEmitter(os.Stdout, logMirror)
202 em.setProtocolIdentity(os.Getenv("ZUT_SUBAGENT_AGENT_ID"))
203 em.emit("agent.ready", map[string]any{
204 "version": version,
205 "cwd": r.CWD,
206 "model": r.Model,
207 "lifetime_turns": args.SubagentLifetimeTurns,
208 "current_run_turns": args.SubagentRunTurns,
209 "max_turns": args.SubagentMaxTurns,
210 })
211
212 // Heartbeats make a live-but-detached child distinguishable from a
213 // failed turn after supervisor reconnect. The parent still treats
214 // unknown event types as forward-compatible.
215 heartbeatDone := make(chan struct{})
216 defer close(heartbeatDone)
217 go func() {
218 interval := 10 * time.Second
219 if raw := os.Getenv("ZUT_SUBAGENT_HEARTBEAT_INTERVAL"); raw != "" {
220 if parsed, parseErr := time.ParseDuration(raw); parseErr == nil && parsed > 0 {
221 interval = parsed
222 }
223 }
224 ticker := time.NewTicker(interval)
225 defer ticker.Stop()
226 for {
227 select {
228 case <-heartbeatDone:
229 return
230 case <-ticker.C:
231 em.emit("agent.heartbeat", map[string]any{"activity": "alive"})
232 }
233 }
234 }()
235
236 // Keep a per-turn cancel so the "cancel" inbox message can interrupt
237 // an in-flight turn without tearing down the whole daemon.
238 var (
239 mu sync.Mutex
240 cancelFn context.CancelFunc
241 busyTurn bool
242 budget = workerTurnBudget{
243 sequence: initialTurnNumber(os.Getenv("ZUT_SUBAGENT_EVENT_LOG")),
244 lifetime: args.SubagentLifetimeTurns,
245 current: args.SubagentRunTurns,
246 }
247 turnWG sync.WaitGroup
248 closing bool
249 turnPending bool
250 seenCommands = make(map[string]struct{})
251 shutdown = make(chan struct{})
252 shutdownOnce sync.Once
253 shutdownOrigin subagents.ShutdownOrigin
254 )
255 maxOutputBytes, maxOutputLines := workerOutputLimits()
256
257 runOne := func(prompt string, newRun bool, commandID string) {
258 mu.Lock()
259 // launchTurn admits the goroutine to turnWG before it starts so
260 // shutdown can join it. A shutdown can win the scheduling race
261 // after that admission but before this goroutine acquires mu; in
262 // that case do not start a new Prompt after waitForActiveTurn has
263 // declared the worker closing.
264 if closing || busyTurn {
265 mu.Unlock()
266 em.emit("error", map[string]any{"code": "turn_rejected", "reason": "busy", "command_id": commandID, "message": "worker is busy"})
267 return
268 }
269 busyTurn = true
270 turnPending = false
271 step, lifetime, currentRun, admitted := budget.start(args.SubagentMaxTurns, newRun)
272 if !admitted {
273 busyTurn = false
274 mu.Unlock()
275 turnID := fmt.Sprintf("turn-%d", step)
276 errPayload := map[string]any{"code": "turn_rejected", "reason": "max_turns", "command_id": commandID, "message": "maximum subagent turns reached"}
277 em.emit("error", errPayload)
278 em.emit("turn.result", map[string]any{"status": "failed", "turn_id": turnID, "error": errPayload})
279 em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": errPayload})
280 em.emit("turn_end", map[string]any{"step": step, "turn_id": turnID, "error": errPayload["message"]})
281 // Max-turn rejection ends the current run, but the worker remains
282 // alive and can accept a queued follow-up as a fresh run.
283 em.emit("agent.idle", map[string]any{
284 "turn_id": turnID,
285 "lifetime_turns": lifetime,
286 "current_run_turns": currentRun,
287 })
288 return
289 }
290 turnID := fmt.Sprintf("turn-%d", step)
291 c, cancel := subagentTurnContext(ctx, args.SubagentTurnTimeout)
292 cancelFn = cancel
293 mu.Unlock()
294
295 turnStarted := false
296 sink := func(ev core.AgentEvent) {
297 data := subagentEventData(ev)
298 if ev.Type() == "user_message" && !turnStarted {
299 // Prompt invokes OnMessageAppended before delivering this sink
300 // event, so the session row is durable before the lifecycle
301 // acknowledgement is written.
302 em.emit(ev.Type(), data)
303 if !sessionPersistence.failed() {
304 em.emit("turn.started", map[string]any{"step": step, "turn_id": turnID, "command_id": commandID, "lifetime_turns": lifetime, "current_run_turns": currentRun})
305 turnStarted = true
306 }
307 return
308 }
309 if ev.Type() == "turn_start" {
310 // A provider/model-loop turn is nested inside the one delegated
311 // message turn admitted above. Mark it on the wire so the
312 // supervisor cannot mistake it for a new worker turn.
313 data["nested_turn"] = true
314 }
315 em.emit(ev.Type(), data)
316 }
317
318 baseline := workerTranscriptState{
319 messages: ag.Messages(),
320 cost: ag.Cost(),
321 lastTurn: ag.LastTurnUsage(),
322 }
323 var persistCompaction func([]provider.Message) error
324 if sess != nil {
325 // Buffer the checkpoint until the continued turn completes. The final
326 // transcript and cumulative usage are then persisted in one JSONL row.
327 persistCompaction = func([]provider.Message) error {
328 return c.Err()
329 }
330 }
331 recovery, err := promptWithContextRecovery(c, ag, prompt, sink, modes.ContextRecoveryOptions{
332 PersistCompaction: persistCompaction,
333 ContextWindow: contextWindow,
334 AutoCompactThreshold: cfg.AutoCompactThreshold,
335 })
336 output := finalAssistantText(ag.Messages()[recovery.OutputStart:])
337 // Message and usage callbacks persist ordinary turns incrementally. A
338 // compacted turn still needs one atomic replacement checkpoint so the
339 // durable transcript cannot mix generations.
340 if recovery.Compacted || ag.OnMessageAppended == nil {
341 if persistErr := persistWorkerTranscript(c, ag, sess, recovery.OutputStart, recovery.Compacted, baseline); persistErr != nil {
342 if err != nil {
343 err = errors.Join(err, persistErr)
344 } else {
345 err = persistErr
346 }
347 }
348 }
349 if sess != nil {
350 if persistErr := sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost()); persistErr != nil {
351 err = errors.Join(err, persistErr)
352 }
353 }
354 if sessionErr := sessionPersistence.err(); err == nil && sessionErr != nil {
355 err = sessionErr
356 }
357
358 resultStatus := "succeeded"
359 if err != nil {
360 if errors.Is(err, context.Canceled) {
361 resultStatus = "canceled"
362 } else {
363 resultStatus = "failed"
364 }
365 }
366 mu.Lock()
367 resultShutdownOrigin := shutdownOrigin.Sanitized()
368 mu.Unlock()
369 switch {
370 case errors.Is(err, context.DeadlineExceeded):
371 resultShutdownOrigin = subagents.ShutdownOriginDeadline
372 case !errors.Is(err, context.Canceled):
373 resultShutdownOrigin = ""
374 }
375 resultPayload := map[string]any{
376 "status": resultStatus,
377 "turn_id": turnID,
378 "summary": boundedResultSummary(output),
379 "output": boundedResultOutput(output, maxOutputBytes, maxOutputLines),
380 "error": resultErrorPayload(err, resultShutdownOrigin),
381 }
382 if resultShutdownOrigin != "" {
383 resultPayload["shutdown_origin"] = resultShutdownOrigin
384 }
385 em.emit("turn.result", resultPayload)
386 if err != nil {
387 em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": resultErrorPayload(err, resultShutdownOrigin)})
388 }
389 em.emit("turn_end", map[string]any{
390 "step": step,
391 "turn_id": turnID,
392 "error": turnErrorMessage(err, resultShutdownOrigin),
393 })
394
395 cancel()
396 mu.Lock()
397 busyTurn = false
398 cancelFn = nil
399 mu.Unlock()
400 // The supervisor treats agent.idle as permission to start a follow-up.
401 // Publish it only after the busy gate has opened, otherwise a manager
402 // can send a turn that the inbox loop drops as still busy.
403 em.emit("agent.idle", map[string]any{"turn_id": turnID, "lifetime_turns": lifetime, "current_run_turns": currentRun})
404 }
405
406 launchTurn := func(prompt string, newRun bool, commandID string) bool {
407 mu.Lock()
408 if closing || busyTurn || turnPending {
409 mu.Unlock()
410 em.emit("error", map[string]any{"code": "turn_rejected", "reason": "busy", "command_id": commandID, "message": "worker is busy"})
411 return false
412 }
413 // Reserve the worker before launching the goroutine. A second command
414 // cannot be accepted in the gap before runOne acquires mu.
415 turnPending = true
416 turnWG.Add(1)
417 mu.Unlock()
418 go func() {
419 defer turnWG.Done()
420 runOne(prompt, newRun, commandID)
421 }()
422 return true
423 }
424
425 waitForActiveTurn := func(cancel bool) {
426 mu.Lock()
427 closing = true
428 activeCancel := cancelFn
429 mu.Unlock()
430 if cancel && activeCancel != nil {
431 activeCancel()
432 }
433 turnWG.Wait()
434 }
435
436 requestShutdown := func(origin subagents.ShutdownOrigin) {
437 shutdownOnce.Do(func() {
438 mu.Lock()
439 closing = true
440 shutdownOrigin = origin.Sanitized()
441 mu.Unlock()
442 close(shutdown)
443 })
444 }
445
446 // Initial task: run before processing the inbox so the agent
447 // "starts working" the moment it boots, matching what users
448 // expect from `/subagents new <task>`.
449 if args.Prompt != "" {
450 launchTurn(args.Prompt, false, "")
451 }
452
453 // Inbox loop: one supervisor message at a time. We don't spawn
454 // a goroutine per turn because runOne already serialises them
455 // via the busyTurn flag; doing the dispatch on the main
456 // goroutine keeps the daemon's lifecycle easy to follow.
457 for {
458 select {
459 case <-ctx.Done():
460 waitForActiveTurn(true)
461 em.emit("agent.exited", map[string]any{"reason": "cancelled"})
462 return ctx.Err()
463 case <-shutdown:
464 // Shutdown is an explicit supervisor stop. Cancel an active
465 // turn before joining it so a detached worker cannot remain
466 // alive indefinitely while the host waits for its socket to
467 // disappear.
468 waitForActiveTurn(true)
469 em.emit("agent.exited", shutdownExitPayload(shutdownOrigin))
470 return nil
471 case msg, ok := <-ln.Lines():
472 if !ok {
473 waitForActiveTurn(true)
474 em.emit("agent.exited", map[string]any{"reason": "inbox-closed"})
475 return nil
476 }
477 command, parseErr := subagents.ParseCommand(msg)
478 if parseErr != nil {
479 em.emit("error", map[string]any{"message": "invalid supervisor command"})
480 continue
481 }
482 switch command.Type {
483 case subagents.CommandAgentShutdown:
484 var payload subagents.AgentShutdownPayload
485 if err := command.DecodePayload(&payload); err != nil {
486 em.emit("error", map[string]any{"message": "invalid agent.shutdown payload"})
487 continue
488 }
489 requestShutdown(payload.Origin)
490 case subagents.CommandTurnCancel:
491 mu.Lock()
492 if cancelFn != nil {
493 cancelFn()
494 }
495 mu.Unlock()
496 case subagents.CommandAgentPing:
497 em.emit("agent.heartbeat", map[string]any{"activity": "alive"})
498 case subagents.CommandTurnStart:
499 var payload subagents.TurnStartPayload
500 if err := command.DecodePayload(&payload); err != nil {
501 em.emit("error", map[string]any{"code": "turn_rejected", "reason": "invalid_payload", "command_id": command.MessageID, "message": "invalid turn.start payload"})
502 continue
503 }
504 mu.Lock()
505 _, duplicate := seenCommands[command.MessageID]
506 mu.Unlock()
507 if duplicate && command.MessageID != "" {
508 continue
509 }
510 if launchTurn(payload.Prompt, payload.NewRun, command.MessageID) && command.MessageID != "" {
511 mu.Lock()
512 seenCommands[command.MessageID] = struct{}{}
513 mu.Unlock()
514 }
515 default:
516 // Unknown commands are rejected rather than interpreted as
517 // child-originated control requests.
518 em.emit("error", map[string]any{"message": "unknown supervisor command"})
519 }
520 }
521 }
522}
523
524// subagentEmitter serialises events to stdout and (optionally) to a
525// durable log file. Concurrent goroutines call emit so we have to
526// hold a mutex around the encoder.
527type subagentEmitter struct {
528 mu sync.Mutex
529 w *os.File
530 mirror *subagents.EventLog
531 agentID string
532
533 // orphan flips true the first time a stdout write fails (broken
534 // pipe — the supervisor died). Until then the mirror stays
535 // dormant: the supervisor is the canonical writer to events.jsonl
536 // (it parses our stdout and Append()s each event itself). Writing
537 // from both sides used to land every event in the log twice,
538 // which showed up as a fully-duplicated transcript the next time
539 // the agent was reloaded — the exact "why is everything doubled"
540 // bug. Once orphaned the mirror takes over so the events still
541 // land on disk for the next reload.
542 orphan bool
543}
544
545func newSubagentEmitter(w *os.File, mirror *subagents.EventLog) *subagentEmitter {
546 return &subagentEmitter{w: w, mirror: mirror}
547}
548
549func (e *subagentEmitter) setProtocolIdentity(agentID string) {
550 e.mu.Lock()
551 e.agentID = agentID
552 e.mu.Unlock()
553}
554
555func (e *subagentEmitter) emit(typ string, data map[string]any) {
556 if data == nil {
557 data = map[string]any{}
558 }
559 payload := make(map[string]any, len(data))
560 for key, value := range data {
561 payload[key] = value
562 }
563 turnID, _ := payload["turn_id"].(string)
564
565 e.mu.Lock()
566 defer e.mu.Unlock()
567
568 wireType := canonicalWorkerEvent(typ)
569 envelope := subagents.NewEventEnvelope(wireType, e.agentID, turnID, payload)
570 line, err := subagents.MarshalJSONL(envelope)
571
572 if !e.orphan && err == nil {
573 if _, werr := e.w.Write(line); werr != nil {
574 // Supervisor's stdout pipe is gone (parent zut exited but we
575 // kept running). Switch to mirror-only mode and preserve this
576 // event in the durable log.
577 e.orphan = true
578 }
579 }
580
581 if e.orphan && e.mirror != nil {
582 _ = e.mirror.Append(subagents.NewEvent(wireType, payload))
583 }
584}
585
586func subagentEventData(ev core.AgentEvent) map[string]any {
587 data := modes.EventToJSON(ev)
588 if turnEnd, ok := ev.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
589 data["error"] = subagentContextLimitMessage
590 }
591 return data
592}
593
594func canonicalWorkerEvent(typ string) string {
595 switch typ {
596 case "agent_ready":
597 return subagents.EventAgentReady
598 case "agent_stopped":
599 return subagents.EventAgentExited
600 case "turn_start":
601 return subagents.EventTurnStarted
602 case "turn_progress":
603 return subagents.EventTurnProgress
604 case "tool_call":
605 return subagents.EventToolStarted
606 case "tool_result":
607 return subagents.EventToolFinished
608 case "text_delta":
609 return subagents.EventMessageDelta
610 default:
611 return typ
612 }
613}
614
615const subagentContextLimitMessage = "request exceeds the model context window; narrow the task or reduce gathered context"
616
617func loadSubagentWorkerConfig() (Config, error) {
618 cfg, err := LoadConfig()
619 if err != nil {
620 return Config{}, fmt.Errorf("load subagent worker config: %w", err)
621 }
622 return cfg, nil
623}
624
625// workerTranscriptState is restored when an atomic post-compaction checkpoint
626// cannot be persisted.
627type workerTranscriptState struct {
628 messages []provider.Message
629 cost provider.Usage
630 lastTurn provider.Usage
631}
632
633func persistWorkerTranscript(ctx context.Context, ag *core.Agent, sess *core.Session, from int, compacted bool, baseline workerTranscriptState) error {
634 if sess == nil || ag == nil {
635 return nil
636 }
637 if !compacted {
638 if err := ctx.Err(); err != nil {
639 return err
640 }
641 if err := WriteNewTranscript(ag, sess, from); err != nil {
642 return fmt.Errorf("persist subagent transcript: %w", err)
643 }
644 return nil
645 }
646
647 rollback := func() {
648 ag.SetMessages(baseline.messages)
649 ag.SeedCost(baseline.cost)
650 ag.SeedLastTurnUsage(baseline.lastTurn)
651 }
652 if err := ctx.Err(); err != nil {
653 rollback()
654 return err
655 }
656 if err := sess.AppendCompactionWithUsage(ag.Messages(), ag.Cost()); err != nil {
657 rollback()
658 return fmt.Errorf("persist compacted subagent transcript: %w", err)
659 }
660 return nil
661}
662
663// promptWithContextRecovery preserves the worker's existing helper boundary
664// while delegating proactive compaction and one-shot overflow recovery to
665// modes. The result includes both the output index and compaction state.
666func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), options modes.ContextRecoveryOptions) (modes.ContextRecoveryResult, error) {
667 return modes.PromptWithContextRecovery(ctx, ag, prompt, nil, sink, options)
668}
669
670func finalAssistantText(messages []provider.Message) string {
671 var out strings.Builder
672 for _, message := range messages {
673 if message.Role != provider.RoleAssistant {
674 continue
675 }
676 for _, block := range message.Content {
677 if text, ok := block.(provider.TextBlock); ok {
678 if out.Len() > 0 {
679 out.WriteByte('\n')
680 }
681 out.WriteString(text.Text)
682 }
683 }
684 }
685 return out.String()
686}
687
688func firstLine(text string) string {
689 return strings.Split(strings.TrimSpace(text), "\n")[0]
690}
691
692func boundedResultSummary(text string) string {
693 return boundedResultOutput(firstLine(text), 4*1024, 1)
694}
695
696func shutdownExitPayload(origin subagents.ShutdownOrigin) map[string]any {
697 payload := map[string]any{"reason": "shutdown"}
698 if origin = origin.Sanitized(); origin != "" {
699 payload["origin"] = origin
700 }
701 return payload
702}
703
704func workerOutputLimits() (maxBytes, maxLines int) {
705 maxBytes, maxLines = 500_000, 5_000
706 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {
707 maxBytes = value
708 }
709 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_LINES"))); err == nil && value > 0 {
710 maxLines = value
711 }
712 return maxBytes, maxLines
713}
714
715func boundedResultOutput(text string, maxBytes, maxLines int) string {
716 const marker = "...[output truncated]"
717 lines := strings.Split(text, "\n")
718 if maxLines > 0 && len(lines) > maxLines {
719 if maxLines == 1 {
720 text = marker
721 } else {
722 text = strings.Join(append(lines[:maxLines-1], marker), "\n")
723 }
724 }
725 if maxBytes > 0 && len([]byte(text)) > maxBytes {
726 if maxBytes <= len(marker) {
727 return truncateOutputUTF8(text, maxBytes)
728 }
729 return truncateOutputUTF8(text, maxBytes-len(marker)) + marker
730 }
731 return text
732}
733
734func truncateOutputUTF8(value string, maxBytes int) string {
735 if maxBytes <= 0 {
736 return ""
737 }
738 data := []byte(value)
739 if len(data) <= maxBytes {
740 return value
741 }
742 data = data[:maxBytes]
743 for len(data) > 0 && !utf8.Valid(data) {
744 data = data[:len(data)-1]
745 }
746 return string(data)
747}
748
749func resultErrorPayload(err error, shutdownOrigin subagents.ShutdownOrigin) map[string]any {
750 if err == nil {
751 return nil
752 }
753 if provider.IsContextOverflowError(err) {
754 return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
755 }
756 if errors.Is(err, context.DeadlineExceeded) || shutdownOrigin == subagents.ShutdownOriginDeadline {
757 return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}
758 }
759 if errors.Is(err, context.Canceled) {
760 if resultErr := subagents.ShutdownResultError(shutdownOrigin); resultErr != nil {
761 return map[string]any{"code": resultErr.Code, "message": resultErr.Message}
762 }
763 }
764 return map[string]any{"code": "turn_failed", "message": truncateForLog(err.Error(), 500)}
765}
766
767func turnErrorMessage(err error, shutdownOrigin subagents.ShutdownOrigin) string {
768 if provider.IsContextOverflowError(err) {
769 return subagentContextLimitMessage
770 }
771 if err != nil {
772 if resultErr := subagents.ShutdownResultError(shutdownOrigin); resultErr != nil {
773 return resultErr.Message
774 }
775 }
776 return errString(err)
777}
778
779func errString(err error) string {
780 if err == nil {
781 return ""
782 }
783 return err.Error()
784}
785
786func initialTurnNumber(path string) int {
787 if strings.TrimSpace(path) == "" {
788 return 0
789 }
790 events, err := subagents.ReadEventLog(path)
791 if err != nil {
792 return 0
793 }
794 maxStep := 0
795 for _, ev := range events {
796 if !subagents.IsDelegatedTurnStart(ev) {
797 continue
798 }
799 if step, ok := ev.Data["step"].(float64); ok && int(step) > maxStep {
800 maxStep = int(step)
801 }
802 if strings.HasPrefix(ev.TurnID, "turn-") {
803 if n, parseErr := strconv.Atoi(strings.TrimPrefix(ev.TurnID, "turn-")); parseErr == nil && n > maxStep {
804 maxStep = n
805 }
806 }
807 }
808 return maxStep
809}
810
811func truncateForLog(s string, n int) string {
812 if len(s) <= n {
813 return s
814 }
815 if n <= 3 {
816 return strings.Repeat(".", n)
817 }
818 return s[:n-3] + "..."
819}