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 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 mu.Lock()
358 resultShutdownOrigin := shutdownOrigin.Sanitized()
359 mu.Unlock()
360 switch {
361 case errors.Is(err, context.DeadlineExceeded):
362 resultShutdownOrigin = subagents.ShutdownOriginDeadline
363 case !errors.Is(err, context.Canceled):
364 resultShutdownOrigin = ""
365 }
366 resultPayload := map[string]any{
367 "status": resultStatus,
368 "turn_id": turnID,
369 "summary": boundedResultSummary(output),
370 "output": boundedResultOutput(output, maxOutputBytes, maxOutputLines),
371 "error": resultErrorPayload(err, resultShutdownOrigin),
372 }
373 if resultShutdownOrigin != "" {
374 resultPayload["shutdown_origin"] = resultShutdownOrigin
375 }
376 em.emit("turn.result", resultPayload)
377 if err != nil {
378 em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": resultErrorPayload(err, resultShutdownOrigin)})
379 }
380 em.emit("turn_end", map[string]any{
381 "step": step,
382 "turn_id": turnID,
383 "error": turnErrorMessage(err, resultShutdownOrigin),
384 })
385
386 cancel()
387 mu.Lock()
388 busyTurn = false
389 cancelFn = nil
390 mu.Unlock()
391 // The supervisor treats agent.idle as permission to start a follow-up.
392 // Publish it only after the busy gate has opened, otherwise a manager
393 // can send a turn that the inbox loop drops as still busy.
394 em.emit("agent.idle", map[string]any{"turn_id": turnID, "lifetime_turns": lifetime, "current_run_turns": currentRun})
395 }
396
397 launchTurn := func(prompt string, newRun bool, commandID string) bool {
398 mu.Lock()
399 if closing || busyTurn || turnPending {
400 mu.Unlock()
401 em.emit("error", map[string]any{"code": "turn_rejected", "reason": "busy", "command_id": commandID, "message": "worker is busy"})
402 return false
403 }
404 // Reserve the worker before launching the goroutine. A second command
405 // cannot be accepted in the gap before runOne acquires mu.
406 turnPending = true
407 turnWG.Add(1)
408 mu.Unlock()
409 go func() {
410 defer turnWG.Done()
411 runOne(prompt, newRun, commandID)
412 }()
413 return true
414 }
415
416 waitForActiveTurn := func(cancel bool) {
417 mu.Lock()
418 closing = true
419 activeCancel := cancelFn
420 mu.Unlock()
421 if cancel && activeCancel != nil {
422 activeCancel()
423 }
424 turnWG.Wait()
425 }
426
427 requestShutdown := func(origin subagents.ShutdownOrigin) {
428 shutdownOnce.Do(func() {
429 mu.Lock()
430 closing = true
431 shutdownOrigin = origin.Sanitized()
432 mu.Unlock()
433 close(shutdown)
434 })
435 }
436
437 // Initial task: run before processing the inbox so the agent
438 // "starts working" the moment it boots, matching what users
439 // expect from `/subagents new <task>`.
440 if args.Prompt != "" {
441 launchTurn(args.Prompt, false, "")
442 }
443
444 // Inbox loop: one supervisor message at a time. We don't spawn
445 // a goroutine per turn because runOne already serialises them
446 // via the busyTurn flag; doing the dispatch on the main
447 // goroutine keeps the daemon's lifecycle easy to follow.
448 for {
449 select {
450 case <-ctx.Done():
451 waitForActiveTurn(true)
452 em.emit("agent.exited", map[string]any{"reason": "cancelled"})
453 return ctx.Err()
454 case <-shutdown:
455 // Shutdown is an explicit supervisor stop. Cancel an active
456 // turn before joining it so a detached worker cannot remain
457 // alive indefinitely while the host waits for its socket to
458 // disappear.
459 waitForActiveTurn(true)
460 em.emit("agent.exited", shutdownExitPayload(shutdownOrigin))
461 return nil
462 case msg, ok := <-ln.Lines():
463 if !ok {
464 waitForActiveTurn(true)
465 em.emit("agent.exited", map[string]any{"reason": "inbox-closed"})
466 return nil
467 }
468 command, parseErr := subagents.ParseCommand(msg)
469 if parseErr != nil {
470 em.emit("error", map[string]any{"message": "invalid supervisor command"})
471 continue
472 }
473 switch command.Type {
474 case subagents.CommandAgentShutdown:
475 var payload subagents.AgentShutdownPayload
476 if err := command.DecodePayload(&payload); err != nil {
477 em.emit("error", map[string]any{"message": "invalid agent.shutdown payload"})
478 continue
479 }
480 requestShutdown(payload.Origin)
481 case subagents.CommandTurnCancel:
482 mu.Lock()
483 if cancelFn != nil {
484 cancelFn()
485 }
486 mu.Unlock()
487 case subagents.CommandAgentPing:
488 em.emit("agent.heartbeat", map[string]any{"activity": "alive"})
489 case subagents.CommandTurnStart:
490 var payload subagents.TurnStartPayload
491 if err := command.DecodePayload(&payload); err != nil {
492 em.emit("error", map[string]any{"code": "turn_rejected", "reason": "invalid_payload", "command_id": command.MessageID, "message": "invalid turn.start payload"})
493 continue
494 }
495 mu.Lock()
496 _, duplicate := seenCommands[command.MessageID]
497 mu.Unlock()
498 if duplicate && command.MessageID != "" {
499 continue
500 }
501 if launchTurn(payload.Prompt, payload.NewRun, command.MessageID) && command.MessageID != "" {
502 mu.Lock()
503 seenCommands[command.MessageID] = struct{}{}
504 mu.Unlock()
505 }
506 default:
507 // Unknown commands are rejected rather than interpreted as
508 // child-originated control requests.
509 em.emit("error", map[string]any{"message": "unknown supervisor command"})
510 }
511 }
512 }
513}
514
515// subagentEmitter serialises events to stdout and (optionally) to a
516// durable log file. Concurrent goroutines call emit so we have to
517// hold a mutex around the encoder.
518type subagentEmitter struct {
519 mu sync.Mutex
520 w *os.File
521 mirror *subagents.EventLog
522 agentID string
523
524 // orphan flips true the first time a stdout write fails (broken
525 // pipe — the supervisor died). Until then the mirror stays
526 // dormant: the supervisor is the canonical writer to events.jsonl
527 // (it parses our stdout and Append()s each event itself). Writing
528 // from both sides used to land every event in the log twice,
529 // which showed up as a fully-duplicated transcript the next time
530 // the agent was reloaded — the exact "why is everything doubled"
531 // bug. Once orphaned the mirror takes over so the events still
532 // land on disk for the next reload.
533 orphan bool
534}
535
536func newSubagentEmitter(w *os.File, mirror *subagents.EventLog) *subagentEmitter {
537 return &subagentEmitter{w: w, mirror: mirror}
538}
539
540func (e *subagentEmitter) setProtocolIdentity(agentID string) {
541 e.mu.Lock()
542 e.agentID = agentID
543 e.mu.Unlock()
544}
545
546func (e *subagentEmitter) emit(typ string, data map[string]any) {
547 if data == nil {
548 data = map[string]any{}
549 }
550 payload := make(map[string]any, len(data))
551 for key, value := range data {
552 payload[key] = value
553 }
554 turnID, _ := payload["turn_id"].(string)
555
556 e.mu.Lock()
557 defer e.mu.Unlock()
558
559 wireType := canonicalWorkerEvent(typ)
560 envelope := subagents.NewEventEnvelope(wireType, e.agentID, turnID, payload)
561 line, err := subagents.MarshalJSONL(envelope)
562
563 if !e.orphan && err == nil {
564 if _, werr := e.w.Write(line); werr != nil {
565 // Supervisor's stdout pipe is gone (parent zut exited but we
566 // kept running). Switch to mirror-only mode and preserve this
567 // event in the durable log.
568 e.orphan = true
569 }
570 }
571
572 if e.orphan && e.mirror != nil {
573 _ = e.mirror.Append(subagents.NewEvent(wireType, payload))
574 }
575}
576
577func subagentEventData(ev core.AgentEvent) map[string]any {
578 data := modes.EventToJSON(ev)
579 if turnEnd, ok := ev.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
580 data["error"] = subagentContextLimitMessage
581 }
582 return data
583}
584
585func canonicalWorkerEvent(typ string) string {
586 switch typ {
587 case "agent_ready":
588 return subagents.EventAgentReady
589 case "agent_stopped":
590 return subagents.EventAgentExited
591 case "turn_start":
592 return subagents.EventTurnStarted
593 case "turn_progress":
594 return subagents.EventTurnProgress
595 case "tool_call":
596 return subagents.EventToolStarted
597 case "tool_result":
598 return subagents.EventToolFinished
599 case "text_delta":
600 return subagents.EventMessageDelta
601 default:
602 return typ
603 }
604}
605
606const subagentContextLimitMessage = "request exceeds the model context window; narrow the task or reduce gathered context"
607
608func loadSubagentWorkerConfig() (Config, error) {
609 cfg, err := LoadConfig()
610 if err != nil {
611 return Config{}, fmt.Errorf("load subagent worker config: %w", err)
612 }
613 return cfg, nil
614}
615
616// workerTranscriptState is restored when an atomic post-compaction checkpoint
617// cannot be persisted.
618type workerTranscriptState struct {
619 messages []provider.Message
620 cost provider.Usage
621 lastTurn provider.Usage
622}
623
624func persistWorkerTranscript(ctx context.Context, ag *core.Agent, sess *core.Session, from int, compacted bool, baseline workerTranscriptState) error {
625 if sess == nil || ag == nil {
626 return nil
627 }
628 if !compacted {
629 if err := ctx.Err(); err != nil {
630 return err
631 }
632 if err := WriteNewTranscript(ag, sess, from); err != nil {
633 return fmt.Errorf("persist subagent transcript: %w", err)
634 }
635 return nil
636 }
637
638 rollback := func() {
639 ag.SetMessages(baseline.messages)
640 ag.SeedCost(baseline.cost)
641 ag.SeedLastTurnUsage(baseline.lastTurn)
642 }
643 if err := ctx.Err(); err != nil {
644 rollback()
645 return err
646 }
647 if err := sess.AppendCompactionWithUsage(ag.Messages(), ag.Cost()); err != nil {
648 rollback()
649 return fmt.Errorf("persist compacted subagent transcript: %w", err)
650 }
651 return nil
652}
653
654// promptWithContextRecovery preserves the worker's existing helper boundary
655// while delegating proactive compaction and one-shot overflow recovery to
656// modes. The result includes both the output index and compaction state.
657func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), options modes.ContextRecoveryOptions) (modes.ContextRecoveryResult, error) {
658 return modes.PromptWithContextRecovery(ctx, ag, prompt, nil, sink, options)
659}
660
661func finalAssistantText(messages []provider.Message) string {
662 var out strings.Builder
663 for _, message := range messages {
664 if message.Role != provider.RoleAssistant {
665 continue
666 }
667 for _, block := range message.Content {
668 if text, ok := block.(provider.TextBlock); ok {
669 if out.Len() > 0 {
670 out.WriteByte('\n')
671 }
672 out.WriteString(text.Text)
673 }
674 }
675 }
676 return out.String()
677}
678
679func firstLine(text string) string {
680 return strings.Split(strings.TrimSpace(text), "\n")[0]
681}
682
683func boundedResultSummary(text string) string {
684 return boundedResultOutput(firstLine(text), 4*1024, 1)
685}
686
687func shutdownExitPayload(origin subagents.ShutdownOrigin) map[string]any {
688 payload := map[string]any{"reason": "shutdown"}
689 if origin = origin.Sanitized(); origin != "" {
690 payload["origin"] = origin
691 }
692 return payload
693}
694
695func workerOutputLimits() (maxBytes, maxLines int) {
696 maxBytes, maxLines = 500_000, 5_000
697 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {
698 maxBytes = value
699 }
700 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_LINES"))); err == nil && value > 0 {
701 maxLines = value
702 }
703 return maxBytes, maxLines
704}
705
706func boundedResultOutput(text string, maxBytes, maxLines int) string {
707 const marker = "...[output truncated]"
708 lines := strings.Split(text, "\n")
709 if maxLines > 0 && len(lines) > maxLines {
710 if maxLines == 1 {
711 text = marker
712 } else {
713 text = strings.Join(append(lines[:maxLines-1], marker), "\n")
714 }
715 }
716 if maxBytes > 0 && len([]byte(text)) > maxBytes {
717 if maxBytes <= len(marker) {
718 return truncateOutputUTF8(text, maxBytes)
719 }
720 return truncateOutputUTF8(text, maxBytes-len(marker)) + marker
721 }
722 return text
723}
724
725func truncateOutputUTF8(value string, maxBytes int) string {
726 if maxBytes <= 0 {
727 return ""
728 }
729 data := []byte(value)
730 if len(data) <= maxBytes {
731 return value
732 }
733 data = data[:maxBytes]
734 for len(data) > 0 && !utf8.Valid(data) {
735 data = data[:len(data)-1]
736 }
737 return string(data)
738}
739
740func resultErrorPayload(err error, shutdownOrigin subagents.ShutdownOrigin) map[string]any {
741 if err == nil {
742 return nil
743 }
744 if provider.IsContextOverflowError(err) {
745 return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
746 }
747 if errors.Is(err, context.DeadlineExceeded) || shutdownOrigin == subagents.ShutdownOriginDeadline {
748 return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}
749 }
750 if errors.Is(err, context.Canceled) {
751 if resultErr := subagents.ShutdownResultError(shutdownOrigin); resultErr != nil {
752 return map[string]any{"code": resultErr.Code, "message": resultErr.Message}
753 }
754 }
755 return map[string]any{"code": "turn_failed", "message": truncateForLog(err.Error(), 500)}
756}
757
758func turnErrorMessage(err error, shutdownOrigin subagents.ShutdownOrigin) string {
759 if provider.IsContextOverflowError(err) {
760 return subagentContextLimitMessage
761 }
762 if err != nil {
763 if resultErr := subagents.ShutdownResultError(shutdownOrigin); resultErr != nil {
764 return resultErr.Message
765 }
766 }
767 return errString(err)
768}
769
770func errString(err error) string {
771 if err == nil {
772 return ""
773 }
774 return err.Error()
775}
776
777func initialTurnNumber(path string) int {
778 if strings.TrimSpace(path) == "" {
779 return 0
780 }
781 events, err := subagents.ReadEventLog(path)
782 if err != nil {
783 return 0
784 }
785 maxStep := 0
786 for _, ev := range events {
787 if !subagents.IsDelegatedTurnStart(ev) {
788 continue
789 }
790 if step, ok := ev.Data["step"].(float64); ok && int(step) > maxStep {
791 maxStep = int(step)
792 }
793 if strings.HasPrefix(ev.TurnID, "turn-") {
794 if n, parseErr := strconv.Atoi(strings.TrimPrefix(ev.TurnID, "turn-")); parseErr == nil && n > maxStep {
795 maxStep = n
796 }
797 }
798 }
799 return maxStep
800}
801
802func truncateForLog(s string, n int) string {
803 if len(s) <= n {
804 return s
805 }
806 if n <= 3 {
807 return strings.Repeat(".", n)
808 }
809 return s[:n-3] + "..."
810}