返回 DeepSeek-Reasonix
rewind.go
根目录 / internal / control / rewind.go
1 package control
2
3 import (
4 "encoding/json"
5 "errors"
6 "fmt"
7 "log/slog"
8 "sync/atomic"
9
10 "reasonix/internal/agent"
11 "reasonix/internal/checkpoint"
12 "reasonix/internal/diff"
13 "reasonix/internal/event"
14 "reasonix/internal/provider"
15 )
16
17 // ErrRewindCoverageConfirmationRequired is returned by the compatibility
18 // Rewind path when restoring files from a partially covered checkpoint. New
19 // callers should preview with PrepareRewind, show the coverage warning, and
20 // commit only after the user explicitly confirms it.
21 var ErrRewindCoverageConfirmationRequired = errors.New("partial checkpoint coverage requires explicit confirmation")
22
23 // RewindPlanRequiresConfirmation reports whether a prepared plan can restore
24 // files but cannot guarantee that every workspace mutation was captured.
25 func RewindPlanRequiresConfirmation(plan checkpoint.RewindPlan) bool {
26 wantsFiles := plan.Scope == checkpoint.RewindCode || plan.Scope == checkpoint.RewindBoth
27 if !wantsFiles || !plan.CanFiles {
28 return false
29 }
30 if plan.ExpiredFilePayload || plan.Legacy {
31 return true
32 }
33 return checkpoint.HasProjectCoverageGap(plan.CoverageGaps)
34 }
35
36 // conversationApplier bridges checkpoint transactions to controller session state.
37 type conversationApplier struct {
38 c *Controller
39 }
40
41 func (a conversationApplier) ApplyConversationTruncate(boundary int, forward []byte) error {
42 c := a.c
43 if c.executor == nil {
44 return fmt.Errorf("executor unavailable")
45 }
46 s := c.executor.Session()
47 msgs := s.Snapshot()
48 if boundary > len(msgs) {
49 return fmt.Errorf("conversation rewind unavailable: the conversation was compacted past this point")
50 }
51 if len(forward) == 0 {
52 var err error
53 forward, err = json.Marshal(msgs)
54 if err != nil {
55 return err
56 }
57 }
58 s.Rewrite(msgs[:boundary], "rewind_truncate")
59 // Drop the projection only when the truncation reached into the folded
60 // prefix; a tail-only rewind keeps the covered prefix byte-identical.
61 c.executor.InvalidateProjectionIfStale()
62 if err := c.SnapshotRewrite(); err != nil {
63 _ = a.RestoreConversation(forward)
64 return fmt.Errorf("persist conversation after rewind: %w", err)
65 }
66 return nil
67 }
68
69 func (a conversationApplier) RestoreConversation(forward []byte) error {
70 c := a.c
71 if c.executor == nil {
72 return fmt.Errorf("executor unavailable")
73 }
74 var msgs []provider.Message
75 if err := json.Unmarshal(forward, &msgs); err != nil {
76 return err
77 }
78 c.executor.Session().Rewrite(msgs, "rewind_restore")
79 c.executor.InvalidateProjectionIfStale()
80 if err := c.SnapshotRewrite(); err != nil {
81 return fmt.Errorf("restore conversation: %w", err)
82 }
83 return nil
84 }
85
86 func (a conversationApplier) TruncateCheckpoints(fromTurn int) error {
87 return a.c.checkpoints.truncateFrom(fromTurn)
88 }
89
90 func (a conversationApplier) RestoreCheckpoints(backup []byte) error {
91 store := a.c.checkpoints.storeRef()
92 if store == nil {
93 return fmt.Errorf("checkpoints unavailable")
94 }
95 if err := store.RestoreCheckpointBackupPublic(backup); err != nil {
96 return err
97 }
98 bounds := store.Bounds()
99 a.c.checkpoints.mu.Lock()
100 a.c.checkpoints.bound = bounds
101 a.c.checkpoints.turn = store.NextTurn()
102 a.c.checkpoints.mu.Unlock()
103 return nil
104 }
105
106 // PrepareRewind validates that a rewind can proceed without mutating state.
107 func (c *Controller) PrepareRewind(turn int, scope RewindScope) (checkpoint.RewindPlan, error) {
108 if !c.checkpoints.enabled() || c.executor == nil {
109 return checkpoint.RewindPlan{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
110 }
111 if err := c.beginRotation(); err != nil {
112 if errors.Is(err, errTurnRunningRotation) {
113 return checkpoint.RewindPlan{}, c.rewindFail(fmt.Errorf("cannot rewind while a turn is running"))
114 }
115 return checkpoint.RewindPlan{}, c.rewindFail(err)
116 }
117 // Release rotation before file precheck I/O.
118 c.endRotation()
119
120 boundary, hasBound := c.checkpoints.boundary(turn)
121 store := c.checkpoints.storeRef()
122 if store == nil {
123 return checkpoint.RewindPlan{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
124 }
125 if obs := c.mutationObserver; obs != nil {
126 store.SetActiveWriters(obs.ActiveWriters())
127 }
128 rev := atomic.LoadInt64(&c.sessionRevision)
129 plan, err := store.PrepareRewind(turn, checkpoint.RewindScope(scope), rev, boundary, hasBound)
130 if err != nil {
131 return plan, c.rewindFail(err)
132 }
133 if scope == RewindConversation || scope == RewindBoth {
134 store.MarkPlanConversationFork(plan.PlanID)
135 plan.ConversationAction = "fork"
136 }
137 if scope == RewindBoth && !plan.CanConversation {
138 plan.CanFiles = false
139 if plan.DisabledReason == "" {
140 plan.DisabledReason = "conversation boundary unavailable"
141 }
142 }
143 return plan, nil
144 }
145
146 // CommitRewind executes a prepared plan under rotation gate + mutation barrier.
147 // Conversation forks are returned detached so multi-tab frontends can keep the
148 // parent controller; single-session frontends must activate result.Branch.
149 func (c *Controller) CommitRewind(planID string) (checkpoint.RewindResult, error) {
150 return c.commitRewind(planID, false)
151 }
152
153 // CommitRewindInPlace commits a prepared plan and moves this controller onto
154 // the rewound conversation: a new head of the same schema-2 log, or the fork
155 // file for a schema-1 session. Desktop tabs use it so the tab stays put.
156 func (c *Controller) CommitRewindInPlace(planID string) (checkpoint.RewindResult, error) {
157 return c.commitRewind(planID, true)
158 }
159
160 func (c *Controller) commitRewind(planID string, switchToFork bool) (checkpoint.RewindResult, error) {
161 if !c.checkpoints.enabled() || c.executor == nil {
162 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
163 }
164 if err := c.beginRotation(); err != nil {
165 if errors.Is(err, errTurnRunningRotation) {
166 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("cannot rewind while a turn is running"))
167 }
168 return checkpoint.RewindResult{}, c.rewindFail(err)
169 }
170 defer c.endRotation()
171
172 store := c.checkpoints.storeRef()
173 if store == nil {
174 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
175 }
176 if err := store.ValidatePlanSessionRevision(planID, atomic.LoadInt64(&c.sessionRevision)); err != nil {
177 conflict := checkpoint.RewindConflict{Reason: checkpoint.ConflictStalePlan}
178 return checkpoint.RewindResult{OK: false, Error: err.Error(), Conflicts: []checkpoint.RewindConflict{conflict}}, c.rewindFail(err)
179 }
180
181 forward, err := json.Marshal(c.executor.Session().Snapshot())
182 if err != nil {
183 return checkpoint.RewindResult{}, c.rewindFail(err)
184 }
185
186 result, err := c.commitRewindReady(store, planID, forward, false, switchToFork)
187 if err != nil {
188 return result, c.rewindFail(err)
189 }
190 if result.OK {
191 if len(result.Written) > 0 || len(result.Deleted) > 0 {
192 c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo,
193 Text: fmt.Sprintf("rewound code — %d file(s) restored, %d removed", len(result.Written), len(result.Deleted))})
194 }
195 if result.ConversationForked && !switchToFork {
196 c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo,
197 Text: "forked conversation; parent session is unchanged"})
198 }
199 atomic.AddInt64(&c.sessionRevision, 1)
200 }
201 return result, nil
202 }
203
204 // CommitFileRewind commits only the file half of a prepared plan.
205 func (c *Controller) CommitFileRewind(planID string) (checkpoint.RewindResult, error) {
206 if !c.checkpoints.enabled() || c.executor == nil {
207 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
208 }
209 if err := c.beginRotation(); err != nil {
210 if errors.Is(err, errTurnRunningRotation) {
211 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("cannot rewind while a turn is running"))
212 }
213 return checkpoint.RewindResult{}, c.rewindFail(err)
214 }
215 defer c.endRotation()
216 store := c.checkpoints.storeRef()
217 if store == nil {
218 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
219 }
220 return c.commitRewindReady(store, planID, nil, true, false)
221 }
222
223 func (c *Controller) commitRewindReady(store *checkpoint.Store, planID string, forward []byte, filesOnly, switchToFork bool) (checkpoint.RewindResult, error) {
224 plan, ok := store.PeekPlan(planID)
225 if !ok {
226 return checkpoint.RewindResult{OK: false, Error: "unknown or expired plan"}, fmt.Errorf("unknown or expired plan %q", planID)
227 }
228 result := checkpoint.RewindResult{}
229 wantConv := !filesOnly && (plan.Scope == checkpoint.RewindConversation || plan.Scope == checkpoint.RewindBoth)
230 wantFiles := plan.Scope == checkpoint.RewindCode || plan.Scope == checkpoint.RewindBoth
231 if wantConv {
232 path, err := c.forkNamedReady(plan.Turn, "", switchToFork, agent.HeadKindRewind)
233 if err != nil {
234 return result, err
235 }
236 result.ConversationForked = true
237 result.ConversationOK = true
238 result.Branch = path
239 }
240 if wantFiles {
241 fileResult, err := store.CommitRewindWithForward(planID, forward, conversationApplier{c: c}, nil)
242 if err != nil {
243 if result.ConversationForked {
244 fileResult.Partial = true
245 fileResult.OK = true
246 fileResult.ConversationForked = true
247 fileResult.ConversationOK = true
248 fileResult.Branch = result.Branch
249 fileResult.Error = err.Error()
250 return fileResult, nil
251 }
252 return fileResult, err
253 }
254 fileResult.ConversationForked = result.ConversationForked
255 fileResult.ConversationOK = result.ConversationOK || fileResult.ConversationOK
256 fileResult.Branch = result.Branch
257 if fileResult.OperationID == "" {
258 fileResult.OperationID = fileResult.TransactionID
259 }
260 return fileResult, nil
261 }
262 _ = store.DiscardPlan(planID)
263 result.OK = result.ConversationForked
264 return result, nil
265 }
266
267 // UndoRewind reverses the last committed rewind transaction when still available.
268 func (c *Controller) UndoRewind(transactionID string) (checkpoint.RewindResult, error) {
269 if !c.checkpoints.enabled() || c.executor == nil {
270 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
271 }
272 if err := c.beginRotation(); err != nil {
273 if errors.Is(err, errTurnRunningRotation) {
274 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("cannot undo rewind while a turn is running"))
275 }
276 return checkpoint.RewindResult{}, c.rewindFail(err)
277 }
278 defer c.endRotation()
279
280 store := c.checkpoints.storeRef()
281 if store == nil {
282 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
283 }
284 result, err := store.UndoRewind(transactionID, conversationApplier{c: c})
285 if err != nil {
286 return result, c.rewindFail(err)
287 }
288 if result.OK {
289 if c.undoHeadRewind() {
290 result.ConversationOK = true
291 }
292 atomic.AddInt64(&c.sessionRevision, 1)
293 c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo, Text: "undid last rewind"})
294 }
295 return result, nil
296 }
297
298 // PrepareFileRevert prepares a single-file restore to the session's first-touch preimage.
299 func (c *Controller) PrepareFileRevert(path string) (checkpoint.RewindPlan, error) {
300 if !c.checkpoints.enabled() || c.executor == nil {
301 return checkpoint.RewindPlan{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
302 }
303 store := c.checkpoints.storeRef()
304 if store == nil {
305 return checkpoint.RewindPlan{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
306 }
307 state, ok := store.FileState(path)
308 if !ok {
309 return checkpoint.RewindPlan{
310 Path: path, CanFiles: false, DisabledReason: "file is not session-owned",
311 }, nil
312 }
313 _ = state
314 return store.PrepareFileRevert(path, atomic.LoadInt64(&c.sessionRevision))
315 }
316
317 // CommitFileRevert commits a single-file restore with optional conflict resolution.
318 func (c *Controller) CommitFileRevert(planID string, resolution checkpoint.ConflictResolution) (checkpoint.RewindResult, error) {
319 if !c.checkpoints.enabled() || c.executor == nil {
320 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
321 }
322 if err := c.beginRotation(); err != nil {
323 if errors.Is(err, errTurnRunningRotation) {
324 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("cannot revert file while a turn is running"))
325 }
326 return checkpoint.RewindResult{}, c.rewindFail(err)
327 }
328 defer c.endRotation()
329
330 store := c.checkpoints.storeRef()
331 if store == nil {
332 return checkpoint.RewindResult{}, c.rewindFail(fmt.Errorf("checkpoints unavailable"))
333 }
334 if err := store.ValidatePlanSessionRevision(planID, atomic.LoadInt64(&c.sessionRevision)); err != nil {
335 conflict := checkpoint.RewindConflict{Reason: checkpoint.ConflictStalePlan}
336 return checkpoint.RewindResult{OK: false, Error: err.Error(), Conflicts: []checkpoint.RewindConflict{conflict}}, c.rewindFail(err)
337 }
338 result, err := store.CommitFileRevert(planID, resolution)
339 if err != nil {
340 return result, c.rewindFail(err)
341 }
342 if result.OK {
343 atomic.AddInt64(&c.sessionRevision, 1)
344 }
345 return result, nil
346 }
347
348 // Rewind is the compatibility wrapper used by CLI and existing desktop paths.
349 // Conversation failures never leave files half-applied for both-scope: files are
350 // captured first, restored second, and conversation is persisted last with full
351 // compensation on failure.
352 func (c *Controller) Rewind(turn int, scope RewindScope) error {
353 if !c.checkpoints.enabled() || c.executor == nil {
354 return c.rewindFail(fmt.Errorf("checkpoints unavailable"))
355 }
356 if err := c.beginRotation(); err != nil {
357 if errors.Is(err, errTurnRunningRotation) {
358 return c.rewindFail(fmt.Errorf("cannot rewind while a turn is running"))
359 }
360 return c.rewindFail(err)
361 }
362 defer c.endRotation()
363
364 boundary, hasBound := c.checkpoints.boundary(turn)
365 var forward []byte
366 if scope == RewindConversation || scope == RewindBoth {
367 if !hasBound {
368 return c.rewindFail(fmt.Errorf("conversation rewind unavailable for turn %d (resumed session)", turn))
369 }
370 msgs := c.executor.Session().Snapshot()
371 if boundary > len(msgs) {
372 return c.rewindFail(fmt.Errorf("conversation rewind unavailable for turn %d: the conversation was compacted past this point", turn))
373 }
374 var err error
375 forward, err = json.Marshal(msgs)
376 if err != nil {
377 return c.rewindFail(err)
378 }
379 }
380 store := c.checkpoints.storeRef()
381 if store == nil {
382 return c.rewindFail(fmt.Errorf("checkpoints unavailable"))
383 }
384 if obs := c.mutationObserver; obs != nil {
385 store.SetActiveWriters(obs.ActiveWriters())
386 }
387 rev := atomic.LoadInt64(&c.sessionRevision)
388 plan, err := store.PrepareRewind(turn, checkpoint.RewindScope(scope), rev, boundary, hasBound)
389 if err != nil {
390 return c.rewindFail(err)
391 }
392 if scope == RewindConversation || scope == RewindBoth {
393 store.MarkPlanConversationFork(plan.PlanID)
394 plan.ConversationAction = "fork"
395 }
396 if (scope == RewindCode || scope == RewindBoth) && !plan.CanFiles {
397 return c.rewindFail(fmt.Errorf("%s", plan.DisabledReason))
398 }
399 if (scope == RewindConversation || scope == RewindBoth) && !plan.CanConversation {
400 return c.rewindFail(fmt.Errorf("%s", plan.DisabledReason))
401 }
402 if RewindPlanRequiresConfirmation(plan) {
403 return c.rewindFail(fmt.Errorf("%w (%d coverage gap(s))", ErrRewindCoverageConfirmationRequired, len(plan.CoverageGaps)))
404 }
405 if forward == nil {
406 forward, err = json.Marshal(c.executor.Session().Snapshot())
407 if err != nil {
408 return c.rewindFail(err)
409 }
410 }
411 result, err := c.commitRewindReady(store, plan.PlanID, forward, scope == RewindCode, true)
412 if err != nil {
413 return c.rewindFail(err)
414 }
415 if len(result.Written) > 0 || len(result.Deleted) > 0 {
416 c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo,
417 Text: fmt.Sprintf("rewound code to turn %d — %d file(s) restored, %d removed", turn, len(result.Written), len(result.Deleted))})
418 }
419 if result.ConversationForked {
420 c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo,
421 Text: fmt.Sprintf("forked conversation at turn %d; parent session unchanged (%s)", turn, result.Branch)})
422 }
423 if result.Partial {
424 c.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn,
425 Text: "session branch created; code was not fully restored because of a conflict"})
426 }
427 atomic.AddInt64(&c.sessionRevision, 1)
428 return nil
429 }
430
431 func (c *Controller) recoverCheckpointTransactions() {
432 store := c.checkpoints.storeRef()
433 if store == nil || c.executor == nil {
434 return
435 }
436 for _, note := range store.RecoverTransactionsWithApplier(conversationApplier{c: c}) {
437 slog.Info("controller: checkpoint transaction recovery", "result", note)
438 }
439 }
440
441 // wireMutationObserver installs the v2 observer on the executor.
442 func (c *Controller) wireMutationObserver() {
443 store := c.checkpoints.storeRef()
444 if store == nil || c.executor == nil {
445 return
446 }
447 obs := checkpoint.NewMutationObserver(checkpoint.ObserverOptions{
448 Store: store,
449 WriterID: "root",
450 })
451 c.mutationObserver = obs
452 c.executor.SetMutationObserver(obs)
453 // Keep legacy pre-edit hook as a secondary path when observer is absent on
454 // a cloned agent; with observer set, BeforeMutation is preferred.
455 c.executor.SetPreEditHook(func(ch diff.Change) {
456 if c.mutationObserver != nil {
457 c.mutationObserver.BeforeMutationFromChange(ch, "legacy_hook")
458 return
459 }
460 c.checkpoints.snapshot(ch)
461 })
462 }
463
463 lines GO