1package subagents
2
3import (
4 "bufio"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "os"
10 "path/filepath"
11 "sort"
12 "strings"
13 "sync"
14 "time"
15 "unicode/utf8"
16
17 "github.com/bnema/zut/packages/core"
18 "github.com/bnema/zut/packages/provider"
19)
20
21const (
22 residentMetadataName = "metadata.json"
23 residentTranscriptName = "transcript.jsonl"
24 residentResultName = "result.json"
25 residentPatchName = "patch.diff"
26 residentRecordAccepted = "child.accepted"
27 residentRecordTurnAccepted = "turn.accepted"
28 residentRecordTurnStarted = "turn.started"
29 residentRecordTurnFinished = "turn.finished"
30 residentRecordInterrupted = "child.interrupted"
31 residentRecordFailed = "child.failed"
32 residentRecordUser = "message.user"
33 residentRecordAssistant = "message.assistant"
34 residentRecordToolCall = "tool.call"
35 residentRecordToolResult = "tool.result"
36 residentJournalVersion = 1
37 residentMaxRecordBytes = 2 << 20
38 residentResultSummaryBytes = 256 << 10
39)
40
41// Artifact references are logical names; they do not disclose journal paths.
42func AgentRef(id string) string { return "subagent://" + id }
43func HistoryRef(id string) string { return AgentRef(id) + "/history" }
44func ResultRef(id string) string { return AgentRef(id) + "/result" }
45func PatchRef(id string) string { return AgentRef(id) + "/patch" }
46
47// ResidentState is the durable lifecycle used by the in-process child
48// runtime. It deliberately has no process-liveness states.
49type ResidentState string
50
51const (
52 ResidentQueued ResidentState = "queued"
53 ResidentRunning ResidentState = "running"
54 ResidentIdle ResidentState = "idle"
55 ResidentCompleted ResidentState = "completed"
56 ResidentFailed ResidentState = "failed"
57 ResidentStopped ResidentState = "stopped"
58 ResidentInterrupted ResidentState = "interrupted"
59)
60
61// ResidentChildSpec is the non-secret configuration required to reconstruct
62// one resident child. Credentials are intentionally excluded; the selected
63// provider endpoint and TLS mode are retained so explicit resume reconstructs
64// the same transport contract.
65type ResidentChildSpec struct {
66 ID string `json:"id"`
67 SessionID string `json:"session_id"`
68 InitialTurnID string `json:"initial_turn_id,omitempty"`
69 ParentSessionID string `json:"parent_session_id,omitempty"`
70 Profile string `json:"profile,omitempty"`
71 Provider string `json:"provider"`
72 BaseURL string `json:"base_url,omitempty"`
73 InsecureTLS bool `json:"insecure_tls,omitempty"`
74 Model string `json:"model"`
75 Reasoning string `json:"reasoning,omitempty"`
76 FastMode bool `json:"fast_mode,omitempty"`
77 Tools []string `json:"tools,omitempty"`
78 RepositoryRoot string `json:"repository_root,omitempty"`
79 Workspace string `json:"workspace,omitempty"`
80 WorkspaceMode WorkspaceMode `json:"workspace_mode,omitempty"`
81 WorkspaceBase string `json:"workspace_base,omitempty"`
82 WorkspaceCapture CaptureMode `json:"workspace_capture,omitempty"`
83 SystemPrompt string `json:"system_prompt,omitempty"`
84 SystemPromptMode string `json:"system_prompt_mode,omitempty"`
85 InheritProjectContext *bool `json:"inherit_project_context,omitempty"`
86 InheritSkills *bool `json:"inherit_skills,omitempty"`
87 Permissions []string `json:"permissions,omitempty"`
88 Required bool `json:"required,omitempty"`
89}
90
91type residentRecord struct {
92 Version int `json:"version"`
93 Type string `json:"type"`
94 Time time.Time `json:"time"`
95 Spec *ResidentChildSpec `json:"spec,omitempty"`
96 TurnID string `json:"turn_id,omitempty"`
97 Prompt string `json:"prompt,omitempty"`
98 Outcome string `json:"outcome,omitempty"`
99 Message json.RawMessage `json:"message,omitempty"`
100 ToolID string `json:"tool_id,omitempty"`
101 ToolName string `json:"tool_name,omitempty"`
102 ToolArgs json.RawMessage `json:"tool_args,omitempty"`
103 ToolResult json.RawMessage `json:"tool_result,omitempty"`
104}
105
106// RecordAgentEvent persists finalized provider-neutral transcript events. It
107// deliberately excludes streamed deltas and hidden reasoning, keeping the
108// journal authoritative for resumable visible history without an fsync per
109// token.
110func (j *ResidentJournal) RecordAgentEvent(event core.AgentEvent) error {
111 if j == nil || event == nil {
112 return nil
113 }
114 var record residentRecord
115 switch value := event.(type) {
116 case core.EvUserMessage:
117 message, err := json.Marshal(value.Message)
118 if err != nil {
119 return fmt.Errorf("resident journal encode user message: %w", err)
120 }
121 record = residentRecord{Type: residentRecordUser, Message: message}
122 case core.EvAssistantMessage:
123 message, err := json.Marshal(value.Message)
124 if err != nil {
125 return fmt.Errorf("resident journal encode assistant message: %w", err)
126 }
127 record = residentRecord{Type: residentRecordAssistant, Message: message}
128 case core.EvToolCall:
129 record = residentRecord{Type: residentRecordToolCall, ToolID: value.ID, ToolName: value.Name, ToolArgs: append(json.RawMessage(nil), value.Args...)}
130 case core.EvToolResult:
131 result, err := json.Marshal(value.Result)
132 if err != nil {
133 return fmt.Errorf("resident journal encode tool result: %w", err)
134 }
135 record = residentRecord{Type: residentRecordToolResult, ToolID: value.ID, ToolResult: result}
136 default:
137 j.publishAgentEvent(event)
138 return nil
139 }
140 record.Version = residentJournalVersion
141 record.Time = time.Now().UTC()
142 j.mu.Lock()
143 if j.file == nil {
144 j.mu.Unlock()
145 return errors.New("resident journal: closed")
146 }
147 err := j.appendSync(record)
148 j.mu.Unlock()
149 if err == nil {
150 switch message := event.(type) {
151 case core.EvAssistantMessage:
152 j.recordAssistantSummary(message.Message)
153 case core.EvToolCall:
154 // Core emits tool calls immediately after the assistant message that
155 // requested them. That text is an intermediate thought, not this
156 // turn's final response.
157 j.clearLatestSummary()
158 }
159 j.publishAgentEvent(event)
160 }
161 return err
162}
163
164// ResidentMetadata is a bounded, rebuildable projection. The transcript is
165// the authority for acceptance; callers must never infer acceptance solely
166// from this file.
167type ResidentMetadata struct {
168 Version int `json:"version"`
169 ID string `json:"id"`
170 SessionID string `json:"session_id"`
171 ParentSessionID string `json:"parent_session_id,omitempty"`
172 State ResidentState `json:"state"`
173 UpdatedAt time.Time `json:"updated_at"`
174}
175
176// ResidentResult is the bounded latest-turn projection. The transcript
177// remains authoritative for detailed content; this file allows status and
178// resource readers to locate a terminal outcome without reading it all.
179type ResidentResult struct {
180 Version int `json:"version"`
181 ID string `json:"id"`
182 TurnID string `json:"turn_id"`
183 State ResidentState `json:"state"`
184 Summary string `json:"summary,omitempty"`
185 ErrorCode string `json:"error_code,omitempty"`
186 PatchRef string `json:"patch_ref,omitempty"`
187 ChangedFiles []string `json:"changed_files,omitempty"`
188 CreatedAt time.Time `json:"created_at"`
189}
190
191// ResidentJournal serializes durable child records. It is intentionally a
192// narrow storage primitive; the manager owns scheduling and provider work.
193type ResidentJournal struct {
194 mu sync.Mutex
195 eventMu sync.RWMutex
196 eventObserver func(core.AgentEvent)
197 summaryMu sync.RWMutex
198 latestSummary string
199 dir string
200 file *os.File
201}
202
203// SetEventObserver publishes in-memory agent events without creating another
204// durable transcript. Finalized events are observed only after journal append.
205func (j *ResidentJournal) SetEventObserver(observer func(core.AgentEvent)) {
206 if j == nil {
207 return
208 }
209 j.eventMu.Lock()
210 j.eventObserver = observer
211 j.eventMu.Unlock()
212}
213
214func (j *ResidentJournal) publishAgentEvent(event core.AgentEvent) {
215 if j == nil || event == nil {
216 return
217 }
218 j.eventMu.RLock()
219 observer := j.eventObserver
220 j.eventMu.RUnlock()
221 if observer != nil {
222 observer(event)
223 }
224}
225
226func OpenResidentJournal(root, childID string) (*ResidentJournal, error) {
227 if !residentChildID(childID) {
228 return nil, errors.New("resident journal: invalid child ID")
229 }
230 dir := filepath.Join(root, childID)
231 if err := os.MkdirAll(dir, 0o700); err != nil {
232 return nil, fmt.Errorf("resident journal directory: %w", err)
233 }
234 _ = os.Chmod(dir, 0o700)
235 f, err := os.OpenFile(filepath.Join(dir, residentTranscriptName), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
236 if err != nil {
237 return nil, fmt.Errorf("resident journal transcript: %w", err)
238 }
239 _ = f.Chmod(0o600)
240 return &ResidentJournal{dir: dir, file: f}, nil
241}
242
243// Dir is the child-owned state directory. It is exposed only to host factory
244// code rebuilding a resident core.Agent from its authoritative transcript.
245func (j *ResidentJournal) Dir() string {
246 if j == nil {
247 return ""
248 }
249 return j.dir
250}
251
252// Accept establishes the authoritative spawn commit point. The transcript is
253// synced before metadata is projected so a crash cannot fabricate acceptance.
254func (j *ResidentJournal) Accept(spec ResidentChildSpec, prompt string) error {
255 if spec.ID == "" || spec.SessionID == "" || spec.Provider == "" || spec.Model == "" || strings.TrimSpace(prompt) == "" {
256 return errors.New("resident journal: incomplete accepted child")
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{Version: residentJournalVersion, Type: residentRecordAccepted, Time: time.Now().UTC(), Spec: &spec, Prompt: prompt}
264 if err := j.appendSync(record); err != nil {
265 return err
266 }
267 return writeResidentMetadata(j.dir, ResidentMetadata{Version: residentJournalVersion, ID: spec.ID, SessionID: spec.SessionID, ParentSessionID: spec.ParentSessionID, State: ResidentQueued, UpdatedAt: record.Time})
268}
269
270// RecordFailure makes a post-acceptance construction failure durable.
271func (j *ResidentJournal) RecordFailure(spec ResidentChildSpec) error {
272 j.mu.Lock()
273 defer j.mu.Unlock()
274 if j.file == nil {
275 return errors.New("resident journal: closed")
276 }
277 now := time.Now().UTC()
278 if err := j.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordFailed, Time: now}); err != nil {
279 return err
280 }
281 return writeResidentMetadata(j.dir, ResidentMetadata{Version: residentJournalVersion, ID: spec.ID, SessionID: spec.SessionID, ParentSessionID: spec.ParentSessionID, State: ResidentFailed, UpdatedAt: now})
282}
283
284// AcceptFollowUp establishes a durable follow-up before a live child is told
285// about it. A restart therefore has enough information to classify the prompt
286// as unstarted or interrupted, but never to replay it automatically.
287func (j *ResidentJournal) AcceptFollowUp(spec ResidentChildSpec, turnID, prompt string) error {
288 if strings.TrimSpace(turnID) == "" || strings.TrimSpace(prompt) == "" {
289 return errors.New("resident journal: incomplete accepted turn")
290 }
291 j.mu.Lock()
292 defer j.mu.Unlock()
293 if j.file == nil {
294 return errors.New("resident journal: closed")
295 }
296 now := time.Now().UTC()
297 if err := j.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordTurnAccepted, Time: now, TurnID: turnID, Prompt: prompt}); err != nil {
298 return err
299 }
300 return writeResidentMetadata(j.dir, ResidentMetadata{Version: residentJournalVersion, ID: spec.ID, SessionID: spec.SessionID, ParentSessionID: spec.ParentSessionID, State: ResidentQueued, UpdatedAt: now})
301}
302
303func (j *ResidentJournal) RecordTurnStarted(spec ResidentChildSpec, turnID string) error {
304 if err := j.recordTurnBoundary(spec, residentRecordTurnStarted, turnID, ResidentRunning, ""); err != nil {
305 return err
306 }
307 j.summaryMu.Lock()
308 j.latestSummary = ""
309 j.summaryMu.Unlock()
310 return nil
311}
312
313func (j *ResidentJournal) RecordTurnFinished(spec ResidentChildSpec, turnID string, err error) error {
314 return j.RecordTurnFinishedWithCapture(spec, turnID, err, nil)
315}
316
317// RecordTurnFinishedWithCapture publishes bounded workspace artifacts before
318// the terminal projection. The transcript remains the lifecycle authority.
319func (j *ResidentJournal) RecordTurnFinishedWithCapture(spec ResidentChildSpec, turnID string, turnErr error, capture *WorkspaceCapture) error {
320 state, outcome := ResidentIdle, "completed"
321 if turnErr != nil {
322 state, outcome = ResidentFailed, "failed"
323 }
324 result := ResidentResult{Version: residentJournalVersion, ID: spec.ID, TurnID: turnID, State: state, Summary: j.latestAssistantSummary(), CreatedAt: time.Now().UTC()}
325 if turnErr != nil {
326 result.ErrorCode = "turn_failed"
327 }
328 if capture != nil {
329 result.ChangedFiles = append([]string(nil), capture.ChangedFiles...)
330 if len(capture.Patch) > 0 {
331 if err := writeResidentPatch(j.dir, capture.Patch); err != nil {
332 return fmt.Errorf("resident journal write workspace patch: %w", err)
333 }
334 result.PatchRef = PatchRef(spec.ID)
335 }
336 }
337 if recordErr := j.recordTurnBoundary(spec, residentRecordTurnFinished, turnID, state, outcome); recordErr != nil {
338 return recordErr
339 }
340 return writeResidentResult(j.dir, result)
341}
342
343func (j *ResidentJournal) recordAssistantSummary(message provider.Message) {
344 summary := residentAssistantSummary(message)
345 if summary == "" {
346 return
347 }
348 j.summaryMu.Lock()
349 j.latestSummary = summary
350 j.summaryMu.Unlock()
351}
352
353func residentAssistantSummary(message provider.Message) string {
354 var text strings.Builder
355 for _, content := range message.Content {
356 if block, ok := content.(provider.TextBlock); ok {
357 text.WriteString(block.Text)
358 }
359 }
360 return truncateResidentResultSummary(text.String())
361}
362
363func (j *ResidentJournal) latestAssistantSummary() string {
364 if j == nil {
365 return ""
366 }
367 j.summaryMu.RLock()
368 summary := j.latestSummary
369 j.summaryMu.RUnlock()
370 return summary
371}
372
373func (j *ResidentJournal) clearLatestSummary() {
374 if j == nil {
375 return
376 }
377 j.summaryMu.Lock()
378 j.latestSummary = ""
379 j.summaryMu.Unlock()
380}
381
382func truncateResidentResultSummary(text string) string {
383 text = strings.TrimSpace(text)
384 if len(text) <= residentResultSummaryBytes {
385 return text
386 }
387 limit := residentResultSummaryBytes - len("…")
388 for limit > 0 && !utf8.RuneStart(text[limit]) {
389 limit--
390 }
391 return strings.TrimSpace(text[:limit]) + "…"
392}
393
394func (j *ResidentJournal) RecordTurnInterrupted(spec ResidentChildSpec, turnID string) error {
395 return j.recordTurnBoundary(spec, residentRecordInterrupted, turnID, ResidentInterrupted, "interrupted")
396}
397
398func (j *ResidentJournal) recordTurnBoundary(spec ResidentChildSpec, recordType, turnID string, state ResidentState, outcome string) error {
399 if strings.TrimSpace(turnID) == "" {
400 return errors.New("resident journal: missing turn ID")
401 }
402 j.mu.Lock()
403 defer j.mu.Unlock()
404 if j.file == nil {
405 return errors.New("resident journal: closed")
406 }
407 now := time.Now().UTC()
408 if err := j.appendSync(residentRecord{Version: residentJournalVersion, Type: recordType, Time: now, TurnID: turnID, Outcome: outcome}); err != nil {
409 return err
410 }
411 return writeResidentMetadata(j.dir, ResidentMetadata{Version: residentJournalVersion, ID: spec.ID, SessionID: spec.SessionID, ParentSessionID: spec.ParentSessionID, State: state, UpdatedAt: now})
412}
413
414func (j *ResidentJournal) appendSync(record residentRecord) error {
415 data, err := json.Marshal(record)
416 if err != nil {
417 return fmt.Errorf("resident journal encode: %w", err)
418 }
419 if len(data) > residentMaxRecordBytes {
420 return errors.New("resident journal: record too large")
421 }
422 data = append(data, '\n')
423 if _, err := j.file.Write(data); err != nil {
424 return fmt.Errorf("resident journal append: %w", err)
425 }
426 if err := j.file.Sync(); err != nil {
427 return fmt.Errorf("resident journal sync: %w", err)
428 }
429 return nil
430}
431
432func (j *ResidentJournal) Close() error {
433 j.mu.Lock()
434 defer j.mu.Unlock()
435 if j.file == nil {
436 return nil
437 }
438 err := j.file.Close()
439 j.file = nil
440 return err
441}
442
443func ReadResidentJournal(path string) ([]residentRecord, error) {
444 f, err := os.Open(path)
445 if err != nil {
446 return nil, err
447 }
448 defer f.Close()
449 reader := bufio.NewReader(f)
450 var records []residentRecord
451 for {
452 line, err := reader.ReadBytes('\n')
453 if len(line) > residentMaxRecordBytes {
454 return nil, errors.New("resident journal: record too large")
455 }
456 if len(line) > 0 && line[len(line)-1] == '\n' {
457 var record residentRecord
458 if decodeErr := json.Unmarshal(line, &record); decodeErr != nil {
459 return nil, fmt.Errorf("resident journal decode: %w", decodeErr)
460 }
461 records = append(records, record)
462 }
463 if err == io.EOF {
464 return records, nil
465 }
466 if err != nil {
467 return nil, err
468 }
469 }
470}
471
472func ReadResidentMetadata(path string) (ResidentMetadata, error) {
473 data, err := os.ReadFile(path)
474 if err != nil {
475 return ResidentMetadata{}, err
476 }
477 var metadata ResidentMetadata
478 if err := json.Unmarshal(data, &metadata); err != nil {
479 return ResidentMetadata{}, err
480 }
481 return metadata, nil
482}
483
484func ReadResidentResult(path string) (ResidentResult, error) {
485 data, err := os.ReadFile(path)
486 if err != nil {
487 return ResidentResult{}, err
488 }
489 var result ResidentResult
490 if err := json.Unmarshal(data, &result); err != nil {
491 return ResidentResult{}, err
492 }
493 if result.Version != residentJournalVersion || result.ID == "" || result.TurnID == "" {
494 return ResidentResult{}, errors.New("resident result: invalid projection")
495 }
496 return result, nil
497}
498
499// Result reads this child's bounded latest-turn projection.
500func (j *ResidentJournal) Result() (ResidentResult, error) {
501 if j == nil {
502 return ResidentResult{}, errors.New("resident journal: unavailable")
503 }
504 return ReadResidentResult(filepath.Join(j.dir, residentResultName))
505}
506
507// ReconcileResidentJournal rebuilds the durable projection from the
508// authoritative acceptance record. Work that was queued or running when the
509// host disappeared is recorded as interrupted; it is never made runnable by
510// reconciliation and must receive an explicit new resume prompt later.
511func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
512 records, err := ReadResidentJournal(filepath.Join(dir, residentTranscriptName))
513 if err != nil {
514 return ResidentMetadata{}, err
515 }
516 if len(records) == 0 || records[0].Type != residentRecordAccepted || records[0].Spec == nil {
517 return ResidentMetadata{}, errors.New("resident journal: missing accepted child record")
518 }
519 spec := *records[0].Spec
520 if spec.ID == "" || spec.SessionID == "" {
521 return ResidentMetadata{}, errors.New("resident journal: invalid accepted child record")
522 }
523 state := ResidentQueued
524 lastStateAt := records[0].Time
525 lastFinishedTurn, lastFinishedOutcome, lastAssistantSummary := "", "", ""
526 seenTurns := make(map[string]string)
527 // The initial prompt is accepted atomically with child.accepted rather than
528 // through a separate turn.accepted record. Seed it so interruption and
529 // repeated reconciliation obey the same turn-state invariant as follow-ups.
530 if spec.InitialTurnID != "" {
531 seenTurns[spec.InitialTurnID] = residentRecordTurnAccepted
532 }
533 toolCalls := make(map[string]struct{})
534 toolResults := make(map[string]struct{})
535 for index, record := range records {
536 if index > 0 && record.Type == residentRecordAccepted {
537 return ResidentMetadata{}, errors.New("resident journal: duplicate accepted child record")
538 }
539 switch record.Type {
540 case residentRecordTurnAccepted:
541 if record.TurnID == "" || seenTurns[record.TurnID] != "" {
542 return ResidentMetadata{}, errors.New("resident journal: invalid accepted turn record")
543 }
544 seenTurns[record.TurnID] = residentRecordTurnAccepted
545 lastStateAt = record.Time
546 case residentRecordTurnStarted:
547 if record.TurnID == "" || (seenTurns[record.TurnID] != residentRecordTurnAccepted && record.TurnID != spec.InitialTurnID) {
548 return ResidentMetadata{}, errors.New("resident journal: turn started without acceptance")
549 }
550 seenTurns[record.TurnID] = residentRecordTurnStarted
551 state = ResidentRunning
552 lastStateAt = record.Time
553 lastAssistantSummary = ""
554 case residentRecordTurnFinished:
555 if record.TurnID == "" || seenTurns[record.TurnID] != residentRecordTurnStarted {
556 return ResidentMetadata{}, errors.New("resident journal: turn finished without start")
557 }
558 seenTurns[record.TurnID] = residentRecordTurnFinished
559 lastFinishedTurn, lastFinishedOutcome = record.TurnID, record.Outcome
560 if record.Outcome == "failed" {
561 state = ResidentFailed
562 } else {
563 state = ResidentIdle
564 }
565 lastStateAt = record.Time
566 case residentRecordAssistant:
567 message, err := core.DecodeMessageJSON(record.Message)
568 if err != nil {
569 return ResidentMetadata{}, fmt.Errorf("resident journal: invalid assistant message: %w", err)
570 }
571 if summary := residentAssistantSummary(message); summary != "" {
572 lastAssistantSummary = summary
573 }
574 case residentRecordInterrupted:
575 if record.TurnID != "" && seenTurns[record.TurnID] != residentRecordTurnAccepted && seenTurns[record.TurnID] != residentRecordTurnStarted {
576 return ResidentMetadata{}, errors.New("resident journal: turn interrupted without acceptance")
577 }
578 if record.TurnID != "" {
579 seenTurns[record.TurnID] = residentRecordInterrupted
580 }
581 state = ResidentInterrupted
582 lastStateAt = record.Time
583 case residentRecordFailed:
584 state = ResidentFailed
585 lastStateAt = record.Time
586 case residentRecordToolCall:
587 if record.ToolID == "" || record.ToolName == "" || !json.Valid(record.ToolArgs) {
588 return ResidentMetadata{}, errors.New("resident journal: invalid tool call record")
589 }
590 if _, duplicate := toolCalls[record.ToolID]; duplicate {
591 return ResidentMetadata{}, errors.New("resident journal: duplicate tool call")
592 }
593 toolCalls[record.ToolID] = struct{}{}
594 case residentRecordToolResult:
595 if record.ToolID == "" || !json.Valid(record.ToolResult) {
596 return ResidentMetadata{}, errors.New("resident journal: invalid tool result record")
597 }
598 if _, exists := toolCalls[record.ToolID]; !exists {
599 return ResidentMetadata{}, errors.New("resident journal: orphan tool result")
600 }
601 if _, duplicate := toolResults[record.ToolID]; duplicate {
602 return ResidentMetadata{}, errors.New("resident journal: duplicate tool result")
603 }
604 toolResults[record.ToolID] = struct{}{}
605 }
606 }
607 if lastStateAt.IsZero() {
608 lastStateAt = time.Now().UTC()
609 }
610 metadata := ResidentMetadata{Version: residentJournalVersion, ID: spec.ID, SessionID: spec.SessionID, ParentSessionID: spec.ParentSessionID, State: state, UpdatedAt: lastStateAt}
611 needsInterruption := state == ResidentQueued || state == ResidentRunning
612 needsToolRepair := len(toolCalls) != len(toolResults)
613 if needsInterruption || needsToolRepair {
614 // A recorded repair is new observable work. A no-op reconciliation,
615 // however, must retain the last durable lifecycle time.
616 metadata.UpdatedAt = time.Now().UTC()
617 journal, err := OpenResidentJournal(filepath.Dir(dir), filepath.Base(dir))
618 if err != nil {
619 return ResidentMetadata{}, err
620 }
621 defer journal.Close()
622 dangling := make([]string, 0, len(toolCalls))
623 for toolID := range toolCalls {
624 if _, finished := toolResults[toolID]; !finished {
625 dangling = append(dangling, toolID)
626 }
627 }
628 sort.Strings(dangling)
629 for _, toolID := range dangling {
630 result, marshalErr := json.Marshal(core.ToolResult{IsError: true, Content: []provider.Content{provider.TextBlock{Text: "tool interrupted by resident host restart"}}})
631 if marshalErr != nil {
632 return ResidentMetadata{}, fmt.Errorf("resident journal encode interrupted tool result: %w", marshalErr)
633 }
634 if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: metadata.UpdatedAt, ToolID: toolID, ToolResult: result}); err != nil {
635 return ResidentMetadata{}, err
636 }
637 }
638 if !needsInterruption {
639 if err := writeResidentMetadata(dir, metadata); err != nil {
640 return ResidentMetadata{}, err
641 }
642 if err := rebuildResidentResult(dir, spec, metadata, lastFinishedTurn, lastFinishedOutcome, lastAssistantSummary); err != nil {
643 return ResidentMetadata{}, err
644 }
645 return metadata, nil
646 }
647 interrupted := false
648 turnIDs := make([]string, 0, len(seenTurns))
649 for turnID, turnState := range seenTurns {
650 if turnState != residentRecordTurnAccepted && turnState != residentRecordTurnStarted {
651 continue
652 }
653 turnIDs = append(turnIDs, turnID)
654 }
655 sort.Strings(turnIDs)
656 for _, turnID := range turnIDs {
657 if err := journal.RecordTurnInterrupted(spec, turnID); err != nil {
658 return ResidentMetadata{}, err
659 }
660 seenTurns[turnID] = residentRecordInterrupted
661 interrupted = true
662 }
663 if !interrupted {
664 if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordInterrupted, Time: metadata.UpdatedAt}); err != nil {
665 return ResidentMetadata{}, err
666 }
667 }
668 metadata.State = ResidentInterrupted
669 }
670 if err := writeResidentMetadata(dir, metadata); err != nil {
671 return ResidentMetadata{}, err
672 }
673 if err := rebuildResidentResult(dir, spec, metadata, lastFinishedTurn, lastFinishedOutcome, lastAssistantSummary); err != nil {
674 return ResidentMetadata{}, err
675 }
676 return metadata, nil
677}
678
679func rebuildResidentResult(dir string, spec ResidentChildSpec, metadata ResidentMetadata, turnID, outcome, summary string) error {
680 if turnID == "" {
681 return nil
682 }
683 resultPath := filepath.Join(dir, residentResultName)
684 if _, err := ReadResidentResult(resultPath); err == nil {
685 return nil
686 } else if !errors.Is(err, os.ErrNotExist) {
687 if removeErr := os.Remove(resultPath); removeErr != nil {
688 return fmt.Errorf("resident journal remove corrupt result projection: %w", removeErr)
689 }
690 }
691 result := ResidentResult{Version: residentJournalVersion, ID: spec.ID, TurnID: turnID, State: metadata.State, Summary: summary, CreatedAt: metadata.UpdatedAt}
692 if outcome == "failed" {
693 result.ErrorCode = "turn_failed"
694 }
695 return writeResidentResult(dir, result)
696}
697
698func writeResidentMetadata(dir string, metadata ResidentMetadata) error {
699 return writeResidentProjection(dir, residentMetadataName, metadata)
700}
701
702func writeResidentResult(dir string, result ResidentResult) error {
703 return writeResidentProjection(dir, residentResultName, result)
704}
705
706func writeResidentPatch(dir string, patch []byte) error {
707 tmp, err := os.CreateTemp(dir, ".patch-*")
708 if err != nil {
709 return err
710 }
711 tmpName := tmp.Name()
712 defer os.Remove(tmpName)
713 if err := tmp.Chmod(0o600); err != nil {
714 _ = tmp.Close()
715 return err
716 }
717 if _, err := tmp.Write(patch); err != nil {
718 _ = tmp.Close()
719 return err
720 }
721 if err := tmp.Sync(); err != nil {
722 _ = tmp.Close()
723 return err
724 }
725 if err := tmp.Close(); err != nil {
726 return err
727 }
728 return os.Rename(tmpName, filepath.Join(dir, residentPatchName))
729}
730
731func residentChildID(childID string) bool {
732 childID = strings.TrimSpace(childID)
733 return childID != "" && childID != "." && childID != ".." && filepath.Base(childID) == childID
734}
735
736func writeResidentProjection(dir, name string, value any) error {
737 data, err := json.Marshal(value)
738 if err != nil {
739 return err
740 }
741 tmp, err := os.CreateTemp(dir, ".metadata-*")
742 if err != nil {
743 return err
744 }
745 tmpName := tmp.Name()
746 defer os.Remove(tmpName)
747 if err := tmp.Chmod(0o600); err != nil {
748 _ = tmp.Close()
749 return fmt.Errorf("resident metadata permissions: %w", err)
750 }
751 _, err = tmp.Write(data)
752 if err == nil {
753 err = tmp.Sync()
754 }
755 if closeErr := tmp.Close(); err == nil {
756 err = closeErr
757 }
758 if err != nil {
759 return fmt.Errorf("resident metadata write: %w", err)
760 }
761 if err := os.Rename(tmpName, filepath.Join(dir, name)); err != nil {
762 return fmt.Errorf("resident projection rename: %w", err)
763 }
764 if err := syncDirectory(dir); err != nil {
765 return fmt.Errorf("resident projection directory sync: %w", err)
766 }
767 return nil
768}