Wait for review terminal before cleanup

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)

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<Cancellation: Sendable>(
+        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(

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<Void, any Error> {
+            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 {