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