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