1package subagents
2
3import (
4 "bufio"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "os"
10 "path/filepath"
11 "sort"
12 "strings"
13 "sync"
14 "time"
15 "unicode/utf8"
16
17 "github.com/bnema/zut/packages/core"
18 "github.com/bnema/zut/packages/provider"
19)
20
21const (
22 residentMetadataName = "metadata.json"
23 residentTranscriptName = "transcript.jsonl"
24 residentResultName = "result.json"
25 residentPatchName = "patch.diff"
26 residentRecordAccepted = "child.accepted"
27 residentRecordTurnAccepted = "turn.accepted"
28 residentRecordTurnStarted = "turn.started"
29 residentRecordTurnFinished = "turn.finished"
30 residentRecordInterrupted = "child.interrupted"
31 residentRecordFailed = "child.failed"
32 residentRecordUser = "message.user"
33 residentRecordAssistant = "message.assistant"
34 residentRecordToolCall = "tool.call"
35 residentRecordToolResult = "tool.result"
36 residentRecordUsage = "usage"
37 residentJournalVersion = 1
38 residentInterruptedText = "tool interrupted by resident host restart"
39 residentMaxRecordBytes = 2 << 20
40 residentResultSummaryBytes = 256 << 10
41)
42
43// Artifact references are logical names; they do not disclose journal paths.
44func AgentRef(id string) string { return "subagent://" + id }
45func HistoryRef(id string) string { return AgentRef(id) + "/history" }
46func ResultRef(id string) string { return AgentRef(id) + "/result" }
47func PatchRef(id string) string { return AgentRef(id) + "/patch" }
48
49// ResidentState is the durable lifecycle used by the in-process child
50// runtime. It deliberately has no process-liveness states.
51type ResidentState string
52
53const (
54 ResidentQueued ResidentState = "queued"
55 ResidentRunning ResidentState = "running"
56 ResidentIdle ResidentState = "idle"
57 ResidentCompleted ResidentState = "completed"
58 ResidentFailed ResidentState = "failed"
59 ResidentStopped ResidentState = "stopped"
60 ResidentInterrupted ResidentState = "interrupted"
61)
62
63// ResidentChildSpec is the non-secret configuration required to reconstruct
64// one resident child. Credentials are intentionally excluded; the selected
65// provider endpoint and TLS mode are retained so explicit resume reconstructs
66// the same transport contract.
67type ResidentChildSpec struct {
68 ID string `json:"id"`
69 // SessionID identifies the child's conversation thread. RootCacheID is the
70 // root cache affinity shared by every child of the same parent session.
71 SessionID string `json:"session_id"`
72 RootCacheID string `json:"root_cache_id,omitempty"`
73 InitialTurnID string `json:"initial_turn_id,omitempty"`
74 ParentSessionID string `json:"parent_session_id,omitempty"`
75 Profile string `json:"profile,omitempty"`
76 Provider string `json:"provider"`
77 BaseURL string `json:"base_url,omitempty"`
78 InsecureTLS bool `json:"insecure_tls,omitempty"`
79 Model string `json:"model"`
80 Reasoning string `json:"reasoning,omitempty"`
81 FastMode bool `json:"fast_mode,omitempty"`
82 Tools []string `json:"tools,omitempty"`
83 RepositoryRoot string `json:"repository_root,omitempty"`
84 Workspace string `json:"workspace,omitempty"`
85 WorkspaceMode WorkspaceMode `json:"workspace_mode,omitempty"`
86 WorkspaceBase string `json:"workspace_base,omitempty"`
87 WorkspaceCapture CaptureMode `json:"workspace_capture,omitempty"`
88 SystemPrompt string `json:"system_prompt,omitempty"`
89 SystemPromptMode string `json:"system_prompt_mode,omitempty"`
90 InheritProjectContext *bool `json:"inherit_project_context,omitempty"`
91 InheritSkills *bool `json:"inherit_skills,omitempty"`
92 Permissions []string `json:"permissions,omitempty"`
93 Required bool `json:"required,omitempty"`
94}
95
96type residentRecord struct {
97 Version int `json:"version"`
98 Type string `json:"type"`
99 Time time.Time `json:"time"`
100 Spec *ResidentChildSpec `json:"spec,omitempty"`
101 TurnID string `json:"turn_id,omitempty"`
102 Prompt string `json:"prompt,omitempty"`
103 Outcome string `json:"outcome,omitempty"`
104 Message json.RawMessage `json:"message,omitempty"`
105 ToolID string `json:"tool_id,omitempty"`
106 ToolName string `json:"tool_name,omitempty"`
107 ToolArgs json.RawMessage `json:"tool_args,omitempty"`
108 ToolResult json.RawMessage `json:"tool_result,omitempty"`
109 Usage *provider.Usage `json:"usage,omitempty"`
110 ContextUsed int `json:"context_used,omitempty"`
111 ContextMax int `json:"context_max,omitempty"`
112 Subscription bool `json:"subscription,omitempty"`
113 raw json.RawMessage `json:"-"`
114}
115
116// RecordAgentEvent persists finalized provider-neutral transcript events. It
117// deliberately excludes streamed deltas and hidden reasoning, keeping the
118// journal authoritative for resumable visible history without an fsync per
119// token.
120func (j *ResidentJournal) RecordAgentEvent(event core.AgentEvent) error {
121 if j == nil || event == nil {
122 return nil
123 }
124 var record residentRecord
125 switch value := event.(type) {
126 case core.EvUserMessage:
127 message, err := json.Marshal(value.Message)
128 if err != nil {
129 return fmt.Errorf("resident journal encode user message: %w", err)
130 }
131 record = residentRecord{Type: residentRecordUser, Message: message}
132 case core.EvAssistantMessage:
133 message, err := json.Marshal(value.Message)
134 if err != nil {
135 return fmt.Errorf("resident journal encode assistant message: %w", err)
136 }
137 record = residentRecord{Type: residentRecordAssistant, Message: message}
138 case core.EvToolCall:
139 record = residentRecord{Type: residentRecordToolCall, ToolID: value.ID, ToolName: value.Name, ToolArgs: append(json.RawMessage(nil), value.Args...)}
140 case core.EvToolResult:
141 result, err := json.Marshal(value.Result)
142 if err != nil {
143 return fmt.Errorf("resident journal encode tool result: %w", err)
144 }
145 record = residentRecord{Type: residentRecordToolResult, ToolID: value.ID, ToolResult: result}
146 case core.EvUsage:
147 j.usageMu.RLock()
148 contextUsed, contextMax, subscription := j.contextUsed, j.contextMax, j.subscription
149 j.usageMu.RUnlock()
150 if current := value.Usage.PromptTokens(); current > 0 {
151 contextUsed = current
152 }
153 usage := value.Cumulative
154 record = residentRecord{Type: residentRecordUsage, Usage: &usage, ContextUsed: contextUsed, ContextMax: contextMax, Subscription: subscription}
155 default:
156 j.publishAgentEvent(event)
157 return nil
158 }
159 record.Version = residentJournalVersion
160 record.Time = time.Now().UTC()
161 j.mu.Lock()
162 if j.file == nil {
163 j.mu.Unlock()
164 return errors.New("resident journal: closed")
165 }
166 if record.Type == residentRecordUsage {
167 durable, metadataPersisted := j.usageRecordPersistence(record)
168 if durable {
169 var err error
170 if !metadataPersisted {
171 err = j.projectUsageMetadata()
172 if err == nil {
173 j.markUsageMetadataPersisted(record)
174 }
175 }
176 j.mu.Unlock()
177 if err != nil {
178 return err
179 }
180 j.publishAgentEvent(event)
181 return nil
182 }
183 }
184 err := j.appendSync(record)
185 if usageEvent, ok := event.(core.EvUsage); err == nil && ok {
186 j.usageMu.Lock()
187 j.usage = usageEvent.Cumulative
188 j.usageRecordPersisted = true
189 j.usageMetadataPersisted = false
190 if current := usageEvent.Usage.PromptTokens(); current > 0 {
191 j.contextUsed = current
192 }
193 j.usageMu.Unlock()
194 err = j.projectUsageMetadata()
195 if err == nil {
196 j.markUsageMetadataPersisted(record)
197 }
198 }
199 j.mu.Unlock()
200 if err == nil {
201 switch message := event.(type) {
202 case core.EvAssistantMessage:
203 j.recordAssistantSummary(message.Message)
204 case core.EvToolCall:
205 // Core emits tool calls immediately after the assistant message that
206 // requested them. That text is an intermediate thought, not this
207 // turn's final response.
208 j.clearLatestSummary()
209 }
210 j.publishAgentEvent(event)
211 }
212 return err
213}
214
215// ResidentMetadata is a bounded, rebuildable projection. The transcript is
216// the authority for acceptance; callers must never infer acceptance solely
217// from this file.
218type ResidentMetadata struct {
219 Version int `json:"version"`
220 ID string `json:"id"`
221 SessionID string `json:"session_id"`
222 RootCacheID string `json:"root_cache_id,omitempty"`
223 ParentSessionID string `json:"parent_session_id,omitempty"`
224 State ResidentState `json:"state"`
225 UpdatedAt time.Time `json:"updated_at"`
226 Usage provider.Usage `json:"usage,omitempty"`
227 ContextUsed int `json:"context_used,omitempty"`
228 ContextMax int `json:"context_max,omitempty"`
229 Subscription bool `json:"subscription,omitempty"`
230}
231
232// ResidentResult is the bounded latest-turn projection. The transcript
233// remains authoritative for detailed content; this file allows status and
234// resource readers to locate a terminal outcome without reading it all.
235type ResidentResult struct {
236 Version int `json:"version"`
237 ID string `json:"id"`
238 TurnID string `json:"turn_id"`
239 State ResidentState `json:"state"`
240 Summary string `json:"summary,omitempty"`
241 ErrorCode string `json:"error_code,omitempty"`
242 PatchRef string `json:"patch_ref,omitempty"`
243 ChangedFiles []string `json:"changed_files,omitempty"`
244 CreatedAt time.Time `json:"created_at"`
245}
246
247// ResidentJournal serializes durable child records. It is intentionally a
248// narrow storage primitive; the manager owns scheduling and provider work.
249type ResidentJournal struct {
250 mu sync.Mutex
251 eventMu sync.RWMutex
252 eventObserver func(core.AgentEvent)
253 summaryMu sync.RWMutex
254 latestSummary string
255 usageMu sync.RWMutex
256 usage provider.Usage
257 usageRecordPersisted bool
258 usageMetadataPersisted bool
259 contextUsed int
260 contextMax int
261 subscription bool
262 dir string
263 file *os.File
264 lease *residentLease
265}
266
267// ResidentUsageSnapshot is the bounded usage projection shared by durable and
268// live resident state.
269type ResidentUsageSnapshot struct {
270 Usage provider.Usage
271 ContextUsed int
272 ContextMax int
273 Subscription bool
274}
275
276// ConfigureUsage records resolved model metadata and restores the latest
277// durable usage baseline when a resident child is reconstructed.
278func (j *ResidentJournal) ConfigureUsage(contextMax int, subscription bool) ResidentUsageSnapshot {
279 if j == nil {
280 return ResidentUsageSnapshot{}
281 }
282 j.usageMu.Lock()
283 if metadata, err := ReadResidentMetadata(filepath.Join(j.dir, residentMetadataName)); err == nil {
284 j.usage = metadata.Usage
285 j.contextUsed = metadata.ContextUsed
286 }
287 j.contextMax = contextMax
288 j.subscription = subscription
289 snapshot := ResidentUsageSnapshot{Usage: j.usage, ContextUsed: j.contextUsed, ContextMax: j.contextMax, Subscription: j.subscription}
290 j.usageMu.Unlock()
291 return snapshot
292}
293
294func (j *ResidentJournal) usageSnapshot() ResidentUsageSnapshot {
295 if j == nil {
296 return ResidentUsageSnapshot{}
297 }
298 j.usageMu.RLock()
299 defer j.usageMu.RUnlock()
300 return ResidentUsageSnapshot{Usage: j.usage, ContextUsed: j.contextUsed, ContextMax: j.contextMax, Subscription: j.subscription}
301}
302
303func (j *ResidentJournal) usageRecordPersistence(record residentRecord) (durable, metadataPersisted bool) {
304 if record.Usage == nil {
305 return false, false
306 }
307 j.usageMu.RLock()
308 defer j.usageMu.RUnlock()
309 matches := j.usage == *record.Usage &&
310 j.contextUsed == record.ContextUsed &&
311 j.contextMax == record.ContextMax &&
312 j.subscription == record.Subscription
313 return j.usageRecordPersisted && matches, j.usageMetadataPersisted && matches
314}
315
316func (j *ResidentJournal) markUsageMetadataPersisted(record residentRecord) {
317 if record.Usage == nil {
318 return
319 }
320 j.usageMu.Lock()
321 if j.usage == *record.Usage &&
322 j.contextUsed == record.ContextUsed &&
323 j.contextMax == record.ContextMax &&
324 j.subscription == record.Subscription {
325 j.usageMetadataPersisted = true
326 }
327 j.usageMu.Unlock()
328}
329
330// SetEventObserver publishes in-memory agent events without creating another
331// durable transcript. Finalized events are observed only after journal append.
332func (j *ResidentJournal) SetEventObserver(observer func(core.AgentEvent)) {
333 if j == nil {
334 return
335 }
336 j.eventMu.Lock()
337 j.eventObserver = observer
338 j.eventMu.Unlock()
339}
340
341func (j *ResidentJournal) publishAgentEvent(event core.AgentEvent) {
342 if j == nil || event == nil {
343 return
344 }
345 j.eventMu.RLock()
346 observer := j.eventObserver
347 j.eventMu.RUnlock()
348 if observer != nil {
349 observer(event)
350 }
351}
352
353func OpenResidentJournal(root, childID string) (*ResidentJournal, error) {
354 if !residentChildID(childID) {
355 return nil, errors.New("resident journal: invalid child ID")
356 }
357 dir := filepath.Join(root, childID)
358 if err := os.MkdirAll(dir, 0o700); err != nil {
359 return nil, fmt.Errorf("resident journal directory: %w", err)
360 }
361 _ = os.Chmod(dir, 0o700)
362 lease, err := acquireResidentLease(dir)
363 if err != nil {
364 return nil, err
365 }
366 f, err := os.OpenFile(filepath.Join(dir, residentTranscriptName), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
367 if err != nil {
368 _ = lease.Close()
369 return nil, fmt.Errorf("resident journal transcript: %w", err)
370 }
371 _ = f.Chmod(0o600)
372 return &ResidentJournal{dir: dir, file: f, lease: lease}, nil
373}
374
375// Dir is the child-owned state directory. It is exposed only to host factory
376// code rebuilding a resident core.Agent from its authoritative transcript.
377func (j *ResidentJournal) Dir() string {
378 if j == nil {
379 return ""
380 }
381 return j.dir
382}
383
384// Accept establishes the authoritative spawn commit point. The transcript is
385// synced before metadata is projected so a crash cannot fabricate acceptance.
386func (j *ResidentJournal) Accept(spec ResidentChildSpec, prompt string) error {
387 if spec.ID == "" || spec.SessionID == "" || spec.Provider == "" || spec.Model == "" || strings.TrimSpace(prompt) == "" {
388 return errors.New("resident journal: incomplete accepted child")
389 }
390 j.mu.Lock()
391 defer j.mu.Unlock()
392 if j.file == nil {
393 return errors.New("resident journal: closed")
394 }
395 record := residentRecord{Version: residentJournalVersion, Type: residentRecordAccepted, Time: time.Now().UTC(), Spec: &spec, Prompt: prompt}
396 if err := j.appendSync(record); err != nil {
397 return err
398 }
399 return writeResidentMetadata(j.dir, j.metadata(spec, ResidentQueued, record.Time))
400}
401
402// RecordFailure makes a post-acceptance construction failure durable.
403func (j *ResidentJournal) RecordFailure(spec ResidentChildSpec) error {
404 j.mu.Lock()
405 defer j.mu.Unlock()
406 if j.file == nil {
407 return errors.New("resident journal: closed")
408 }
409 now := time.Now().UTC()
410 if err := j.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordFailed, Time: now}); err != nil {
411 return err
412 }
413 return writeResidentMetadata(j.dir, j.metadata(spec, ResidentFailed, now))
414}
415
416// AcceptFollowUp establishes a durable follow-up before a live child is told
417// about it. A restart therefore has enough information to classify the prompt
418// as unstarted or interrupted, but never to replay it automatically.
419func (j *ResidentJournal) AcceptFollowUp(spec ResidentChildSpec, turnID, prompt string) error {
420 if strings.TrimSpace(turnID) == "" || strings.TrimSpace(prompt) == "" {
421 return errors.New("resident journal: incomplete accepted turn")
422 }
423 j.mu.Lock()
424 defer j.mu.Unlock()
425 if j.file == nil {
426 return errors.New("resident journal: closed")
427 }
428 now := time.Now().UTC()
429 if err := j.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordTurnAccepted, Time: now, TurnID: turnID, Prompt: prompt}); err != nil {
430 return err
431 }
432 return writeResidentMetadata(j.dir, j.metadata(spec, ResidentQueued, now))
433}
434
435func (j *ResidentJournal) RecordTurnStarted(spec ResidentChildSpec, turnID string) error {
436 if err := j.recordTurnBoundary(spec, residentRecordTurnStarted, turnID, ResidentRunning, ""); err != nil {
437 return err
438 }
439 j.summaryMu.Lock()
440 j.latestSummary = ""
441 j.summaryMu.Unlock()
442 return nil
443}
444
445func (j *ResidentJournal) RecordTurnFinished(spec ResidentChildSpec, turnID string, err error) error {
446 return j.RecordTurnFinishedWithCapture(spec, turnID, err, nil)
447}
448
449// RecordTurnFinishedWithCapture publishes bounded workspace artifacts before
450// the terminal projection. The transcript remains the lifecycle authority.
451func (j *ResidentJournal) RecordTurnFinishedWithCapture(spec ResidentChildSpec, turnID string, turnErr error, capture *WorkspaceCapture) error {
452 state, outcome := ResidentIdle, "completed"
453 if turnErr != nil {
454 state, outcome = ResidentFailed, "failed"
455 }
456 result := ResidentResult{Version: residentJournalVersion, ID: spec.ID, TurnID: turnID, State: state, Summary: j.latestAssistantSummary(), CreatedAt: time.Now().UTC()}
457 if turnErr != nil {
458 result.ErrorCode = "turn_failed"
459 }
460 if capture != nil {
461 result.ChangedFiles = append([]string(nil), capture.ChangedFiles...)
462 if len(capture.Patch) > 0 {
463 if err := writeResidentPatch(j.dir, capture.Patch); err != nil {
464 return fmt.Errorf("resident journal write workspace patch: %w", err)
465 }
466 result.PatchRef = PatchRef(spec.ID)
467 }
468 }
469 if recordErr := j.recordTurnBoundary(spec, residentRecordTurnFinished, turnID, state, outcome); recordErr != nil {
470 return recordErr
471 }
472 return writeResidentResult(j.dir, result)
473}
474
475func (j *ResidentJournal) recordAssistantSummary(message provider.Message) {
476 summary := residentAssistantSummary(message)
477 if summary == "" {
478 return
479 }
480 j.summaryMu.Lock()
481 j.latestSummary = summary
482 j.summaryMu.Unlock()
483}
484
485func residentAssistantSummary(message provider.Message) string {
486 var text strings.Builder
487 for _, content := range message.Content {
488 if block, ok := content.(provider.TextBlock); ok {
489 text.WriteString(block.Text)
490 }
491 }
492 return truncateResidentResultSummary(text.String())
493}
494
495func (j *ResidentJournal) latestAssistantSummary() string {
496 if j == nil {
497 return ""
498 }
499 j.summaryMu.RLock()
500 summary := j.latestSummary
501 j.summaryMu.RUnlock()
502 return summary
503}
504
505func (j *ResidentJournal) clearLatestSummary() {
506 if j == nil {
507 return
508 }
509 j.summaryMu.Lock()
510 j.latestSummary = ""
511 j.summaryMu.Unlock()
512}
513
514func truncateResidentResultSummary(text string) string {
515 text = strings.TrimSpace(text)
516 if len(text) <= residentResultSummaryBytes {
517 return text
518 }
519 limit := residentResultSummaryBytes - len("…")
520 for limit > 0 && !utf8.RuneStart(text[limit]) {
521 limit--
522 }
523 return strings.TrimSpace(text[:limit]) + "…"
524}
525
526func (j *ResidentJournal) RecordTurnInterrupted(spec ResidentChildSpec, turnID string) error {
527 return j.recordTurnBoundary(spec, residentRecordInterrupted, turnID, ResidentInterrupted, "interrupted")
528}
529
530func (j *ResidentJournal) recordTurnBoundary(spec ResidentChildSpec, recordType, turnID string, state ResidentState, outcome string) error {
531 if strings.TrimSpace(turnID) == "" {
532 return errors.New("resident journal: missing turn ID")
533 }
534 j.mu.Lock()
535 defer j.mu.Unlock()
536 if j.file == nil {
537 return errors.New("resident journal: closed")
538 }
539 now := time.Now().UTC()
540 if err := j.appendSync(residentRecord{Version: residentJournalVersion, Type: recordType, Time: now, TurnID: turnID, Outcome: outcome}); err != nil {
541 return err
542 }
543 return writeResidentMetadata(j.dir, j.metadata(spec, state, now))
544}
545
546func (j *ResidentJournal) metadata(spec ResidentChildSpec, state ResidentState, updatedAt time.Time) ResidentMetadata {
547 usage := j.usageSnapshot()
548 return ResidentMetadata{
549 Version: residentJournalVersion, ID: spec.ID, SessionID: spec.SessionID,
550 RootCacheID: spec.RootCacheID, ParentSessionID: spec.ParentSessionID, State: state, UpdatedAt: updatedAt,
551 Usage: usage.Usage, ContextUsed: usage.ContextUsed, ContextMax: usage.ContextMax, Subscription: usage.Subscription,
552 }
553}
554
555// projectUsageMetadata updates the rebuildable projection while j.mu holds the
556// lifecycle ordering lock. The transcript remains authoritative if this write
557// fails and reconciliation can reconstruct the same values.
558func (j *ResidentJournal) projectUsageMetadata() error {
559 metadata, err := ReadResidentMetadata(filepath.Join(j.dir, residentMetadataName))
560 if err != nil {
561 return err
562 }
563 usage := j.usageSnapshot()
564 metadata.Usage = usage.Usage
565 metadata.ContextUsed = usage.ContextUsed
566 metadata.ContextMax = usage.ContextMax
567 metadata.Subscription = usage.Subscription
568 return writeResidentMetadata(j.dir, metadata)
569}
570
571func (j *ResidentJournal) appendSync(record residentRecord) error {
572 if j == nil || j.file == nil {
573 return errors.New("resident journal: closed")
574 }
575 data, err := json.Marshal(record)
576 if err != nil {
577 return fmt.Errorf("resident journal encode: %w", err)
578 }
579 if len(data) > residentMaxRecordBytes {
580 return errors.New("resident journal: record too large")
581 }
582 data = append(data, '\n')
583 if _, err := j.file.Write(data); err != nil {
584 return fmt.Errorf("resident journal append: %w", err)
585 }
586 if err := j.file.Sync(); err != nil {
587 return fmt.Errorf("resident journal sync: %w", err)
588 }
589 return nil
590}
591
592func (j *ResidentJournal) Close() error {
593 if j == nil {
594 return nil
595 }
596 j.mu.Lock()
597 file := j.file
598 lease := j.lease
599 j.file = nil
600 j.lease = nil
601 j.mu.Unlock()
602 var errs []error
603 if file != nil {
604 errs = append(errs, file.Close())
605 }
606 if lease != nil {
607 errs = append(errs, lease.Close())
608 }
609 return errors.Join(errs...)
610}
611
612func ReadResidentJournal(path string) ([]residentRecord, error) {
613 f, err := os.Open(path)
614 if err != nil {
615 return nil, err
616 }
617 defer f.Close()
618 reader := bufio.NewReader(f)
619 var records []residentRecord
620 for {
621 line, err := reader.ReadBytes('\n')
622 if len(line) > residentMaxRecordBytes {
623 return nil, errors.New("resident journal: record too large")
624 }
625 if len(line) > 0 && line[len(line)-1] == '\n' {
626 var record residentRecord
627 if decodeErr := json.Unmarshal(line, &record); decodeErr != nil {
628 return nil, fmt.Errorf("resident journal decode: %w", decodeErr)
629 }
630 record.raw = append(record.raw[:0], line...)
631 records = append(records, record)
632 }
633 if err == io.EOF {
634 return records, nil
635 }
636 if err != nil {
637 return nil, err
638 }
639 }
640}
641
642func ReadResidentMetadata(path string) (ResidentMetadata, error) {
643 data, err := os.ReadFile(path)
644 if err != nil {
645 return ResidentMetadata{}, err
646 }
647 var metadata ResidentMetadata
648 if err := json.Unmarshal(data, &metadata); err != nil {
649 return ResidentMetadata{}, err
650 }
651 return metadata, nil
652}
653
654func ReadResidentResult(path string) (ResidentResult, error) {
655 data, err := os.ReadFile(path)
656 if err != nil {
657 return ResidentResult{}, err
658 }
659 var result ResidentResult
660 if err := json.Unmarshal(data, &result); err != nil {
661 return ResidentResult{}, err
662 }
663 if result.Version != residentJournalVersion || result.ID == "" || result.TurnID == "" {
664 return ResidentResult{}, errors.New("resident result: invalid projection")
665 }
666 return result, nil
667}
668
669// Result reads this child's bounded latest-turn projection.
670func (j *ResidentJournal) Result() (ResidentResult, error) {
671 if j == nil {
672 return ResidentResult{}, errors.New("resident journal: unavailable")
673 }
674 return ReadResidentResult(filepath.Join(j.dir, residentResultName))
675}
676
677// ReconcileResidentJournal rebuilds the durable projection from the
678// authoritative acceptance record. Work that was queued or running when the
679// host disappeared is recorded as interrupted; it is never made runnable by
680// reconciliation and must receive an explicit new resume prompt later.
681//
682// It first takes exclusive resident ownership. A busy lease means another
683// host may still append records, so this function intentionally performs no
684// transcript inspection or recovery.
685func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
686 metadata, _, err := reconcileResidentJournalWithSpec(dir)
687 return metadata, err
688}
689
690// reconcileResidentJournalWithSpec returns the accepted specification from the
691// same ownership interval as reconciliation. Callers must not use a prior
692// snapshot to resume a child because another host could have changed the
693// journal after that snapshot was built.
694func reconcileResidentJournalWithSpec(dir string) (ResidentMetadata, ResidentChildSpec, error) {
695 journal, err := OpenResidentJournal(filepath.Dir(dir), filepath.Base(dir))
696 if err != nil {
697 return ResidentMetadata{}, ResidentChildSpec{}, err
698 }
699 metadata, reconcileErr := reconcileOwnedResidentJournal(journal)
700 var spec ResidentChildSpec
701 if reconcileErr == nil {
702 records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
703 if err != nil || len(records) == 0 || records[0].Spec == nil {
704 reconcileErr = errors.New("resident journal: invalid accepted child record")
705 } else {
706 spec = *records[0].Spec
707 }
708 }
709 if closeErr := journal.Close(); closeErr != nil {
710 reconcileErr = errors.Join(reconcileErr, closeErr)
711 }
712 if reconcileErr != nil {
713 return ResidentMetadata{}, ResidentChildSpec{}, reconcileErr
714 }
715 return metadata, spec, nil
716}
717
718// reconcileOwnedResidentJournal requires the resident journal lease held by
719// journal. It is the sole reconciliation implementation so reads and writes
720// share one ownership interval.
721func reconcileOwnedResidentJournal(journal *ResidentJournal) (ResidentMetadata, error) {
722 if journal == nil {
723 return ResidentMetadata{}, errors.New("resident journal: unavailable")
724 }
725 dir := journal.Dir()
726 records, err := ReadResidentJournal(filepath.Join(dir, residentTranscriptName))
727 if err != nil {
728 return ResidentMetadata{}, err
729 }
730 if repaired, ok := repairLegacyFalseRecovery(records); ok {
731 if err := journal.rewriteTranscript(repaired); err != nil {
732 return ResidentMetadata{}, err
733 }
734 records = repaired
735 }
736 if len(records) == 0 || records[0].Type != residentRecordAccepted || records[0].Spec == nil {
737 return ResidentMetadata{}, errors.New("resident journal: missing accepted child record")
738 }
739 spec := *records[0].Spec
740 if spec.ID == "" || spec.SessionID == "" {
741 return ResidentMetadata{}, errors.New("resident journal: invalid accepted child record")
742 }
743 state := ResidentQueued
744 lastStateAt := records[0].Time
745 lastFinishedTurn, lastFinishedOutcome, lastAssistantSummary := "", "", ""
746 usage := ResidentUsageSnapshot{}
747 seenTurns := make(map[string]string)
748 // The initial prompt is accepted atomically with child.accepted rather than
749 // through a separate turn.accepted record. Seed it so interruption and
750 // repeated reconciliation obey the same turn-state invariant as follow-ups.
751 if spec.InitialTurnID != "" {
752 seenTurns[spec.InitialTurnID] = residentRecordTurnAccepted
753 }
754 toolCalls := make(map[string]struct{})
755 toolResults := make(map[string]struct{})
756 for index, record := range records {
757 if index > 0 && record.Type == residentRecordAccepted {
758 return ResidentMetadata{}, errors.New("resident journal: duplicate accepted child record")
759 }
760 switch record.Type {
761 case residentRecordTurnAccepted:
762 if record.TurnID == "" || seenTurns[record.TurnID] != "" {
763 return ResidentMetadata{}, errors.New("resident journal: invalid accepted turn record")
764 }
765 seenTurns[record.TurnID] = residentRecordTurnAccepted
766 lastStateAt = record.Time
767 case residentRecordTurnStarted:
768 if record.TurnID == "" || (seenTurns[record.TurnID] != residentRecordTurnAccepted && record.TurnID != spec.InitialTurnID) {
769 return ResidentMetadata{}, errors.New("resident journal: turn started without acceptance")
770 }
771 seenTurns[record.TurnID] = residentRecordTurnStarted
772 state = ResidentRunning
773 lastStateAt = record.Time
774 lastAssistantSummary = ""
775 case residentRecordTurnFinished:
776 if record.TurnID == "" || seenTurns[record.TurnID] != residentRecordTurnStarted {
777 return ResidentMetadata{}, errors.New("resident journal: turn finished without start")
778 }
779 seenTurns[record.TurnID] = residentRecordTurnFinished
780 lastFinishedTurn, lastFinishedOutcome = record.TurnID, record.Outcome
781 if record.Outcome == "failed" {
782 state = ResidentFailed
783 } else {
784 state = ResidentIdle
785 }
786 lastStateAt = record.Time
787 case residentRecordAssistant:
788 message, err := core.DecodeMessageJSON(record.Message)
789 if err != nil {
790 return ResidentMetadata{}, fmt.Errorf("resident journal: invalid assistant message: %w", err)
791 }
792 if summary := residentAssistantSummary(message); summary != "" {
793 lastAssistantSummary = summary
794 }
795 case residentRecordInterrupted:
796 if record.TurnID != "" && seenTurns[record.TurnID] != residentRecordTurnAccepted && seenTurns[record.TurnID] != residentRecordTurnStarted {
797 return ResidentMetadata{}, errors.New("resident journal: turn interrupted without acceptance")
798 }
799 if record.TurnID != "" {
800 seenTurns[record.TurnID] = residentRecordInterrupted
801 }
802 state = ResidentInterrupted
803 lastStateAt = record.Time
804 case residentRecordFailed:
805 state = ResidentFailed
806 lastStateAt = record.Time
807 case residentRecordToolCall:
808 if record.ToolID == "" || record.ToolName == "" || !json.Valid(record.ToolArgs) {
809 return ResidentMetadata{}, errors.New("resident journal: invalid tool call record")
810 }
811 if _, duplicate := toolCalls[record.ToolID]; duplicate {
812 return ResidentMetadata{}, errors.New("resident journal: duplicate tool call")
813 }
814 toolCalls[record.ToolID] = struct{}{}
815 case residentRecordToolResult:
816 if record.ToolID == "" || !json.Valid(record.ToolResult) {
817 return ResidentMetadata{}, errors.New("resident journal: invalid tool result record")
818 }
819 if _, exists := toolCalls[record.ToolID]; !exists {
820 return ResidentMetadata{}, errors.New("resident journal: orphan tool result")
821 }
822 if _, duplicate := toolResults[record.ToolID]; duplicate {
823 return ResidentMetadata{}, errors.New("resident journal: duplicate tool result")
824 }
825 toolResults[record.ToolID] = struct{}{}
826 case residentRecordUsage:
827 if record.Usage == nil {
828 return ResidentMetadata{}, errors.New("resident journal: invalid usage record")
829 }
830 usage = ResidentUsageSnapshot{Usage: *record.Usage, ContextUsed: record.ContextUsed, ContextMax: record.ContextMax, Subscription: record.Subscription}
831 }
832 }
833 if lastStateAt.IsZero() {
834 lastStateAt = time.Now().UTC()
835 }
836 metadata := ResidentMetadata{
837 Version: residentJournalVersion, ID: spec.ID, SessionID: spec.SessionID,
838 RootCacheID: spec.RootCacheID, ParentSessionID: spec.ParentSessionID, State: state, UpdatedAt: lastStateAt,
839 Usage: usage.Usage, ContextUsed: usage.ContextUsed, ContextMax: usage.ContextMax, Subscription: usage.Subscription,
840 }
841 needsInterruption := state == ResidentQueued || state == ResidentRunning
842 needsToolRepair := len(toolCalls) != len(toolResults)
843 if needsInterruption || needsToolRepair {
844 // A recorded repair is new observable work. A no-op reconciliation,
845 // however, must retain the last durable lifecycle time.
846 metadata.UpdatedAt = time.Now().UTC()
847 dangling := make([]string, 0, len(toolCalls))
848 for toolID := range toolCalls {
849 if _, finished := toolResults[toolID]; !finished {
850 dangling = append(dangling, toolID)
851 }
852 }
853 sort.Strings(dangling)
854 for _, toolID := range dangling {
855 result, marshalErr := json.Marshal(core.ToolResult{IsError: true, Content: []provider.Content{provider.TextBlock{Text: residentInterruptedText}}})
856 if marshalErr != nil {
857 return ResidentMetadata{}, fmt.Errorf("resident journal encode interrupted tool result: %w", marshalErr)
858 }
859 if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: metadata.UpdatedAt, ToolID: toolID, ToolResult: result}); err != nil {
860 return ResidentMetadata{}, err
861 }
862 }
863 if !needsInterruption {
864 if err := writeResidentMetadata(dir, metadata); err != nil {
865 return ResidentMetadata{}, err
866 }
867 if err := rebuildResidentResult(dir, spec, metadata, lastFinishedTurn, lastFinishedOutcome, lastAssistantSummary); err != nil {
868 return ResidentMetadata{}, err
869 }
870 return metadata, nil
871 }
872 interrupted := false
873 turnIDs := make([]string, 0, len(seenTurns))
874 for turnID, turnState := range seenTurns {
875 if turnState != residentRecordTurnAccepted && turnState != residentRecordTurnStarted {
876 continue
877 }
878 turnIDs = append(turnIDs, turnID)
879 }
880 sort.Strings(turnIDs)
881 for _, turnID := range turnIDs {
882 if err := journal.RecordTurnInterrupted(spec, turnID); err != nil {
883 return ResidentMetadata{}, err
884 }
885 seenTurns[turnID] = residentRecordInterrupted
886 interrupted = true
887 }
888 if !interrupted {
889 if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordInterrupted, Time: metadata.UpdatedAt}); err != nil {
890 return ResidentMetadata{}, err
891 }
892 }
893 metadata.State = ResidentInterrupted
894 }
895 if err := writeResidentMetadata(dir, metadata); err != nil {
896 return ResidentMetadata{}, err
897 }
898 if err := rebuildResidentResult(dir, spec, metadata, lastFinishedTurn, lastFinishedOutcome, lastAssistantSummary); err != nil {
899 return ResidentMetadata{}, err
900 }
901 return metadata, nil
902}
903
904// repairLegacyFalseRecovery removes the exact false-recovery sequence produced
905// before resident journals had host ownership: a synthetic interruption result,
906// followed only by interruption records, before the live owner records the real
907// result for the same call. It deliberately refuses broader or ambiguous
908// duplicate histories.
909func repairLegacyFalseRecovery(records []residentRecord) ([]residentRecord, bool) {
910 for start, record := range records {
911 if record.Type != residentRecordToolResult || !isSyntheticInterruption(record.ToolResult) {
912 continue
913 }
914 turnID := activeResidentTurn(records[:start])
915 if turnID == "" {
916 continue
917 }
918 end := -1
919 interrupted := false
920 for index := start + 1; index < len(records); index++ {
921 next := records[index]
922 if next.Type == residentRecordInterrupted && next.TurnID == turnID {
923 interrupted = true
924 }
925 if next.Type == residentRecordToolResult && next.ToolID == record.ToolID && !isSyntheticInterruption(next.ToolResult) {
926 end = index
927 break
928 }
929 if next.Type != residentRecordInterrupted && (next.Type != residentRecordToolResult || next.ToolID != record.ToolID || !isSyntheticInterruption(next.ToolResult)) {
930 break
931 }
932 }
933 if end < 0 || !interrupted {
934 continue
935 }
936 repaired := make([]residentRecord, 0, len(records)-(end-start))
937 repaired = append(repaired, records[:start]...)
938 for _, candidate := range records[start:end] {
939 if candidate.Type == residentRecordInterrupted && candidate.TurnID != turnID {
940 return nil, false
941 }
942 }
943 repaired = append(repaired, records[end:]...)
944 return repaired, true
945 }
946 return nil, false
947}
948
949func activeResidentTurn(records []residentRecord) string {
950 active := ""
951 for _, record := range records {
952 switch record.Type {
953 case residentRecordTurnStarted:
954 active = record.TurnID
955 case residentRecordTurnFinished, residentRecordInterrupted:
956 if record.TurnID == active {
957 active = ""
958 }
959 }
960 }
961 return active
962}
963
964func isSyntheticInterruption(raw json.RawMessage) bool {
965 var result struct {
966 Content []struct {
967 Text string
968 }
969 IsError bool
970 }
971 if err := json.Unmarshal(raw, &result); err != nil || !result.IsError || len(result.Content) != 1 {
972 return false
973 }
974 return result.Content[0].Text == residentInterruptedText
975}
976
977// rewriteTranscript replaces the authoritative transcript while the caller
978// owns its lease. The old transcript is retained as a private recovery backup.
979func (j *ResidentJournal) rewriteTranscript(records []residentRecord) error {
980 if j == nil {
981 return errors.New("resident journal: unavailable")
982 }
983 j.mu.Lock()
984 defer j.mu.Unlock()
985 if j.file == nil || j.lease == nil {
986 return errors.New("resident journal: closed")
987 }
988 transcript := filepath.Join(j.dir, residentTranscriptName)
989 backup, err := os.CreateTemp(j.dir, ".transcript-backup-*")
990 if err != nil {
991 return fmt.Errorf("resident journal backup: %w", err)
992 }
993 backupName := backup.Name()
994 if err := backup.Chmod(0o600); err != nil {
995 _ = backup.Close()
996 _ = os.Remove(backupName)
997 return fmt.Errorf("resident journal backup permissions: %w", err)
998 }
999 source, err := os.Open(transcript)
1000 if err != nil {
1001 _ = backup.Close()
1002 _ = os.Remove(backupName)
1003 return fmt.Errorf("resident journal backup source: %w", err)
1004 }
1005 _, copyErr := io.Copy(backup, source)
1006 closeSourceErr := source.Close()
1007 if copyErr == nil {
1008 copyErr = backup.Sync()
1009 }
1010 if closeErr := backup.Close(); copyErr == nil {
1011 copyErr = closeErr
1012 }
1013 if copyErr != nil {
1014 _ = os.Remove(backupName)
1015 return fmt.Errorf("resident journal backup write: %w", errors.Join(copyErr, closeSourceErr))
1016 }
1017
1018 tmp, err := os.CreateTemp(j.dir, ".transcript-repair-*")
1019 if err != nil {
1020 return fmt.Errorf("resident journal repair: %w", err)
1021 }
1022 tmpName := tmp.Name()
1023 defer os.Remove(tmpName)
1024 if err := tmp.Chmod(0o600); err != nil {
1025 _ = tmp.Close()
1026 return fmt.Errorf("resident journal repair permissions: %w", err)
1027 }
1028 encoder := json.NewEncoder(tmp)
1029 for _, record := range records {
1030 if len(record.raw) > 0 {
1031 if _, err := tmp.Write(record.raw); err != nil {
1032 _ = tmp.Close()
1033 return fmt.Errorf("resident journal repair write: %w", err)
1034 }
1035 continue
1036 }
1037 if err := encoder.Encode(record); err != nil {
1038 _ = tmp.Close()
1039 return fmt.Errorf("resident journal repair encode: %w", err)
1040 }
1041 }
1042 if err := tmp.Sync(); err != nil {
1043 _ = tmp.Close()
1044 return fmt.Errorf("resident journal repair sync: %w", err)
1045 }
1046 if err := tmp.Close(); err != nil {
1047 return fmt.Errorf("resident journal repair close: %w", err)
1048 }
1049 if err := j.file.Close(); err != nil {
1050 return fmt.Errorf("resident journal repair close transcript: %w", err)
1051 }
1052 j.file = nil
1053 if err := os.Rename(tmpName, transcript); err != nil {
1054 return fmt.Errorf("resident journal repair rename: %w", err)
1055 }
1056 if err := syncDirectory(j.dir); err != nil {
1057 return fmt.Errorf("resident journal repair directory sync: %w", err)
1058 }
1059 file, err := os.OpenFile(transcript, os.O_APPEND|os.O_WRONLY, 0o600)
1060 if err != nil {
1061 return fmt.Errorf("resident journal repair reopen: %w", err)
1062 }
1063 j.file = file
1064 return nil
1065}
1066
1067func rebuildResidentResult(dir string, spec ResidentChildSpec, metadata ResidentMetadata, turnID, outcome, summary string) error {
1068 if turnID == "" {
1069 return nil
1070 }
1071 resultPath := filepath.Join(dir, residentResultName)
1072 if existing, err := ReadResidentResult(resultPath); err == nil && existing.ID == spec.ID && existing.TurnID == turnID && existing.State == metadata.State {
1073 return nil
1074 } else if err != nil && !errors.Is(err, os.ErrNotExist) {
1075 if removeErr := os.Remove(resultPath); removeErr != nil {
1076 return fmt.Errorf("resident journal remove corrupt result projection: %w", removeErr)
1077 }
1078 }
1079 result := ResidentResult{Version: residentJournalVersion, ID: spec.ID, TurnID: turnID, State: metadata.State, Summary: summary, CreatedAt: metadata.UpdatedAt}
1080 if outcome == "failed" {
1081 result.ErrorCode = "turn_failed"
1082 }
1083 return writeResidentResult(dir, result)
1084}
1085
1086func writeResidentMetadata(dir string, metadata ResidentMetadata) error {
1087 return writeResidentProjection(dir, residentMetadataName, metadata)
1088}
1089
1090func writeResidentResult(dir string, result ResidentResult) error {
1091 return writeResidentProjection(dir, residentResultName, result)
1092}
1093
1094func writeResidentPatch(dir string, patch []byte) error {
1095 tmp, err := os.CreateTemp(dir, ".patch-*")
1096 if err != nil {
1097 return err
1098 }
1099 tmpName := tmp.Name()
1100 defer os.Remove(tmpName)
1101 if err := tmp.Chmod(0o600); err != nil {
1102 _ = tmp.Close()
1103 return err
1104 }
1105 if _, err := tmp.Write(patch); err != nil {
1106 _ = tmp.Close()
1107 return err
1108 }
1109 if err := tmp.Sync(); err != nil {
1110 _ = tmp.Close()
1111 return err
1112 }
1113 if err := tmp.Close(); err != nil {
1114 return err
1115 }
1116 return os.Rename(tmpName, filepath.Join(dir, residentPatchName))
1117}
1118
1119func residentChildID(childID string) bool {
1120 childID = strings.TrimSpace(childID)
1121 return childID != "" && childID != "." && childID != ".." && filepath.Base(childID) == childID
1122}
1123
1124func writeResidentProjection(dir, name string, value any) error {
1125 data, err := json.Marshal(value)
1126 if err != nil {
1127 return err
1128 }
1129 path := filepath.Join(dir, name)
1130 if residentProjectionCurrent(path, data) {
1131 return nil
1132 }
1133 tmp, err := os.CreateTemp(dir, ".metadata-*")
1134 if err != nil {
1135 return err
1136 }
1137 tmpName := tmp.Name()
1138 defer os.Remove(tmpName)
1139 if err := tmp.Chmod(0o600); err != nil {
1140 _ = tmp.Close()
1141 return fmt.Errorf("resident metadata permissions: %w", err)
1142 }
1143 _, err = tmp.Write(data)
1144 if err == nil {
1145 err = tmp.Sync()
1146 }
1147 if closeErr := tmp.Close(); err == nil {
1148 err = closeErr
1149 }
1150 if err != nil {
1151 return fmt.Errorf("resident metadata write: %w", err)
1152 }
1153 if err := os.Rename(tmpName, path); err != nil {
1154 return fmt.Errorf("resident projection rename: %w", err)
1155 }
1156 if err := syncDirectory(dir); err != nil {
1157 return fmt.Errorf("resident projection directory sync: %w", err)
1158 }
1159 return nil
1160}