1package subagents
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "os"
10 "path/filepath"
11 "strings"
12 "time"
13 "unicode/utf8"
14)
15
16// TurnStatus is the durable outcome of one delegated turn.
17type TurnStatus string
18
19const (
20 ResultSucceeded TurnStatus = "succeeded"
21 ResultFailed TurnStatus = "failed"
22 ResultCanceled TurnStatus = "canceled"
23)
24
25// ArtifactRef points to a durable file without exposing the supervisor's
26// local state layout to callers.
27type ArtifactRef struct {
28 Name string `json:"name,omitempty"`
29 Ref string `json:"ref"`
30 MediaType string `json:"media_type,omitempty"`
31 Size int64 `json:"size,omitempty"`
32}
33
34// ResultError is intentionally free of credentials and argv details.
35type ResultError struct {
36 Code string `json:"code,omitempty"`
37 Message string `json:"message,omitempty"`
38}
39
40// ShutdownResultError returns the stable, user-facing error associated with
41// an intentional worker shutdown. Unknown origins are not attributed.
42func ShutdownResultError(origin ShutdownOrigin) *ResultError {
43 switch origin.Sanitized() {
44 case ShutdownOriginTargeted:
45 return &ResultError{Code: "shutdown", Message: "subagent stopped by request"}
46 case ShutdownOriginSession:
47 return &ResultError{Code: "shutdown", Message: "subagent stopped during session shutdown"}
48 case ShutdownOriginDeadline:
49 return &ResultError{Code: "deadline_exceeded", Message: "subagent turn deadline exceeded; partial output is preserved in the result and history"}
50 case ShutdownOriginProcess:
51 return &ResultError{Code: "shutdown", Message: "subagent stopped during process shutdown"}
52 default:
53 return nil
54 }
55}
56
57// TurnResult is the protocol-level result for a delegated turn. Output is a
58// bounded inline preview; the full session/history remains available through
59// HistoryRef and the complete JSON result through ResultRef.
60type TurnResult struct {
61 Version int `json:"version"`
62 AgentID string `json:"agent_id,omitempty"`
63 TurnID string `json:"turn_id,omitempty"`
64 Status TurnStatus `json:"status"`
65 Summary string `json:"summary,omitempty"`
66 Output string `json:"output,omitempty"`
67 Structured json.RawMessage `json:"structured,omitempty"`
68 Artifacts []ArtifactRef `json:"artifacts,omitempty"`
69 ChangedFiles []string `json:"changed_files,omitempty"`
70 Usage map[string]any `json:"usage,omitempty"`
71 Error *ResultError `json:"error,omitempty"`
72 ShutdownOrigin ShutdownOrigin `json:"shutdown_origin,omitempty"`
73 CreatedAt time.Time `json:"created_at"`
74}
75
76func (r *TurnResult) Validate(maxBytes, maxLines int) error {
77 if r == nil {
78 return errors.New("subagents: nil turn result")
79 }
80 if r.Version == 0 {
81 r.Version = ProtocolVersion
82 }
83 if r.Version != ProtocolVersion {
84 return fmt.Errorf("subagents: unsupported result version %d", r.Version)
85 }
86 if r.Status != ResultSucceeded && r.Status != ResultFailed && r.Status != ResultCanceled {
87 return fmt.Errorf("subagents: invalid result status %q", r.Status)
88 }
89 if r.AgentID == "" {
90 return errors.New("subagents: result missing agent id")
91 }
92 if r.ShutdownOrigin != "" && r.ShutdownOrigin.Sanitized() == "" {
93 return errors.New("subagents: result has invalid shutdown origin")
94 }
95 if r.TurnID == "" {
96 return errors.New("subagents: result missing turn id")
97 }
98 if len(r.Structured) > 0 && !json.Valid(r.Structured) {
99 return errors.New("subagents: result structured payload is invalid JSON")
100 }
101 if maxBytes > 0 && len([]byte(r.Output)) > maxBytes {
102 return fmt.Errorf("subagents: result output exceeds %d bytes", maxBytes)
103 }
104 if maxLines > 0 && countLines(r.Output) > maxLines {
105 return fmt.Errorf("subagents: result output exceeds %d lines", maxLines)
106 }
107 return nil
108}
109
110// Bounded returns a copy with inline output capped by bytes and lines. It
111// never mutates the durable result's caller-owned value.
112func (r *TurnResult) Bounded(maxBytes, maxLines int) *TurnResult {
113 out := cloneTurnResult(r)
114 if out == nil {
115 return nil
116 }
117 if len([]byte(out.Summary)) > 4*1024 {
118 out.Summary = truncateUTF8(out.Summary, 4*1024)
119 }
120 out.Output = boundInlineText(out.Output, maxBytes, maxLines)
121 return out
122}
123
124// boundInlineText applies the shared line, byte, and UTF-8-safe bounds used
125// for inline assistant output and durable turn-result previews.
126func boundInlineText(value string, maxBytes, maxLines int) string {
127 const marker = "...[output truncated]"
128 if maxLines > 0 {
129 lines := strings.Split(value, "\n")
130 if len(lines) > maxLines {
131 if maxLines == 1 {
132 value = marker
133 } else {
134 value = strings.Join(lines[:maxLines-1], "\n") + "\n" + marker
135 }
136 }
137 }
138 if maxBytes > 0 && len([]byte(value)) > maxBytes {
139 if maxBytes <= len(marker) {
140 return truncateUTF8(value, maxBytes)
141 }
142 return truncateUTF8(value, maxBytes-len(marker)) + marker
143 }
144 return value
145}
146
147func truncateUTF8(value string, maxBytes int) string {
148 if maxBytes <= 0 {
149 return ""
150 }
151 data := []byte(value)
152 if len(data) <= maxBytes {
153 return value
154 }
155
156 cut := maxBytes
157 // Only inspect the short suffix that can belong to a UTF-8 rune ending
158 // at the cut. Validating the entire prefix would make this quadratic for
159 // invalid input and would incorrectly discard valid bytes before an
160 // earlier invalid byte.
161 start := cut - 1
162 minimum := cut - utf8.UTFMax
163 if minimum < 0 {
164 minimum = 0
165 }
166 for start >= minimum && isUTF8Continuation(data[start]) {
167 start--
168 }
169 if start >= minimum {
170 if size := utf8LeadSize(data[start]); size > cut-start && isUTF8Prefix(data[start:cut], size) {
171 cut = start
172 }
173 }
174 return string(data[:cut])
175}
176
177func isUTF8Continuation(b byte) bool {
178 return b&0xc0 == 0x80
179}
180
181func utf8LeadSize(b byte) int {
182 switch {
183 case b < utf8.RuneSelf:
184 return 1
185 case b >= 0xc2 && b <= 0xdf:
186 return 2
187 case b >= 0xe0 && b <= 0xef:
188 return 3
189 case b >= 0xf0 && b <= 0xf4:
190 return 4
191 default:
192 return 0
193 }
194}
195
196func isUTF8Prefix(data []byte, size int) bool {
197 if len(data) <= 1 {
198 return true
199 }
200 if !isUTF8Continuation(data[1]) {
201 return false
202 }
203 switch data[0] {
204 case 0xe0:
205 if data[1] < 0xa0 {
206 return false
207 }
208 case 0xed:
209 if data[1] > 0x9f {
210 return false
211 }
212 case 0xf0:
213 if data[1] < 0x90 {
214 return false
215 }
216 case 0xf4:
217 if data[1] > 0x8f {
218 return false
219 }
220 }
221 for _, b := range data[2:] {
222 if !isUTF8Continuation(b) {
223 return false
224 }
225 }
226 return len(data) < size
227}
228
229func validateTurnResultAgent(result *TurnResult, agentID string) error {
230 if result == nil {
231 return errors.New("subagents: nil turn result")
232 }
233 if agentID != "" && result.AgentID != agentID {
234 return fmt.Errorf("subagents: turn result belongs to agent %q, want %q", result.AgentID, agentID)
235 }
236 return nil
237}
238
239func decodeTurnResultEvent(ev Event, agentID string, maxBytes, maxLines int) (*TurnResult, error) {
240 data, err := json.Marshal(ev.Data)
241 if err != nil {
242 return nil, fmt.Errorf("subagents: encode turn result event: %w", err)
243 }
244 var result TurnResult
245 if err := json.Unmarshal(data, &result); err != nil {
246 return nil, fmt.Errorf("subagents: decode turn result event: %w", err)
247 }
248 if result.AgentID == "" {
249 result.AgentID = firstNonEmpty(ev.AgentID, agentID)
250 }
251 if err := validateTurnResultAgent(&result, agentID); err != nil {
252 return nil, err
253 }
254 if result.TurnID == "" {
255 result.TurnID = ev.TurnID
256 }
257 if result.Version == 0 {
258 result.Version = ProtocolVersion
259 }
260 if result.CreatedAt.IsZero() {
261 result.CreatedAt = ev.Time
262 if result.CreatedAt.IsZero() {
263 result.CreatedAt = time.Now().UTC()
264 }
265 }
266 resultPtr := result.Bounded(maxBytes, maxLines)
267 if err := resultPtr.Validate(0, 0); err != nil {
268 return nil, err
269 }
270 return resultPtr, nil
271}
272
273func cloneTurnResult(r *TurnResult) *TurnResult {
274 if r == nil {
275 return nil
276 }
277 copy := *r
278 copy.Structured = append(json.RawMessage(nil), r.Structured...)
279 copy.Artifacts = append([]ArtifactRef(nil), r.Artifacts...)
280 copy.ChangedFiles = append([]string(nil), r.ChangedFiles...)
281 if r.Usage != nil {
282 copy.Usage = make(map[string]any, len(r.Usage))
283 for key, value := range r.Usage {
284 copy.Usage[key] = value
285 }
286 }
287 if r.Error != nil {
288 errCopy := *r.Error
289 copy.Error = &errCopy
290 }
291 return ©
292}
293
294func countLines(s string) int {
295 if s == "" {
296 return 0
297 }
298 return 1 + strings.Count(s, "\n")
299}
300
301func AgentRef(id string) string { return "subagent://" + id }
302func HistoryRef(id string) string { return AgentRef(id) + "/history" }
303func ResultRef(id string) string { return AgentRef(id) + "/result" }
304func PatchRef(id string) string { return AgentRef(id) + "/patch" }
305
306const maxResultFileBytes = 8 * 1024 * 1024
307
308const (
309 workspaceCaptureErrorCode = "workspace_capture_failed"
310 workspaceCaptureErrorMessage = "workspace capture failed"
311)
312
313func resultPath(stateDir string) string { return filepath.Join(stateDir, "result.json") }
314func patchPath(stateDir string) string { return filepath.Join(stateDir, "patch.diff") }
315
316// writeSyncedFile writes a complete temporary file and closes it after the
317// contents have been synced. Keeping this separate makes it harder for an
318// atomic rename caller to accidentally publish an unflushed or open file.
319func writeSyncedFile(path string, data []byte, perm os.FileMode) (err error) {
320 file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, perm)
321 if err != nil {
322 return fmt.Errorf("open: %w", err)
323 }
324 defer func() {
325 closeErr := file.Close()
326 if closeErr == nil {
327 return
328 }
329 closeErr = fmt.Errorf("close: %w", closeErr)
330 if err == nil {
331 err = closeErr
332 } else {
333 err = errors.Join(err, closeErr)
334 }
335 }()
336
337 for written := 0; written < len(data); {
338 n, writeErr := file.Write(data[written:])
339 written += n
340 if writeErr != nil {
341 return fmt.Errorf("write: %w", writeErr)
342 }
343 if n == 0 {
344 return fmt.Errorf("write: %w", io.ErrShortWrite)
345 }
346 }
347 if err := file.Sync(); err != nil {
348 return fmt.Errorf("sync: %w", err)
349 }
350 return nil
351}
352
353func writeTurnResult(stateDir string, result *TurnResult) (err error) {
354 if result == nil {
355 return errors.New("subagents: nil result")
356 }
357 if result.CreatedAt.IsZero() {
358 result.CreatedAt = time.Now().UTC()
359 }
360 if result.Version == 0 {
361 result.Version = ProtocolVersion
362 }
363 if err := result.Validate(0, 0); err != nil {
364 return err
365 }
366 data, err := json.MarshalIndent(result, "", " ")
367 if err != nil {
368 return fmt.Errorf("subagents result marshal: %w", err)
369 }
370 if len(data) > maxResultFileBytes {
371 return fmt.Errorf("subagents result exceeds %d bytes", maxResultFileBytes)
372 }
373 if err := os.MkdirAll(stateDir, 0o700); err != nil {
374 return fmt.Errorf("subagents result dir: %w", err)
375 }
376 tmp := resultPath(stateDir) + ".tmp"
377 defer func() {
378 if err != nil {
379 _ = os.Remove(tmp)
380 }
381 }()
382 if err := writeSyncedFile(tmp, append(data, '\n'), 0o600); err != nil {
383 return fmt.Errorf("subagents result write: %w", err)
384 }
385 if err := os.Rename(tmp, resultPath(stateDir)); err != nil {
386 return fmt.Errorf("subagents result rename: %w", err)
387 }
388 return nil
389}
390
391func (f *Supervisor) captureWorkspace(a *Agent) {
392 if a == nil || a.workspaceCapture == nil {
393 return
394 }
395 stateDir := a.stateDirectory(f.cfg.Root)
396 if _, err := os.Stat(stateDir); err != nil {
397 recordWorkspaceCaptureFailure(a)
398 return
399 }
400 capture, err := a.workspaceCapture()
401 if err != nil {
402 recordWorkspaceCaptureFailure(a)
403 return
404 }
405 a.lifecycleMu.Lock()
406 a.changedFiles = append([]string(nil), capture.ChangedFiles...)
407 a.lifecycleMu.Unlock()
408 if len(capture.Patch) == 0 {
409 return
410 }
411 path := patchPath(stateDir)
412 tmp := path + ".tmp"
413 if err := writeSyncedFile(tmp, capture.Patch, 0o600); err != nil {
414 _ = os.Remove(tmp)
415 recordWorkspaceCaptureFailure(a)
416 return
417 }
418 if err := os.Rename(tmp, path); err != nil {
419 _ = os.Remove(tmp)
420 recordWorkspaceCaptureFailure(a)
421 return
422 }
423 a.lifecycleMu.Lock()
424 a.patchRef = PatchRef(a.ID)
425 a.lifecycleMu.Unlock()
426}
427
428// recordWorkspaceCaptureFailure records a stable, non-sensitive error in the
429// result already received from the child (or creates the durable fallback).
430// The underlying filesystem/capture error can contain private paths or data,
431// so it is deliberately not copied into ResultError.
432func recordWorkspaceCaptureFailure(a *Agent) {
433 if a == nil {
434 return
435 }
436 result := a.Result()
437 if result == nil || validateTurnResultAgent(result, a.ID) != nil {
438 result = &TurnResult{AgentID: a.ID}
439 }
440 result.Version = ProtocolVersion
441 if result.AgentID == "" {
442 result.AgentID = a.ID
443 }
444 if result.TurnID == "" {
445 turnID := a.CurrentTurnID()
446 if turnID == "" {
447 turnID = fmt.Sprintf("turn-%d", a.AttemptValue())
448 }
449 result.TurnID = turnID
450 }
451 if result.CreatedAt.IsZero() {
452 result.CreatedAt = time.Now().UTC()
453 }
454 result.Status = ResultFailed
455 result.Error = &ResultError{
456 Code: workspaceCaptureErrorCode,
457 Message: workspaceCaptureErrorMessage,
458 }
459 a.setResult(result)
460}
461
462func (f *Supervisor) ensureResult(a *Agent, status Status, runErr error) {
463 if a == nil {
464 return
465 }
466 attempt := a.AttemptValue()
467 shutdownOrigin := a.shutdownOriginValue()
468 if errors.Is(runErr, context.DeadlineExceeded) {
469 shutdownOrigin = ShutdownOriginDeadline
470 }
471 a.mu.Lock()
472 partialOutput := a.lastAssistant
473 a.mu.Unlock()
474 result := a.Result()
475 if result != nil {
476 if err := validateTurnResultAgent(result, a.ID); err != nil {
477 result = nil
478 }
479 }
480 if result != nil {
481 // A child may omit agent/turn metadata in a result event; the
482 // supervisor supplies it before validating the durable fallback.
483 if result.AgentID == "" {
484 result.AgentID = a.ID
485 }
486 if result.TurnID == "" {
487 result.TurnID = firstNonEmpty(a.CurrentTurnID(), fmt.Sprintf("turn-%d", attempt))
488 }
489 if result.Version == 0 {
490 result.Version = ProtocolVersion
491 }
492 if result.CreatedAt.IsZero() {
493 result.CreatedAt = time.Now().UTC()
494 }
495 if err := result.Validate(0, 0); err != nil {
496 result = nil
497 }
498 }
499 if result != nil && result.Output == "" && partialOutput != "" &&
500 (result.Status == ResultFailed || result.Status == ResultCanceled || status == StatusFailed || status == StatusKilled) {
501 // A canceled or failed worker turn may emit its terminal result before
502 // its provider turns accumulated text into a final assistant message.
503 // The event stream has already projected that text onto the Agent, so
504 // retain it in the durable result rather than returning an empty failure.
505 result.Output = partialOutput
506 }
507 if result == nil {
508 turnID := a.CurrentTurnID()
509 if turnID == "" {
510 turnID = fmt.Sprintf("turn-%d", attempt)
511 }
512 output := partialOutput
513 resultStatus := ResultSucceeded
514 switch {
515 case status == StatusFailed && errors.Is(runErr, context.Canceled):
516 resultStatus = ResultCanceled
517 case status == StatusFailed:
518 resultStatus = ResultFailed
519 case status == StatusKilled:
520 resultStatus = ResultCanceled
521 }
522 result = &TurnResult{
523 Version: ProtocolVersion,
524 AgentID: a.ID,
525 TurnID: turnID,
526 Status: resultStatus,
527 Output: output,
528 CreatedAt: time.Now().UTC(),
529 }
530 }
531 if result.AgentID == "" {
532 result.AgentID = a.ID
533 }
534 if result.TurnID == "" {
535 result.TurnID = firstNonEmpty(a.CurrentTurnID(), fmt.Sprintf("turn-%d", attempt))
536 }
537 if result.Version == 0 {
538 result.Version = ProtocolVersion
539 }
540 if result.CreatedAt.IsZero() {
541 result.CreatedAt = time.Now().UTC()
542 }
543 if shutdownOrigin != "" {
544 result.ShutdownOrigin = shutdownOrigin
545 result.Error = ShutdownResultError(shutdownOrigin)
546 if shutdownOrigin == ShutdownOriginDeadline {
547 result.Status = ResultFailed
548 } else {
549 result.Status = ResultCanceled
550 }
551 }
552 if runErr != nil {
553 if status == StatusKilled || errors.Is(runErr, context.Canceled) {
554 result.Status = ResultCanceled
555 } else {
556 result.Status = ResultFailed
557 }
558 if errors.Is(runErr, context.DeadlineExceeded) {
559 result.Status = ResultFailed
560 result.ShutdownOrigin = ShutdownOriginDeadline
561 } else if result.Error == nil {
562 result.Error = &ResultError{Code: "runner_failed", Message: truncate(runErr.Error(), 500)}
563 }
564 }
565 if len(result.ChangedFiles) == 0 {
566 a.lifecycleMu.Lock()
567 result.ChangedFiles = append([]string(nil), a.changedFiles...)
568 a.lifecycleMu.Unlock()
569 }
570 if result.Summary == "" && result.Output != "" {
571 result.Summary = strings.Split(strings.TrimSpace(result.Output), "\n")[0]
572 }
573 result = result.Bounded(a.maxOutputBytes, a.maxOutputLines)
574 stateDir := a.stateDirectory(f.cfg.Root)
575 if err := writeTurnResult(stateDir, result); err != nil {
576 a.recordPersistenceError(fmt.Errorf("write turn result: %w", err))
577 failed := cloneTurnResult(result)
578 failed.Status = ResultFailed
579 failed.Error = &ResultError{Code: "result_persistence_failed", Message: "failed to persist delegated turn result"}
580 a.setResult(failed)
581 a.resolveRequirement(0, nil, fmt.Sprintf("write turn result: %v", err), true)
582 return
583 }
584 a.setResult(result)
585 a.lifecycleMu.Lock()
586 a.resultRef = ResultRef(a.ID)
587 a.lifecycleMu.Unlock()
588 a.recordTrace(TraceEvent{Type: "result.available", TurnID: result.TurnID, Data: map[string]any{"ref": ResultRef(a.ID)}})
589 a.resolveRequirement(0, result, "", true)
590}
591
592// ReadResult returns the durable structured result for an agent.
593func (f *Supervisor) ReadResult(id string) (*TurnResult, error) {
594 a := f.Get(id)
595 if a == nil {
596 return nil, fmt.Errorf("subagents: no such agent %q", id)
597 }
598 if result := a.Result(); result != nil {
599 if err := validateTurnResultAgent(result, a.ID); err != nil {
600 return nil, err
601 }
602 return result, nil
603 }
604 result, err := readTurnResult(a.stateDirectory(f.cfg.Root))
605 if err != nil {
606 return nil, err
607 }
608 if err := validateTurnResultAgent(result, a.ID); err != nil {
609 return nil, err
610 }
611 return result.Bounded(f.cfg.Policy.MaxOutputBytes, f.cfg.Policy.MaxOutputLines), nil
612}
613
614func (f *Supervisor) ResultReference(id string) string { return ResultRef(id) }
615func (f *Supervisor) HistoryReference(id string) string { return HistoryRef(id) }
616func (f *Supervisor) PatchReference(id string) string { return PatchRef(id) }
617
618func readTurnResult(stateDir string) (*TurnResult, error) {
619 file, err := os.Open(resultPath(stateDir))
620 if err != nil {
621 return nil, err
622 }
623 defer file.Close()
624 data, err := io.ReadAll(io.LimitReader(file, maxResultFileBytes+1))
625 if err != nil {
626 return nil, err
627 }
628 if len(data) > maxResultFileBytes {
629 return nil, fmt.Errorf("subagents result exceeds %d bytes", maxResultFileBytes)
630 }
631 var result TurnResult
632 if err := json.Unmarshal(data, &result); err != nil {
633 return nil, fmt.Errorf("subagents result parse: %w", err)
634 }
635 if err := result.Validate(0, 0); err != nil {
636 return nil, err
637 }
638 return &result, nil
639}