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