detached 960446d

fix(subagents): lease resident journals across hosts

commit
960446d0c59ebcac899861516fab933a383d55f8
author
brice · · 2026-08-14 06:34 UTC
committer
brice · · 2026-08-14 07:51 UTC
parent
d275f18
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
10 files · +737 -33
M docs/subagents.md +15 -0
@@ -51,6 +51,8 @@ The manager records each accepted child under its managed state root:
5151 ```text
5252 subagents/<child-id>/
5353 transcript.jsonl # authoritative accepted turns and finalized messages
54+ owner.lock # host-ownership coordination; do not modify or remove
55+ .transcript-backup-* # retained private pre-repair transcript, when repaired
5456 metadata.json # rebuildable bounded state
5557 result.json # latest state plus a bounded final assistant summary
5658 patch.diff # optional worktree capture
@@ -66,6 +68,19 @@ turns are marked `interrupted`; zut never replays their task. Resume only with
6668 an explicit new prompt through `subagent_resume`, the child-session composer,
6769 or `/subagents resume <id> <prompt>`.
6870
71+Multiple zut processes may share a state root. Each resident journal has one
72+host owner at a time. A process that finds a child owned elsewhere reports it
73+as such and does not inspect, recover, or modify its transcript. Resume that
74+child from its owning host or after the owner exits. On a subsequent safe
75+reconciliation, zut repairs only the known legacy false-recovery sequence; it
76+leaves any ambiguous journal corruption untouched. A repair retains the
77+pre-repair transcript as `.transcript-backup-*`; it can contain private session
78+data and is intentionally not removed automatically.
79+
80+`subagent_status` reports `owned_elsewhere: true` when another process owns a
81+child. The field is omitted when false. A foreign-owned child cannot be resumed,
82+or have its history or result read, from this process.
83+
6984 The global scheduler admits the oldest eligible accepted prompt, runs at most
7085 one turn for each child, and defaults to six concurrent turns. Set
7186 `subagents.max_concurrent` to a positive limit to override it. Queueing is
@@ -51,6 +51,8 @@ The manager records each accepted child under its managed state root:
51 ```text51 ```text
52 subagents/<child-id>/52 subagents/<child-id>/
53 transcript.jsonl # authoritative accepted turns and finalized messages53 transcript.jsonl # authoritative accepted turns and finalized messages
54+ owner.lock # host-ownership coordination; do not modify or remove
55+ .transcript-backup-* # retained private pre-repair transcript, when repaired
54 metadata.json # rebuildable bounded state56 metadata.json # rebuildable bounded state
55 result.json # latest state plus a bounded final assistant summary57 result.json # latest state plus a bounded final assistant summary
56 patch.diff # optional worktree capture58 patch.diff # optional worktree capture
@@ -66,6 +68,19 @@ turns are marked `interrupted`; zut never replays their task. Resume only with
66 an explicit new prompt through `subagent_resume`, the child-session composer,68 an explicit new prompt through `subagent_resume`, the child-session composer,
67 or `/subagents resume <id> <prompt>`.69 or `/subagents resume <id> <prompt>`.
68 70
71+Multiple zut processes may share a state root. Each resident journal has one
72+host owner at a time. A process that finds a child owned elsewhere reports it
73+as such and does not inspect, recover, or modify its transcript. Resume that
74+child from its owning host or after the owner exits. On a subsequent safe
75+reconciliation, zut repairs only the known legacy false-recovery sequence; it
76+leaves any ambiguous journal corruption untouched. A repair retains the
77+pre-repair transcript as `.transcript-backup-*`; it can contain private session
78+data and is intentionally not removed automatically.
79+
80+`subagent_status` reports `owned_elsewhere: true` when another process owns a
81+child. The field is omitted when false. A foreign-owned child cannot be resumed,
82+or have its history or result read, from this process.
83+
69 The global scheduler admits the oldest eligible accepted prompt, runs at most84 The global scheduler admits the oldest eligible accepted prompt, runs at most
70 one turn for each child, and defaults to six concurrent turns. Set85 one turn for each child, and defaults to six concurrent turns. Set
71 `subagents.max_concurrent` to a positive limit to override it. Queueing is86 `subagents.max_concurrent` to a positive limit to override it. Queueing is
M packages/agent/subagents/resident_journal.go +243 -14
@@ -34,6 +34,7 @@ const (
3434 residentRecordToolCall = "tool.call"
3535 residentRecordToolResult = "tool.result"
3636 residentJournalVersion = 1
37+ residentInterruptedText = "tool interrupted by resident host restart"
3738 residentMaxRecordBytes = 2 << 20
3839 residentResultSummaryBytes = 256 << 10
3940 )
@@ -101,6 +102,7 @@ type residentRecord struct {
101102 ToolName string `json:"tool_name,omitempty"`
102103 ToolArgs json.RawMessage `json:"tool_args,omitempty"`
103104 ToolResult json.RawMessage `json:"tool_result,omitempty"`
105+ raw json.RawMessage `json:"-"`
104106 }
105107
106108 // RecordAgentEvent persists finalized provider-neutral transcript events. It
@@ -198,6 +200,7 @@ type ResidentJournal struct {
198200 latestSummary string
199201 dir string
200202 file *os.File
203+ lease *residentLease
201204 }
202205
203206 // SetEventObserver publishes in-memory agent events without creating another
@@ -232,12 +235,17 @@ func OpenResidentJournal(root, childID string) (*ResidentJournal, error) {
232235 return nil, fmt.Errorf("resident journal directory: %w", err)
233236 }
234237 _ = os.Chmod(dir, 0o700)
238+ lease, err := acquireResidentLease(dir)
239+ if err != nil {
240+ return nil, err
241+ }
235242 f, err := os.OpenFile(filepath.Join(dir, residentTranscriptName), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
236243 if err != nil {
244+ _ = lease.Close()
237245 return nil, fmt.Errorf("resident journal transcript: %w", err)
238246 }
239247 _ = f.Chmod(0o600)
240- return &ResidentJournal{dir: dir, file: f}, nil
248+ return &ResidentJournal{dir: dir, file: f, lease: lease}, nil
241249 }
242250
243251 // Dir is the child-owned state directory. It is exposed only to host factory
@@ -412,6 +420,9 @@ func (j *ResidentJournal) recordTurnBoundary(spec ResidentChildSpec, recordType,
412420 }
413421
414422 func (j *ResidentJournal) appendSync(record residentRecord) error {
423+ if j == nil || j.file == nil {
424+ return errors.New("resident journal: closed")
425+ }
415426 data, err := json.Marshal(record)
416427 if err != nil {
417428 return fmt.Errorf("resident journal encode: %w", err)
@@ -430,14 +441,23 @@ func (j *ResidentJournal) appendSync(record residentRecord) error {
430441 }
431442
432443 func (j *ResidentJournal) Close() error {
433- j.mu.Lock()
434- defer j.mu.Unlock()
435- if j.file == nil {
444+ if j == nil {
436445 return nil
437446 }
438- err := j.file.Close()
447+ j.mu.Lock()
448+ file := j.file
449+ lease := j.lease
439450 j.file = nil
440- return err
451+ j.lease = nil
452+ j.mu.Unlock()
453+ var errs []error
454+ if file != nil {
455+ errs = append(errs, file.Close())
456+ }
457+ if lease != nil {
458+ errs = append(errs, lease.Close())
459+ }
460+ return errors.Join(errs...)
441461 }
442462
443463 func ReadResidentJournal(path string) ([]residentRecord, error) {
@@ -458,6 +478,7 @@ func ReadResidentJournal(path string) ([]residentRecord, error) {
458478 if decodeErr := json.Unmarshal(line, &record); decodeErr != nil {
459479 return nil, fmt.Errorf("resident journal decode: %w", decodeErr)
460480 }
481+ record.raw = append(record.raw[:0], line...)
461482 records = append(records, record)
462483 }
463484 if err == io.EOF {
@@ -508,11 +529,61 @@ func (j *ResidentJournal) Result() (ResidentResult, error) {
508529 // authoritative acceptance record. Work that was queued or running when the
509530 // host disappeared is recorded as interrupted; it is never made runnable by
510531 // reconciliation and must receive an explicit new resume prompt later.
532+//
533+// It first takes exclusive resident ownership. A busy lease means another
534+// host may still append records, so this function intentionally performs no
535+// transcript inspection or recovery.
511536 func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
537+ metadata, _, err := reconcileResidentJournalWithSpec(dir)
538+ return metadata, err
539+}
540+
541+// reconcileResidentJournalWithSpec returns the accepted specification from the
542+// same ownership interval as reconciliation. Callers must not use a prior
543+// snapshot to resume a child because another host could have changed the
544+// journal after that snapshot was built.
545+func reconcileResidentJournalWithSpec(dir string) (ResidentMetadata, ResidentChildSpec, error) {
546+ journal, err := OpenResidentJournal(filepath.Dir(dir), filepath.Base(dir))
547+ if err != nil {
548+ return ResidentMetadata{}, ResidentChildSpec{}, err
549+ }
550+ metadata, reconcileErr := reconcileOwnedResidentJournal(journal)
551+ var spec ResidentChildSpec
552+ if reconcileErr == nil {
553+ records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
554+ if err != nil || len(records) == 0 || records[0].Spec == nil {
555+ reconcileErr = errors.New("resident journal: invalid accepted child record")
556+ } else {
557+ spec = *records[0].Spec
558+ }
559+ }
560+ if closeErr := journal.Close(); closeErr != nil {
561+ reconcileErr = errors.Join(reconcileErr, closeErr)
562+ }
563+ if reconcileErr != nil {
564+ return ResidentMetadata{}, ResidentChildSpec{}, reconcileErr
565+ }
566+ return metadata, spec, nil
567+}
568+
569+// reconcileOwnedResidentJournal requires the resident journal lease held by
570+// journal. It is the sole reconciliation implementation so reads and writes
571+// share one ownership interval.
572+func reconcileOwnedResidentJournal(journal *ResidentJournal) (ResidentMetadata, error) {
573+ if journal == nil {
574+ return ResidentMetadata{}, errors.New("resident journal: unavailable")
575+ }
576+ dir := journal.Dir()
512577 records, err := ReadResidentJournal(filepath.Join(dir, residentTranscriptName))
513578 if err != nil {
514579 return ResidentMetadata{}, err
515580 }
581+ if repaired, ok := repairLegacyFalseRecovery(records); ok {
582+ if err := journal.rewriteTranscript(repaired); err != nil {
583+ return ResidentMetadata{}, err
584+ }
585+ records = repaired
586+ }
516587 if len(records) == 0 || records[0].Type != residentRecordAccepted || records[0].Spec == nil {
517588 return ResidentMetadata{}, errors.New("resident journal: missing accepted child record")
518589 }
@@ -614,11 +685,6 @@ func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
614685 // A recorded repair is new observable work. A no-op reconciliation,
615686 // however, must retain the last durable lifecycle time.
616687 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()
622688 dangling := make([]string, 0, len(toolCalls))
623689 for toolID := range toolCalls {
624690 if _, finished := toolResults[toolID]; !finished {
@@ -627,7 +693,7 @@ func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
627693 }
628694 sort.Strings(dangling)
629695 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"}}})
696+ result, marshalErr := json.Marshal(core.ToolResult{IsError: true, Content: []provider.Content{provider.TextBlock{Text: residentInterruptedText}}})
631697 if marshalErr != nil {
632698 return ResidentMetadata{}, fmt.Errorf("resident journal encode interrupted tool result: %w", marshalErr)
633699 }
@@ -676,14 +742,177 @@ func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
676742 return metadata, nil
677743 }
678744
745+// repairLegacyFalseRecovery removes the exact false-recovery sequence produced
746+// before resident journals had host ownership: a synthetic interruption result,
747+// followed only by interruption records, before the live owner records the real
748+// result for the same call. It deliberately refuses broader or ambiguous
749+// duplicate histories.
750+func repairLegacyFalseRecovery(records []residentRecord) ([]residentRecord, bool) {
751+ for start, record := range records {
752+ if record.Type != residentRecordToolResult || !isSyntheticInterruption(record.ToolResult) {
753+ continue
754+ }
755+ turnID := activeResidentTurn(records[:start])
756+ if turnID == "" {
757+ continue
758+ }
759+ end := -1
760+ interrupted := false
761+ for index := start + 1; index < len(records); index++ {
762+ next := records[index]
763+ if next.Type == residentRecordInterrupted && next.TurnID == turnID {
764+ interrupted = true
765+ }
766+ if next.Type == residentRecordToolResult && next.ToolID == record.ToolID && !isSyntheticInterruption(next.ToolResult) {
767+ end = index
768+ break
769+ }
770+ if next.Type != residentRecordInterrupted && (next.Type != residentRecordToolResult || next.ToolID != record.ToolID || !isSyntheticInterruption(next.ToolResult)) {
771+ break
772+ }
773+ }
774+ if end < 0 || !interrupted {
775+ continue
776+ }
777+ repaired := make([]residentRecord, 0, len(records)-(end-start))
778+ repaired = append(repaired, records[:start]...)
779+ for _, candidate := range records[start:end] {
780+ if candidate.Type == residentRecordInterrupted && candidate.TurnID != turnID {
781+ return nil, false
782+ }
783+ }
784+ repaired = append(repaired, records[end:]...)
785+ return repaired, true
786+ }
787+ return nil, false
788+}
789+
790+func activeResidentTurn(records []residentRecord) string {
791+ active := ""
792+ for _, record := range records {
793+ switch record.Type {
794+ case residentRecordTurnStarted:
795+ active = record.TurnID
796+ case residentRecordTurnFinished, residentRecordInterrupted:
797+ if record.TurnID == active {
798+ active = ""
799+ }
800+ }
801+ }
802+ return active
803+}
804+
805+func isSyntheticInterruption(raw json.RawMessage) bool {
806+ var result struct {
807+ Content []struct {
808+ Text string
809+ }
810+ IsError bool
811+ }
812+ if err := json.Unmarshal(raw, &result); err != nil || !result.IsError || len(result.Content) != 1 {
813+ return false
814+ }
815+ return result.Content[0].Text == residentInterruptedText
816+}
817+
818+// rewriteTranscript replaces the authoritative transcript while the caller
819+// owns its lease. The old transcript is retained as a private recovery backup.
820+func (j *ResidentJournal) rewriteTranscript(records []residentRecord) error {
821+ if j == nil {
822+ return errors.New("resident journal: unavailable")
823+ }
824+ j.mu.Lock()
825+ defer j.mu.Unlock()
826+ if j.file == nil || j.lease == nil {
827+ return errors.New("resident journal: closed")
828+ }
829+ transcript := filepath.Join(j.dir, residentTranscriptName)
830+ backup, err := os.CreateTemp(j.dir, ".transcript-backup-*")
831+ if err != nil {
832+ return fmt.Errorf("resident journal backup: %w", err)
833+ }
834+ backupName := backup.Name()
835+ if err := backup.Chmod(0o600); err != nil {
836+ _ = backup.Close()
837+ _ = os.Remove(backupName)
838+ return fmt.Errorf("resident journal backup permissions: %w", err)
839+ }
840+ source, err := os.Open(transcript)
841+ if err != nil {
842+ _ = backup.Close()
843+ _ = os.Remove(backupName)
844+ return fmt.Errorf("resident journal backup source: %w", err)
845+ }
846+ _, copyErr := io.Copy(backup, source)
847+ closeSourceErr := source.Close()
848+ if copyErr == nil {
849+ copyErr = backup.Sync()
850+ }
851+ if closeErr := backup.Close(); copyErr == nil {
852+ copyErr = closeErr
853+ }
854+ if copyErr != nil {
855+ _ = os.Remove(backupName)
856+ return fmt.Errorf("resident journal backup write: %w", errors.Join(copyErr, closeSourceErr))
857+ }
858+
859+ tmp, err := os.CreateTemp(j.dir, ".transcript-repair-*")
860+ if err != nil {
861+ return fmt.Errorf("resident journal repair: %w", err)
862+ }
863+ tmpName := tmp.Name()
864+ defer os.Remove(tmpName)
865+ if err := tmp.Chmod(0o600); err != nil {
866+ _ = tmp.Close()
867+ return fmt.Errorf("resident journal repair permissions: %w", err)
868+ }
869+ encoder := json.NewEncoder(tmp)
870+ for _, record := range records {
871+ if len(record.raw) > 0 {
872+ if _, err := tmp.Write(record.raw); err != nil {
873+ _ = tmp.Close()
874+ return fmt.Errorf("resident journal repair write: %w", err)
875+ }
876+ continue
877+ }
878+ if err := encoder.Encode(record); err != nil {
879+ _ = tmp.Close()
880+ return fmt.Errorf("resident journal repair encode: %w", err)
881+ }
882+ }
883+ if err := tmp.Sync(); err != nil {
884+ _ = tmp.Close()
885+ return fmt.Errorf("resident journal repair sync: %w", err)
886+ }
887+ if err := tmp.Close(); err != nil {
888+ return fmt.Errorf("resident journal repair close: %w", err)
889+ }
890+ if err := j.file.Close(); err != nil {
891+ return fmt.Errorf("resident journal repair close transcript: %w", err)
892+ }
893+ j.file = nil
894+ if err := os.Rename(tmpName, transcript); err != nil {
895+ return fmt.Errorf("resident journal repair rename: %w", err)
896+ }
897+ if err := syncDirectory(j.dir); err != nil {
898+ return fmt.Errorf("resident journal repair directory sync: %w", err)
899+ }
900+ file, err := os.OpenFile(transcript, os.O_APPEND|os.O_WRONLY, 0o600)
901+ if err != nil {
902+ return fmt.Errorf("resident journal repair reopen: %w", err)
903+ }
904+ j.file = file
905+ return nil
906+}
907+
679908 func rebuildResidentResult(dir string, spec ResidentChildSpec, metadata ResidentMetadata, turnID, outcome, summary string) error {
680909 if turnID == "" {
681910 return nil
682911 }
683912 resultPath := filepath.Join(dir, residentResultName)
684- if _, err := ReadResidentResult(resultPath); err == nil {
913+ if existing, err := ReadResidentResult(resultPath); err == nil && existing.ID == spec.ID && existing.TurnID == turnID && existing.State == metadata.State {
685914 return nil
686- } else if !errors.Is(err, os.ErrNotExist) {
915+ } else if err != nil && !errors.Is(err, os.ErrNotExist) {
687916 if removeErr := os.Remove(resultPath); removeErr != nil {
688917 return fmt.Errorf("resident journal remove corrupt result projection: %w", removeErr)
689918 }
@@ -34,6 +34,7 @@ const (
34 residentRecordToolCall = "tool.call"34 residentRecordToolCall = "tool.call"
35 residentRecordToolResult = "tool.result"35 residentRecordToolResult = "tool.result"
36 residentJournalVersion = 136 residentJournalVersion = 1
37+ residentInterruptedText = "tool interrupted by resident host restart"
37 residentMaxRecordBytes = 2 << 2038 residentMaxRecordBytes = 2 << 20
38 residentResultSummaryBytes = 256 << 1039 residentResultSummaryBytes = 256 << 10
39 )40 )
@@ -101,6 +102,7 @@ type residentRecord struct {
101 ToolName string `json:"tool_name,omitempty"`102 ToolName string `json:"tool_name,omitempty"`
102 ToolArgs json.RawMessage `json:"tool_args,omitempty"`103 ToolArgs json.RawMessage `json:"tool_args,omitempty"`
103 ToolResult json.RawMessage `json:"tool_result,omitempty"`104 ToolResult json.RawMessage `json:"tool_result,omitempty"`
105+ raw json.RawMessage `json:"-"`
104 }106 }
105 107
106 // RecordAgentEvent persists finalized provider-neutral transcript events. It108 // RecordAgentEvent persists finalized provider-neutral transcript events. It
@@ -198,6 +200,7 @@ type ResidentJournal struct {
198 latestSummary string200 latestSummary string
199 dir string201 dir string
200 file *os.File202 file *os.File
203+ lease *residentLease
201 }204 }
202 205
203 // SetEventObserver publishes in-memory agent events without creating another206 // SetEventObserver publishes in-memory agent events without creating another
@@ -232,12 +235,17 @@ func OpenResidentJournal(root, childID string) (*ResidentJournal, error) {
232 return nil, fmt.Errorf("resident journal directory: %w", err)235 return nil, fmt.Errorf("resident journal directory: %w", err)
233 }236 }
234 _ = os.Chmod(dir, 0o700)237 _ = os.Chmod(dir, 0o700)
238+ lease, err := acquireResidentLease(dir)
239+ if err != nil {
240+ return nil, err
241+ }
235 f, err := os.OpenFile(filepath.Join(dir, residentTranscriptName), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)242 f, err := os.OpenFile(filepath.Join(dir, residentTranscriptName), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
236 if err != nil {243 if err != nil {
244+ _ = lease.Close()
237 return nil, fmt.Errorf("resident journal transcript: %w", err)245 return nil, fmt.Errorf("resident journal transcript: %w", err)
238 }246 }
239 _ = f.Chmod(0o600)247 _ = f.Chmod(0o600)
240- return &ResidentJournal{dir: dir, file: f}, nil248+ return &ResidentJournal{dir: dir, file: f, lease: lease}, nil
241 }249 }
242 250
243 // Dir is the child-owned state directory. It is exposed only to host factory251 // Dir is the child-owned state directory. It is exposed only to host factory
@@ -412,6 +420,9 @@ func (j *ResidentJournal) recordTurnBoundary(spec ResidentChildSpec, recordType,
412 }420 }
413 421
414 func (j *ResidentJournal) appendSync(record residentRecord) error {422 func (j *ResidentJournal) appendSync(record residentRecord) error {
423+ if j == nil || j.file == nil {
424+ return errors.New("resident journal: closed")
425+ }
415 data, err := json.Marshal(record)426 data, err := json.Marshal(record)
416 if err != nil {427 if err != nil {
417 return fmt.Errorf("resident journal encode: %w", err)428 return fmt.Errorf("resident journal encode: %w", err)
@@ -430,14 +441,23 @@ func (j *ResidentJournal) appendSync(record residentRecord) error {
430 }441 }
431 442
432 func (j *ResidentJournal) Close() error {443 func (j *ResidentJournal) Close() error {
433- j.mu.Lock()444+ if j == nil {
434- defer j.mu.Unlock()
435- if j.file == nil {
436 return nil445 return nil
437 }446 }
438- err := j.file.Close()447+ j.mu.Lock()
448+ file := j.file
449+ lease := j.lease
439 j.file = nil450 j.file = nil
440- return err451+ j.lease = nil
452+ j.mu.Unlock()
453+ var errs []error
454+ if file != nil {
455+ errs = append(errs, file.Close())
456+ }
457+ if lease != nil {
458+ errs = append(errs, lease.Close())
459+ }
460+ return errors.Join(errs...)
441 }461 }
442 462
443 func ReadResidentJournal(path string) ([]residentRecord, error) {463 func ReadResidentJournal(path string) ([]residentRecord, error) {
@@ -458,6 +478,7 @@ func ReadResidentJournal(path string) ([]residentRecord, error) {
458 if decodeErr := json.Unmarshal(line, &record); decodeErr != nil {478 if decodeErr := json.Unmarshal(line, &record); decodeErr != nil {
459 return nil, fmt.Errorf("resident journal decode: %w", decodeErr)479 return nil, fmt.Errorf("resident journal decode: %w", decodeErr)
460 }480 }
481+ record.raw = append(record.raw[:0], line...)
461 records = append(records, record)482 records = append(records, record)
462 }483 }
463 if err == io.EOF {484 if err == io.EOF {
@@ -508,11 +529,61 @@ func (j *ResidentJournal) Result() (ResidentResult, error) {
508 // authoritative acceptance record. Work that was queued or running when the529 // authoritative acceptance record. Work that was queued or running when the
509 // host disappeared is recorded as interrupted; it is never made runnable by530 // host disappeared is recorded as interrupted; it is never made runnable by
510 // reconciliation and must receive an explicit new resume prompt later.531 // reconciliation and must receive an explicit new resume prompt later.
532+//
533+// It first takes exclusive resident ownership. A busy lease means another
534+// host may still append records, so this function intentionally performs no
535+// transcript inspection or recovery.
511 func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {536 func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
537+ metadata, _, err := reconcileResidentJournalWithSpec(dir)
538+ return metadata, err
539+}
540+
541+// reconcileResidentJournalWithSpec returns the accepted specification from the
542+// same ownership interval as reconciliation. Callers must not use a prior
543+// snapshot to resume a child because another host could have changed the
544+// journal after that snapshot was built.
545+func reconcileResidentJournalWithSpec(dir string) (ResidentMetadata, ResidentChildSpec, error) {
546+ journal, err := OpenResidentJournal(filepath.Dir(dir), filepath.Base(dir))
547+ if err != nil {
548+ return ResidentMetadata{}, ResidentChildSpec{}, err
549+ }
550+ metadata, reconcileErr := reconcileOwnedResidentJournal(journal)
551+ var spec ResidentChildSpec
552+ if reconcileErr == nil {
553+ records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
554+ if err != nil || len(records) == 0 || records[0].Spec == nil {
555+ reconcileErr = errors.New("resident journal: invalid accepted child record")
556+ } else {
557+ spec = *records[0].Spec
558+ }
559+ }
560+ if closeErr := journal.Close(); closeErr != nil {
561+ reconcileErr = errors.Join(reconcileErr, closeErr)
562+ }
563+ if reconcileErr != nil {
564+ return ResidentMetadata{}, ResidentChildSpec{}, reconcileErr
565+ }
566+ return metadata, spec, nil
567+}
568+
569+// reconcileOwnedResidentJournal requires the resident journal lease held by
570+// journal. It is the sole reconciliation implementation so reads and writes
571+// share one ownership interval.
572+func reconcileOwnedResidentJournal(journal *ResidentJournal) (ResidentMetadata, error) {
573+ if journal == nil {
574+ return ResidentMetadata{}, errors.New("resident journal: unavailable")
575+ }
576+ dir := journal.Dir()
512 records, err := ReadResidentJournal(filepath.Join(dir, residentTranscriptName))577 records, err := ReadResidentJournal(filepath.Join(dir, residentTranscriptName))
513 if err != nil {578 if err != nil {
514 return ResidentMetadata{}, err579 return ResidentMetadata{}, err
515 }580 }
581+ if repaired, ok := repairLegacyFalseRecovery(records); ok {
582+ if err := journal.rewriteTranscript(repaired); err != nil {
583+ return ResidentMetadata{}, err
584+ }
585+ records = repaired
586+ }
516 if len(records) == 0 || records[0].Type != residentRecordAccepted || records[0].Spec == nil {587 if len(records) == 0 || records[0].Type != residentRecordAccepted || records[0].Spec == nil {
517 return ResidentMetadata{}, errors.New("resident journal: missing accepted child record")588 return ResidentMetadata{}, errors.New("resident journal: missing accepted child record")
518 }589 }
@@ -614,11 +685,6 @@ func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
614 // A recorded repair is new observable work. A no-op reconciliation,685 // A recorded repair is new observable work. A no-op reconciliation,
615 // however, must retain the last durable lifecycle time.686 // however, must retain the last durable lifecycle time.
616 metadata.UpdatedAt = time.Now().UTC()687 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))688 dangling := make([]string, 0, len(toolCalls))
623 for toolID := range toolCalls {689 for toolID := range toolCalls {
624 if _, finished := toolResults[toolID]; !finished {690 if _, finished := toolResults[toolID]; !finished {
@@ -627,7 +693,7 @@ func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
627 }693 }
628 sort.Strings(dangling)694 sort.Strings(dangling)
629 for _, toolID := range dangling {695 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"}}})696+ result, marshalErr := json.Marshal(core.ToolResult{IsError: true, Content: []provider.Content{provider.TextBlock{Text: residentInterruptedText}}})
631 if marshalErr != nil {697 if marshalErr != nil {
632 return ResidentMetadata{}, fmt.Errorf("resident journal encode interrupted tool result: %w", marshalErr)698 return ResidentMetadata{}, fmt.Errorf("resident journal encode interrupted tool result: %w", marshalErr)
633 }699 }
@@ -676,14 +742,177 @@ func ReconcileResidentJournal(dir string) (ResidentMetadata, error) {
676 return metadata, nil742 return metadata, nil
677 }743 }
678 744
745+// repairLegacyFalseRecovery removes the exact false-recovery sequence produced
746+// before resident journals had host ownership: a synthetic interruption result,
747+// followed only by interruption records, before the live owner records the real
748+// result for the same call. It deliberately refuses broader or ambiguous
749+// duplicate histories.
750+func repairLegacyFalseRecovery(records []residentRecord) ([]residentRecord, bool) {
751+ for start, record := range records {
752+ if record.Type != residentRecordToolResult || !isSyntheticInterruption(record.ToolResult) {
753+ continue
754+ }
755+ turnID := activeResidentTurn(records[:start])
756+ if turnID == "" {
757+ continue
758+ }
759+ end := -1
760+ interrupted := false
761+ for index := start + 1; index < len(records); index++ {
762+ next := records[index]
763+ if next.Type == residentRecordInterrupted && next.TurnID == turnID {
764+ interrupted = true
765+ }
766+ if next.Type == residentRecordToolResult && next.ToolID == record.ToolID && !isSyntheticInterruption(next.ToolResult) {
767+ end = index
768+ break
769+ }
770+ if next.Type != residentRecordInterrupted && (next.Type != residentRecordToolResult || next.ToolID != record.ToolID || !isSyntheticInterruption(next.ToolResult)) {
771+ break
772+ }
773+ }
774+ if end < 0 || !interrupted {
775+ continue
776+ }
777+ repaired := make([]residentRecord, 0, len(records)-(end-start))
778+ repaired = append(repaired, records[:start]...)
779+ for _, candidate := range records[start:end] {
780+ if candidate.Type == residentRecordInterrupted && candidate.TurnID != turnID {
781+ return nil, false
782+ }
783+ }
784+ repaired = append(repaired, records[end:]...)
785+ return repaired, true
786+ }
787+ return nil, false
788+}
789+
790+func activeResidentTurn(records []residentRecord) string {
791+ active := ""
792+ for _, record := range records {
793+ switch record.Type {
794+ case residentRecordTurnStarted:
795+ active = record.TurnID
796+ case residentRecordTurnFinished, residentRecordInterrupted:
797+ if record.TurnID == active {
798+ active = ""
799+ }
800+ }
801+ }
802+ return active
803+}
804+
805+func isSyntheticInterruption(raw json.RawMessage) bool {
806+ var result struct {
807+ Content []struct {
808+ Text string
809+ }
810+ IsError bool
811+ }
812+ if err := json.Unmarshal(raw, &result); err != nil || !result.IsError || len(result.Content) != 1 {
813+ return false
814+ }
815+ return result.Content[0].Text == residentInterruptedText
816+}
817+
818+// rewriteTranscript replaces the authoritative transcript while the caller
819+// owns its lease. The old transcript is retained as a private recovery backup.
820+func (j *ResidentJournal) rewriteTranscript(records []residentRecord) error {
821+ if j == nil {
822+ return errors.New("resident journal: unavailable")
823+ }
824+ j.mu.Lock()
825+ defer j.mu.Unlock()
826+ if j.file == nil || j.lease == nil {
827+ return errors.New("resident journal: closed")
828+ }
829+ transcript := filepath.Join(j.dir, residentTranscriptName)
830+ backup, err := os.CreateTemp(j.dir, ".transcript-backup-*")
831+ if err != nil {
832+ return fmt.Errorf("resident journal backup: %w", err)
833+ }
834+ backupName := backup.Name()
835+ if err := backup.Chmod(0o600); err != nil {
836+ _ = backup.Close()
837+ _ = os.Remove(backupName)
838+ return fmt.Errorf("resident journal backup permissions: %w", err)
839+ }
840+ source, err := os.Open(transcript)
841+ if err != nil {
842+ _ = backup.Close()
843+ _ = os.Remove(backupName)
844+ return fmt.Errorf("resident journal backup source: %w", err)
845+ }
846+ _, copyErr := io.Copy(backup, source)
847+ closeSourceErr := source.Close()
848+ if copyErr == nil {
849+ copyErr = backup.Sync()
850+ }
851+ if closeErr := backup.Close(); copyErr == nil {
852+ copyErr = closeErr
853+ }
854+ if copyErr != nil {
855+ _ = os.Remove(backupName)
856+ return fmt.Errorf("resident journal backup write: %w", errors.Join(copyErr, closeSourceErr))
857+ }
858+
859+ tmp, err := os.CreateTemp(j.dir, ".transcript-repair-*")
860+ if err != nil {
861+ return fmt.Errorf("resident journal repair: %w", err)
862+ }
863+ tmpName := tmp.Name()
864+ defer os.Remove(tmpName)
865+ if err := tmp.Chmod(0o600); err != nil {
866+ _ = tmp.Close()
867+ return fmt.Errorf("resident journal repair permissions: %w", err)
868+ }
869+ encoder := json.NewEncoder(tmp)
870+ for _, record := range records {
871+ if len(record.raw) > 0 {
872+ if _, err := tmp.Write(record.raw); err != nil {
873+ _ = tmp.Close()
874+ return fmt.Errorf("resident journal repair write: %w", err)
875+ }
876+ continue
877+ }
878+ if err := encoder.Encode(record); err != nil {
879+ _ = tmp.Close()
880+ return fmt.Errorf("resident journal repair encode: %w", err)
881+ }
882+ }
883+ if err := tmp.Sync(); err != nil {
884+ _ = tmp.Close()
885+ return fmt.Errorf("resident journal repair sync: %w", err)
886+ }
887+ if err := tmp.Close(); err != nil {
888+ return fmt.Errorf("resident journal repair close: %w", err)
889+ }
890+ if err := j.file.Close(); err != nil {
891+ return fmt.Errorf("resident journal repair close transcript: %w", err)
892+ }
893+ j.file = nil
894+ if err := os.Rename(tmpName, transcript); err != nil {
895+ return fmt.Errorf("resident journal repair rename: %w", err)
896+ }
897+ if err := syncDirectory(j.dir); err != nil {
898+ return fmt.Errorf("resident journal repair directory sync: %w", err)
899+ }
900+ file, err := os.OpenFile(transcript, os.O_APPEND|os.O_WRONLY, 0o600)
901+ if err != nil {
902+ return fmt.Errorf("resident journal repair reopen: %w", err)
903+ }
904+ j.file = file
905+ return nil
906+}
907+
679 func rebuildResidentResult(dir string, spec ResidentChildSpec, metadata ResidentMetadata, turnID, outcome, summary string) error {908 func rebuildResidentResult(dir string, spec ResidentChildSpec, metadata ResidentMetadata, turnID, outcome, summary string) error {
680 if turnID == "" {909 if turnID == "" {
681 return nil910 return nil
682 }911 }
683 resultPath := filepath.Join(dir, residentResultName)912 resultPath := filepath.Join(dir, residentResultName)
684- if _, err := ReadResidentResult(resultPath); err == nil {913+ if existing, err := ReadResidentResult(resultPath); err == nil && existing.ID == spec.ID && existing.TurnID == turnID && existing.State == metadata.State {
685 return nil914 return nil
686- } else if !errors.Is(err, os.ErrNotExist) {915+ } else if err != nil && !errors.Is(err, os.ErrNotExist) {
687 if removeErr := os.Remove(resultPath); removeErr != nil {916 if removeErr := os.Remove(resultPath); removeErr != nil {
688 return fmt.Errorf("resident journal remove corrupt result projection: %w", removeErr)917 return fmt.Errorf("resident journal remove corrupt result projection: %w", removeErr)
689 }918 }
M packages/agent/subagents/resident_journal_test.go +87 -0
@@ -355,6 +355,93 @@ func TestTruncateResidentResultSummaryPreservesUTF8Boundary(t *testing.T) {
355355 }
356356 }
357357
358+func TestReconcileResidentJournalRepairsLegacyFalseRecovery(t *testing.T) {
359+ root := t.TempDir()
360+ journal, err := OpenResidentJournal(root, "legacy-race")
361+ if err != nil {
362+ t.Fatal(err)
363+ }
364+ spec := ResidentChildSpec{ID: "legacy-race", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
365+ if err := journal.Accept(spec, "task"); err != nil {
366+ t.Fatal(err)
367+ }
368+ if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
369+ t.Fatal(err)
370+ }
371+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolCall, Time: time.Now().UTC(), ToolID: "call-1", ToolName: "bash", ToolArgs: json.RawMessage(`{"command":"pwd"}`)}); err != nil {
372+ t.Fatal(err)
373+ }
374+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: time.Now().UTC(), ToolID: "call-1", ToolResult: json.RawMessage(`{"Content":[{"text":"tool interrupted by resident host restart"}],"IsError":true}`)}); err != nil {
375+ t.Fatal(err)
376+ }
377+ if err := journal.RecordTurnInterrupted(spec, spec.InitialTurnID); err != nil {
378+ t.Fatal(err)
379+ }
380+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: time.Now().UTC(), ToolID: "call-1", ToolResult: json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)}); err != nil {
381+ t.Fatal(err)
382+ }
383+ if err := journal.RecordTurnFinished(spec, spec.InitialTurnID, nil); err != nil {
384+ t.Fatal(err)
385+ }
386+ if err := journal.Close(); err != nil {
387+ t.Fatal(err)
388+ }
389+
390+ metadata, err := ReconcileResidentJournal(filepath.Join(root, spec.ID))
391+ if err != nil {
392+ t.Fatal(err)
393+ }
394+ if metadata.State != ResidentIdle {
395+ t.Fatalf("state = %q, want idle", metadata.State)
396+ }
397+ records, err := ReadResidentJournal(filepath.Join(root, spec.ID, residentTranscriptName))
398+ if err != nil {
399+ t.Fatal(err)
400+ }
401+ for _, record := range records {
402+ if record.Type == residentRecordInterrupted || isSyntheticInterruption(record.ToolResult) {
403+ t.Fatalf("repaired records retained false recovery: %#v", record)
404+ }
405+ }
406+ backups, err := filepath.Glob(filepath.Join(root, spec.ID, ".transcript-backup-*"))
407+ if err != nil {
408+ t.Fatal(err)
409+ }
410+ if len(backups) != 1 {
411+ t.Fatalf("backups = %v, want one", backups)
412+ }
413+}
414+
415+func TestRepairLegacyFalseRecoveryRefusesAmbiguousBlock(t *testing.T) {
416+ synthetic := json.RawMessage(`{"Content":[{"text":"tool interrupted by resident host restart"}],"IsError":true}`)
417+ real := json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)
418+ records := []residentRecord{
419+ {Type: residentRecordTurnStarted, TurnID: "turn-1"},
420+ {Type: residentRecordToolResult, ToolID: "call-1", ToolResult: synthetic},
421+ {Type: residentRecordToolResult, ToolID: "call-2", ToolResult: synthetic},
422+ {Type: residentRecordInterrupted, TurnID: "turn-1"},
423+ {Type: residentRecordToolResult, ToolID: "call-1", ToolResult: real},
424+ }
425+ if repaired, ok := repairLegacyFalseRecovery(records); ok || repaired != nil {
426+ t.Fatalf("repairLegacyFalseRecovery = %#v, %v; want no repair", repaired, ok)
427+ }
428+}
429+
430+func TestRepairLegacyFalseRecoveryPreservesOtherTurnInterruption(t *testing.T) {
431+ synthetic := json.RawMessage(`{"Content":[{"text":"tool interrupted by resident host restart"}],"IsError":true}`)
432+ real := json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)
433+ records := []residentRecord{
434+ {Type: residentRecordTurnStarted, TurnID: "turn-1"},
435+ {Type: residentRecordToolResult, ToolID: "call-1", ToolResult: synthetic},
436+ {Type: residentRecordInterrupted, TurnID: "turn-1"},
437+ {Type: residentRecordInterrupted, TurnID: "turn-2"},
438+ {Type: residentRecordToolResult, ToolID: "call-1", ToolResult: real},
439+ }
440+ if repaired, ok := repairLegacyFalseRecovery(records); ok || repaired != nil {
441+ t.Fatalf("repairLegacyFalseRecovery = %#v, %v; want no repair", repaired, ok)
442+ }
443+}
444+
358445 func TestReconcileResidentJournalRejectsConflictingToolPairs(t *testing.T) {
359446 for _, tc := range []struct {
360447 name string
@@ -355,6 +355,93 @@ func TestTruncateResidentResultSummaryPreservesUTF8Boundary(t *testing.T) {
355 }355 }
356 }356 }
357 357
358+func TestReconcileResidentJournalRepairsLegacyFalseRecovery(t *testing.T) {
359+ root := t.TempDir()
360+ journal, err := OpenResidentJournal(root, "legacy-race")
361+ if err != nil {
362+ t.Fatal(err)
363+ }
364+ spec := ResidentChildSpec{ID: "legacy-race", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
365+ if err := journal.Accept(spec, "task"); err != nil {
366+ t.Fatal(err)
367+ }
368+ if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
369+ t.Fatal(err)
370+ }
371+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolCall, Time: time.Now().UTC(), ToolID: "call-1", ToolName: "bash", ToolArgs: json.RawMessage(`{"command":"pwd"}`)}); err != nil {
372+ t.Fatal(err)
373+ }
374+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: time.Now().UTC(), ToolID: "call-1", ToolResult: json.RawMessage(`{"Content":[{"text":"tool interrupted by resident host restart"}],"IsError":true}`)}); err != nil {
375+ t.Fatal(err)
376+ }
377+ if err := journal.RecordTurnInterrupted(spec, spec.InitialTurnID); err != nil {
378+ t.Fatal(err)
379+ }
380+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: time.Now().UTC(), ToolID: "call-1", ToolResult: json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)}); err != nil {
381+ t.Fatal(err)
382+ }
383+ if err := journal.RecordTurnFinished(spec, spec.InitialTurnID, nil); err != nil {
384+ t.Fatal(err)
385+ }
386+ if err := journal.Close(); err != nil {
387+ t.Fatal(err)
388+ }
389+
390+ metadata, err := ReconcileResidentJournal(filepath.Join(root, spec.ID))
391+ if err != nil {
392+ t.Fatal(err)
393+ }
394+ if metadata.State != ResidentIdle {
395+ t.Fatalf("state = %q, want idle", metadata.State)
396+ }
397+ records, err := ReadResidentJournal(filepath.Join(root, spec.ID, residentTranscriptName))
398+ if err != nil {
399+ t.Fatal(err)
400+ }
401+ for _, record := range records {
402+ if record.Type == residentRecordInterrupted || isSyntheticInterruption(record.ToolResult) {
403+ t.Fatalf("repaired records retained false recovery: %#v", record)
404+ }
405+ }
406+ backups, err := filepath.Glob(filepath.Join(root, spec.ID, ".transcript-backup-*"))
407+ if err != nil {
408+ t.Fatal(err)
409+ }
410+ if len(backups) != 1 {
411+ t.Fatalf("backups = %v, want one", backups)
412+ }
413+}
414+
415+func TestRepairLegacyFalseRecoveryRefusesAmbiguousBlock(t *testing.T) {
416+ synthetic := json.RawMessage(`{"Content":[{"text":"tool interrupted by resident host restart"}],"IsError":true}`)
417+ real := json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)
418+ records := []residentRecord{
419+ {Type: residentRecordTurnStarted, TurnID: "turn-1"},
420+ {Type: residentRecordToolResult, ToolID: "call-1", ToolResult: synthetic},
421+ {Type: residentRecordToolResult, ToolID: "call-2", ToolResult: synthetic},
422+ {Type: residentRecordInterrupted, TurnID: "turn-1"},
423+ {Type: residentRecordToolResult, ToolID: "call-1", ToolResult: real},
424+ }
425+ if repaired, ok := repairLegacyFalseRecovery(records); ok || repaired != nil {
426+ t.Fatalf("repairLegacyFalseRecovery = %#v, %v; want no repair", repaired, ok)
427+ }
428+}
429+
430+func TestRepairLegacyFalseRecoveryPreservesOtherTurnInterruption(t *testing.T) {
431+ synthetic := json.RawMessage(`{"Content":[{"text":"tool interrupted by resident host restart"}],"IsError":true}`)
432+ real := json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)
433+ records := []residentRecord{
434+ {Type: residentRecordTurnStarted, TurnID: "turn-1"},
435+ {Type: residentRecordToolResult, ToolID: "call-1", ToolResult: synthetic},
436+ {Type: residentRecordInterrupted, TurnID: "turn-1"},
437+ {Type: residentRecordInterrupted, TurnID: "turn-2"},
438+ {Type: residentRecordToolResult, ToolID: "call-1", ToolResult: real},
439+ }
440+ if repaired, ok := repairLegacyFalseRecovery(records); ok || repaired != nil {
441+ t.Fatalf("repairLegacyFalseRecovery = %#v, %v; want no repair", repaired, ok)
442+ }
443+}
444+
358 func TestReconcileResidentJournalRejectsConflictingToolPairs(t *testing.T) {445 func TestReconcileResidentJournalRejectsConflictingToolPairs(t *testing.T) {
359 for _, tc := range []struct {446 for _, tc := range []struct {
360 name string447 name string
A packages/agent/subagents/resident_lease.go +48 -0
@@ -0,0 +1,48 @@
1+package subagents
2+
3+import (
4+ "errors"
5+ "fmt"
6+ "os"
7+ "path/filepath"
8+)
9+
10+const residentLeaseName = "owner.lock"
11+
12+// ErrResidentLeaseBusy means another zut process owns this resident journal.
13+// Callers must not read the authoritative transcript or mutate its projections.
14+var ErrResidentLeaseBusy = errors.New("resident journal: owned by another process")
15+
16+// residentLease keeps the OS-owned exclusive lock alive for a resident's full
17+// writer lifetime. The lock file is deliberately stable: deleting or replacing
18+// it would permit separate processes to lock different filesystem objects.
19+type residentLease struct {
20+ file *os.File
21+}
22+
23+func acquireResidentLease(dir string) (*residentLease, error) {
24+ file, err := os.OpenFile(filepath.Join(dir, residentLeaseName), os.O_CREATE|os.O_RDWR, 0o600)
25+ if err != nil {
26+ return nil, fmt.Errorf("resident journal lease: %w", err)
27+ }
28+ _ = file.Chmod(0o600)
29+ if err := lockResidentLease(file); err != nil {
30+ _ = file.Close()
31+ if errors.Is(err, ErrResidentLeaseBusy) {
32+ return nil, err
33+ }
34+ return nil, fmt.Errorf("resident journal lease lock: %w", err)
35+ }
36+ return &residentLease{file: file}, nil
37+}
38+
39+func (l *residentLease) Close() error {
40+ if l == nil || l.file == nil {
41+ return nil
42+ }
43+ file := l.file
44+ l.file = nil
45+ unlockErr := unlockResidentLease(file)
46+ closeErr := file.Close()
47+ return errors.Join(unlockErr, closeErr)
48+}
@@ -0,0 +1,48 @@
1+package subagents
2+
3+import (
4+ "errors"
5+ "fmt"
6+ "os"
7+ "path/filepath"
8+)
9+
10+const residentLeaseName = "owner.lock"
11+
12+// ErrResidentLeaseBusy means another zut process owns this resident journal.
13+// Callers must not read the authoritative transcript or mutate its projections.
14+var ErrResidentLeaseBusy = errors.New("resident journal: owned by another process")
15+
16+// residentLease keeps the OS-owned exclusive lock alive for a resident's full
17+// writer lifetime. The lock file is deliberately stable: deleting or replacing
18+// it would permit separate processes to lock different filesystem objects.
19+type residentLease struct {
20+ file *os.File
21+}
22+
23+func acquireResidentLease(dir string) (*residentLease, error) {
24+ file, err := os.OpenFile(filepath.Join(dir, residentLeaseName), os.O_CREATE|os.O_RDWR, 0o600)
25+ if err != nil {
26+ return nil, fmt.Errorf("resident journal lease: %w", err)
27+ }
28+ _ = file.Chmod(0o600)
29+ if err := lockResidentLease(file); err != nil {
30+ _ = file.Close()
31+ if errors.Is(err, ErrResidentLeaseBusy) {
32+ return nil, err
33+ }
34+ return nil, fmt.Errorf("resident journal lease lock: %w", err)
35+ }
36+ return &residentLease{file: file}, nil
37+}
38+
39+func (l *residentLease) Close() error {
40+ if l == nil || l.file == nil {
41+ return nil
42+ }
43+ file := l.file
44+ l.file = nil
45+ unlockErr := unlockResidentLease(file)
46+ closeErr := file.Close()
47+ return errors.Join(unlockErr, closeErr)
48+}
A packages/agent/subagents/resident_lease_test.go +100 -0
@@ -0,0 +1,100 @@
1+package subagents
2+
3+import (
4+ "bufio"
5+ "context"
6+ "errors"
7+ "fmt"
8+ "os"
9+ "os/exec"
10+ "testing"
11+ "time"
12+)
13+
14+func TestResidentJournalLeaseExcludesOtherProcess(t *testing.T) {
15+ if os.Getenv("ZUT_RESIDENT_LEASE_HELPER") == "1" {
16+ journal, err := OpenResidentJournal(os.Getenv("ZUT_RESIDENT_LEASE_ROOT"), "child")
17+ if err != nil {
18+ t.Fatal(err)
19+ }
20+ fmt.Fprintln(os.Stdout, "ready")
21+ if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil {
22+ t.Fatal(err)
23+ }
24+ if err := journal.Close(); err != nil {
25+ t.Fatal(err)
26+ }
27+ return
28+ }
29+
30+ root := t.TempDir()
31+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
32+ defer cancel()
33+ cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestResidentJournalLeaseExcludesOtherProcess$")
34+ defer func() {
35+ if cmd.Process != nil {
36+ _ = cmd.Process.Kill()
37+ }
38+ _ = cmd.Wait()
39+ }()
40+ cmd.Env = append(os.Environ(), "ZUT_RESIDENT_LEASE_HELPER=1", "ZUT_RESIDENT_LEASE_ROOT="+root)
41+ stdout, err := cmd.StdoutPipe()
42+ if err != nil {
43+ t.Fatal(err)
44+ }
45+ stdin, err := cmd.StdinPipe()
46+ if err != nil {
47+ t.Fatal(err)
48+ }
49+ if err := cmd.Start(); err != nil {
50+ t.Fatal(err)
51+ }
52+ ready, err := bufio.NewReader(stdout).ReadString('\n')
53+ if err != nil {
54+ t.Fatal(err)
55+ }
56+ if ready != "ready\n" {
57+ t.Fatalf("helper readiness = %q", ready)
58+ }
59+ if _, err := OpenResidentJournal(root, "child"); !errors.Is(err, ErrResidentLeaseBusy) {
60+ t.Fatalf("OpenResidentJournal while other process owns lease = %v, want ErrResidentLeaseBusy", err)
61+ }
62+ if _, err := fmt.Fprintln(stdin); err != nil {
63+ t.Fatal(err)
64+ }
65+ if err := stdin.Close(); err != nil {
66+ t.Fatal(err)
67+ }
68+ if err := cmd.Wait(); err != nil {
69+ t.Fatal(err)
70+ }
71+ journal, err := OpenResidentJournal(root, "child")
72+ if err != nil {
73+ t.Fatalf("OpenResidentJournal after owner exit: %v", err)
74+ }
75+ if err := journal.Close(); err != nil {
76+ t.Fatal(err)
77+ }
78+}
79+
80+func TestResidentJournalLeaseExcludesSecondOwner(t *testing.T) {
81+ root := t.TempDir()
82+ first, err := OpenResidentJournal(root, "child")
83+ if err != nil {
84+ t.Fatal(err)
85+ }
86+ defer first.Close()
87+ if _, err := OpenResidentJournal(root, "child"); !errors.Is(err, ErrResidentLeaseBusy) {
88+ t.Fatalf("second OpenResidentJournal error = %v, want ErrResidentLeaseBusy", err)
89+ }
90+ if err := first.Close(); err != nil {
91+ t.Fatal(err)
92+ }
93+ second, err := OpenResidentJournal(root, "child")
94+ if err != nil {
95+ t.Fatalf("OpenResidentJournal after Close: %v", err)
96+ }
97+ if err := second.Close(); err != nil {
98+ t.Fatal(err)
99+ }
100+}
@@ -0,0 +1,100 @@
1+package subagents
2+
3+import (
4+ "bufio"
5+ "context"
6+ "errors"
7+ "fmt"
8+ "os"
9+ "os/exec"
10+ "testing"
11+ "time"
12+)
13+
14+func TestResidentJournalLeaseExcludesOtherProcess(t *testing.T) {
15+ if os.Getenv("ZUT_RESIDENT_LEASE_HELPER") == "1" {
16+ journal, err := OpenResidentJournal(os.Getenv("ZUT_RESIDENT_LEASE_ROOT"), "child")
17+ if err != nil {
18+ t.Fatal(err)
19+ }
20+ fmt.Fprintln(os.Stdout, "ready")
21+ if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil {
22+ t.Fatal(err)
23+ }
24+ if err := journal.Close(); err != nil {
25+ t.Fatal(err)
26+ }
27+ return
28+ }
29+
30+ root := t.TempDir()
31+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
32+ defer cancel()
33+ cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestResidentJournalLeaseExcludesOtherProcess$")
34+ defer func() {
35+ if cmd.Process != nil {
36+ _ = cmd.Process.Kill()
37+ }
38+ _ = cmd.Wait()
39+ }()
40+ cmd.Env = append(os.Environ(), "ZUT_RESIDENT_LEASE_HELPER=1", "ZUT_RESIDENT_LEASE_ROOT="+root)
41+ stdout, err := cmd.StdoutPipe()
42+ if err != nil {
43+ t.Fatal(err)
44+ }
45+ stdin, err := cmd.StdinPipe()
46+ if err != nil {
47+ t.Fatal(err)
48+ }
49+ if err := cmd.Start(); err != nil {
50+ t.Fatal(err)
51+ }
52+ ready, err := bufio.NewReader(stdout).ReadString('\n')
53+ if err != nil {
54+ t.Fatal(err)
55+ }
56+ if ready != "ready\n" {
57+ t.Fatalf("helper readiness = %q", ready)
58+ }
59+ if _, err := OpenResidentJournal(root, "child"); !errors.Is(err, ErrResidentLeaseBusy) {
60+ t.Fatalf("OpenResidentJournal while other process owns lease = %v, want ErrResidentLeaseBusy", err)
61+ }
62+ if _, err := fmt.Fprintln(stdin); err != nil {
63+ t.Fatal(err)
64+ }
65+ if err := stdin.Close(); err != nil {
66+ t.Fatal(err)
67+ }
68+ if err := cmd.Wait(); err != nil {
69+ t.Fatal(err)
70+ }
71+ journal, err := OpenResidentJournal(root, "child")
72+ if err != nil {
73+ t.Fatalf("OpenResidentJournal after owner exit: %v", err)
74+ }
75+ if err := journal.Close(); err != nil {
76+ t.Fatal(err)
77+ }
78+}
79+
80+func TestResidentJournalLeaseExcludesSecondOwner(t *testing.T) {
81+ root := t.TempDir()
82+ first, err := OpenResidentJournal(root, "child")
83+ if err != nil {
84+ t.Fatal(err)
85+ }
86+ defer first.Close()
87+ if _, err := OpenResidentJournal(root, "child"); !errors.Is(err, ErrResidentLeaseBusy) {
88+ t.Fatalf("second OpenResidentJournal error = %v, want ErrResidentLeaseBusy", err)
89+ }
90+ if err := first.Close(); err != nil {
91+ t.Fatal(err)
92+ }
93+ second, err := OpenResidentJournal(root, "child")
94+ if err != nil {
95+ t.Fatalf("OpenResidentJournal after Close: %v", err)
96+ }
97+ if err := second.Close(); err != nil {
98+ t.Fatal(err)
99+ }
100+}
A packages/agent/subagents/resident_lease_unix.go +22 -0
@@ -0,0 +1,22 @@
1+//go:build !windows
2+
3+package subagents
4+
5+import (
6+ "errors"
7+ "os"
8+
9+ "golang.org/x/sys/unix"
10+)
11+
12+func lockResidentLease(file *os.File) error {
13+ err := unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB)
14+ if errors.Is(err, unix.EWOULDBLOCK) || errors.Is(err, unix.EAGAIN) {
15+ return ErrResidentLeaseBusy
16+ }
17+ return err
18+}
19+
20+func unlockResidentLease(file *os.File) error {
21+ return unix.Flock(int(file.Fd()), unix.LOCK_UN)
22+}
@@ -0,0 +1,22 @@
1+//go:build !windows
2+
3+package subagents
4+
5+import (
6+ "errors"
7+ "os"
8+
9+ "golang.org/x/sys/unix"
10+)
11+
12+func lockResidentLease(file *os.File) error {
13+ err := unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB)
14+ if errors.Is(err, unix.EWOULDBLOCK) || errors.Is(err, unix.EAGAIN) {
15+ return ErrResidentLeaseBusy
16+ }
17+ return err
18+}
19+
20+func unlockResidentLease(file *os.File) error {
21+ return unix.Flock(int(file.Fd()), unix.LOCK_UN)
22+}
A packages/agent/subagents/resident_lease_windows.go +29 -0
@@ -0,0 +1,29 @@
1+//go:build windows
2+
3+package subagents
4+
5+import (
6+ "errors"
7+ "os"
8+
9+ "golang.org/x/sys/windows"
10+)
11+
12+func lockResidentLease(file *os.File) error {
13+ err := windows.LockFileEx(
14+ windows.Handle(file.Fd()),
15+ windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY,
16+ 0,
17+ 1,
18+ 0,
19+ &windows.Overlapped{},
20+ )
21+ if errors.Is(err, windows.ERROR_LOCK_VIOLATION) || errors.Is(err, windows.ERROR_SHARING_VIOLATION) {
22+ return ErrResidentLeaseBusy
23+ }
24+ return err
25+}
26+
27+func unlockResidentLease(file *os.File) error {
28+ return windows.UnlockFileEx(windows.Handle(file.Fd()), 0, 1, 0, &windows.Overlapped{})
29+}
@@ -0,0 +1,29 @@
1+//go:build windows
2+
3+package subagents
4+
5+import (
6+ "errors"
7+ "os"
8+
9+ "golang.org/x/sys/windows"
10+)
11+
12+func lockResidentLease(file *os.File) error {
13+ err := windows.LockFileEx(
14+ windows.Handle(file.Fd()),
15+ windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY,
16+ 0,
17+ 1,
18+ 0,
19+ &windows.Overlapped{},
20+ )
21+ if errors.Is(err, windows.ERROR_LOCK_VIOLATION) || errors.Is(err, windows.ERROR_SHARING_VIOLATION) {
22+ return ErrResidentLeaseBusy
23+ }
24+ return err
25+}
26+
27+func unlockResidentLease(file *os.File) error {
28+ return windows.UnlockFileEx(windows.Handle(file.Fd()), 0, 1, 0, &windows.Overlapped{})
29+}
M packages/agent/subagents/resident_manager.go +68 -11
@@ -156,6 +156,7 @@ type ResidentSnapshot struct {
156156 TurnStartedAt time.Time
157157 ActivityUpdatedAt time.Time
158158 WaitingForModel bool
159+ OwnedElsewhere bool
159160 }
160161
161162 func residentSnapshot(child *ResidentChild) ResidentSnapshot {
@@ -400,6 +401,12 @@ func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) er
400401 }
401402 child := m.children[childID]
402403 spec, recovered := m.recoveredSpec[childID]
404+ snapshot := m.recovered[childID]
405+ // A recovered foreign snapshot is advisory. Its owner may have exited since
406+ // Reconcile; OpenResidentJournal below makes the authoritative fresh check.
407+ if child == nil && snapshot.OwnedElsewhere {
408+ recovered = true
409+ }
403410 if child == nil && recovered {
404411 if _, pending := m.pending[childID]; pending {
405412 m.mu.Unlock()
@@ -412,8 +419,21 @@ func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) er
412419 defer func() { m.mu.Lock(); delete(m.pending, childID); m.mu.Unlock() }()
413420 journal, err := OpenResidentJournal(m.root, childID)
414421 if err != nil {
422+ if errors.Is(err, ErrResidentLeaseBusy) {
423+ return errors.New("resident manager: child is owned by another zut process")
424+ }
425+ return err
426+ }
427+ if _, err := reconcileOwnedResidentJournal(journal); err != nil {
428+ _ = journal.Close()
415429 return err
416430 }
431+ records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
432+ if err != nil || len(records) == 0 || records[0].Spec == nil {
433+ _ = journal.Close()
434+ return errors.New("resident manager: invalid accepted specification")
435+ }
436+ spec = *records[0].Spec
417437 var workspace WorkspaceHandle
418438 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
419439 if workspaceRoot == "" {
@@ -750,21 +770,33 @@ func (m *ResidentManager) Reconcile() []error {
750770 if live || pending {
751771 continue
752772 }
753- metadata, reconcileErr := ReconcileResidentJournal(filepath.Join(m.root, childID))
754- if reconcileErr != nil {
755- errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
773+ childDir := filepath.Join(m.root, childID)
774+ metadata, spec, reconcileErr := reconcileResidentJournalWithSpec(childDir)
775+ if errors.Is(reconcileErr, ErrResidentLeaseBusy) {
776+ foreign, readErr := ReadResidentMetadata(filepath.Join(childDir, residentMetadataName))
777+ if readErr != nil && !errors.Is(readErr, os.ErrNotExist) {
778+ errs = append(errs, fmt.Errorf("resident child %s: read foreign metadata: %w", childID, readErr))
779+ continue
780+ }
781+ state, updatedAt := foreign.State, foreign.UpdatedAt
782+ if readErr != nil || state == "" {
783+ // Ownership begins before child.accepted is durable. Do not read
784+ // that live transcript merely to improve this transient snapshot.
785+ state, updatedAt = ResidentQueued, time.Now().UTC()
786+ }
787+ m.mu.Lock()
788+ m.recovered[childID] = ResidentSnapshot{ID: childID, State: state, UpdatedAt: updatedAt, OwnedElsewhere: true}
789+ m.mu.Unlock()
756790 continue
757791 }
758- records, readErr := ReadResidentJournal(filepath.Join(m.root, childID, residentTranscriptName))
759- if readErr != nil || len(records) == 0 || records[0].Spec == nil {
760- errs = append(errs, fmt.Errorf("resident child %s: invalid accepted specification", childID))
792+ if reconcileErr != nil {
793+ errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
761794 continue
762795 }
763- spec := records[0].Spec
764796 m.mu.Lock()
765797 if _, live := m.children[childID]; !live {
766798 m.recovered[childID] = ResidentSnapshot{ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model, WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt}
767- m.recoveredSpec[childID] = *spec
799+ m.recoveredSpec[childID] = spec
768800 }
769801 m.mu.Unlock()
770802 }
@@ -824,10 +856,11 @@ func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryI
824856 if m == nil {
825857 return nil, errors.New("resident manager: unavailable")
826858 }
827- dir, err := ResidentHistoryDir(m.root, childID)
859+ dir, release, err := m.readableResidentDir(childID)
828860 if err != nil {
829861 return nil, err
830862 }
863+ defer release()
831864 return ReadResidentHistory(dir, limit)
832865 }
833866
@@ -838,10 +871,11 @@ func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (R
838871 if m == nil {
839872 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
840873 }
841- dir, err := ResidentHistoryDir(m.root, childID)
874+ dir, release, err := m.readableResidentDir(childID)
842875 if err != nil {
843876 return ResidentHistoryPage{}, err
844877 }
878+ defer release()
845879 return ReadResidentHistoryPage(dir, olderCursor, limit)
846880 }
847881
@@ -850,13 +884,36 @@ func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
850884 if m == nil {
851885 return ResidentResult{}, errors.New("resident manager: unavailable")
852886 }
853- dir, err := ResidentHistoryDir(m.root, childID)
887+ dir, release, err := m.readableResidentDir(childID)
854888 if err != nil {
855889 return ResidentResult{}, err
856890 }
891+ defer release()
857892 return ReadResidentResult(filepath.Join(dir, residentResultName))
858893 }
859894
895+func (m *ResidentManager) readableResidentDir(childID string) (string, func(), error) {
896+ dir, err := ResidentHistoryDir(m.root, childID)
897+ if err != nil {
898+ return "", nil, err
899+ }
900+ childID = strings.TrimSpace(childID)
901+ m.mu.Lock()
902+ live := m.children[childID] != nil
903+ m.mu.Unlock()
904+ if live {
905+ return dir, func() {}, nil
906+ }
907+ lease, err := acquireResidentLease(dir)
908+ if err != nil {
909+ if errors.Is(err, ErrResidentLeaseBusy) {
910+ return "", nil, errors.New("resident manager: child is owned by another zut process")
911+ }
912+ return "", nil, err
913+ }
914+ return dir, func() { _ = lease.Close() }, nil
915+}
916+
860917 func (m *ResidentManager) Close(ctx context.Context) error {
861918 if m == nil {
862919 return nil
@@ -156,6 +156,7 @@ type ResidentSnapshot struct {
156 TurnStartedAt time.Time156 TurnStartedAt time.Time
157 ActivityUpdatedAt time.Time157 ActivityUpdatedAt time.Time
158 WaitingForModel bool158 WaitingForModel bool
159+ OwnedElsewhere bool
159 }160 }
160 161
161 func residentSnapshot(child *ResidentChild) ResidentSnapshot {162 func residentSnapshot(child *ResidentChild) ResidentSnapshot {
@@ -400,6 +401,12 @@ func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) er
400 }401 }
401 child := m.children[childID]402 child := m.children[childID]
402 spec, recovered := m.recoveredSpec[childID]403 spec, recovered := m.recoveredSpec[childID]
404+ snapshot := m.recovered[childID]
405+ // A recovered foreign snapshot is advisory. Its owner may have exited since
406+ // Reconcile; OpenResidentJournal below makes the authoritative fresh check.
407+ if child == nil && snapshot.OwnedElsewhere {
408+ recovered = true
409+ }
403 if child == nil && recovered {410 if child == nil && recovered {
404 if _, pending := m.pending[childID]; pending {411 if _, pending := m.pending[childID]; pending {
405 m.mu.Unlock()412 m.mu.Unlock()
@@ -412,8 +419,21 @@ func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) er
412 defer func() { m.mu.Lock(); delete(m.pending, childID); m.mu.Unlock() }()419 defer func() { m.mu.Lock(); delete(m.pending, childID); m.mu.Unlock() }()
413 journal, err := OpenResidentJournal(m.root, childID)420 journal, err := OpenResidentJournal(m.root, childID)
414 if err != nil {421 if err != nil {
422+ if errors.Is(err, ErrResidentLeaseBusy) {
423+ return errors.New("resident manager: child is owned by another zut process")
424+ }
425+ return err
426+ }
427+ if _, err := reconcileOwnedResidentJournal(journal); err != nil {
428+ _ = journal.Close()
415 return err429 return err
416 }430 }
431+ records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
432+ if err != nil || len(records) == 0 || records[0].Spec == nil {
433+ _ = journal.Close()
434+ return errors.New("resident manager: invalid accepted specification")
435+ }
436+ spec = *records[0].Spec
417 var workspace WorkspaceHandle437 var workspace WorkspaceHandle
418 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)438 workspaceRoot := strings.TrimSpace(spec.RepositoryRoot)
419 if workspaceRoot == "" {439 if workspaceRoot == "" {
@@ -750,21 +770,33 @@ func (m *ResidentManager) Reconcile() []error {
750 if live || pending {770 if live || pending {
751 continue771 continue
752 }772 }
753- metadata, reconcileErr := ReconcileResidentJournal(filepath.Join(m.root, childID))773+ childDir := filepath.Join(m.root, childID)
754- if reconcileErr != nil {774+ metadata, spec, reconcileErr := reconcileResidentJournalWithSpec(childDir)
755- errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))775+ if errors.Is(reconcileErr, ErrResidentLeaseBusy) {
776+ foreign, readErr := ReadResidentMetadata(filepath.Join(childDir, residentMetadataName))
777+ if readErr != nil && !errors.Is(readErr, os.ErrNotExist) {
778+ errs = append(errs, fmt.Errorf("resident child %s: read foreign metadata: %w", childID, readErr))
779+ continue
780+ }
781+ state, updatedAt := foreign.State, foreign.UpdatedAt
782+ if readErr != nil || state == "" {
783+ // Ownership begins before child.accepted is durable. Do not read
784+ // that live transcript merely to improve this transient snapshot.
785+ state, updatedAt = ResidentQueued, time.Now().UTC()
786+ }
787+ m.mu.Lock()
788+ m.recovered[childID] = ResidentSnapshot{ID: childID, State: state, UpdatedAt: updatedAt, OwnedElsewhere: true}
789+ m.mu.Unlock()
756 continue790 continue
757 }791 }
758- records, readErr := ReadResidentJournal(filepath.Join(m.root, childID, residentTranscriptName))792+ if reconcileErr != nil {
759- if readErr != nil || len(records) == 0 || records[0].Spec == nil {793+ errs = append(errs, fmt.Errorf("resident child %s: %w", childID, reconcileErr))
760- errs = append(errs, fmt.Errorf("resident child %s: invalid accepted specification", childID))
761 continue794 continue
762 }795 }
763- spec := records[0].Spec
764 m.mu.Lock()796 m.mu.Lock()
765 if _, live := m.children[childID]; !live {797 if _, live := m.children[childID]; !live {
766 m.recovered[childID] = ResidentSnapshot{ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model, WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt}798 m.recovered[childID] = ResidentSnapshot{ID: childID, State: metadata.State, Profile: spec.Profile, Provider: spec.Provider, Model: spec.Model, WorkspaceMode: spec.WorkspaceMode, Required: spec.Required, UpdatedAt: metadata.UpdatedAt}
767- m.recoveredSpec[childID] = *spec799+ m.recoveredSpec[childID] = spec
768 }800 }
769 m.mu.Unlock()801 m.mu.Unlock()
770 }802 }
@@ -824,10 +856,11 @@ func (m *ResidentManager) History(childID string, limit int) ([]ResidentHistoryI
824 if m == nil {856 if m == nil {
825 return nil, errors.New("resident manager: unavailable")857 return nil, errors.New("resident manager: unavailable")
826 }858 }
827- dir, err := ResidentHistoryDir(m.root, childID)859+ dir, release, err := m.readableResidentDir(childID)
828 if err != nil {860 if err != nil {
829 return nil, err861 return nil, err
830 }862 }
863+ defer release()
831 return ReadResidentHistory(dir, limit)864 return ReadResidentHistory(dir, limit)
832 }865 }
833 866
@@ -838,10 +871,11 @@ func (m *ResidentManager) HistoryPage(childID, olderCursor string, limit int) (R
838 if m == nil {871 if m == nil {
839 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")872 return ResidentHistoryPage{}, errors.New("resident manager: unavailable")
840 }873 }
841- dir, err := ResidentHistoryDir(m.root, childID)874+ dir, release, err := m.readableResidentDir(childID)
842 if err != nil {875 if err != nil {
843 return ResidentHistoryPage{}, err876 return ResidentHistoryPage{}, err
844 }877 }
878+ defer release()
845 return ReadResidentHistoryPage(dir, olderCursor, limit)879 return ReadResidentHistoryPage(dir, olderCursor, limit)
846 }880 }
847 881
@@ -850,13 +884,36 @@ func (m *ResidentManager) Result(childID string) (ResidentResult, error) {
850 if m == nil {884 if m == nil {
851 return ResidentResult{}, errors.New("resident manager: unavailable")885 return ResidentResult{}, errors.New("resident manager: unavailable")
852 }886 }
853- dir, err := ResidentHistoryDir(m.root, childID)887+ dir, release, err := m.readableResidentDir(childID)
854 if err != nil {888 if err != nil {
855 return ResidentResult{}, err889 return ResidentResult{}, err
856 }890 }
891+ defer release()
857 return ReadResidentResult(filepath.Join(dir, residentResultName))892 return ReadResidentResult(filepath.Join(dir, residentResultName))
858 }893 }
859 894
895+func (m *ResidentManager) readableResidentDir(childID string) (string, func(), error) {
896+ dir, err := ResidentHistoryDir(m.root, childID)
897+ if err != nil {
898+ return "", nil, err
899+ }
900+ childID = strings.TrimSpace(childID)
901+ m.mu.Lock()
902+ live := m.children[childID] != nil
903+ m.mu.Unlock()
904+ if live {
905+ return dir, func() {}, nil
906+ }
907+ lease, err := acquireResidentLease(dir)
908+ if err != nil {
909+ if errors.Is(err, ErrResidentLeaseBusy) {
910+ return "", nil, errors.New("resident manager: child is owned by another zut process")
911+ }
912+ return "", nil, err
913+ }
914+ return dir, func() { _ = lease.Close() }, nil
915+}
916+
860 func (m *ResidentManager) Close(ctx context.Context) error {917 func (m *ResidentManager) Close(ctx context.Context) error {
861 if m == nil {918 if m == nil {
862 return nil919 return nil
M packages/agent/subagents/resident_manager_test.go +116 -0
@@ -1,7 +1,9 @@
11 package subagents
22
33 import (
4+ "bufio"
45 "context"
6+ "encoding/json"
57 "errors"
68 "fmt"
79 "os"
@@ -589,6 +591,120 @@ func TestResidentManagerReconcileMakesInterruptedJournalDiscoverableWithoutRepla
589591 }
590592 }
591593
594+func TestResidentManagerReconcileDoesNotMutateForeignProcessJournal(t *testing.T) {
595+ if os.Getenv("ZUT_RESIDENT_MANAGER_HELPER") == "1" {
596+ root := os.Getenv("ZUT_RESIDENT_MANAGER_ROOT")
597+ journal, err := OpenResidentJournal(root, "foreign-process")
598+ if err != nil {
599+ t.Fatal(err)
600+ }
601+ spec := ResidentChildSpec{ID: "foreign-process", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
602+ if err := journal.Accept(spec, "active task"); err != nil {
603+ t.Fatal(err)
604+ }
605+ if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
606+ t.Fatal(err)
607+ }
608+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolCall, Time: time.Now().UTC(), ToolID: "call-1", ToolName: "bash", ToolArgs: json.RawMessage(`{"command":"pwd"}`)}); err != nil {
609+ t.Fatal(err)
610+ }
611+ fmt.Fprintln(os.Stdout, "ready")
612+ if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil {
613+ t.Fatal(err)
614+ }
615+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: time.Now().UTC(), ToolID: "call-1", ToolResult: json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)}); err != nil {
616+ t.Fatal(err)
617+ }
618+ if err := journal.RecordTurnFinished(spec, spec.InitialTurnID, nil); err != nil {
619+ t.Fatal(err)
620+ }
621+ if err := journal.Close(); err != nil {
622+ t.Fatal(err)
623+ }
624+ return
625+ }
626+
627+ root := t.TempDir()
628+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
629+ defer cancel()
630+ cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestResidentManagerReconcileDoesNotMutateForeignProcessJournal$")
631+ defer func() {
632+ if cmd.Process != nil {
633+ _ = cmd.Process.Kill()
634+ }
635+ _ = cmd.Wait()
636+ }()
637+ cmd.Env = append(os.Environ(), "ZUT_RESIDENT_MANAGER_HELPER=1", "ZUT_RESIDENT_MANAGER_ROOT="+root)
638+ stdout, err := cmd.StdoutPipe()
639+ if err != nil {
640+ t.Fatal(err)
641+ }
642+ stdin, err := cmd.StdinPipe()
643+ if err != nil {
644+ t.Fatal(err)
645+ }
646+ if err := cmd.Start(); err != nil {
647+ t.Fatal(err)
648+ }
649+ ready, err := bufio.NewReader(stdout).ReadString('\n')
650+ if err != nil {
651+ t.Fatal(err)
652+ }
653+ if ready != "ready\n" {
654+ t.Fatalf("helper readiness = %q", ready)
655+ }
656+ transcript := filepath.Join(root, "foreign-process", residentTranscriptName)
657+ before, err := os.ReadFile(transcript)
658+ if err != nil {
659+ t.Fatal(err)
660+ }
661+
662+ observer := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
663+ return func(context.Context, string) error { return nil }, nil
664+ })
665+ if errs := observer.Reconcile(); len(errs) != 0 {
666+ t.Fatalf("Reconcile errors = %v", errs)
667+ }
668+ after, err := os.ReadFile(transcript)
669+ if err != nil {
670+ t.Fatal(err)
671+ }
672+ if string(after) != string(before) {
673+ t.Fatal("foreign reconciliation modified the transcript")
674+ }
675+ snapshot, ok := observer.SnapshotFor("foreign-process")
676+ if !ok || !snapshot.OwnedElsewhere || snapshot.State != ResidentRunning {
677+ t.Fatalf("foreign snapshot = %#v, found=%v", snapshot, ok)
678+ }
679+ if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err == nil || !strings.Contains(err.Error(), "owned by another") {
680+ t.Fatalf("foreign Resume error = %v", err)
681+ }
682+ if _, err := observer.History("foreign-process", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
683+ t.Fatalf("foreign History error = %v", err)
684+ }
685+ if _, err := observer.HistoryPage("foreign-process", "", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
686+ t.Fatalf("foreign HistoryPage error = %v", err)
687+ }
688+ if _, err := observer.Result("foreign-process"); err == nil || !strings.Contains(err.Error(), "owned by another") {
689+ t.Fatalf("foreign Result error = %v", err)
690+ }
691+ if _, err := fmt.Fprintln(stdin); err != nil {
692+ t.Fatal(err)
693+ }
694+ if err := stdin.Close(); err != nil {
695+ t.Fatal(err)
696+ }
697+ if err := cmd.Wait(); err != nil {
698+ t.Fatal(err)
699+ }
700+ if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err != nil {
701+ t.Fatalf("Resume after owner exit: %v", err)
702+ }
703+ if err := observer.Stop(context.Background(), "foreign-process"); err != nil {
704+ t.Fatalf("Stop rebuilt child: %v", err)
705+ }
706+}
707+
592708 func TestResidentManagerExplicitlyResumesRecoveredChildWithoutReplayingTask(t *testing.T) {
593709 root := t.TempDir()
594710 journal, err := OpenResidentJournal(root, "resume-recovered")
@@ -1,7 +1,9 @@
1 package subagents1 package subagents
2 2
3 import (3 import (
4+ "bufio"
4 "context"5 "context"
6+ "encoding/json"
5 "errors"7 "errors"
6 "fmt"8 "fmt"
7 "os"9 "os"
@@ -589,6 +591,120 @@ func TestResidentManagerReconcileMakesInterruptedJournalDiscoverableWithoutRepla
589 }591 }
590 }592 }
591 593
594+func TestResidentManagerReconcileDoesNotMutateForeignProcessJournal(t *testing.T) {
595+ if os.Getenv("ZUT_RESIDENT_MANAGER_HELPER") == "1" {
596+ root := os.Getenv("ZUT_RESIDENT_MANAGER_ROOT")
597+ journal, err := OpenResidentJournal(root, "foreign-process")
598+ if err != nil {
599+ t.Fatal(err)
600+ }
601+ spec := ResidentChildSpec{ID: "foreign-process", SessionID: "child-session", InitialTurnID: "turn-1", Provider: "openai", Model: "gpt-5"}
602+ if err := journal.Accept(spec, "active task"); err != nil {
603+ t.Fatal(err)
604+ }
605+ if err := journal.RecordTurnStarted(spec, spec.InitialTurnID); err != nil {
606+ t.Fatal(err)
607+ }
608+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolCall, Time: time.Now().UTC(), ToolID: "call-1", ToolName: "bash", ToolArgs: json.RawMessage(`{"command":"pwd"}`)}); err != nil {
609+ t.Fatal(err)
610+ }
611+ fmt.Fprintln(os.Stdout, "ready")
612+ if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil {
613+ t.Fatal(err)
614+ }
615+ if err := journal.appendSync(residentRecord{Version: residentJournalVersion, Type: residentRecordToolResult, Time: time.Now().UTC(), ToolID: "call-1", ToolResult: json.RawMessage(`{"Content":[{"text":"ok"}],"IsError":false}`)}); err != nil {
616+ t.Fatal(err)
617+ }
618+ if err := journal.RecordTurnFinished(spec, spec.InitialTurnID, nil); err != nil {
619+ t.Fatal(err)
620+ }
621+ if err := journal.Close(); err != nil {
622+ t.Fatal(err)
623+ }
624+ return
625+ }
626+
627+ root := t.TempDir()
628+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
629+ defer cancel()
630+ cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestResidentManagerReconcileDoesNotMutateForeignProcessJournal$")
631+ defer func() {
632+ if cmd.Process != nil {
633+ _ = cmd.Process.Kill()
634+ }
635+ _ = cmd.Wait()
636+ }()
637+ cmd.Env = append(os.Environ(), "ZUT_RESIDENT_MANAGER_HELPER=1", "ZUT_RESIDENT_MANAGER_ROOT="+root)
638+ stdout, err := cmd.StdoutPipe()
639+ if err != nil {
640+ t.Fatal(err)
641+ }
642+ stdin, err := cmd.StdinPipe()
643+ if err != nil {
644+ t.Fatal(err)
645+ }
646+ if err := cmd.Start(); err != nil {
647+ t.Fatal(err)
648+ }
649+ ready, err := bufio.NewReader(stdout).ReadString('\n')
650+ if err != nil {
651+ t.Fatal(err)
652+ }
653+ if ready != "ready\n" {
654+ t.Fatalf("helper readiness = %q", ready)
655+ }
656+ transcript := filepath.Join(root, "foreign-process", residentTranscriptName)
657+ before, err := os.ReadFile(transcript)
658+ if err != nil {
659+ t.Fatal(err)
660+ }
661+
662+ observer := NewResidentManager(root, func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
663+ return func(context.Context, string) error { return nil }, nil
664+ })
665+ if errs := observer.Reconcile(); len(errs) != 0 {
666+ t.Fatalf("Reconcile errors = %v", errs)
667+ }
668+ after, err := os.ReadFile(transcript)
669+ if err != nil {
670+ t.Fatal(err)
671+ }
672+ if string(after) != string(before) {
673+ t.Fatal("foreign reconciliation modified the transcript")
674+ }
675+ snapshot, ok := observer.SnapshotFor("foreign-process")
676+ if !ok || !snapshot.OwnedElsewhere || snapshot.State != ResidentRunning {
677+ t.Fatalf("foreign snapshot = %#v, found=%v", snapshot, ok)
678+ }
679+ if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err == nil || !strings.Contains(err.Error(), "owned by another") {
680+ t.Fatalf("foreign Resume error = %v", err)
681+ }
682+ if _, err := observer.History("foreign-process", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
683+ t.Fatalf("foreign History error = %v", err)
684+ }
685+ if _, err := observer.HistoryPage("foreign-process", "", 1); err == nil || !strings.Contains(err.Error(), "owned by another") {
686+ t.Fatalf("foreign HistoryPage error = %v", err)
687+ }
688+ if _, err := observer.Result("foreign-process"); err == nil || !strings.Contains(err.Error(), "owned by another") {
689+ t.Fatalf("foreign Result error = %v", err)
690+ }
691+ if _, err := fmt.Fprintln(stdin); err != nil {
692+ t.Fatal(err)
693+ }
694+ if err := stdin.Close(); err != nil {
695+ t.Fatal(err)
696+ }
697+ if err := cmd.Wait(); err != nil {
698+ t.Fatal(err)
699+ }
700+ if err := observer.Resume(context.Background(), "foreign-process", "follow up"); err != nil {
701+ t.Fatalf("Resume after owner exit: %v", err)
702+ }
703+ if err := observer.Stop(context.Background(), "foreign-process"); err != nil {
704+ t.Fatalf("Stop rebuilt child: %v", err)
705+ }
706+}
707+
592 func TestResidentManagerExplicitlyResumesRecoveredChildWithoutReplayingTask(t *testing.T) {708 func TestResidentManagerExplicitlyResumesRecoveredChildWithoutReplayingTask(t *testing.T) {
593 root := t.TempDir()709 root := t.TempDir()
594 journal, err := OpenResidentJournal(root, "resume-recovered")710 journal, err := OpenResidentJournal(root, "resume-recovered")
M packages/agent/tools/subagent_status.go +9 -8
@@ -32,13 +32,14 @@ type subagentStatusResponse struct {
3232 }
3333
3434 type subagentStatusEntry struct {
35- ID string `json:"agent_id"`
36- State subagents.ResidentState `json:"state"`
37- Profile string `json:"profile,omitempty"`
38- Provider string `json:"provider"`
39- Model string `json:"model"`
40- Workspace subagents.WorkspaceMode `json:"workspace_mode,omitempty"`
41- Required bool `json:"required,omitempty"`
35+ ID string `json:"agent_id"`
36+ State subagents.ResidentState `json:"state"`
37+ Profile string `json:"profile,omitempty"`
38+ Provider string `json:"provider"`
39+ Model string `json:"model"`
40+ Workspace subagents.WorkspaceMode `json:"workspace_mode,omitempty"`
41+ Required bool `json:"required,omitempty"`
42+ OwnedElsewhere bool `json:"owned_elsewhere,omitempty"`
4243 }
4344
4445 const subagentStatusSchema = `{
@@ -116,7 +117,7 @@ func findResidentStatusSnapshot(snapshots []subagents.ResidentSnapshot, id strin
116117 }
117118
118119 func publicResidentStatus(snapshot subagents.ResidentSnapshot) subagentStatusEntry {
119- return subagentStatusEntry{ID: snapshot.ID, State: snapshot.State, Profile: snapshot.Profile, Provider: snapshot.Provider, Model: snapshot.Model, Workspace: snapshot.WorkspaceMode, Required: snapshot.Required}
120+ return subagentStatusEntry{ID: snapshot.ID, State: snapshot.State, Profile: snapshot.Profile, Provider: snapshot.Provider, Model: snapshot.Model, Workspace: snapshot.WorkspaceMode, Required: snapshot.Required, OwnedElsewhere: snapshot.OwnedElsewhere}
120121 }
121122
122123 func renderSubagentStatus(response subagentStatusResponse) (core.ToolResult, error) {
@@ -32,13 +32,14 @@ type subagentStatusResponse struct {
32 }32 }
33 33
34 type subagentStatusEntry struct {34 type subagentStatusEntry struct {
35- ID string `json:"agent_id"`35+ ID string `json:"agent_id"`
36- State subagents.ResidentState `json:"state"`36+ State subagents.ResidentState `json:"state"`
37- Profile string `json:"profile,omitempty"`37+ Profile string `json:"profile,omitempty"`
38- Provider string `json:"provider"`38+ Provider string `json:"provider"`
39- Model string `json:"model"`39+ Model string `json:"model"`
40- Workspace subagents.WorkspaceMode `json:"workspace_mode,omitempty"`40+ Workspace subagents.WorkspaceMode `json:"workspace_mode,omitempty"`
41- Required bool `json:"required,omitempty"`41+ Required bool `json:"required,omitempty"`
42+ OwnedElsewhere bool `json:"owned_elsewhere,omitempty"`
42 }43 }
43 44
44 const subagentStatusSchema = `{45 const subagentStatusSchema = `{
@@ -116,7 +117,7 @@ func findResidentStatusSnapshot(snapshots []subagents.ResidentSnapshot, id strin
116 }117 }
117 118
118 func publicResidentStatus(snapshot subagents.ResidentSnapshot) subagentStatusEntry {119 func publicResidentStatus(snapshot subagents.ResidentSnapshot) subagentStatusEntry {
119- return subagentStatusEntry{ID: snapshot.ID, State: snapshot.State, Profile: snapshot.Profile, Provider: snapshot.Provider, Model: snapshot.Model, Workspace: snapshot.WorkspaceMode, Required: snapshot.Required}120+ return subagentStatusEntry{ID: snapshot.ID, State: snapshot.State, Profile: snapshot.Profile, Provider: snapshot.Provider, Model: snapshot.Model, Workspace: snapshot.WorkspaceMode, Required: snapshot.Required, OwnedElsewhere: snapshot.OwnedElsewhere}
120 }121 }
121 122
122 func renderSubagentStatus(response subagentStatusResponse) (core.ToolResult, error) {123 func renderSubagentStatus(response subagentStatusResponse) (core.ToolResult, error) {

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 476

↑↓ moveenter openesc close