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 goes to the parent supervisor, which
31// owns the optional execution trace. Workers never persist trace data.
32//
33// The runner in packages/agent/subagents/runner.go is the only caller in
34// production; tests use the stubchild binary under
35// packages/agent/subagents/testdata/cmd/stubchild instead of the real model
36// loop.
37// workerTurnCounters track stable delegated-turn identity and observability.
38// Retries, compaction, and provider/tool loops remain inside one message turn.
39type workerTurnCounters struct {
40 sequence int
41 lifetime int
42 current int
43}
44
45type sessionPersistenceState struct {
46 mu sync.Mutex
47 errValue error
48}
49
50func (s *sessionPersistenceState) record(err error) {
51 if err == nil {
52 return
53 }
54 s.mu.Lock()
55 if s.errValue == nil {
56 s.errValue = err
57 }
58 s.mu.Unlock()
59}
60
61func (s *sessionPersistenceState) failed() bool {
62 s.mu.Lock()
63 defer s.mu.Unlock()
64 return s.errValue != nil
65}
66
67func (s *sessionPersistenceState) err() error {
68 s.mu.Lock()
69 defer s.mu.Unlock()
70 return s.errValue
71}
72
73func (b *workerTurnCounters) start(newRun bool) (step, lifetime, current int) {
74 if newRun {
75 b.current = 0
76 }
77 b.sequence++
78 b.lifetime++
79 b.current++
80 return b.sequence, b.lifetime, b.current
81}
82
83func subagentTurnContext(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) {
84 if timeout > 0 {
85 return context.WithTimeout(parent, timeout)
86 }
87 return context.WithCancel(parent)
88}
89
90func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runErr error) {
91 if args.SubagentWorker == "" {
92 return fmt.Errorf("--subagent-worker requires a socket path")
93 }
94 if os.Getenv("ZUT_SUBAGENT_CREDENTIAL_STDIN") == "1" {
95 var inherited subagents.Credential
96 dec := json.NewDecoder(io.LimitReader(os.Stdin, 1<<20))
97 if err := dec.Decode(&inherited); err != nil {
98 return fmt.Errorf("read inherited subagent credential: %w", err)
99 }
100 if inherited.Value == "" || inherited.Method == "" {
101 return fmt.Errorf("read inherited subagent credential: missing value or method")
102 }
103 args.inheritedCredential = inherited.Value
104 args.inheritedAuthMethod = inherited.Method
105 args.inheritedAccountID = inherited.AccountID
106 }
107
108 cfg, err := loadSubagentWorkerConfig()
109 if err != nil {
110 return err
111 }
112 r, err := Resolve(args, true)
113 if err != nil {
114 return err
115 }
116 // A delegated worker has one bounded task and must not inherit arbitrary
117 // local extensions, their tools, hooks, or persisted extension state.
118 // Extensions remain available to explicitly selected skills only through
119 // the parent process; the worker runs with its resolved built-in registry.
120 ag := r.NewAgent()
121 initialAg := ag
122 defer func() {
123 closeAgentLSP(ag)
124 if ag != initialAg {
125 closeAgentLSP(initialAg)
126 }
127 }()
128 sess, err := openOrCreateSession(ctx, args, r, ag, version)
129 if err != nil {
130 return err
131 }
132 var sessionPersistence sessionPersistenceState
133 if sess != nil {
134 var providerName, model string
135 sess, ag, providerName, model, err = applyInitialSessionResume(ctx, args, r, nil, sess, ag)
136 if err != nil {
137 return err
138 }
139 r.Provider, r.Model = providerName, model
140 // A prompt is acknowledged by the supervisor only after this callback
141 // has appended and synced the matching user message. This closes the
142 // crash window between accepting a follow-up and writing session.json.
143 ag.OnMessageAppended = func(message provider.Message) {
144 if err := sess.AppendMessage(message); err != nil {
145 sessionPersistence.record(err)
146 return
147 }
148 if err := sess.Sync(); err != nil {
149 sessionPersistence.record(err)
150 }
151 }
152 ag.OnUsage = func(usage provider.Usage) {
153 if err := sess.AppendUsage(usage, usage); err != nil {
154 sessionPersistence.record(err)
155 }
156 }
157 ag.OnRetryLifecycle = sess.RecordRetryLifecycle
158 defer func() {
159 runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost()))
160 runErr = joinSessionCloseError(runErr, sess)
161 }()
162 }
163 // Resolve retains metadata for catalog models and for valid synthesized
164 // local/routed/open-catalog models. The active agent also survives a session
165 // resume that rebuilt it for a different provider/model pair.
166 contextWindow := ag.ContextWindow
167
168 // Open the inbox listener BEFORE emitting agent_ready so the
169 // supervisor can dial through on the very first send. The
170 // subagent supervisor's Inbox retries dialing for a
171 // short window, but emitting ready first and then listening
172 // would still race in tight loops.
173 ln, err := subagents.Listen(args.SubagentWorker)
174 if err != nil {
175 return fmt.Errorf("subagent-worker listen: %w", err)
176 }
177 defer ln.Close()
178
179 em := newSubagentEmitter(os.Stdout)
180 em.setProtocolIdentity(os.Getenv("ZUT_SUBAGENT_AGENT_ID"))
181 em.emit("agent.ready", map[string]any{
182 "version": version,
183 "cwd": r.CWD,
184 "model": r.Model,
185 "lifetime_turns": args.SubagentLifetimeTurns,
186 "current_run_turns": args.SubagentRunTurns,
187 })
188
189 // Keep a per-turn cancel so the "cancel" inbox message can interrupt
190 // an in-flight turn without tearing down the whole daemon.
191 var (
192 mu sync.Mutex
193 cancelFn context.CancelFunc
194 busyTurn bool
195 counters = workerTurnCounters{
196 sequence: args.SubagentLifetimeTurns,
197 lifetime: args.SubagentLifetimeTurns,
198 current: args.SubagentRunTurns,
199 }
200 turnWG sync.WaitGroup
201 closing bool
202 turnPending bool
203 seenCommands = make(map[string]struct{})
204 shutdown = make(chan struct{})
205 shutdownOnce sync.Once
206 shutdownOrigin subagents.ShutdownOrigin
207 )
208 maxOutputBytes, maxOutputLines := workerOutputLimits()
209
210 runOne := func(prompt string, newRun bool, commandID string) {
211 mu.Lock()
212 // launchTurn admits the goroutine to turnWG before it starts so
213 // shutdown can join it. A shutdown can win the scheduling race
214 // after that admission but before this goroutine acquires mu; in
215 // that case do not start a new Prompt after waitForActiveTurn has
216 // declared the worker closing.
217 if closing || busyTurn {
218 mu.Unlock()
219 em.emit("error", map[string]any{"code": "turn_rejected", "reason": "busy", "command_id": commandID, "message": "worker is busy"})
220 return
221 }
222 busyTurn = true
223 turnPending = false
224 step, lifetime, currentRun := counters.start(newRun)
225 turnID := fmt.Sprintf("turn-%d", step)
226 em.setTurnID(turnID)
227 c, cancel := subagentTurnContext(ctx, args.SubagentTurnTimeout)
228 cancelFn = cancel
229 mu.Unlock()
230
231 turnStarted := false
232 sink := func(ev core.AgentEvent) {
233 data := subagentEventData(ev)
234 if ev.Type() == "user_message" && !turnStarted {
235 // Prompt invokes OnMessageAppended before delivering this sink
236 // event, so the session row is durable before the lifecycle
237 // acknowledgement is written.
238 em.emit(ev.Type(), data)
239 if !sessionPersistence.failed() {
240 em.emit("turn.started", map[string]any{"step": step, "turn_id": turnID, "command_id": commandID, "lifetime_turns": lifetime, "current_run_turns": currentRun})
241 turnStarted = true
242 }
243 return
244 }
245 if ev.Type() == "turn_start" {
246 // A provider/model-loop turn is nested inside the one delegated
247 // message turn admitted above. Mark it on the wire so the
248 // supervisor cannot mistake it for a new worker turn.
249 data["nested_turn"] = true
250 }
251 em.emit(ev.Type(), data)
252 }
253
254 baseline := workerTranscriptState{
255 messages: ag.Messages(),
256 cost: ag.Cost(),
257 lastTurn: ag.LastTurnUsage(),
258 }
259 var persistCompaction func([]provider.Message) error
260 if sess != nil {
261 // Buffer the checkpoint until the continued turn completes. The final
262 // transcript and cumulative usage are then persisted in one JSONL row.
263 persistCompaction = func([]provider.Message) error {
264 return c.Err()
265 }
266 }
267 recovery, err := promptWithContextRecovery(c, ag, prompt, sink, modes.ContextRecoveryOptions{
268 PersistCompaction: persistCompaction,
269 ContextWindow: contextWindow,
270 AutoCompactThreshold: cfg.AutoCompactThreshold,
271 })
272 output := finalAssistantText(ag.Messages()[recovery.OutputStart:])
273 // Message and usage callbacks persist ordinary turns incrementally. A
274 // compacted turn still needs one atomic replacement checkpoint so the
275 // durable transcript cannot mix generations.
276 if recovery.Compacted || ag.OnMessageAppended == nil {
277 if persistErr := persistWorkerTranscript(c, ag, sess, recovery.OutputStart, recovery.Compacted, baseline); persistErr != nil {
278 if err != nil {
279 err = errors.Join(err, persistErr)
280 } else {
281 err = persistErr
282 }
283 }
284 }
285 if sess != nil {
286 if persistErr := sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost()); persistErr != nil {
287 err = errors.Join(err, persistErr)
288 }
289 }
290 if sessionErr := sessionPersistence.err(); err == nil && sessionErr != nil {
291 err = sessionErr
292 }
293
294 resultStatus := "succeeded"
295 if err != nil {
296 if errors.Is(err, context.Canceled) {
297 resultStatus = "canceled"
298 } else {
299 resultStatus = "failed"
300 }
301 }
302 mu.Lock()
303 resultShutdownOrigin := shutdownOrigin.Sanitized()
304 mu.Unlock()
305 switch {
306 case errors.Is(err, context.DeadlineExceeded):
307 resultShutdownOrigin = subagents.ShutdownOriginDeadline
308 case !errors.Is(err, context.Canceled):
309 resultShutdownOrigin = ""
310 }
311 resultPayload := map[string]any{
312 "status": resultStatus,
313 "turn_id": turnID,
314 "summary": boundedResultSummary(output),
315 "output": boundedResultOutput(output, maxOutputBytes, maxOutputLines),
316 "error": resultErrorPayload(err, resultShutdownOrigin),
317 }
318 if resultShutdownOrigin != "" {
319 resultPayload["shutdown_origin"] = resultShutdownOrigin
320 }
321 em.emit("turn.result", resultPayload)
322 if err != nil {
323 em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": resultErrorPayload(err, resultShutdownOrigin)})
324 }
325 em.emit("turn_end", map[string]any{
326 "step": step,
327 "turn_id": turnID,
328 "error": turnErrorMessage(err, resultShutdownOrigin),
329 })
330
331 cancel()
332 mu.Lock()
333 // Clear the emitter identity before reopening the busy gate: a newly
334 // admitted turn must not race this completed turn's cleanup.
335 em.setTurnID("")
336 busyTurn = false
337 cancelFn = nil
338 mu.Unlock()
339 // The supervisor treats agent.idle as permission to start a follow-up.
340 // Publish it only after the busy gate has opened, otherwise a manager
341 // can send a turn that the inbox loop drops as still busy.
342 em.emit("agent.idle", map[string]any{"turn_id": turnID, "lifetime_turns": lifetime, "current_run_turns": currentRun})
343 }
344
345 launchTurn := func(prompt string, newRun bool, commandID string) bool {
346 mu.Lock()
347 if closing || busyTurn || turnPending {
348 mu.Unlock()
349 em.emit("error", map[string]any{"code": "turn_rejected", "reason": "busy", "command_id": commandID, "message": "worker is busy"})
350 return false
351 }
352 // Reserve the worker before launching the goroutine. A second command
353 // cannot be accepted in the gap before runOne acquires mu.
354 turnPending = true
355 turnWG.Add(1)
356 mu.Unlock()
357 go func() {
358 defer turnWG.Done()
359 runOne(prompt, newRun, commandID)
360 }()
361 return true
362 }
363
364 waitForActiveTurn := func(cancel bool) {
365 mu.Lock()
366 closing = true
367 activeCancel := cancelFn
368 mu.Unlock()
369 if cancel && activeCancel != nil {
370 activeCancel()
371 }
372 turnWG.Wait()
373 }
374
375 requestShutdown := func(origin subagents.ShutdownOrigin) {
376 shutdownOnce.Do(func() {
377 mu.Lock()
378 closing = true
379 shutdownOrigin = origin.Sanitized()
380 mu.Unlock()
381 close(shutdown)
382 })
383 }
384
385 // Initial task: run before processing the inbox so the agent
386 // "starts working" the moment it boots, matching what users
387 // expect from `/subagents new <task>`.
388 if args.Prompt != "" {
389 launchTurn(args.Prompt, false, "")
390 }
391
392 // Inbox loop: one supervisor message at a time. We don't spawn
393 // a goroutine per turn because runOne already serialises them
394 // via the busyTurn flag; doing the dispatch on the main
395 // goroutine keeps the daemon's lifecycle easy to follow.
396 for {
397 select {
398 case <-ctx.Done():
399 waitForActiveTurn(true)
400 em.emit("agent.exited", map[string]any{"reason": "cancelled"})
401 return ctx.Err()
402 case <-shutdown:
403 // Shutdown is an explicit supervisor stop. Cancel an active
404 // turn before joining it so a detached worker cannot remain
405 // alive indefinitely while the host waits for its socket to
406 // disappear.
407 waitForActiveTurn(true)
408 em.emit("agent.exited", shutdownExitPayload(shutdownOrigin))
409 return nil
410 case msg, ok := <-ln.Lines():
411 if !ok {
412 waitForActiveTurn(true)
413 em.emit("agent.exited", map[string]any{"reason": "inbox-closed"})
414 return nil
415 }
416 command, parseErr := subagents.ParseCommand(msg)
417 if parseErr != nil {
418 em.emit("error", map[string]any{"message": "invalid supervisor command"})
419 continue
420 }
421 switch command.Type {
422 case subagents.CommandAgentShutdown:
423 var payload subagents.AgentShutdownPayload
424 if err := command.DecodePayload(&payload); err != nil {
425 em.emit("error", map[string]any{"message": "invalid agent.shutdown payload"})
426 continue
427 }
428 requestShutdown(payload.Origin)
429 case subagents.CommandTurnCancel:
430 mu.Lock()
431 if cancelFn != nil {
432 cancelFn()
433 }
434 mu.Unlock()
435 case subagents.CommandAgentPing:
436 em.emit("agent.heartbeat", map[string]any{"activity": "alive"})
437 case subagents.CommandTurnStart:
438 var payload subagents.TurnStartPayload
439 if err := command.DecodePayload(&payload); err != nil {
440 em.emit("error", map[string]any{"code": "turn_rejected", "reason": "invalid_payload", "command_id": command.MessageID, "message": "invalid turn.start payload"})
441 continue
442 }
443 mu.Lock()
444 _, duplicate := seenCommands[command.MessageID]
445 mu.Unlock()
446 if duplicate && command.MessageID != "" {
447 continue
448 }
449 if launchTurn(payload.Prompt, payload.NewRun, command.MessageID) && command.MessageID != "" {
450 mu.Lock()
451 seenCommands[command.MessageID] = struct{}{}
452 mu.Unlock()
453 }
454 default:
455 // Unknown commands are rejected rather than interpreted as
456 // child-originated control requests.
457 em.emit("error", map[string]any{"message": "unknown supervisor command"})
458 }
459 }
460 }
461}
462
463// subagentEmitter serialises protocol events to stdout. The supervisor is
464// the single trace writer; workers never persist execution events themselves.
465type subagentEmitter struct {
466 mu sync.Mutex
467 w *os.File
468 agentID string
469 turnID string
470}
471
472func newSubagentEmitter(w *os.File) *subagentEmitter {
473 return &subagentEmitter{w: w}
474}
475
476func (e *subagentEmitter) setProtocolIdentity(agentID string) {
477 e.mu.Lock()
478 e.agentID = agentID
479 e.mu.Unlock()
480}
481
482// setTurnID associates subsequent worker events with the currently admitted
483// delegated turn. Core events do not carry the worker protocol turn ID.
484func (e *subagentEmitter) setTurnID(turnID string) {
485 e.mu.Lock()
486 e.turnID = turnID
487 e.mu.Unlock()
488}
489
490func (e *subagentEmitter) emit(typ string, data map[string]any) {
491 if data == nil {
492 data = map[string]any{}
493 }
494 payload := make(map[string]any, len(data))
495 for key, value := range data {
496 payload[key] = value
497 }
498 turnID, _ := payload["turn_id"].(string)
499
500 e.mu.Lock()
501 defer e.mu.Unlock()
502 if turnID == "" {
503 turnID = e.turnID
504 }
505
506 wireType := canonicalWorkerEvent(typ)
507 envelope := subagents.NewEventEnvelope(wireType, e.agentID, turnID, payload)
508 line, err := subagents.MarshalJSONL(envelope)
509
510 if err == nil {
511 _, _ = e.w.Write(line)
512 }
513}
514
515func subagentEventData(ev core.AgentEvent) map[string]any {
516 data := modes.EventToJSON(ev)
517 if turnEnd, ok := ev.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
518 data["error"] = subagentContextLimitMessage
519 }
520 return data
521}
522
523func canonicalWorkerEvent(typ string) string {
524 switch typ {
525 case "agent_ready":
526 return subagents.EventAgentReady
527 case "agent_stopped":
528 return subagents.EventAgentExited
529 case "turn_start":
530 return subagents.EventTurnStarted
531 case "turn_progress":
532 return subagents.EventTurnProgress
533 case "tool_execution_started":
534 return subagents.EventToolStarted
535 case "tool_result":
536 return subagents.EventToolFinished
537 case "text_delta":
538 return subagents.EventMessageDelta
539 default:
540 return typ
541 }
542}
543
544const subagentContextLimitMessage = "request exceeds the model context window; narrow the task or reduce gathered context"
545
546func loadSubagentWorkerConfig() (Config, error) {
547 cfg, err := LoadConfig()
548 if err != nil {
549 return Config{}, fmt.Errorf("load subagent worker config: %w", err)
550 }
551 return cfg, nil
552}
553
554// workerTranscriptState is restored when an atomic post-compaction checkpoint
555// cannot be persisted.
556type workerTranscriptState struct {
557 messages []provider.Message
558 cost provider.Usage
559 lastTurn provider.Usage
560}
561
562func persistWorkerTranscript(ctx context.Context, ag *core.Agent, sess *core.Session, from int, compacted bool, baseline workerTranscriptState) error {
563 if sess == nil || ag == nil {
564 return nil
565 }
566 if !compacted {
567 if err := ctx.Err(); err != nil {
568 return err
569 }
570 if err := WriteNewTranscript(ag, sess, from); err != nil {
571 return fmt.Errorf("persist subagent transcript: %w", err)
572 }
573 return nil
574 }
575
576 rollback := func() {
577 ag.SetMessages(baseline.messages)
578 ag.SeedCost(baseline.cost)
579 ag.SeedLastTurnUsage(baseline.lastTurn)
580 }
581 if err := ctx.Err(); err != nil {
582 rollback()
583 return err
584 }
585 if err := sess.AppendCompactionWithUsage(ag.Messages(), ag.Cost()); err != nil {
586 rollback()
587 return fmt.Errorf("persist compacted subagent transcript: %w", err)
588 }
589 return nil
590}
591
592// promptWithContextRecovery preserves the worker's existing helper boundary
593// while delegating proactive compaction and one-shot overflow recovery to
594// modes. The result includes both the output index and compaction state.
595func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), options modes.ContextRecoveryOptions) (modes.ContextRecoveryResult, error) {
596 return modes.PromptWithContextRecovery(ctx, ag, prompt, nil, sink, options)
597}
598
599func finalAssistantText(messages []provider.Message) string {
600 var out strings.Builder
601 for _, message := range messages {
602 if message.Role != provider.RoleAssistant {
603 continue
604 }
605 for _, block := range message.Content {
606 if text, ok := block.(provider.TextBlock); ok {
607 if out.Len() > 0 {
608 out.WriteByte('\n')
609 }
610 out.WriteString(text.Text)
611 }
612 }
613 }
614 return out.String()
615}
616
617func firstLine(text string) string {
618 return strings.Split(strings.TrimSpace(text), "\n")[0]
619}
620
621func boundedResultSummary(text string) string {
622 return boundedResultOutput(firstLine(text), 4*1024, 1)
623}
624
625func shutdownExitPayload(origin subagents.ShutdownOrigin) map[string]any {
626 payload := map[string]any{"reason": "shutdown"}
627 if origin = origin.Sanitized(); origin != "" {
628 payload["origin"] = origin
629 }
630 return payload
631}
632
633func workerOutputLimits() (maxBytes, maxLines int) {
634 maxBytes, maxLines = 500_000, 5_000
635 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {
636 maxBytes = value
637 }
638 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_LINES"))); err == nil && value > 0 {
639 maxLines = value
640 }
641 return maxBytes, maxLines
642}
643
644func boundedResultOutput(text string, maxBytes, maxLines int) string {
645 const marker = "...[output truncated]"
646 lines := strings.Split(text, "\n")
647 if maxLines > 0 && len(lines) > maxLines {
648 if maxLines == 1 {
649 text = marker
650 } else {
651 text = strings.Join(append(lines[:maxLines-1], marker), "\n")
652 }
653 }
654 if maxBytes > 0 && len([]byte(text)) > maxBytes {
655 if maxBytes <= len(marker) {
656 return truncateOutputUTF8(text, maxBytes)
657 }
658 return truncateOutputUTF8(text, maxBytes-len(marker)) + marker
659 }
660 return text
661}
662
663func truncateOutputUTF8(value string, maxBytes int) string {
664 if maxBytes <= 0 {
665 return ""
666 }
667 data := []byte(value)
668 if len(data) <= maxBytes {
669 return value
670 }
671 data = data[:maxBytes]
672 for len(data) > 0 && !utf8.Valid(data) {
673 data = data[:len(data)-1]
674 }
675 return string(data)
676}
677
678func resultErrorPayload(err error, shutdownOrigin subagents.ShutdownOrigin) map[string]any {
679 if err == nil {
680 return nil
681 }
682 if provider.IsContextOverflowError(err) {
683 return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
684 }
685 if errors.Is(err, core.ErrStreamIdleTimeout) {
686 return map[string]any{"code": "stream_idle_timeout", "message": "provider stream produced no events for five minutes; the delegated turn stopped safely"}
687 }
688 if errors.Is(err, context.DeadlineExceeded) || shutdownOrigin == subagents.ShutdownOriginDeadline {
689 return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"}
690 }
691 if errors.Is(err, context.Canceled) {
692 if resultErr := subagents.ShutdownResultError(shutdownOrigin); resultErr != nil {
693 return map[string]any{"code": resultErr.Code, "message": resultErr.Message}
694 }
695 }
696 return map[string]any{"code": "turn_failed", "message": truncateForLog(err.Error(), 500)}
697}
698
699func turnErrorMessage(err error, shutdownOrigin subagents.ShutdownOrigin) string {
700 if provider.IsContextOverflowError(err) {
701 return subagentContextLimitMessage
702 }
703 if err != nil {
704 if resultErr := subagents.ShutdownResultError(shutdownOrigin); resultErr != nil {
705 return resultErr.Message
706 }
707 }
708 return errString(err)
709}
710
711func errString(err error) string {
712 if err == nil {
713 return ""
714 }
715 return err.Error()
716}
717
718func truncateForLog(s string, n int) string {
719 if len(s) <= n {
720 return s
721 }
722 if n <= 3 {
723 return strings.Repeat(".", n)
724 }
725 return s[:n-3] + "..."
726}