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.
39func runSubagentWorkerMode(ctx context.Context, args Args, version string) error {
40 if args.SubagentWorker == "" {
41 return fmt.Errorf("--subagent-worker requires a socket path")
42 }
43 if os.Getenv("ZUT_SUBAGENT_CREDENTIAL_STDIN") == "1" {
44 var inherited subagents.Credential
45 dec := json.NewDecoder(io.LimitReader(os.Stdin, 1<<20))
46 if err := dec.Decode(&inherited); err != nil {
47 return fmt.Errorf("read inherited subagent credential: %w", err)
48 }
49 if inherited.Value == "" || inherited.Method == "" {
50 return fmt.Errorf("read inherited subagent credential: missing value or method")
51 }
52 args.inheritedCredential = inherited.Value
53 args.inheritedAuthMethod = inherited.Method
54 args.inheritedAccountID = inherited.AccountID
55 }
56
57 r, err := Resolve(args, true)
58 if err != nil {
59 return err
60 }
61 extMgr, stopExt := setupNonInteractiveExtensions(ctx, args, &r, version)
62 defer stopExt()
63
64 ag := r.NewAgent()
65 initialAg := ag
66 defer func() {
67 closeAgentLSP(ag)
68 if ag != initialAg {
69 closeAgentLSP(initialAg)
70 }
71 }()
72 wireNonInteractiveAgentExtHooks(ctx, ag, extMgr)
73 sess, err := openOrCreateSession(args, r, ag, version)
74 if err != nil {
75 return err
76 }
77 if sess != nil {
78 var providerName, model string
79 sess, ag, providerName, model, err = applyInitialSessionResume(ctx, args, r, extMgr, sess, ag)
80 if err != nil {
81 return err
82 }
83 r.Provider, r.Model = providerName, model
84 ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) }
85 defer sess.Close()
86 }
87 announceSession(extMgr, sess)
88
89 // Open the inbox listener BEFORE emitting agent_ready so the
90 // supervisor can dial through on the very first send. The
91 // subagent supervisor's Inbox retries dialing for a
92 // short window, but emitting ready first and then listening
93 // would still race in tight loops.
94 ln, err := subagents.Listen(args.SubagentWorker)
95 if err != nil {
96 return fmt.Errorf("subagent-worker listen: %w", err)
97 }
98 defer ln.Close()
99
100 // Event log is owned by the supervisor's runner via stdout, but
101 // the daemon also writes a redundant copy here when the runner's
102 // pipe is closed (e.g. parent zut exited but the agent is still
103 // running headless). The env var is set by the runner; if it's
104 // empty we silently skip the second mirror.
105 var logMirror *subagents.EventLog
106 if path := os.Getenv("ZUT_SUBAGENT_EVENT_LOG"); path != "" {
107 logMirror, _ = subagents.OpenEventLog(path)
108 }
109 if logMirror != nil {
110 defer logMirror.Close()
111 }
112
113 em := newSubagentEmitter(os.Stdout, logMirror)
114 em.setProtocolIdentity(os.Getenv("ZUT_SUBAGENT_AGENT_ID"))
115 em.emit("agent.ready", map[string]any{
116 "version": version,
117 "cwd": r.CWD,
118 "model": r.Model,
119 })
120
121 // Heartbeats make a live-but-detached child distinguishable from a
122 // failed turn after supervisor reconnect. The parent still treats
123 // unknown event types as forward-compatible.
124 heartbeatDone := make(chan struct{})
125 defer close(heartbeatDone)
126 go func() {
127 interval := 10 * time.Second
128 if raw := os.Getenv("ZUT_SUBAGENT_HEARTBEAT_INTERVAL"); raw != "" {
129 if parsed, parseErr := time.ParseDuration(raw); parseErr == nil && parsed > 0 {
130 interval = parsed
131 }
132 }
133 ticker := time.NewTicker(interval)
134 defer ticker.Stop()
135 for {
136 select {
137 case <-heartbeatDone:
138 return
139 case <-ticker.C:
140 em.emit("agent.heartbeat", map[string]any{"activity": "alive"})
141 }
142 }
143 }()
144
145 // Keep a per-turn cancel so the "cancel" inbox message can interrupt
146 // an in-flight turn without tearing down the whole daemon.
147 var (
148 mu sync.Mutex
149 cancelFn context.CancelFunc
150 busyTurn bool
151 turnNo = initialTurnNumber(os.Getenv("ZUT_SUBAGENT_EVENT_LOG"))
152 turnWG sync.WaitGroup
153 closing bool
154 shutdown = make(chan struct{})
155 shutdownOnce sync.Once
156 )
157 maxOutputBytes, maxOutputLines := workerOutputLimits()
158
159 runOne := func(prompt string) {
160 mu.Lock()
161 // launchTurn admits the goroutine to turnWG before it starts so
162 // shutdown can join it. A shutdown can win the scheduling race
163 // after that admission but before this goroutine acquires mu; in
164 // that case do not start a new Prompt after waitForActiveTurn has
165 // declared the worker closing.
166 if closing {
167 mu.Unlock()
168 return
169 }
170 if args.SubagentMaxTurns > 0 && turnNo >= args.SubagentMaxTurns {
171 // Consume a unique rejected attempt so repeated prompts after
172 // the limit cannot reuse the same turn id in the event log.
173 turnNo++
174 step := turnNo
175 mu.Unlock()
176 turnID := fmt.Sprintf("turn-%d", step)
177 errPayload := map[string]any{"code": "max_turns", "message": "maximum subagent turns reached"}
178 em.emit("turn.result", map[string]any{"status": "failed", "turn_id": turnID, "error": errPayload})
179 em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": errPayload})
180 em.emit("turn_end", map[string]any{"step": step, "turn_id": turnID, "error": errPayload["message"]})
181 em.emit("agent.idle", map[string]any{"turn_id": turnID})
182 return
183 }
184 if busyTurn {
185 // Drop concurrent turns rather than queuing. The
186 // supervisor protocol assumes one outstanding turn per
187 // agent; if a user really wants to interrupt and start
188 // another, they should send "cancel" first.
189 mu.Unlock()
190 em.emit("error", map[string]any{"message": "agent busy; send 'cancel' first"})
191 return
192 }
193 busyTurn = true
194 turnNo++
195 step := turnNo
196 c, cancel := context.WithCancel(ctx)
197 cancelFn = cancel
198 mu.Unlock()
199
200 turnID := fmt.Sprintf("turn-%d", step)
201 em.emit("turn.started", map[string]any{"step": step, "turn_id": turnID})
202
203 sink := func(ev core.AgentEvent) {
204 em.emit(ev.Type(), subagentEventData(ev))
205 }
206
207 var persistCompaction func([]provider.Message) error
208 if sess != nil {
209 persistCompaction = sess.AppendCompaction
210 }
211 start, err := promptWithContextRecovery(c, ag, prompt, sink, persistCompaction)
212 WriteNewTranscript(ag, sess, start)
213
214 output := finalAssistantText(ag.Messages()[start:])
215 resultStatus := "succeeded"
216 if err != nil {
217 if errors.Is(err, context.Canceled) {
218 resultStatus = "canceled"
219 } else {
220 resultStatus = "failed"
221 }
222 }
223 resultPayload := map[string]any{
224 "status": resultStatus,
225 "turn_id": turnID,
226 "summary": boundedResultSummary(output),
227 "output": boundedResultOutput(output, maxOutputBytes, maxOutputLines),
228 "error": resultErrorPayload(err),
229 }
230 em.emit("turn.result", resultPayload)
231 if err != nil {
232 em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": resultErrorPayload(err)})
233 }
234 em.emit("turn_end", map[string]any{
235 "step": step,
236 "turn_id": turnID,
237 "error": turnErrorMessage(err),
238 })
239 em.emit("agent.idle", map[string]any{"turn_id": turnID})
240
241 cancel()
242 mu.Lock()
243 busyTurn = false
244 cancelFn = nil
245 mu.Unlock()
246 }
247
248 launchTurn := func(prompt string) {
249 mu.Lock()
250 if closing {
251 mu.Unlock()
252 return
253 }
254 // Register the goroutine before launching it. Shutdown can then
255 // close the gate and wait for every turn that was admitted, even
256 // if runOne has not acquired mu and set busyTurn yet.
257 turnWG.Add(1)
258 mu.Unlock()
259 go func() {
260 defer turnWG.Done()
261 runOne(prompt)
262 }()
263 }
264
265 waitForActiveTurn := func(cancel bool) {
266 mu.Lock()
267 closing = true
268 activeCancel := cancelFn
269 mu.Unlock()
270 if cancel && activeCancel != nil {
271 activeCancel()
272 }
273 turnWG.Wait()
274 }
275
276 requestShutdown := func() {
277 mu.Lock()
278 closing = true
279 mu.Unlock()
280 shutdownOnce.Do(func() { close(shutdown) })
281 }
282
283 // Initial task: run before processing the inbox so the agent
284 // "starts working" the moment it boots, matching what users
285 // expect from `/subagents new <task>`.
286 if args.Prompt != "" {
287 launchTurn(args.Prompt)
288 }
289
290 // Inbox loop: one supervisor message at a time. We don't spawn
291 // a goroutine per turn because runOne already serialises them
292 // via the busyTurn flag; doing the dispatch on the main
293 // goroutine keeps the daemon's lifecycle easy to follow.
294 for {
295 select {
296 case <-ctx.Done():
297 waitForActiveTurn(true)
298 em.emit("agent.exited", map[string]any{"reason": "cancelled"})
299 return ctx.Err()
300 case <-shutdown:
301 // Shutdown is an explicit supervisor stop. Cancel an active
302 // turn before joining it so a detached worker cannot remain
303 // alive indefinitely while the host waits for its socket to
304 // disappear.
305 waitForActiveTurn(true)
306 em.emit("agent.exited", map[string]any{"reason": "shutdown"})
307 return nil
308 case msg, ok := <-ln.Lines():
309 if !ok {
310 waitForActiveTurn(true)
311 em.emit("agent.exited", map[string]any{"reason": "inbox-closed"})
312 return nil
313 }
314 command, parseErr := subagents.ParseCommand(msg)
315 if parseErr != nil {
316 em.emit("error", map[string]any{"message": "invalid supervisor command"})
317 continue
318 }
319 switch command.Type {
320 case subagents.CommandAgentShutdown:
321 requestShutdown()
322 case subagents.CommandTurnCancel:
323 mu.Lock()
324 if cancelFn != nil {
325 cancelFn()
326 }
327 mu.Unlock()
328 case subagents.CommandAgentPing:
329 em.emit("agent.heartbeat", map[string]any{"activity": "alive"})
330 case subagents.CommandTurnStart:
331 var payload subagents.TurnStartPayload
332 if err := command.DecodePayload(&payload); err != nil {
333 em.emit("error", map[string]any{"message": "invalid turn.start payload"})
334 continue
335 }
336 launchTurn(payload.Prompt)
337 default:
338 // Unknown commands are rejected rather than interpreted as
339 // child-originated control requests.
340 em.emit("error", map[string]any{"message": "unknown supervisor command"})
341 }
342 }
343 }
344}
345
346// subagentEmitter serialises events to stdout and (optionally) to a
347// durable log file. Concurrent goroutines call emit so we have to
348// hold a mutex around the encoder.
349type subagentEmitter struct {
350 mu sync.Mutex
351 w *os.File
352 mirror *subagents.EventLog
353 agentID string
354
355 // orphan flips true the first time a stdout write fails (broken
356 // pipe — the supervisor died). Until then the mirror stays
357 // dormant: the supervisor is the canonical writer to events.jsonl
358 // (it parses our stdout and Append()s each event itself). Writing
359 // from both sides used to land every event in the log twice,
360 // which showed up as a fully-duplicated transcript the next time
361 // the agent was reloaded — the exact "why is everything doubled"
362 // bug. Once orphaned the mirror takes over so the events still
363 // land on disk for the next reload.
364 orphan bool
365}
366
367func newSubagentEmitter(w *os.File, mirror *subagents.EventLog) *subagentEmitter {
368 return &subagentEmitter{w: w, mirror: mirror}
369}
370
371func (e *subagentEmitter) setProtocolIdentity(agentID string) {
372 e.mu.Lock()
373 e.agentID = agentID
374 e.mu.Unlock()
375}
376
377func (e *subagentEmitter) emit(typ string, data map[string]any) {
378 if data == nil {
379 data = map[string]any{}
380 }
381 payload := make(map[string]any, len(data))
382 for key, value := range data {
383 payload[key] = value
384 }
385 turnID, _ := payload["turn_id"].(string)
386
387 e.mu.Lock()
388 defer e.mu.Unlock()
389
390 wireType := canonicalWorkerEvent(typ)
391 envelope := subagents.NewEventEnvelope(wireType, e.agentID, turnID, payload)
392 line, err := subagents.MarshalJSONL(envelope)
393
394 if !e.orphan && err == nil {
395 if _, werr := e.w.Write(line); werr != nil {
396 // Supervisor's stdout pipe is gone (parent zut exited but we
397 // kept running). Switch to mirror-only mode and preserve this
398 // event in the durable log.
399 e.orphan = true
400 }
401 }
402
403 if e.orphan && e.mirror != nil {
404 _ = e.mirror.Append(subagents.NewEvent(wireType, payload))
405 }
406}
407
408func subagentEventData(ev core.AgentEvent) map[string]any {
409 data := modes.EventToJSON(ev)
410 if turnEnd, ok := ev.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
411 data["error"] = subagentContextLimitMessage
412 }
413 return data
414}
415
416func canonicalWorkerEvent(typ string) string {
417 switch typ {
418 case "agent_ready":
419 return subagents.EventAgentReady
420 case "agent_stopped":
421 return subagents.EventAgentExited
422 case "turn_start":
423 return subagents.EventTurnStarted
424 case "turn_progress":
425 return subagents.EventTurnProgress
426 case "tool_call":
427 return subagents.EventToolStarted
428 case "tool_result":
429 return subagents.EventToolFinished
430 case "text_delta":
431 return subagents.EventMessageDelta
432 default:
433 return typ
434 }
435}
436
437const subagentContextLimitMessage = "request exceeds the model context window; narrow the task or reduce gathered context"
438
439// promptWithContextRecovery appends prompt and runs it once. When the provider
440// rejects that request for exceeding its context limit, it compacts the current
441// transcript and continues the already-appended prompt exactly once.
442//
443// The returned index marks the first message that can contribute to the turn
444// result or be appended after a compaction checkpoint. Compact replaces the
445// transcript, so callers must not reuse the index captured before Prompt.
446func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), persistCompaction func([]provider.Message) error) (outputStart int, err error) {
447 outputStart = len(ag.Messages())
448 if err = ag.Prompt(ctx, prompt, nil, sink); err == nil || ctx.Err() != nil || !provider.IsContextOverflowError(err) {
449 return outputStart, err
450 }
451
452 messages := ag.Messages()
453 if len(messages) <= 1 {
454 // The only message is the failed prompt. Summarizing it would replace
455 // the user's task with an LLM-generated paraphrase, so report the
456 // provider's limit and ask the caller to narrow the task instead.
457 return outputStart, err
458 }
459 keepTail := min(4, len(messages)-1)
460 if _, compactErr := ag.Compact(ctx, keepTail, nil); compactErr != nil {
461 return outputStart, fmt.Errorf("compact transcript after initial overflow: %w", compactErr)
462 }
463
464 compacted := ag.Messages()
465 if persistCompaction != nil {
466 if persistErr := persistCompaction(compacted); persistErr != nil {
467 // Keep the original history plus the user's pending task so the
468 // outer transcript flush can persist that task without committing
469 // an in-memory compaction that the session did not receive.
470 ag.SetMessages(messages)
471 return outputStart, fmt.Errorf("persist compacted transcript: %w", persistErr)
472 }
473 }
474
475 outputStart = len(compacted)
476 return outputStart, ag.Continue(ctx, sink)
477}
478
479func finalAssistantText(messages []provider.Message) string {
480 var out strings.Builder
481 for _, message := range messages {
482 if message.Role != provider.RoleAssistant {
483 continue
484 }
485 for _, block := range message.Content {
486 if text, ok := block.(provider.TextBlock); ok {
487 if out.Len() > 0 {
488 out.WriteByte('\n')
489 }
490 out.WriteString(text.Text)
491 }
492 }
493 }
494 return out.String()
495}
496
497func firstLine(text string) string {
498 return strings.Split(strings.TrimSpace(text), "\n")[0]
499}
500
501func boundedResultSummary(text string) string {
502 return boundedResultOutput(firstLine(text), 4*1024, 1)
503}
504
505func workerOutputLimits() (maxBytes, maxLines int) {
506 maxBytes, maxLines = 500_000, 5_000
507 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {
508 maxBytes = value
509 }
510 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_LINES"))); err == nil && value > 0 {
511 maxLines = value
512 }
513 return maxBytes, maxLines
514}
515
516func boundedResultOutput(text string, maxBytes, maxLines int) string {
517 const marker = "...[output truncated]"
518 lines := strings.Split(text, "\n")
519 if maxLines > 0 && len(lines) > maxLines {
520 if maxLines == 1 {
521 text = marker
522 } else {
523 text = strings.Join(append(lines[:maxLines-1], marker), "\n")
524 }
525 }
526 if maxBytes > 0 && len([]byte(text)) > maxBytes {
527 if maxBytes <= len(marker) {
528 return truncateOutputUTF8(text, maxBytes)
529 }
530 return truncateOutputUTF8(text, maxBytes-len(marker)) + marker
531 }
532 return text
533}
534
535func truncateOutputUTF8(value string, maxBytes int) string {
536 if maxBytes <= 0 {
537 return ""
538 }
539 data := []byte(value)
540 if len(data) <= maxBytes {
541 return value
542 }
543 data = data[:maxBytes]
544 for len(data) > 0 && !utf8.Valid(data) {
545 data = data[:len(data)-1]
546 }
547 return string(data)
548}
549
550func resultErrorPayload(err error) map[string]any {
551 if err == nil {
552 return nil
553 }
554 if provider.IsContextOverflowError(err) {
555 return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
556 }
557 return map[string]any{"code": "turn_failed", "message": truncateForLog(err.Error(), 500)}
558}
559
560func turnErrorMessage(err error) string {
561 if provider.IsContextOverflowError(err) {
562 return subagentContextLimitMessage
563 }
564 return errString(err)
565}
566
567func errString(err error) string {
568 if err == nil {
569 return ""
570 }
571 return err.Error()
572}
573
574func initialTurnNumber(path string) int {
575 if strings.TrimSpace(path) == "" {
576 return 0
577 }
578 events, err := subagents.ReadEventLog(path)
579 if err != nil {
580 return 0
581 }
582 maxStep := 0
583 for _, ev := range events {
584 if step, ok := ev.Data["step"].(float64); ok && int(step) > maxStep {
585 maxStep = int(step)
586 }
587 if strings.HasPrefix(ev.TurnID, "turn-") {
588 if n, parseErr := strconv.Atoi(strings.TrimPrefix(ev.TurnID, "turn-")); parseErr == nil && n > maxStep {
589 maxStep = n
590 }
591 }
592 }
593 return maxStep
594}
595
596func truncateForLog(s string, n int) string {
597 if len(s) <= n {
598 return s
599 }
600 if n <= 3 {
601 return strings.Repeat(".", n)
602 }
603 return s[:n-3] + "..."
604}