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