diff --git a/Package.resolved b/Package.resolved index 1a33d0c..6845d35 100644 --- a/Package.resolved +++ b/Package.resolved @@ -1,12 +1,12 @@ { - "originHash" : "77addbcb28a7261cbf082f3e63d433de400e80eda0c78ecf87bb7c653bb9da3a", + "originHash" : "db686e8128086fbeb71f9dc90e21b592d1c55f433180924062def874be45aa52", "pins" : [ { "identity" : "codexkit", "kind" : "remoteSourceControl", "location" : "https://github.com/lynnswap/CodexKit.git", "state" : { - "revision" : "c18e0636ef11d47c508836bdeaa424ae1be5c72b" + "revision" : "99ef48d1306435c0bb801b1b1c233f31685421c6" } }, { diff --git a/Package.swift b/Package.swift index 1cfcc44..f5fe9d8 100644 --- a/Package.swift +++ b/Package.swift @@ -8,7 +8,7 @@ let packageDirectory = URL(fileURLWithPath: #filePath) let localCodexKitPath = packageDirectory .appendingPathComponent("dependencies/CodexKit", isDirectory: true) .path -let codexKitFallbackRevision = "c18e0636ef11d47c508836bdeaa424ae1be5c72b" +let codexKitFallbackRevision = "99ef48d1306435c0bb801b1b1c233f31685421c6" let codexKitDependency: Package.Dependency = FileManager.default.fileExists(atPath: "\(localCodexKitPath)/Package.swift") ? .package(path: localCodexKitPath) diff --git a/Sources/CodexReviewAppServer/AppServerCodexReviewBackend.swift b/Sources/CodexReviewAppServer/AppServerCodexReviewBackend.swift index e532ccc..378ef68 100644 --- a/Sources/CodexReviewAppServer/AppServerCodexReviewBackend.swift +++ b/Sources/CodexReviewAppServer/AppServerCodexReviewBackend.swift @@ -504,7 +504,7 @@ package actor AppServerCodexReviewBackend: CodexReviewBackend, CodexModelActor { ) async { await Task { [appServer] in do { - _ = try await session.cancel() + _ = try await Self.interruptAndAwaitTerminal(session) } catch { appServerBackendLogger.error( "Failed to cancel a review with invalid identity before cleanup: \(error.localizedDescription, privacy: .public)" @@ -599,7 +599,71 @@ package actor AppServerCodexReviewBackend: CodexReviewBackend, CodexModelActor { message: "Interrupt requires the active SDK review session for its attempt." ) } - return try await activeReview.session.cancel() + return try await Self.interruptAndAwaitTerminal(activeReview.session) + } + + private nonisolated static func interruptAndAwaitTerminal( + _ session: CodexReviewSession + ) async throws -> CodexTurnCancellation { + try await interruptAndAwaitTerminal( + interrupt: { try await session.cancel() }, + awaitTerminal: { _ = try await session.collect() }, + terminalMayStillArriveAfterInterruptFailure: { error in + Self.terminalMayStillArrive(afterInterruptFailure: error) + } + ) + } + + nonisolated static func interruptAndAwaitTerminal( + interrupt: @escaping @Sendable () async throws -> Cancellation, + awaitTerminal: @escaping @Sendable () async throws -> Void, + terminalMayStillArriveAfterInterruptFailure: + @escaping @Sendable (any Error) -> Bool + ) async throws -> Cancellation { + // Cleanup callers may themselves be cancelled while tearing down a run. + // Keep terminal ownership only when a live connection can still deliver + // the accepted interrupt's terminal after its acknowledgement failed. + let interruption = Task { + let cancellation: Cancellation + do { + cancellation = try await interrupt() + } catch { + let interruptError = error + if terminalMayStillArriveAfterInterruptFailure(interruptError) { + // The barrier owns lifecycle completion, not error selection: + // the interrupt request failure remains the operation result. + _ = try? await awaitTerminal() + } + throw interruptError + } + try await awaitTerminal() + return cancellation + } + return try await interruption.value + } + + private nonisolated static func terminalMayStillArrive( + afterInterruptFailure error: any Error + ) -> Bool { + guard case .request(let failure) = error as? CodexAppServerError else { + return false + } + return terminalMayStillArrive(afterInterruptRequestFailure: failure.kind) + } + + nonisolated static func terminalMayStillArrive( + afterInterruptRequestFailure failure: CodexRequestFailure.Kind + ) -> Bool { + switch failure { + case .invalidResponse: + return true + case .encode, .server, .overloadRetryExhausted: + return false + case .write, .transport, .deadlineExceeded: + // CodexKit terminates the connection before surfacing a post-write + // failure in these paths; their pre-write forms were never accepted. + return false + } } private func cleanupAppServerReview( diff --git a/Tests/CodexReviewAppServerTests/AppServerClientTests.swift b/Tests/CodexReviewAppServerTests/AppServerClientTests.swift index 171c486..2f1aff0 100644 --- a/Tests/CodexReviewAppServerTests/AppServerClientTests.swift +++ b/Tests/CodexReviewAppServerTests/AppServerClientTests.swift @@ -681,7 +681,22 @@ struct AppServerClientTests { let backend = await makeBackend(appServer: runtime.server) let attempt = try await backend.startReview(makeReviewStart()) - try await backend.interruptReview(attempt.attempt, reason: .init(message: "Stop")) + let interruptTask = Task { + try await backend.interruptReview(attempt.attempt, reason: .init(message: "Stop")) + } + defer { + interruptTask.cancel() + } + await runtime.transport.waitForRequest(.turnInterrupt) + try await emitTurn( + on: runtime, + threadID: "thread-1", + turnID: "turn-1", + state: .interrupted + ) + try await withTimeout { + try await interruptTask.value + } let requests = await runtime.transport.recordedRequests() #expect(requests.map(\.request.operation) == [ @@ -702,6 +717,34 @@ struct AppServerClientTests { #expect(interrupt.1 == "turn-1") } + @Test func interruptReviewCompletesTerminalWaitAfterCallerCancellation() async throws { + let runtime = try await CodexAppServerTestRuntime.start() + try await runtime.transport.enqueueThreadStart(threadID: "thread-1", model: "gpt-5") + try await runtime.transport.enqueueReviewStart( + turnID: "turn-1", + reviewThreadID: "thread-1" + ) + try await runtime.transport.handleTurnInterrupt { _ in } + let backend = await makeBackend(appServer: runtime.server) + let attempt = try await backend.startReview(makeReviewStart()) + + let interruptTask = Task { + try await backend.interruptReview(attempt.attempt, reason: .init(message: "Stop")) + } + await runtime.transport.waitForRequest(.turnInterrupt) + interruptTask.cancel() + try await emitTurn( + on: runtime, + threadID: "thread-1", + turnID: "turn-1", + state: .interrupted + ) + + try await withTimeout { + try await interruptTask.value + } + } + @Test func startReviewMapsRequestFailureToTypedOperation() async throws { let runtime = try await CodexAppServerTestRuntime.start() try await runtime.transport.enqueueFailure( @@ -754,6 +797,84 @@ struct AppServerClientTests { #expect(await runtime.transport.recordedRequests(for: .threadResume).isEmpty) } + @Test func interruptFailureClassificationWaitsOnlyForLiveInvalidResponse() { + #expect(AppServerCodexReviewBackend.terminalMayStillArrive( + afterInterruptRequestFailure: .invalidResponse( + expectedType: "EmptyResponse", + message: "Malformed response", + rawData: nil + ) + )) + #expect(!AppServerCodexReviewBackend.terminalMayStillArrive( + afterInterruptRequestFailure: .encode(message: "Encoding failed") + )) + #expect(!AppServerCodexReviewBackend.terminalMayStillArrive( + afterInterruptRequestFailure: .write(.closed) + )) + #expect(!AppServerCodexReviewBackend.terminalMayStillArrive( + afterInterruptRequestFailure: .transport(.closed) + )) + #expect(!AppServerCodexReviewBackend.terminalMayStillArrive( + afterInterruptRequestFailure: .server(.init(code: -32_011, message: "Rejected")) + )) + #expect(!AppServerCodexReviewBackend.terminalMayStillArrive( + afterInterruptRequestFailure: .deadlineExceeded(.seconds(1)) + )) + #expect(!AppServerCodexReviewBackend.terminalMayStillArrive( + afterInterruptRequestFailure: .overloadRetryExhausted( + last: .init(code: -32_001, message: "Overloaded"), + attempts: 3 + ) + )) + } + + @Test func definitiveInterruptFailureSkipsTerminalBarrier() async { + do { + try await AppServerCodexReviewBackend.interruptAndAwaitTerminal( + interrupt: { () async throws -> Void in + throw AppServerClientTestInterruptionError.rejected + }, + awaitTerminal: { + Issue.record("A definitive interrupt failure must not enter the terminal barrier.") + }, + terminalMayStillArriveAfterInterruptFailure: { _ in + false + } + ) + Issue.record("Expected the definitive interrupt failure.") + } catch { + #expect(error as? AppServerClientTestInterruptionError == .rejected) + } + } + + @Test func ambiguousInterruptFailureRetainsTerminalBarrierAndOriginalError() async throws { + let terminalGate = CodexAppServerTestGate() + let interruption = Task { + try await AppServerCodexReviewBackend.interruptAndAwaitTerminal( + interrupt: { () async throws -> Void in + throw AppServerClientTestInterruptionError.rejected + }, + awaitTerminal: { + await terminalGate.waitIgnoringCancellation() + throw AppServerClientTestInterruptionError.terminalFailed + }, + terminalMayStillArriveAfterInterruptFailure: { _ in + true + } + ) + } + + await terminalGate.waitUntilBlocked() + await terminalGate.open() + + do { + try await interruption.value + Issue.record("Expected the interrupt failure after the terminal barrier opened.") + } catch { + #expect(error as? AppServerClientTestInterruptionError == .rejected) + } + } + @Test func prepareRestartMapsRequestFailureToTypedOperation() async throws { let runtime = try await CodexAppServerTestRuntime.start() try await runtime.transport.enqueueThreadStart(threadID: "thread-1", model: "gpt-5") @@ -779,6 +900,12 @@ struct AppServerClientTests { code: -32_002 ) } + + let runID = try ReviewRunID(validating: "run-1") + let retained = await backend.discardAllPreparedReviewRestarts( + ownedAttemptsByRunID: [runID: attempt.attempt] + ) + #expect(retained == [runID: [attempt.attempt]]) } @Test func restartReviewMapsUnavailableTokenToTypedOperation() async throws { @@ -844,10 +971,21 @@ struct AppServerClientTests { await runtime.transport.waitForRequest(.turnInterrupt, count: 2) try await emitTurn( on: runtime, - threadID: "thread-1", + threadID: "thread-review-child", turnID: "turn-new", state: .interrupted ) + try await runtime.notificationEmitter.emitItemCompleted( + threadID: "thread-1", + turnID: "turn-old", + item: .agentMessage(id: "review-output", text: "Review interrupted") + ) + try await emitTurn( + on: runtime, + threadID: "thread-1", + turnID: "turn-old", + state: .interrupted + ) let token = try await withTimeout { try await prepareTask.value } @@ -1076,6 +1214,11 @@ private enum AppServerClientTestTimeout: Error { case timedOut } +private enum AppServerClientTestInterruptionError: Error, Equatable { + case rejected + case terminalFailed +} + private extension ReviewBackendFailure { var operationFailure: ReviewBackendOperationFailure? { guard case .operation(let failure) = self else { diff --git a/Tests/CodexReviewHostTests/CodexReviewHostTests.swift b/Tests/CodexReviewHostTests/CodexReviewHostTests.swift index 8c80787..29c01de 100644 --- a/Tests/CodexReviewHostTests/CodexReviewHostTests.swift +++ b/Tests/CodexReviewHostTests/CodexReviewHostTests.swift @@ -5173,7 +5173,18 @@ struct CodexReviewHostTests { store.reviewRuns.first?.core.attempt?.turnID.rawValue == "turn-first" }) - try await store.switchAccount(CodexReviewKit.CodexReviewAccount(email: "second@example.com")) + let switchTask = Task { @MainActor in + try await store.switchAccount( + CodexReviewKit.CodexReviewAccount(email: "second@example.com") + ) + } + await firstTransport.waitForRequest(.turnInterrupt) + try await emitInterruptedTurn( + on: firstTransport, + threadID: "thread-first", + turnID: "turn-first" + ) + try await switchTask.value let result = try await reviewRead await secondTransport.waitForRequestCount(2) await firstTransport.waitForRequestCount(7) @@ -5238,7 +5249,16 @@ struct CodexReviewHostTests { store.reviewRuns.first?.core.attempt?.turnID.rawValue == "turn-active" }) - await store.logout() + let logoutTask = Task { @MainActor in + await store.logout() + } + await firstTransport.waitForRequest(.turnInterrupt) + try await emitInterruptedTurn( + on: firstTransport, + threadID: "thread-active", + turnID: "turn-active" + ) + await logoutTask.value let result = try await reviewRead await secondTransport.waitForRequestCount(2) @@ -5390,6 +5410,11 @@ struct CodexReviewHostTests { } let methodsBeforeInterruptCompletes = await transport.recordedRequests().map(\.request.operation) await interruptGate.open() + try await emitInterruptedTurn( + on: transport, + threadID: "thread-1", + turnID: "turn-1" + ) await stopTask.value let result = try await reviewRead @@ -5442,6 +5467,11 @@ struct CodexReviewHostTests { }) #expect(await stopFinished.isCompleted() == false) await interruptGate.open() + try await emitInterruptedTurn( + on: transport, + threadID: "thread-1", + turnID: "turn-1" + ) await stopTask.value let result = try await reviewRead.value @@ -5493,6 +5523,20 @@ struct CodexReviewHostTests { try #require(await waitUntil(timeout: .seconds(2)) { await transport.recordedRequests().map(\.request.operation).contains(.turnInterrupt) }) + try await emitInterruptedTurn( + on: transport, + threadID: "thread-1", + turnID: "turn-1" + ) + try #require(await waitUntil(timeout: .seconds(2)) { + guard let run = store.reviewRuns.first else { + return false + } + if case .waitingForNetwork = run.presentation.lifecycle { + return true + } + return false + }) let stopFinished = CompletionFlag() let stopTask = Task { @MainActor in @@ -6213,6 +6257,20 @@ struct CodexReviewHostTests { } } +private func emitInterruptedTurn( + on transport: FakeCodexAppServerTransport, + threadID: CodexThreadID, + turnID: CodexTurnID +) async throws { + try await transport.notificationEmitter.emitTurnCompleted( + threadID: threadID, + turn: try CodexAppServerTestTurn( + snapshot: .init(id: turnID, state: .interrupted), + items: [] + ) + ) +} + @MainActor private func exerciseUnknownPrimaryCancellation( previousAccountKey: String?, diff --git a/Tools/ReviewMonitor/CodexReviewMonitor.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved b/Tools/ReviewMonitor/CodexReviewMonitor.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved index e599fba..1317590 100644 --- a/Tools/ReviewMonitor/CodexReviewMonitor.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved +++ b/Tools/ReviewMonitor/CodexReviewMonitor.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved @@ -1,6 +1,14 @@ { - "originHash" : "11d1673255c652381957935c0aa0fcc8f6cba52200a3e3e3c7296d58e6207f76", + "originHash" : "844e7e3b43d92f89827170165177645f6aa45da42a6028dea22353eb9f9cbdde", "pins" : [ + { + "identity" : "codexkit", + "kind" : "remoteSourceControl", + "location" : "https://github.com/lynnswap/CodexKit.git", + "state" : { + "revision" : "99ef48d1306435c0bb801b1b1c233f31685421c6" + } + }, { "identity" : "eventsource", "kind" : "remoteSourceControl",