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