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
240 cancel()
241 mu.Lock()
242 busyTurn = false
243 cancelFn = nil
244 mu.Unlock()
245 // The supervisor treats agent.idle as permission to start a follow-up.
246 // Publish it only after the busy gate has opened, otherwise a manager
247 // can send a turn that the inbox loop drops as still busy.
248 em.emit("agent.idle", map[string]any{"turn_id": turnID})
249 }
250
251 launchTurn := func(prompt string) {
252 mu.Lock()
253 if closing {
254 mu.Unlock()
255 return
256 }
257 // Register the goroutine before launching it. Shutdown can then
258 // close the gate and wait for every turn that was admitted, even
259 // if runOne has not acquired mu and set busyTurn yet.
260 turnWG.Add(1)
261 mu.Unlock()
262 go func() {
263 defer turnWG.Done()
264 runOne(prompt)
265 }()
266 }
267
268 waitForActiveTurn := func(cancel bool) {
269 mu.Lock()
270 closing = true
271 activeCancel := cancelFn
272 mu.Unlock()
273 if cancel && activeCancel != nil {
274 activeCancel()
275 }
276 turnWG.Wait()
277 }
278
279 requestShutdown := func() {
280 mu.Lock()
281 closing = true
282 mu.Unlock()
283 shutdownOnce.Do(func() { close(shutdown) })
284 }
285
286 // Initial task: run before processing the inbox so the agent
287 // "starts working" the moment it boots, matching what users
288 // expect from `/subagents new <task>`.
289 if args.Prompt != "" {
290 launchTurn(args.Prompt)
291 }
292
293 // Inbox loop: one supervisor message at a time. We don't spawn
294 // a goroutine per turn because runOne already serialises them
295 // via the busyTurn flag; doing the dispatch on the main
296 // goroutine keeps the daemon's lifecycle easy to follow.
297 for {
298 select {
299 case <-ctx.Done():
300 waitForActiveTurn(true)
301 em.emit("agent.exited", map[string]any{"reason": "cancelled"})
302 return ctx.Err()
303 case <-shutdown:
304 // Shutdown is an explicit supervisor stop. Cancel an active
305 // turn before joining it so a detached worker cannot remain
306 // alive indefinitely while the host waits for its socket to
307 // disappear.
308 waitForActiveTurn(true)
309 em.emit("agent.exited", map[string]any{"reason": "shutdown"})
310 return nil
311 case msg, ok := <-ln.Lines():
312 if !ok {
313 waitForActiveTurn(true)
314 em.emit("agent.exited", map[string]any{"reason": "inbox-closed"})
315 return nil
316 }
317 command, parseErr := subagents.ParseCommand(msg)
318 if parseErr != nil {
319 em.emit("error", map[string]any{"message": "invalid supervisor command"})
320 continue
321 }
322 switch command.Type {
323 case subagents.CommandAgentShutdown:
324 requestShutdown()
325 case subagents.CommandTurnCancel:
326 mu.Lock()
327 if cancelFn != nil {
328 cancelFn()
329 }
330 mu.Unlock()
331 case subagents.CommandAgentPing:
332 em.emit("agent.heartbeat", map[string]any{"activity": "alive"})
333 case subagents.CommandTurnStart:
334 var payload subagents.TurnStartPayload
335 if err := command.DecodePayload(&payload); err != nil {
336 em.emit("error", map[string]any{"message": "invalid turn.start payload"})
337 continue
338 }
339 launchTurn(payload.Prompt)
340 default:
341 // Unknown commands are rejected rather than interpreted as
342 // child-originated control requests.
343 em.emit("error", map[string]any{"message": "unknown supervisor command"})
344 }
345 }
346 }
347}
348
349// subagentEmitter serialises events to stdout and (optionally) to a
350// durable log file. Concurrent goroutines call emit so we have to
351// hold a mutex around the encoder.
352type subagentEmitter struct {
353 mu sync.Mutex
354 w *os.File
355 mirror *subagents.EventLog
356 agentID string
357
358 // orphan flips true the first time a stdout write fails (broken
359 // pipe — the supervisor died). Until then the mirror stays
360 // dormant: the supervisor is the canonical writer to events.jsonl
361 // (it parses our stdout and Append()s each event itself). Writing
362 // from both sides used to land every event in the log twice,
363 // which showed up as a fully-duplicated transcript the next time
364 // the agent was reloaded — the exact "why is everything doubled"
365 // bug. Once orphaned the mirror takes over so the events still
366 // land on disk for the next reload.
367 orphan bool
368}
369
370func newSubagentEmitter(w *os.File, mirror *subagents.EventLog) *subagentEmitter {
371 return &subagentEmitter{w: w, mirror: mirror}
372}
373
374func (e *subagentEmitter) setProtocolIdentity(agentID string) {
375 e.mu.Lock()
376 e.agentID = agentID
377 e.mu.Unlock()
378}
379
380func (e *subagentEmitter) emit(typ string, data map[string]any) {
381 if data == nil {
382 data = map[string]any{}
383 }
384 payload := make(map[string]any, len(data))
385 for key, value := range data {
386 payload[key] = value
387 }
388 turnID, _ := payload["turn_id"].(string)
389
390 e.mu.Lock()
391 defer e.mu.Unlock()
392
393 wireType := canonicalWorkerEvent(typ)
394 envelope := subagents.NewEventEnvelope(wireType, e.agentID, turnID, payload)
395 line, err := subagents.MarshalJSONL(envelope)
396
397 if !e.orphan && err == nil {
398 if _, werr := e.w.Write(line); werr != nil {
399 // Supervisor's stdout pipe is gone (parent zut exited but we
400 // kept running). Switch to mirror-only mode and preserve this
401 // event in the durable log.
402 e.orphan = true
403 }
404 }
405
406 if e.orphan && e.mirror != nil {
407 _ = e.mirror.Append(subagents.NewEvent(wireType, payload))
408 }
409}
410
411func subagentEventData(ev core.AgentEvent) map[string]any {
412 data := modes.EventToJSON(ev)
413 if turnEnd, ok := ev.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
414 data["error"] = subagentContextLimitMessage
415 }
416 return data
417}
418
419func canonicalWorkerEvent(typ string) string {
420 switch typ {
421 case "agent_ready":
422 return subagents.EventAgentReady
423 case "agent_stopped":
424 return subagents.EventAgentExited
425 case "turn_start":
426 return subagents.EventTurnStarted
427 case "turn_progress":
428 return subagents.EventTurnProgress
429 case "tool_call":
430 return subagents.EventToolStarted
431 case "tool_result":
432 return subagents.EventToolFinished
433 case "text_delta":
434 return subagents.EventMessageDelta
435 default:
436 return typ
437 }
438}
439
440const subagentContextLimitMessage = "request exceeds the model context window; narrow the task or reduce gathered context"
441
442// promptWithContextRecovery preserves the worker's existing helper boundary
443// while delegating its one-shot overflow recovery to modes. The returned index
444// is the first message to include in the result or append after a persisted
445// compaction checkpoint.
446func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), persistCompaction func([]provider.Message) error) (outputStart int, err error) {
447 result, err := modes.PromptWithContextRecovery(ctx, ag, prompt, nil, sink, modes.ContextRecoveryOptions{
448 PersistCompaction: persistCompaction,
449 })
450 return result.OutputStart, err
451}
452
453func finalAssistantText(messages []provider.Message) string {
454 var out strings.Builder
455 for _, message := range messages {
456 if message.Role != provider.RoleAssistant {
457 continue
458 }
459 for _, block := range message.Content {
460 if text, ok := block.(provider.TextBlock); ok {
461 if out.Len() > 0 {
462 out.WriteByte('\n')
463 }
464 out.WriteString(text.Text)
465 }
466 }
467 }
468 return out.String()
469}
470
471func firstLine(text string) string {
472 return strings.Split(strings.TrimSpace(text), "\n")[0]
473}
474
475func boundedResultSummary(text string) string {
476 return boundedResultOutput(firstLine(text), 4*1024, 1)
477}
478
479func workerOutputLimits() (maxBytes, maxLines int) {
480 maxBytes, maxLines = 500_000, 5_000
481 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {
482 maxBytes = value
483 }
484 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_LINES"))); err == nil && value > 0 {
485 maxLines = value
486 }
487 return maxBytes, maxLines
488}
489
490func boundedResultOutput(text string, maxBytes, maxLines int) string {
491 const marker = "...[output truncated]"
492 lines := strings.Split(text, "\n")
493 if maxLines > 0 && len(lines) > maxLines {
494 if maxLines == 1 {
495 text = marker
496 } else {
497 text = strings.Join(append(lines[:maxLines-1], marker), "\n")
498 }
499 }
500 if maxBytes > 0 && len([]byte(text)) > maxBytes {
501 if maxBytes <= len(marker) {
502 return truncateOutputUTF8(text, maxBytes)
503 }
504 return truncateOutputUTF8(text, maxBytes-len(marker)) + marker
505 }
506 return text
507}
508
509func truncateOutputUTF8(value string, maxBytes int) string {
510 if maxBytes <= 0 {
511 return ""
512 }
513 data := []byte(value)
514 if len(data) <= maxBytes {
515 return value
516 }
517 data = data[:maxBytes]
518 for len(data) > 0 && !utf8.Valid(data) {
519 data = data[:len(data)-1]
520 }
521 return string(data)
522}
523
524func resultErrorPayload(err error) map[string]any {
525 if err == nil {
526 return nil
527 }
528 if provider.IsContextOverflowError(err) {
529 return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage}
530 }
531 return map[string]any{"code": "turn_failed", "message": truncateForLog(err.Error(), 500)}
532}
533
534func turnErrorMessage(err error) string {
535 if provider.IsContextOverflowError(err) {
536 return subagentContextLimitMessage
537 }
538 return errString(err)
539}
540
541func errString(err error) string {
542 if err == nil {
543 return ""
544 }
545 return err.Error()
546}
547
548func initialTurnNumber(path string) int {
549 if strings.TrimSpace(path) == "" {
550 return 0
551 }
552 events, err := subagents.ReadEventLog(path)
553 if err != nil {
554 return 0
555 }
556 maxStep := 0
557 for _, ev := range events {
558 if step, ok := ev.Data["step"].(float64); ok && int(step) > maxStep {
559 maxStep = int(step)
560 }
561 if strings.HasPrefix(ev.TurnID, "turn-") {
562 if n, parseErr := strconv.Atoi(strings.TrimPrefix(ev.TurnID, "turn-")); parseErr == nil && n > maxStep {
563 maxStep = n
564 }
565 }
566 }
567 return maxStep
568}
569
570func truncateForLog(s string, n int) string {
571 if len(s) <= n {
572 return s
573 }
574 if n <= 3 {
575 return strings.Repeat(".", n)
576 }
577 return s[:n-3] + "..."
578}