Finish atomic restore-token reset recovery

Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>

Author: minpeter

internal/daemon/daemon.go

@@ -1354,6 +1354,10 @@ func (d *Daemon) getOrStartPreparedCaptureGroup(ctx context.Context, entry *acti
 	if d.captureGroups == nil {
 		d.captureGroups = make(map[videoCaptureKey]*videoCaptureGroup)
 	}
+	if _, err := d.migrateRestoreTokenResetReservationLocked(entry, key); err != nil {
+		d.mu.Unlock()
+		return nil, 0, err
+	}
 	group := d.captureGroups[key]
 	var captureCtx context.Context
 	var captureCancel context.CancelFunc
@@ -1474,6 +1478,10 @@ func (d *Daemon) getOrStartCaptureGroup(entry *activeStream, restoreToken, devic
 	if d.captureGroups == nil {
 		d.captureGroups = make(map[videoCaptureKey]*videoCaptureGroup)
 	}
+	if _, err := d.migrateRestoreTokenResetReservationLocked(entry, key); err != nil {
+		d.mu.Unlock()
+		return nil, err
+	}
 	group := d.captureGroups[key]
 	var captureCtx context.Context
 	var captureCancel context.CancelFunc
@@ -1565,6 +1573,30 @@ func (d *Daemon) getOrStartCaptureGroup(entry *activeStream, restoreToken, devic
 	return newBC, nil
 }
 
+// migrateRestoreTokenResetReservationLocked moves an owner-only reset claim
+// when the receiver negotiates a different canvas after reconnecting. It never
+// joins an occupied destination: reset replacement remains exclusive.
+func (d *Daemon) migrateRestoreTokenResetReservationLocked(entry *activeStream, key videoCaptureKey) (*videoCaptureGroup, error) {
+	reservation := entry.captureGroup
+	if reservation == nil || reservation.resetReservedBy != entry ||
+		reservation.broadcast != nil || reservation.capture != nil || reservation.cancel != nil {
+		return nil, nil
+	}
+	if reservation.key == key {
+		return reservation, nil
+	}
+	if d.captureGroups[reservation.key] != reservation {
+		return nil, context.Canceled
+	}
+	if d.captureGroups[key] != nil {
+		return nil, fmt.Errorf("%w: replacement canvas %dx%d is already active", errCaptureGroupResetReserved, key.maxWidth, key.maxHeight)
+	}
+	delete(d.captureGroups, reservation.key)
+	reservation.key = key
+	d.captureGroups[key] = reservation
+	return reservation, nil
+}
+
 // detachStreamLocked removes a single stream and transfers ownership of its
 // resources, plus an unused final capture group, to a cleanup plan. The caller
 // must unlock d.mu before running the plan.

internal/daemon/reset_restore_token.go

@@ -1,6 +1,10 @@
 package daemon
 
-import "context"
+import (
+	"context"
+
+	"doubletake/internal/airplay"
+)
 
 func (d *Daemon) handleResetRestoreToken(req Request) Response {
 	d.mu.Lock()
@@ -68,22 +72,18 @@ func (d *Daemon) handleResetRestoreToken(req Request) Response {
 			group.resetReservedBy = nil
 		}
 		d.mu.Unlock()
-		if clearErr == nil {
-			_ = credentialReset.Rollback()
-		}
+		rollbackErr := rollbackRestoreTokenReset(credentialReset, clearErr)
 		d.streamWorkers.Done()
-		return Response{OK: false, State: state, Error: "daemon is shutting down"}
+		return Response{OK: false, State: state, Error: resetFailure("daemon is shutting down", rollbackErr)}
 	}
 	if !reservationCurrent {
 		if group.resetReservedBy == entry {
 			group.resetReservedBy = nil
 		}
 		d.mu.Unlock()
-		if clearErr == nil {
-			_ = credentialReset.Rollback()
-		}
+		rollbackErr := rollbackRestoreTokenReset(credentialReset, clearErr)
 		d.streamWorkers.Done()
-		return Response{OK: false, State: state, Error: "restore token reset was canceled for " + target}
+		return Response{OK: false, State: state, Error: resetFailure("restore token reset was canceled for "+target, rollbackErr)}
 	}
 	if clearErr != nil {
 		group.resetReservedBy = nil
@@ -132,12 +132,12 @@ func (d *Daemon) handleResetRestoreToken(req Request) Response {
 		d.mu.Unlock()
 		abandoned.run()
 		cancel()
-		_ = credentialReset.Rollback()
+		rollbackErr := credentialReset.Rollback()
 		d.streamWorkers.Done()
 		if shuttingDown {
-			return Response{OK: false, State: state, Error: "daemon is shutting down"}
+			return Response{OK: false, State: state, Error: resetFailure("daemon is shutting down", rollbackErr)}
 		}
-		return Response{OK: false, State: state, Error: "restore token reset was canceled for " + target}
+		return Response{OK: false, State: state, Error: resetFailure("restore token reset was canceled for "+target, rollbackErr)}
 	}
 	d.clearLastErrorForTargetLocked(target)
 	state = d.overallStateLocked()
@@ -151,6 +151,20 @@ func (d *Daemon) handleResetRestoreToken(req Request) Response {
 	return Response{OK: true, State: state, Device: target, DeviceIP: target}
 }
 
+func rollbackRestoreTokenReset(reset *airplay.RestoreTokenReset, clearErr error) error {
+	if clearErr != nil {
+		return nil
+	}
+	return reset.Rollback()
+}
+
+func resetFailure(message string, rollbackErr error) string {
+	if rollbackErr == nil {
+		return message
+	}
+	return message + "; restore token rollback failed: " + rollbackErr.Error()
+}
+
 // restoreTokenResetReservationCurrentLocked reports whether the exact stream
 // and exclusive capture generation reserved before credential I/O are still
 // current. Must be called with d.mu held.

internal/daemon/reset_restore_token_test.go

@@ -404,6 +404,87 @@ func TestResetRestoreTokenConcurrentDisconnectReportsCancellationAndOverallState
 	}
 }
 
+func TestCanceledResetReportsRollbackPersistenceFailures(t *testing.T) {
+	for _, test := range []struct {
+		name        string
+		lookupErrAt int
+		saveErrAt   int
+	}{
+		{name: "rollback lookup", lookupErrAt: 2},
+		{name: "rollback save", saveErrAt: 2},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			backend := &controlledCredentialBackend{
+				credentials: &airplay.SavedCredentials{RestoreToken: "restore-1"},
+				lookupErrAt: test.lookupErrAt,
+				saveErrAt:   test.saveErrAt,
+			}
+			d, entry, _, _ := newResetTestDaemon(t, backend)
+			cleanupStarted := make(chan struct{})
+			cleanupRelease := make(chan struct{})
+			originalCancel := entry.cancelFn
+			entry.cancelFn = func() {
+				close(cleanupStarted)
+				<-cleanupRelease
+				originalCancel()
+			}
+			defer d.Shutdown()
+
+			resetResponse := make(chan Response, 1)
+			go func() {
+				resetResponse <- d.handleResetRestoreToken(Request{Cmd: "reset-restore-token", Target: resetTestTarget})
+			}()
+			waitForResetSignal(t, cleanupStarted, "cleanup did not start")
+			if response := d.handleDisconnect(Request{Cmd: "disconnect", Target: resetTestTarget}); !response.OK {
+				t.Fatalf("disconnect response = %+v", response)
+			}
+			close(cleanupRelease)
+
+			response := waitForResetResponse(t, resetResponse)
+			if response.OK || !strings.Contains(response.Error, "rollback failed") {
+				t.Fatalf("reset response = %+v, want explicit rollback failure", response)
+			}
+		})
+	}
+}
+
+func TestRestoreTokenResetReservationMigratesChangedCanvasExclusively(t *testing.T) {
+	d, entry, reservation, _ := newResetTestDaemon(t, &controlledCredentialBackend{})
+	oldKey := reservation.key
+	newKey := normalizedVideoCaptureKey(1280, 720, airplay.VideoCodecH264)
+	reservation.broadcast = nil
+	reservation.resetReservedBy = entry
+
+	d.mu.Lock()
+	migrated, err := d.migrateRestoreTokenResetReservationLocked(entry, newKey)
+	oldReleased := d.captureGroups[oldKey] == nil
+	newOwned := d.captureGroups[newKey] == reservation
+	d.mu.Unlock()
+	if err != nil || migrated != reservation || !oldReleased || !newOwned {
+		t.Fatalf("migration = %p, %v, oldReleased=%t newOwned=%t", migrated, err, oldReleased, newOwned)
+	}
+
+	occupiedKey := normalizedVideoCaptureKey(3840, 2160, airplay.VideoCodecH264)
+	occupied := &videoCaptureGroup{key: occupiedKey}
+	d.mu.Lock()
+	d.captureGroups[occupiedKey] = occupied
+	_, err = d.migrateRestoreTokenResetReservationLocked(entry, occupiedKey)
+	stillOwned := d.captureGroups[newKey] == reservation && entry.captureGroup == reservation
+	d.mu.Unlock()
+	if err == nil || !errors.Is(err, errCaptureGroupResetReserved) || !stillOwned {
+		t.Fatalf("occupied migration = %v, stillOwned=%t", err, stillOwned)
+	}
+
+	d.mu.Lock()
+	cleanup := d.detachStreamLocked(resetTestTarget)
+	reservationReleased := d.captureGroups[newKey] == nil
+	d.mu.Unlock()
+	cleanup.run()
+	if !reservationReleased {
+		t.Fatal("disconnect left migrated reservation behind")
+	}
+}
+
 func TestResetRestoreTokenClearsExclusiveTargetAndReconnectsActualPort(t *testing.T) {
 	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
@@ -499,9 +580,17 @@ type controlledCredentialBackend struct {
 	saveStarted chan struct{}
 	saveRelease <-chan struct{}
 	saveOnce    sync.Once
+	lookupCalls int
+	saveCalls   int
+	lookupErrAt int
+	saveErrAt   int
 }
 
 func (b *controlledCredentialBackend) Lookup(string) (*airplay.SavedCredentials, error) {
+	b.lookupCalls++
+	if b.lookupErrAt > 0 && b.lookupCalls == b.lookupErrAt {
+		return nil, errors.New("injected lookup failure")
+	}
 	if b.lookupErr != nil {
 		err := b.lookupErr
 		b.lookupErr = nil
@@ -521,6 +610,10 @@ func (b *controlledCredentialBackend) Save(_ string, credentials *airplay.SavedC
 	if b.saveRelease != nil {
 		<-b.saveRelease
 	}
+	b.saveCalls++
+	if b.saveErrAt > 0 && b.saveCalls == b.saveErrAt {
+		return errors.New("injected save failure")
+	}
 	if b.saveErr != nil {
 		return b.saveErr
 	}