| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "sync" |
| 5 | |
| 6 | "reasonix/internal/checkpoint" |
| 7 | "reasonix/internal/diff" |
| 8 | ) |
| 9 | |
| 10 | // checkpointManager owns the snapshot-based rewind bookkeeping: the per-session |
| 11 | // checkpoint store, the monotonic turn counter, and the conversation-rewind |
| 12 | // boundary map. Like approvalManager it holds only the bookkeeping behind its own |
| 13 | // lock, off the controller's c.mu — the Controller keeps the rewind/fork |
| 14 | // orchestration (truncating the session, restoring code, emitting events) that |
| 15 | // needs its other collaborators. |
| 16 | // |
| 17 | // turn is decoupled from the store so it never collides after a log restructure; |
| 18 | // bound[turn] records len(Session.Messages) at that turn's start — the truncation |
| 19 | // boundary for a conversation rewind/fork. Boundaries are persisted in each |
| 20 | // checkpoint and rebuilt from the store on resume (so a reopened session can still |
| 21 | // rewind conversation / fork), but dropped after a summarize restructures the log |
| 22 | // so those operations report "unavailable" rather than mis-truncating; code |
| 23 | // rewind (file-based) is unaffected. Every store call does its disk I/O off mu — |
| 24 | // mu is taken only to read/swap the store pointer and mutate turn/bound. |
| 25 | type checkpointManager struct { |
| 26 | // mu guards store, turn, and bound; every critical section under it is short |
| 27 | // and non-blocking (no disk I/O). |
| 28 | mu sync.Mutex |
| 29 | store *checkpoint.Store |
| 30 | turn int |
| 31 | bound map[int]int |
| 32 | } |
| 33 | |
| 34 | // rebind points the store at the (possibly new) session, loading any checkpoints |
| 35 | // already on disk, and resets the turn counter and boundaries from them. root is |
| 36 | // the workspace root used to guard restore writes. Called on construction and |
| 37 | // whenever the session path changes (NewSession/Resume/SetSessionPath/fork). |
| 38 | func (m *checkpointManager) rebind(dir, root string) { |
| 39 | store := checkpoint.New(dir, root) |
| 40 | next := store.NextTurn() // continue numbering past any checkpoints on disk |
| 41 | bound := store.Bounds() // rebuilt from persisted checkpoints so a resumed |
| 42 | if bound == nil { // session can still rewind conversation / fork |
| 43 | bound = map[int]int{} |
| 44 | } |
| 45 | m.mu.Lock() |
| 46 | m.store = store |
| 47 | m.turn = next |
| 48 | m.bound = bound |
| 49 | m.mu.Unlock() |
| 50 | } |
| 51 | |
| 52 | // enabled reports whether a checkpoint store is bound. |
| 53 | func (m *checkpointManager) enabled() bool { |
| 54 | m.mu.Lock() |
| 55 | defer m.mu.Unlock() |
| 56 | return m.store != nil |
| 57 | } |
| 58 | |
| 59 | // beginWithObserver opens a checkpoint and updates the mutation observer's |
| 60 | // ownership turn for subsequent captures. |
| 61 | func (m *checkpointManager) beginWithObserver(input string, msgIndex int, obs *checkpoint.MutationObserver) { |
| 62 | m.mu.Lock() |
| 63 | store := m.store |
| 64 | if store == nil { |
| 65 | m.mu.Unlock() |
| 66 | return |
| 67 | } |
| 68 | turn := m.turn |
| 69 | m.turn++ |
| 70 | m.bound[turn] = msgIndex |
| 71 | m.mu.Unlock() |
| 72 | if obs != nil { |
| 73 | obs.NoteCrossTurnBackgroundWriter(turn) |
| 74 | obs.SetOwnershipTurn(turn) |
| 75 | } |
| 76 | store.Begin(turn, input, msgIndex) |
| 77 | } |
| 78 | |
| 79 | // turnsByMessageIndex returns message-log index -> checkpoint turn over live |
| 80 | // boundaries. The desktop transcript uses this authoritative map instead of |
| 81 | // recounting visible user bubbles, which can diverge when synthetic user-role |
| 82 | // messages are hidden from the UI. |
| 83 | func (m *checkpointManager) turnsByMessageIndex() map[int]int { |
| 84 | m.mu.Lock() |
| 85 | defer m.mu.Unlock() |
| 86 | out := make(map[int]int, len(m.bound)) |
| 87 | for turn, index := range m.bound { |
| 88 | if existing, ok := out[index]; ok && existing < turn { |
| 89 | continue |
| 90 | } |
| 91 | out[index] = turn |
| 92 | } |
| 93 | return out |
| 94 | } |
| 95 | |
| 96 | // boundary returns the recorded turn-start message index, if any. |
| 97 | func (m *checkpointManager) boundary(turn int) (int, bool) { |
| 98 | m.mu.Lock() |
| 99 | defer m.mu.Unlock() |
| 100 | b, ok := m.bound[turn] |
| 101 | return b, ok |
| 102 | } |
| 103 | |
| 104 | // list returns the checkpoint metadata (nil when disabled). |
| 105 | func (m *checkpointManager) list() []checkpoint.Meta { |
| 106 | m.mu.Lock() |
| 107 | store := m.store |
| 108 | m.mu.Unlock() |
| 109 | if store == nil { |
| 110 | return nil |
| 111 | } |
| 112 | return store.List() |
| 113 | } |
| 114 | |
| 115 | func (m *checkpointManager) fileState(path string) (checkpoint.FileState, bool) { |
| 116 | m.mu.Lock() |
| 117 | store := m.store |
| 118 | m.mu.Unlock() |
| 119 | if store == nil { |
| 120 | return checkpoint.FileState{}, false |
| 121 | } |
| 122 | return store.FileState(path) |
| 123 | } |
| 124 | |
| 125 | // snapshot records a pre-edit file change into the open checkpoint — the |
| 126 | // executor's pre-edit hook. No-op when disabled. |
| 127 | func (m *checkpointManager) snapshot(ch diff.Change) { |
| 128 | m.mu.Lock() |
| 129 | store := m.store |
| 130 | m.mu.Unlock() |
| 131 | if store != nil { |
| 132 | store.Snapshot(ch) |
| 133 | } |
| 134 | } |
| 135 | |
| 136 | // truncateFrom renumbers future turns from `turn` and drops every boundary at or |
| 137 | // after it — the conversation-rewind renumber after the message log is cut back. |
| 138 | func (m *checkpointManager) truncateFrom(turn int) error { |
| 139 | m.mu.Lock() |
| 140 | store := m.store |
| 141 | m.mu.Unlock() |
| 142 | if store != nil { |
| 143 | if err := store.TruncateFrom(turn); err != nil { |
| 144 | return err |
| 145 | } |
| 146 | } |
| 147 | m.mu.Lock() |
| 148 | m.turn = turn |
| 149 | for k := range m.bound { |
| 150 | if k >= turn { |
| 151 | delete(m.bound, k) |
| 152 | } |
| 153 | } |
| 154 | m.mu.Unlock() |
| 155 | return nil |
| 156 | } |
| 157 | |
| 158 | // clearBounds drops every boundary after a summarize restructures the log (so |
| 159 | // conversation rewind degrades to "unavailable" until fresh turns rebuild them) |
| 160 | // while keeping turn monotonic so new turns don't collide with the store. |
| 161 | func (m *checkpointManager) clearBounds() { |
| 162 | m.mu.Lock() |
| 163 | m.bound = map[int]int{} |
| 164 | m.mu.Unlock() |
| 165 | } |
| 166 | |
| 167 | // storeRef returns the live store pointer without holding mu across caller work. |
| 168 | func (m *checkpointManager) storeRef() *checkpoint.Store { |
| 169 | m.mu.Lock() |
| 170 | defer m.mu.Unlock() |
| 171 | return m.store |
| 172 | } |
| 173 |