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