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