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