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