From c4ff426fcd3102557b4eb2a8d3807301c0dd3e6c Mon Sep 17 00:00:00 2001 From: Kazuki Nakashima <65545348+lynnswap@users.noreply.github.com> Date: Sat, 22 Aug 2026 14:27:53 +0900 Subject: [PATCH 1/3] Own settings cutover commit replay --- .../Settings/CodexReviewSettingsService.swift | 229 ++++++++++++++---- Sources/CodexReviewTesting/TestSupport.swift | 22 ++ ...dexReviewSettingsRuntimeCutoverTests.swift | 115 +++++++++ 3 files changed, 321 insertions(+), 45 deletions(-) diff --git a/Sources/CodexReview/Settings/CodexReviewSettingsService.swift b/Sources/CodexReview/Settings/CodexReviewSettingsService.swift index a4c7fd56..44614314 100644 --- a/Sources/CodexReview/Settings/CodexReviewSettingsService.swift +++ b/Sources/CodexReview/Settings/CodexReviewSettingsService.swift @@ -38,6 +38,7 @@ package final class CodexReviewSettingsService { case foreignToken case tokenAlreadyConsumed case staleToken + case conflictingCommitSnapshot case epochExhausted } @@ -72,6 +73,7 @@ package final class CodexReviewSettingsService { RuntimeCutoverToken, priorErrorMessage: String? ) + case committing(RuntimeCutoverToken) case awaitingRecovery( committedEpoch: UInt64, deferredEpoch: UInt64, @@ -84,7 +86,7 @@ package final class CodexReviewSettingsService { .active case .draining: .draining - case .awaitingCommit: + case .awaitingCommit, .committing: .awaitingCommit case .awaitingRecovery: .awaitingRecovery @@ -95,7 +97,7 @@ package final class CodexReviewSettingsService { switch self { case .active(let epoch, _): epoch - case .draining(let token), .awaitingCommit(let token, _): + case .draining(let token), .awaitingCommit(let token, _), .committing(let token): token.targetEpoch case .awaitingRecovery(_, let deferredEpoch, _): deferredEpoch @@ -108,6 +110,8 @@ package final class CodexReviewSettingsService { activeEpoch == epoch case .draining(let token): token.sourceEpoch == epoch + case .committing(let token): + token.targetEpoch == epoch case .awaitingCommit, .awaitingRecovery: false } @@ -126,12 +130,43 @@ package final class CodexReviewSettingsService { tokenID case .awaitingRecovery(_, _, let tokenID): tokenID + case .committing(let token): + token.id case .draining, .awaitingCommit: nil } } } + private typealias RuntimeCommitResult = Result + + private enum RuntimeCommitOperation { + case running( + RuntimeCutoverToken, + CodexReviewSettings.Snapshot, + Task + ) + case completed( + RuntimeCutoverToken, + CodexReviewSettings.Snapshot, + RuntimeCommitResult + ) + + var token: RuntimeCutoverToken { + switch self { + case .running(let token, _, _), .completed(let token, _, _): + token + } + } + + var snapshot: CodexReviewSettings.Snapshot { + switch self { + case .running(_, let snapshot, _), .completed(_, let snapshot, _): + snapshot + } + } + } + let initialSnapshot: CodexReviewSettings.Snapshot private let backend: any CodexReviewSettingsBackend @@ -145,6 +180,7 @@ package final class CodexReviewSettingsService { private var queuedIntents: [QueuedIntent] = [] private var processingEpoch: UInt64? private var epochDrainWaiters: [UInt64: [CheckedContinuation]] = [:] + private var runtimeCommitOperation: RuntimeCommitOperation? package var runtimeCutoverStatus: RuntimeCutoverStatus { cutoverPhase.status @@ -212,7 +248,7 @@ package final class CodexReviewSettingsService { priorErrorMessage: settingsStore.lastErrorMessage ) - case .draining, .awaitingCommit: + case .draining, .awaitingCommit, .committing: throw RuntimeCutoverError.cutoverAlreadyInProgress } @@ -224,23 +260,104 @@ package final class CodexReviewSettingsService { token: RuntimeCutoverToken, snapshot: CodexReviewSettings.Snapshot ) async throws { + if let operation = runtimeCommitOperation, + operation.token == token + { + guard operation.snapshot == snapshot else { + throw RuntimeCutoverError.conflictingCommitSnapshot + } + let result = await runtimeCommitResult(for: operation) + clearCompletedRuntimeCommit(token: token) + try result.get() + return + } + _ = try requireCurrentCutoverToken(token) - guard let settingsStore else { + guard settingsStore != nil else { throw RuntimeCutoverError.settingsStoreUnavailable } - settingsStore.apply(snapshot: snapshot) - lastPersistedSelection = settingsStore.currentSelection() - cutoverPhase = .active( - epoch: token.targetEpoch, - lastConsumedTokenID: token.id - ) - replayQueuedSelectionIntents( - for: token.targetEpoch, - settingsStore: settingsStore - ) - settingsStore.finishLoading(errorMessage: nil) - await drainIntents(for: token.targetEpoch) + cutoverPhase = .committing(token) + let task = Task { @MainActor [self] in + let result: RuntimeCommitResult + do { + try await performRuntimeCommit(token: token, snapshot: snapshot) + result = .success(()) + } catch { + result = .failure(error) + } + runtimeCommitOperation = .completed(token, snapshot, result) + return result + } + runtimeCommitOperation = .running(token, snapshot, task) + let result = await task.value + clearCompletedRuntimeCommit(token: token) + try result.get() + } + + private func runtimeCommitResult( + for operation: RuntimeCommitOperation + ) async -> RuntimeCommitResult { + switch operation { + case .running(_, _, let task): + await task.value + case .completed(_, _, let result): + result + } + } + + private func clearCompletedRuntimeCommit(token: RuntimeCutoverToken) { + guard case .completed(let completedToken, _, _) = runtimeCommitOperation, + completedToken == token + else { + return + } + runtimeCommitOperation = nil + } + + private func performRuntimeCommit( + token: RuntimeCutoverToken, + snapshot: CodexReviewSettings.Snapshot + ) async throws { + do { + guard let settingsStore else { + throw RuntimeCutoverError.settingsStoreUnavailable + } + + settingsStore.apply(snapshot: snapshot) + lastPersistedSelection = settingsStore.currentSelection() + replayQueuedSelectionIntents( + for: token.targetEpoch, + settingsStore: settingsStore + ) + settingsStore.finishLoading(errorMessage: nil) + + if let error = await drainIntents( + for: token.targetEpoch, + retainingFailedIntents: true + ) { + throw error + } + + cutoverPhase = .active( + epoch: token.targetEpoch, + lastConsumedTokenID: token.id + ) + } catch { + cutoverPhase = .awaitingRecovery( + committedEpoch: token.targetEpoch, + deferredEpoch: token.targetEpoch, + lastConsumedTokenID: token.id + ) + if let settingsStore { + replayQueuedSelectionIntents( + for: token.targetEpoch, + settingsStore: settingsStore + ) + settingsStore.finishLoading(errorMessage: error.localizedDescription) + } + throw error + } } package func abortRuntimeCutover( @@ -344,9 +461,16 @@ package final class CodexReviewSettingsService { } private func drainIntents(for epoch: UInt64) async { + _ = await drainIntents(for: epoch, retainingFailedIntents: false) + } + + private func drainIntents( + for epoch: UInt64, + retainingFailedIntents: Bool + ) async -> (any Error)? { guard processingEpoch == nil else { await waitUntilEpochDrained(epoch) - return + return nil } processingEpoch = epoch @@ -357,27 +481,40 @@ package final class CodexReviewSettingsService { var retainedSelectionIntents: [QueuedIntent] = [] while cutoverPhase.permitsDispatch(for: epoch) { - if takeQueuedRefresh(for: epoch) { + let refreshIntents = takeQueuedRefreshIntents(for: epoch) + if refreshIntents.isEmpty == false { if cutoverPhase.isDrainingSource(epoch) == false { - await performRefresh() + if let error = await performRefresh() { + guard retainingFailedIntents else { + continue + } + queuedIntents.insert(contentsOf: refreshIntents, at: 0) + queuedIntents.insert(contentsOf: retainedSelectionIntents, at: 0) + return error + } } continue } let selectionIntents = takeQueuedSelectionIntents(for: epoch) guard selectionIntents.isEmpty == false else { - return + return nil } retainedSelectionIntents.append(contentsOf: selectionIntents) - if await persistSelectionIntents(retainedSelectionIntents) == false { + if let error = await persistSelectionIntents(retainedSelectionIntents) { + if retainingFailedIntents { + queuedIntents.insert(contentsOf: retainedSelectionIntents, at: 0) + return error + } retainedSelectionIntents.removeAll(keepingCapacity: true) } } + return nil } - private func performRefresh() async { + private func performRefresh() async -> (any Error)? { guard let settingsStore else { - return + return RuntimeCutoverError.settingsStoreUnavailable } settingsStore.beginLoading() @@ -386,14 +523,16 @@ package final class CodexReviewSettingsService { settingsStore.apply(snapshot: snapshot) lastPersistedSelection = settingsStore.currentSelection() settingsStore.finishLoading(errorMessage: nil) + return nil } catch { settingsStore.finishLoading(errorMessage: error.localizedDescription) + return error } } - private func persistSelectionIntents(_ intents: [QueuedIntent]) async -> Bool { + private func persistSelectionIntents(_ intents: [QueuedIntent]) async -> (any Error)? { guard let settingsStore else { - return false + return RuntimeCutoverError.settingsStoreUnavailable } let previous = lastPersistedSelection @@ -408,18 +547,17 @@ package final class CodexReviewSettingsService { candidate: candidate ) guard triggers.isEmpty == false else { - return true + return nil } var appliedSelection = previous for trigger in triggers { - let didPersist = await persistSelectionChange( + if let error = await persistSelectionChange( trigger: trigger, previous: appliedSelection, candidate: candidate - ) - guard didPersist else { - return false + ) { + return error } appliedSelection = settingsStore.selectionAfterPersisting( trigger: trigger, @@ -427,16 +565,16 @@ package final class CodexReviewSettingsService { candidate: candidate ) } - return true + return nil } private func persistSelectionChange( trigger: SettingsStore.SelectionTrigger, previous: SettingsStore.Selection, candidate: SettingsStore.Selection - ) async -> Bool { + ) async -> (any Error)? { guard let settingsStore else { - return false + return RuntimeCutoverError.settingsStoreUnavailable } settingsStore.beginLoading() @@ -452,12 +590,12 @@ package final class CodexReviewSettingsService { candidate: candidate ) settingsStore.finishLoading(errorMessage: nil) - return true + return nil } catch { settingsStore.apply(snapshot: settingsStore.snapshot(selection: previous)) lastPersistedSelection = previous settingsStore.finishLoading(errorMessage: error.localizedDescription) - return false + return error } } @@ -552,17 +690,18 @@ package final class CodexReviewSettingsService { ) } - private func takeQueuedRefresh(for epoch: UInt64) -> Bool { - let hasRefresh = queuedIntents.contains { - $0.epoch == epoch && $0.intent.isRefresh - } - guard hasRefresh else { - return false - } - queuedIntents.removeAll { - $0.epoch == epoch && $0.intent.isRefresh + private func takeQueuedRefreshIntents(for epoch: UInt64) -> [QueuedIntent] { + var intents: [QueuedIntent] = [] + queuedIntents.removeAll { queuedIntent in + guard queuedIntent.epoch == epoch, + queuedIntent.intent.isRefresh + else { + return false + } + intents.append(queuedIntent) + return true } - return true + return intents } private func takeQueuedSelectionIntents(for epoch: UInt64) -> [QueuedIntent] { diff --git a/Sources/CodexReviewTesting/TestSupport.swift b/Sources/CodexReviewTesting/TestSupport.swift index 62b53231..2bae5862 100644 --- a/Sources/CodexReviewTesting/TestSupport.swift +++ b/Sources/CodexReviewTesting/TestSupport.swift @@ -185,8 +185,10 @@ package actor FakeCodexReviewBackend: CodexReviewBackend { private var settings: CodexReviewBackendModel.Settings.Snapshot private var settingsUpdateFailureMessage: String? + private var settingsUpdateIsCancelled = false private var settingsUpdateGate: AsyncGate? private var settingsUpdateStartedGate = AsyncGate() + private var settingsUpdateChecksCancellationAfterGate = false private var auth: CodexReviewBackendModel.Auth.Snapshot private var commands: [Command] = [] private var startAdmissionIdentities: [ObjectIdentifier] = [] @@ -262,9 +264,20 @@ package actor FakeCodexReviewBackend: CodexReviewBackend { settingsUpdateFailureMessage = message } + package func cancelNextSettingsUpdate() { + settingsUpdateIsCancelled = true + } + package func holdNextSettingsUpdate(with gate: AsyncGate) { settingsUpdateGate = gate settingsUpdateStartedGate = AsyncGate() + settingsUpdateChecksCancellationAfterGate = false + } + + package func holdNextSettingsUpdateCheckingCancellationAfterGate(with gate: AsyncGate) { + settingsUpdateGate = gate + settingsUpdateStartedGate = AsyncGate() + settingsUpdateChecksCancellationAfterGate = true } package func waitForSettingsUpdate() async { @@ -465,6 +478,15 @@ package actor FakeCodexReviewBackend: CodexReviewBackend { await settingsUpdateStartedGate.open() await settingsUpdateGate?.waitIgnoringCancellation() settingsUpdateGate = nil + let checksCancellation = settingsUpdateChecksCancellationAfterGate + settingsUpdateChecksCancellationAfterGate = false + if checksCancellation { + try Task.checkCancellation() + } + if settingsUpdateIsCancelled { + settingsUpdateIsCancelled = false + throw CancellationError() + } if let settingsUpdateFailureMessage { self.settingsUpdateFailureMessage = nil throw FakeCodexReviewBackendError(message: settingsUpdateFailureMessage) diff --git a/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift b/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift index fb10b137..64e42a9e 100644 --- a/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift +++ b/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift @@ -157,6 +157,121 @@ struct CodexReviewSettingsRuntimeCutoverTests { #expect(store.settings.selectedModel == "deferred-model") } + @Test func callerCancellationCannotCancelOwnedCommitReplay() async throws { + let initial = settingsSnapshot(model: "initial-model") + let backend = FakeCodexReviewBackend(settings: backendSnapshot(initial)) + let store = makeStore(initial: initial, backend: backend) + let token = try await store.settingsService.beginRuntimeCutover() + await store.updateSettingsModel("deferred-model") + + let replayGate = AsyncGate() + await backend.holdNextSettingsUpdateCheckingCancellationAfterGate(with: replayGate) + let commit = Task { @MainActor in + try await store.settingsService.commitRuntimeSnapshot(token: token, snapshot: initial) + } + await backend.waitForSettingsUpdate() + + let joinedCommit = Task { @MainActor in + try await store.settingsService.commitRuntimeSnapshot(token: token, snapshot: initial) + } + await Task.yield() + commit.cancel() + + await #expect( + throws: CodexReviewSettingsService.RuntimeCutoverError.cutoverAlreadyInProgress + ) { + try await store.settingsService.beginRuntimeCutover() + } + #expect(throws: CodexReviewSettingsService.RuntimeCutoverError.tokenAlreadyConsumed) { + try store.settingsService.cancelRuntimeCutover(token: token) + } + #expect(throws: CodexReviewSettingsService.RuntimeCutoverError.tokenAlreadyConsumed) { + try store.settingsService.abortRuntimeCutover(token: token, message: "Superseded.") + } + await #expect( + throws: CodexReviewSettingsService.RuntimeCutoverError.conflictingCommitSnapshot + ) { + try await store.settingsService.commitRuntimeSnapshot( + token: token, + snapshot: settingsSnapshot(model: "conflicting-model") + ) + } + + await replayGate.open() + try await commit.value + try await joinedCommit.value + + #expect(store.settingsService.runtimeCutoverStatus == .active) + #expect(await backend.settingsSnapshot().model == "deferred-model") + #expect(store.settings.selectedModel == "deferred-model") + #expect(store.settings.lastErrorMessage == nil) + + await backend.failNextSettingsUpdate(message: "Rejected after commit.") + await store.updateSettingsModel("rejected-model") + #expect(await backend.settingsSnapshot().model == "deferred-model") + #expect(store.settings.selectedModel == "deferred-model") + #expect(store.settings.lastErrorMessage == "Rejected after commit.") + } + + @Test func genuineCommitReplayFailureRequeuesAndReprojectsRawIntent() async throws { + let initial = settingsSnapshot(model: "initial-model") + let backend = FakeCodexReviewBackend(settings: backendSnapshot(initial)) + let store = makeStore(initial: initial, backend: backend) + let token = try await store.settingsService.beginRuntimeCutover() + await store.updateSettingsModel("deferred-model") + await backend.failNextSettingsUpdate(message: "Injected replay failure.") + + do { + try await store.settingsService.commitRuntimeSnapshot(token: token, snapshot: initial) + Issue.record("Expected the backend replay failure.") + } catch { + #expect(error.localizedDescription == "Injected replay failure.") + } + + #expect(store.settingsService.runtimeCutoverStatus == .awaitingRecovery) + #expect(await backend.settingsSnapshot().model == "initial-model") + #expect(store.settings.selectedModel == "deferred-model") + #expect(store.settings.lastErrorMessage == "Injected replay failure.") + await #expect(throws: CodexReviewSettingsService.RuntimeCutoverError.tokenAlreadyConsumed) { + try await store.settingsService.commitRuntimeSnapshot(token: token, snapshot: initial) + } + + let recoveryToken = try await store.settingsService.beginRuntimeCutover() + try await store.settingsService.commitRuntimeSnapshot( + token: recoveryToken, + snapshot: initial + ) + + #expect(store.settingsService.runtimeCutoverStatus == .active) + #expect(await backend.settingsSnapshot().model == "deferred-model") + #expect(store.settings.selectedModel == "deferred-model") + #expect(store.settings.lastErrorMessage == nil) + #expect(await backend.recordedCommands().filter(\.isSettingsWrite).count == 2) + } + + @Test func backendCommitCancellationRequeuesIntentForRecovery() async throws { + let initial = settingsSnapshot(model: "initial-model") + let backend = FakeCodexReviewBackend(settings: backendSnapshot(initial)) + let store = makeStore(initial: initial, backend: backend) + let token = try await store.settingsService.beginRuntimeCutover() + await store.updateSettingsModel("deferred-model") + await backend.cancelNextSettingsUpdate() + + await #expect(throws: CancellationError.self) { + try await store.settingsService.commitRuntimeSnapshot(token: token, snapshot: initial) + } + #expect(store.settingsService.runtimeCutoverStatus == .awaitingRecovery) + #expect(await backend.settingsSnapshot().model == "initial-model") + #expect(store.settings.selectedModel == "deferred-model") + #expect(store.settings.lastErrorMessage != nil) + + let recoveryToken = try await store.settingsService.beginRuntimeCutover() + try await store.settingsService.commitRuntimeSnapshot(token: recoveryToken, snapshot: initial) + #expect(await backend.settingsSnapshot().model == "deferred-model") + #expect(store.settings.selectedModel == "deferred-model") + #expect(store.settings.lastErrorMessage == nil) + } + @Test func cancellationTokenMisuseIsTypedAndNeverMutatesState() async throws { let initial = settingsSnapshot(model: "initial-model") let backend = FakeCodexReviewBackend(settings: backendSnapshot(initial)) From 90663b09cfed0c2c6a5ac6f6566c103dddcfb3ae Mon Sep 17 00:00:00 2001 From: Kazuki Nakashima <65545348+lynnswap@users.noreply.github.com> Date: Sat, 22 Aug 2026 15:01:26 +0900 Subject: [PATCH 2/3] Fix settings commit operation ownership --- .../Settings/CodexReviewSettingsService.swift | 162 +++++++++++++----- ...dexReviewSettingsRuntimeCutoverTests.swift | 83 +++++++++ 2 files changed, 202 insertions(+), 43 deletions(-) diff --git a/Sources/CodexReview/Settings/CodexReviewSettingsService.swift b/Sources/CodexReview/Settings/CodexReviewSettingsService.swift index 44614314..84394dd6 100644 --- a/Sources/CodexReview/Settings/CodexReviewSettingsService.swift +++ b/Sources/CodexReview/Settings/CodexReviewSettingsService.swift @@ -117,6 +117,13 @@ package final class CodexReviewSettingsService { } } + func permitsSubmittedIntentDrain(for epoch: UInt64) -> Bool { + guard case .active(let activeEpoch, _) = self else { + return false + } + return activeEpoch == epoch + } + func isDrainingSource(_ epoch: UInt64) -> Bool { guard case .draining(let token) = self else { return false @@ -140,28 +147,43 @@ package final class CodexReviewSettingsService { private typealias RuntimeCommitResult = Result + private enum RuntimeCommitPublication { + case published + case replayPending + case superseded + } + private enum RuntimeCommitOperation { case running( - RuntimeCutoverToken, - CodexReviewSettings.Snapshot, - Task + id: UUID, + token: RuntimeCutoverToken, + snapshot: CodexReviewSettings.Snapshot, + task: Task ) case completed( - RuntimeCutoverToken, - CodexReviewSettings.Snapshot, - RuntimeCommitResult + id: UUID, + token: RuntimeCutoverToken, + snapshot: CodexReviewSettings.Snapshot, + result: RuntimeCommitResult ) + var id: UUID { + switch self { + case .running(let id, _, _, _), .completed(let id, _, _, _): + id + } + } + var token: RuntimeCutoverToken { switch self { - case .running(let token, _, _), .completed(let token, _, _): + case .running(_, let token, _, _), .completed(_, let token, _, _): token } } var snapshot: CodexReviewSettings.Snapshot { switch self { - case .running(_, let snapshot, _), .completed(_, let snapshot, _): + case .running(_, _, let snapshot, _), .completed(_, _, let snapshot, _): snapshot } } @@ -267,7 +289,7 @@ package final class CodexReviewSettingsService { throw RuntimeCutoverError.conflictingCommitSnapshot } let result = await runtimeCommitResult(for: operation) - clearCompletedRuntimeCommit(token: token) + clearCompletedRuntimeCommit(id: operation.id) try result.get() return } @@ -278,20 +300,22 @@ package final class CodexReviewSettingsService { } cutoverPhase = .committing(token) + let operationID = UUID() let task = Task { @MainActor [self] in - let result: RuntimeCommitResult - do { - try await performRuntimeCommit(token: token, snapshot: snapshot) - result = .success(()) - } catch { - result = .failure(error) - } - runtimeCommitOperation = .completed(token, snapshot, result) - return result + await performRuntimeCommit( + id: operationID, + token: token, + snapshot: snapshot + ) } - runtimeCommitOperation = .running(token, snapshot, task) + runtimeCommitOperation = .running( + id: operationID, + token: token, + snapshot: snapshot, + task: task + ) let result = await task.value - clearCompletedRuntimeCommit(token: token) + clearCompletedRuntimeCommit(id: operationID) try result.get() } @@ -299,16 +323,56 @@ package final class CodexReviewSettingsService { for operation: RuntimeCommitOperation ) async -> RuntimeCommitResult { switch operation { - case .running(_, _, let task): + case .running(_, _, _, let task): await task.value - case .completed(_, _, let result): + case .completed(_, _, _, let result): result } } - private func clearCompletedRuntimeCommit(token: RuntimeCutoverToken) { - guard case .completed(let completedToken, _, _) = runtimeCommitOperation, - completedToken == token + private func publishRuntimeCommitCompletion( + id: UUID, + token: RuntimeCutoverToken, + snapshot: CodexReviewSettings.Snapshot, + result: RuntimeCommitResult + ) -> RuntimeCommitPublication { + guard case .running(let runningID, let runningToken, _, _) = runtimeCommitOperation, + runningID == id, + runningToken == token + else { + return .superseded + } + if case .success = result, + processingEpoch != nil || queuedIntents.contains(where: { $0.epoch == token.targetEpoch }) + { + return .replayPending + } + + switch result { + case .success: + cutoverPhase = .active( + epoch: token.targetEpoch, + lastConsumedTokenID: token.id + ) + case .failure: + cutoverPhase = .awaitingRecovery( + committedEpoch: token.targetEpoch, + deferredEpoch: token.targetEpoch, + lastConsumedTokenID: token.id + ) + } + runtimeCommitOperation = .completed( + id: id, + token: token, + snapshot: snapshot, + result: result + ) + return .published + } + + private func clearCompletedRuntimeCommit(id: UUID) { + guard case .completed(let completedID, _, _, _) = runtimeCommitOperation, + completedID == id else { return } @@ -316,9 +380,10 @@ package final class CodexReviewSettingsService { } private func performRuntimeCommit( + id: UUID, token: RuntimeCutoverToken, snapshot: CodexReviewSettings.Snapshot - ) async throws { + ) async -> RuntimeCommitResult { do { guard let settingsStore else { throw RuntimeCutoverError.settingsStoreUnavailable @@ -332,23 +397,27 @@ package final class CodexReviewSettingsService { ) settingsStore.finishLoading(errorMessage: nil) - if let error = await drainIntents( - for: token.targetEpoch, - retainingFailedIntents: true - ) { - throw error + while true { + if let error = await drainIntents( + for: token.targetEpoch, + retainingFailedIntents: true + ) { + throw error + } + let result: RuntimeCommitResult = .success(()) + switch publishRuntimeCommitCompletion( + id: id, + token: token, + snapshot: snapshot, + result: result + ) { + case .published, .superseded: + return result + case .replayPending: + continue + } } - - cutoverPhase = .active( - epoch: token.targetEpoch, - lastConsumedTokenID: token.id - ) } catch { - cutoverPhase = .awaitingRecovery( - committedEpoch: token.targetEpoch, - deferredEpoch: token.targetEpoch, - lastConsumedTokenID: token.id - ) if let settingsStore { replayQueuedSelectionIntents( for: token.targetEpoch, @@ -356,7 +425,14 @@ package final class CodexReviewSettingsService { ) settingsStore.finishLoading(errorMessage: error.localizedDescription) } - throw error + let result: RuntimeCommitResult = .failure(error) + _ = publishRuntimeCommitCompletion( + id: id, + token: token, + snapshot: snapshot, + result: result + ) + return result } } @@ -452,7 +528,7 @@ package final class CodexReviewSettingsService { } queuedIntents.append(.init(epoch: epoch, intent: intent, requiresCatalogRevalidation: cutoverPhase.status != .active)) - guard cutoverPhase.permitsDispatch(for: epoch), + guard cutoverPhase.permitsSubmittedIntentDrain(for: epoch), processingEpoch == nil else { return diff --git a/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift b/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift index 64e42a9e..0d53cfca 100644 --- a/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift +++ b/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift @@ -213,6 +213,89 @@ struct CodexReviewSettingsRuntimeCutoverTests { #expect(store.settings.lastErrorMessage == "Rejected after commit.") } + @Test func queuedSubmissionCannotTakeCommitOwnedDrain() async throws { + let initial = settingsSnapshot(model: "initial-model") + let backend = FakeCodexReviewBackend(settings: backendSnapshot(initial)) + let store = makeStore(initial: initial, backend: backend) + let token = try await store.settingsService.beginRuntimeCutover() + let replayGate = AsyncGate() + await backend.holdNextSettingsUpdateCheckingCancellationAfterGate(with: replayGate) + + let submittedEdit = Task { @MainActor in + await store.updateSettingsModel("deferred-model") + await store.refreshSettings() + } + let releaseReplay = Task { + await backend.waitForSettingsUpdate() + submittedEdit.cancel() + await replayGate.open() + } + + try await store.settingsService.commitRuntimeSnapshot(token: token, snapshot: initial) + await submittedEdit.value + await releaseReplay.value + + #expect(store.settingsService.runtimeCutoverStatus == .active) + #expect(await backend.settingsSnapshot().model == "deferred-model") + #expect(store.settings.selectedModel == "deferred-model") + #expect(store.settings.lastErrorMessage == nil) + #expect(await backend.recordedCommands().filter(\.isSettingsWrite).count == 1) + #expect(await backend.recordedCommands().filter(\.isSettingsRead).count == 1) + } + + @Test func completedCommitCannotOverwriteSuccessorOperation() async throws { + let initial = settingsSnapshot(model: "initial-model") + let secondPublished = settingsSnapshot(model: "first-model") + let backend = FakeCodexReviewBackend(settings: backendSnapshot(initial)) + let store = makeStore(initial: initial, backend: backend) + let firstToken = try await store.settingsService.beginRuntimeCutover() + await store.updateSettingsModel("first-model") + let firstReplayGate = AsyncGate() + await backend.holdNextSettingsUpdate(with: firstReplayGate) + + let firstCommit = Task { @MainActor in + try await store.settingsService.commitRuntimeSnapshot( + token: firstToken, + snapshot: initial + ) + } + await backend.waitForSettingsUpdate() + + let successor = Task { @MainActor in + try await waitForCutoverStatus(.active, service: store.settingsService) + let token = try await store.settingsService.beginRuntimeCutover() + await store.updateSettingsModel("second-model") + let secondReplayGate = AsyncGate() + await backend.holdNextSettingsUpdate(with: secondReplayGate) + let commit = Task { @MainActor in + try await store.settingsService.commitRuntimeSnapshot( + token: token, + snapshot: secondPublished + ) + } + await backend.waitForSettingsUpdate() + let joinedCommit = Task { @MainActor in + try await store.settingsService.commitRuntimeSnapshot( + token: token, + snapshot: secondPublished + ) + } + await Task.yield() + await secondReplayGate.open() + try await commit.value + try await joinedCommit.value + } + + await firstReplayGate.open() + try await firstCommit.value + try await successor.value + + #expect(store.settingsService.runtimeCutoverStatus == .active) + #expect(await backend.settingsSnapshot().model == "second-model") + #expect(store.settings.selectedModel == "second-model") + #expect(store.settings.lastErrorMessage == nil) + } + @Test func genuineCommitReplayFailureRequeuesAndReprojectsRawIntent() async throws { let initial = settingsSnapshot(model: "initial-model") let backend = FakeCodexReviewBackend(settings: backendSnapshot(initial)) From 5260e0fbe892e25ab3408649ba3b64250e1ec48a Mon Sep 17 00:00:00 2001 From: Kazuki Nakashima <65545348+lynnswap@users.noreply.github.com> Date: Sat, 22 Aug 2026 15:14:39 +0900 Subject: [PATCH 3/3] Reject stale settings commit replay --- .../Settings/CodexReviewSettingsService.swift | 25 ++++++++++++++++++- ...dexReviewSettingsRuntimeCutoverTests.swift | 6 +++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/Sources/CodexReview/Settings/CodexReviewSettingsService.swift b/Sources/CodexReview/Settings/CodexReviewSettingsService.swift index 84394dd6..29d0a7cd 100644 --- a/Sources/CodexReview/Settings/CodexReviewSettingsService.swift +++ b/Sources/CodexReview/Settings/CodexReviewSettingsService.swift @@ -124,6 +124,19 @@ package final class CodexReviewSettingsService { return activeEpoch == epoch } + func ownsCommitResult(for token: RuntimeCutoverToken) -> Bool { + switch self { + case .committing(let currentToken): + currentToken == token + case .active(let epoch, let lastConsumedTokenID): + epoch == token.targetEpoch && lastConsumedTokenID == token.id + case .awaitingRecovery(_, _, let lastConsumedTokenID): + lastConsumedTokenID == token.id + case .draining, .awaitingCommit: + false + } + } + func isDrainingSource(_ epoch: UInt64) -> Bool { guard case .draining(let token) = self else { return false @@ -243,6 +256,7 @@ package final class CodexReviewSettingsService { sourceEpoch: epoch, targetEpoch: epoch + 1 ) + discardCompletedRuntimeCommitForSuccessor() cutoverPhase = .draining(token) if processingEpoch == nil { @@ -265,6 +279,7 @@ package final class CodexReviewSettingsService { sourceEpoch: committedEpoch, targetEpoch: deferredEpoch ) + discardCompletedRuntimeCommitForSuccessor() cutoverPhase = .awaitingCommit( token, priorErrorMessage: settingsStore.lastErrorMessage @@ -283,7 +298,8 @@ package final class CodexReviewSettingsService { snapshot: CodexReviewSettings.Snapshot ) async throws { if let operation = runtimeCommitOperation, - operation.token == token + operation.token == token, + cutoverPhase.ownsCommitResult(for: token) { guard operation.snapshot == snapshot else { throw RuntimeCutoverError.conflictingCommitSnapshot @@ -379,6 +395,13 @@ package final class CodexReviewSettingsService { runtimeCommitOperation = nil } + private func discardCompletedRuntimeCommitForSuccessor() { + guard case .completed = runtimeCommitOperation else { + return + } + runtimeCommitOperation = nil + } + private func performRuntimeCommit( id: UUID, token: RuntimeCutoverToken, diff --git a/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift b/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift index 0d53cfca..4d6705ee 100644 --- a/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift +++ b/Tests/CodexReviewTests/CodexReviewSettingsRuntimeCutoverTests.swift @@ -264,6 +264,12 @@ struct CodexReviewSettingsRuntimeCutoverTests { let successor = Task { @MainActor in try await waitForCutoverStatus(.active, service: store.settingsService) let token = try await store.settingsService.beginRuntimeCutover() + await #expect(throws: CodexReviewSettingsService.RuntimeCutoverError.staleToken) { + try await store.settingsService.commitRuntimeSnapshot( + token: firstToken, + snapshot: initial + ) + } await store.updateSettingsModel("second-model") let secondReplayGate = AsyncGate() await backend.holdNextSettingsUpdate(with: secondReplayGate)