fix(talk): preserve relay output lifecycle

Preserve generation-fenced cancellation, terminal teardown on current cancel failure, pause-safe output and capture state, and late relay cleanup in the shared Apple relay owner. Co-authored-by: Zhilong Zheng <zhengzhilong1115@gmail.com> Punchcard-Session: golden-meadow-cedar-dv
2026-08-12 21:53:00 -06:00 · 2026-08-11 18:02:00 +08:00
parent 4ad52b9fd4
commit 39597be775
7 changed files with 964 additions and 81 deletions
@@ -80,6 +80,11 @@ extension RealtimeTalkRelayTransport {
                guard await gateway.currentRoute() == route else { throw CancellationError() }
                return response
            },
-            isCurrent: { await gateway.currentRoute() == route })
+            isCurrent: { await gateway.currentRoute() == route },
+            supportsOutputGeneration: {
+                await gateway.supportsServerCapability(
+                    .talkOutputGeneration,
+                    ifCurrentRoute: route) == true
+            })
    }
 }
@@ -38,6 +38,11 @@ extension GatewayConnection {
            },
            isCurrent: {
                await self.isCurrentServerLease(lease)
+            },
+            supportsOutputGeneration: {
+                await self.supportsServerCapability(
+                    .talkOutputGeneration,
+                    ifCurrentServerLease: lease) == true
            })
    }

@@ -206,6 +206,26 @@ extension TalkModeRuntime {
            relayGeneration: relayGeneration)
    }

+    func setRealtimeInputPaused(
+        _ paused: Bool,
+        session: RealtimeTalkRelaySession,
+        relayGeneration: UInt64) async -> Bool
+    {
+        do {
+            try await MainActor.run {
+                try session.setInputPaused(paused)
+            }
+            return true
+        } catch {
+            self.logger.error(
+                "talk realtime pause transition failed: \(error.localizedDescription, privacy: .public)")
+            await self.handleRealtimeInputRestartFailure(
+                error.localizedDescription,
+                relayGeneration: relayGeneration)
+            return false
+        }
+    }
+
    private func handleRealtimeTermination(
        _ termination: RealtimeTalkRelayTermination,
        relayGeneration: UInt64) async
@@ -258,10 +278,11 @@ extension TalkModeRuntime {
        }
    }

-    private func handleRealtimeSpeakingChanged(_ speaking: Bool, relayGeneration: UInt64) async {
+    func handleRealtimeSpeakingChanged(_ speaking: Bool, relayGeneration: UInt64) async {
        guard self.realtimeRelayGeneration == relayGeneration,
              self.realtimeSession != nil,
-              self.isEnabled
+              self.isEnabled,
+              !self.isPaused
        else { return }
        if speaking {
            self.phase = .speaking
@@ -272,29 +293,32 @@ extension TalkModeRuntime {
        }
    }

-    private func handleRealtimeInputLevel(_ level: Double, relayGeneration: UInt64) async {
+    func handleRealtimeInputLevel(_ level: Double, relayGeneration: UInt64) async {
        guard self.realtimeRelayGeneration == relayGeneration,
              self.realtimeSession != nil,
-              self.isEnabled
+              self.isEnabled,
+              !self.isPaused
        else { return }
        await MainActor.run { TalkModeController.shared.updateLevel(level) }
    }

-    private func handleRealtimeOutputLevel(_ level: Double?, relayGeneration: UInt64) async {
+    func handleRealtimeOutputLevel(_ level: Double?, relayGeneration: UInt64) async {
        guard self.realtimeRelayGeneration == relayGeneration,
              self.realtimeSession != nil,
-              self.isEnabled
+              self.isEnabled,
+              !self.isPaused
        else { return }
        await MainActor.run { TalkModeController.shared.updateSpeakingLevel(level) }
    }

-    private func handleRealtimeTranscript(
+    func handleRealtimeTranscript(
        _ transcript: RealtimeTalkTranscript,
        relayGeneration: UInt64) async
    {
        guard self.realtimeRelayGeneration == relayGeneration,
              self.realtimeSession != nil,
-              self.isEnabled
+              self.isEnabled,
+              !self.isPaused
        else { return }
        let text = transcript.text.trimmingCharacters(in: .whitespacesAndNewlines)
        guard !text.isEmpty else { return }
@@ -104,6 +104,7 @@ actor TalkModeRuntime {
    var bypassRealtimeOnNextStart = false
    var realtimeRelayGeneration: UInt64 = 0
    var realtimeRelayStartGeneration: UInt64?
+    private var pendingRealtimeRelayStartLifecycleGeneration: Int?
    var realtimeRestartGeneration: UInt64 = 0
    var realtimeRestartTask: Task<Void, Never>?
    var startDependencies: StartDependencies?
@@ -157,29 +158,61 @@ actor TalkModeRuntime {

        guard self.isEnabled else { return }

+        if paused {
+            self.pendingRealtimeRelayStartLifecycleGeneration = nil
+            if self.realtimeRelayStartGeneration != nil {
+                self.realtimeRelayGeneration &+= 1
+            }
+        } else if self.realtimeRelayStartGeneration != nil, self.shouldAttemptRealtimeRelay() {
+            self.pendingRealtimeRelayStartLifecycleGeneration = self.lifecycleGeneration
+            return
+        }
+
        if paused, self.realtimeSession == nil {
            self.cancelScheduledRealtimeRecovery()
        }

        if let realtimeSession {
            let relayGeneration = self.realtimeRelayGeneration
-            do {
-                try await MainActor.run {
-                    try realtimeSession.setInputPaused(paused)
+            if paused {
+                await MainActor.run { realtimeSession.setOutputPaused(true) }
+                guard self.isPaused,
+                      self.realtimeRelayGeneration == relayGeneration,
+                      self.realtimeSession === realtimeSession,
+                      await self.setRealtimeInputPaused(
+                          true,
+                          session: realtimeSession,
+                          relayGeneration: relayGeneration)
+                else { return }
+                self.lastTranscript = ""
+                self.lastHeard = nil
+                self.lastSpeechEnergyAt = nil
+                self.phase = .idle
+                await MainActor.run {
+                    TalkModeController.shared.updateLevel(0)
+                    TalkModeController.shared.updateSpeakingLevel(nil)
+                    TalkModeController.shared.updatePartialTranscript("")
+                    TalkModeController.shared.updatePhase(.idle)
                }
-                if !paused {
-                    self.phase = .listening
-                    await MainActor.run { TalkModeController.shared.updatePhase(.listening) }
+            } else {
+                guard await self.setRealtimeInputPaused(
+                    false,
+                    session: realtimeSession,
+                    relayGeneration: relayGeneration),
+                    !self.isPaused,
+                    self.realtimeRelayGeneration == relayGeneration,
+                    self.realtimeSession === realtimeSession
+                else { return }
+                await MainActor.run { realtimeSession.setOutputPaused(false) }
+                guard !self.isPaused,
+                      self.realtimeRelayGeneration == relayGeneration,
+                      self.realtimeSession === realtimeSession
+                else {
+                    await MainActor.run { realtimeSession.setOutputPaused(true) }
+                    return
                }
-            } catch {
-                // Only the unpause branch throws, and the session restores its own input-paused
-                // state on the way out. Without this the runtime would report Talk as running
-                // with no microphone, so route it through the typed close-and-recover path.
-                self.logger.error(
-                    "talk realtime pause transition failed: \(error.localizedDescription, privacy: .public)")
-                await self.handleRealtimeInputRestartFailure(
-                    error.localizedDescription,
-                    relayGeneration: relayGeneration)
+                self.phase = .listening
+                await MainActor.run { TalkModeController.shared.updatePhase(.listening) }
            }
            return
        }
@@ -264,8 +297,12 @@ actor TalkModeRuntime {
                }
                return
            } catch is CancellationError {
+                if self.consumePendingRealtimeRelayStart() {
+                    await self.start()
+                }
                return
            } catch {
+                self.pendingRealtimeRelayStartLifecycleGeneration = nil
                guard self.canOwnNativeFallback(fallbackOwner) else { return }
                self.logger.error(
                    "talk realtime unavailable; using native fallback: " +
@@ -291,6 +328,7 @@ actor TalkModeRuntime {
    }

    private func stop() async {
+        self.pendingRealtimeRelayStartLifecycleGeneration = nil
        self.resetRealtimeRecoveryState()
        self.realtimeRelayGeneration &+= 1
        self.realtimeRelayStartGeneration = nil
@@ -321,6 +359,14 @@ actor TalkModeRuntime {
        }
    }

+    func consumePendingRealtimeRelayStart() -> Bool {
+        guard let generation = self.pendingRealtimeRelayStartLifecycleGeneration else { return false }
+        self.pendingRealtimeRelayStartLifecycleGeneration = nil
+        return generation == self.lifecycleGeneration && self.isEnabled && !self.isPaused &&
+            self.realtimeSession == nil && self.realtimeRelayStartGeneration == nil &&
+            self.shouldAttemptRealtimeRelay()
+    }
+
    #if DEBUG
    func _test_enableRealtimeRelaySelection() {
        (self.macOSRealtimeRelayOptIn, self.hasGatewayRealtimeRelayTuple) = (true, true)
@@ -13,12 +13,14 @@ private enum RuntimeTestAudioCaptureError: Error {
 private final class RuntimeTestAudioCapture: RealtimeTalkAudioCapturing {
    let suppressesInputDuringOutput = false
    var startError: Error?
+    private(set) var startCount = 0

    func start(
        targetSampleRate: Double,
        onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void,
        onFailure: @escaping @MainActor (String) -> Void) throws
    {
+        self.startCount += 1
        if let startError = self.startError {
            throw startError
        }
@@ -148,6 +150,12 @@ private enum RuntimeRelayStartError: Error {
    case failed
 }

+enum RuntimeRelayStartupPauseOutcome: Equatable {
+    case resume
+    case remainPaused
+    case disable
+}
+
@MainActor
 private func makeRuntimeTestRealtimeSession(
    player: RuntimeTestPCMPlayer) -> RealtimeTalkRelaySession
@@ -335,6 +343,84 @@ struct TalkModeRuntimeSpeechTests {
        session.stop()
    }

+    @Test @MainActor func `pausing realtime resets visible state and ignores late callbacks`() async {
+        let runtime = TalkModeRuntime()
+        let player = RuntimeTestPCMPlayer()
+        let session = makeRuntimeTestRealtimeSession(player: player)
+        let relayGeneration = await runtime._test_prepareEnabledRealtimeSessionForClose(session)
+        TalkModeController.shared.updatePhase(.speaking)
+        TalkModeController.shared.updateLevel(0.8)
+        TalkModeController.shared.updatePartialTranscript("stale")
+
+        await runtime.setPaused(true)
+        await runtime.handleRealtimeSpeakingChanged(true, relayGeneration: relayGeneration)
+        await runtime.handleRealtimeInputLevel(0.9, relayGeneration: relayGeneration)
+        await runtime.handleRealtimeOutputLevel(0.8, relayGeneration: relayGeneration)
+        await runtime.handleRealtimeTranscript(
+            .init(role: "user", text: "late transcript", isFinal: false),
+            relayGeneration: relayGeneration)
+
+        #expect(await runtime._test_phase() == .idle)
+        #expect(TalkModeController.shared.phase == .idle)
+        #expect(TalkModeController.shared.level == 0)
+        #expect(TalkModeController.shared.partialTranscript.isEmpty)
+        #expect(player.stopCount == 0)
+
+        await runtime._test_cancelRealtimeRecovery()
+        session.stop()
+    }
+
+    @Test @MainActor func `resuming realtime restarts input and reuses the relay`() async throws {
+        let runtime = TalkModeRuntime()
+        let audioCapture = RuntimeTestAudioCapture()
+        let player = RuntimeTestPCMPlayer()
+        let eventChannel = AsyncStream<EventFrame>.makeStream()
+        let result = TalkSessionCreateResult(
+            sessionid: "talk-session",
+            mode: AnyCodable("realtime"),
+            transport: AnyCodable("gateway-relay"),
+            brain: AnyCodable("agent-consult"),
+            relaysessionid: "relay-1")
+        let resultData = try JSONEncoder().encode(result)
+        let session = RealtimeTalkRelaySession(
+            transport: RealtimeTalkRelayTransport(
+                subscribeServerEvents: { _ in eventChannel.stream },
+                request: { method, _, _ in
+                    if method == "talk.session.create" {
+                        eventChannel.continuation.yield(EventFrame(
+                            type: "event",
+                            event: "talk.event",
+                            payload: AnyCodable([
+                                "relaySessionId": "relay-1",
+                                "type": "ready",
+                            ]),
+                            seq: nil,
+                            stateversion: nil))
+                        return resultData
+                    }
+                    return Data("{\"ok\":true}".utf8)
+                }),
+            options: .init(sessionKey: "main", provider: "openai", model: "gpt-realtime-2", voice: nil),
+            audioCapture: audioCapture,
+            pcmPlayer: player,
+            onStatus: { _ in },
+            onSpeakingChanged: { _ in })
+        try await session.start()
+        let relayGeneration = await runtime._test_prepareEnabledRealtimeSessionForClose(session)
+
+        await runtime.setPaused(true)
+        await runtime.setPaused(false)
+
+        #expect(audioCapture.startCount == 2)
+        #expect(await runtime._test_realtimeSessionIs(session))
+        await runtime.handleRealtimeSpeakingChanged(true, relayGeneration: relayGeneration)
+        #expect(await runtime._test_phase() == .speaking)
+
+        await runtime._test_cancelRealtimeRecovery()
+        session.stop()
+        eventChannel.continuation.finish()
+    }
+
    @Test @MainActor func `disabling during relay startup stops the published session`() async {
        let runtime = TalkModeRuntime()
        let lifecycleGeneration = await runtime._test_prepareEnabledLifecycle()
@@ -368,6 +454,58 @@ struct TalkModeRuntimeSpeechTests {
        #expect(player.stopCount == 1)
    }

+    @Test(arguments: [
+        RuntimeRelayStartupPauseOutcome.resume,
+        .remainPaused,
+        .disable,
+    ])
+    @MainActor
+    func `relay startup pause retries only a matching resume`(
+        outcome: RuntimeRelayStartupPauseOutcome) async
+    {
+        let runtime = TalkModeRuntime()
+        let lifecycleGeneration = await runtime._test_prepareEnabledLifecycle()
+        await runtime._test_enableRealtimeRelaySelection()
+        let barrier = RuntimeContinuationBarrier()
+        let probe = RuntimeCommitProbe()
+        let player = RuntimeTestPCMPlayer()
+        let session = makeRuntimeTestRealtimeSession(player: player)
+        let attempt = Task { @MainActor in
+            do {
+                try await runtime._test_startRealtimeRelay(
+                    lifecycleGeneration: lifecycleGeneration,
+                    makeSession: { session },
+                    start: { _ in
+                        probe.record("start")
+                        await barrier.wait()
+                    })
+                return true
+            } catch {
+                return false
+            }
+        }
+
+        await barrier.waitUntilEntered()
+        #expect(await runtime._test_realtimeSessionIs(session))
+        await runtime.setPaused(true)
+        if outcome != .remainPaused {
+            await runtime.setPaused(false)
+        }
+        if outcome == .disable {
+            await runtime.setEnabled(false)
+        }
+        await barrier.release()
+
+        #expect(await attempt.value == false)
+        #expect(await runtime._test_realtimeSessionIsActive() == false)
+        if await runtime.consumePendingRealtimeRelayStart() { probe.record("retry") }
+        if await runtime.consumePendingRealtimeRelayStart() { probe.record("retry") }
+        #expect(probe.values() == (outcome == .resume ? ["start", "retry"] : ["start"]))
+        #expect(player.stopCount == 1)
+
+        await runtime.setEnabled(false)
+    }
+
    @Test func `processed recognition start failure retries a fresh raw capture`() {
        let probe = RuntimeCommitProbe()

@@ -65,15 +65,18 @@ public struct RealtimeTalkRelayTransport: Sendable {
    public let subscribeServerEvents: @Sendable (Int) async -> AsyncStream<EventFrame>
    public let request: @Sendable (String, [String: AnyCodable]?, Double) async throws -> Data
    public let isCurrent: @Sendable () async -> Bool
+    public let supportsOutputGeneration: @Sendable () async -> Bool

    public init(
        subscribeServerEvents: @escaping @Sendable (Int) async -> AsyncStream<EventFrame>,
        request: @escaping @Sendable (String, [String: AnyCodable]?, Double) async throws -> Data,
-        isCurrent: @escaping @Sendable () async -> Bool = { true })
+        isCurrent: @escaping @Sendable () async -> Bool = { true },
+        supportsOutputGeneration: @escaping @Sendable () async -> Bool = { false })
    {
        self.subscribeServerEvents = subscribeServerEvents
        self.request = request
        self.isCurrent = isCurrent
+        self.supportsOutputGeneration = supportsOutputGeneration
    }
 }

@@ -118,6 +121,7 @@ public enum RealtimeTalkRelayTermination: Equatable, Sendable {
    case remoteClose(reason: String?)
    case eventStreamEnded
    case audioCaptureFailed(message: String)
+    case outputCancellationFailed
 }

 private actor RealtimeAudioSender {
@@ -257,10 +261,17 @@ public final class RealtimeTalkRelaySession {
    private var pendingPlaybackMarks: [String] = []
    private var audioSender: RealtimeAudioSender?
    private var isInputPaused = false
+    private var isOutputPaused = false
    private var audioCaptureGeneration: UInt64 = 0
    private var isClosed = false
    private var lifecycleGeneration: UInt64 = 0
+    private var outputCancellationGeneration: UInt64 = 0
    private var isOutputPlaying = false
+    private var outputIdentity: OutputIdentity?
+    private var suppressedOutputIdentity: OutputIdentity?
+    private var awaitingOutputClear = false
+    private var cancelledOutputGenerationWatermark: Int?
+    private var outputCancellationTask: Task<Void, Never>?
    private var outputStartedAtMs: Double?
    private var outputPlaybackExpectedEndMs: Double = 0
    private var lastBargeInAtMs: Double = 0
@@ -334,6 +345,16 @@ public final class RealtimeTalkRelaySession {
                }
                return
            }
+            if let startupIssue {
+                if let relaySessionId = result.relaysessionid?.trimmingCharacters(in: .whitespacesAndNewlines),
+                   !relaySessionId.isEmpty
+                {
+                    await Self.closeRelaySession(
+                        transport: self.transport,
+                        relaySessionId: relaySessionId)
+                }
+                throw Self.startupFailureError(startupIssue)
+            }
            guard let relaySessionId = result.relaysessionid?.trimmingCharacters(in: .whitespacesAndNewlines),
                  !relaySessionId.isEmpty
            else {
@@ -397,14 +418,13 @@ public final class RealtimeTalkRelaySession {
        for task in self.toolCallTasks.values {
            task.cancel()
        }
-        for task in self.audioSendTasks.values {
-            task.cancel()
-        }
-        self.audioSendTasks.removeAll()
        self.pendingPlaybackMarks.removeAll()
        let audioSender = self.audioSender
        self.audioSender = nil
        Task { await audioSender?.close() }
+        self.retireOutputCancellation()
+        self.cancelledOutputGenerationWatermark = nil
+        self.isOutputPaused = false
        self.stopOutputPlayback()
        if sendClose, let relaySessionId = self.relaySessionId {
            Task { [transport] in
@@ -423,6 +443,12 @@ public final class RealtimeTalkRelaySession {
        ])
    }

+    private nonisolated static func startupFailureError(_ issue: RealtimeTalkRelayIssue) -> NSError {
+        NSError(domain: "RealtimeTalkRelay", code: 6, userInfo: [
+            NSLocalizedDescriptionKey: issue.message,
+        ])
+    }
+
    private nonisolated static func closeRelaySession(
        transport: RealtimeTalkRelayTransport,
        relaySessionId: String) async
@@ -431,18 +457,6 @@ public final class RealtimeTalkRelaySession {
        _ = try? await transport.request("talk.session.close", payload, 8000)
    }

-    public func cancelOutput(reason: String = "user") {
-        self.stopOutputPlayback()
-        guard let relaySessionId else { return }
-        Task { [transport] in
-            let payload: [String: AnyCodable] = [
-                "sessionId": AnyCodable(relaySessionId),
-                "reason": AnyCodable(reason),
-            ]
-            _ = try? await transport.request("talk.session.cancelOutput", payload, 8000)
-        }
-    }
-
    public func setInputPaused(_ paused: Bool) throws {
        guard self.isInputPaused != paused else { return }
        self.isInputPaused = paused
@@ -459,6 +473,14 @@ public final class RealtimeTalkRelaySession {
        }
    }

+    public func setOutputPaused(_ paused: Bool) {
+        guard self.isOutputPaused != paused else { return }
+        self.isOutputPaused = paused
+        if paused, self.isOutputPlaying {
+            self.cancelOutput(reason: "pause")
+        }
+    }
+
    private func createRelaySession() async throws -> TalkSessionCreateResult {
        var payload: [String: AnyCodable] = [
            "sessionKey": AnyCodable(self.options.sessionKey),
@@ -504,7 +526,9 @@ public final class RealtimeTalkRelaySession {
            await self?.handleEventStreamEnded(lifecycleGeneration: lifecycleGeneration)
        }
    }
+}

+extension RealtimeTalkRelaySession {
    private func handleEventStreamEnded(lifecycleGeneration: UInt64) async {
        guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return }
        self.logger.debug("talk realtime: event stream ended")
@@ -550,25 +574,11 @@ public final class RealtimeTalkRelaySession {
            self.finishStartupWait(.ready)
            self.onStatus("Listening (Realtime)")
        case "audio":
-            guard let base64 = payload["audioBase64"]?.stringValue,
-                  let data = Data(base64Encoded: base64)
-            else { return }
-            self.recordOutputAudioChunk(byteCount: data.count)
-            self.markOutputAudioStarted(byteCount: data.count, nowMs: ProcessInfo.processInfo.systemUptime * 1000)
-            self.onSpeakingChanged(true)
-            if self.outputContinuation == nil, self.outputTask != nil {
-                self.pendingOutputChunks.append(data)
-                return
-            }
-            self.ensureOutputPlaybackStarted()
-            self.outputEnvelope?.append(data)
-            self.outputContinuation?.yield(data)
+            self.handleOutputAudio(payload)
        case "audioDone":
            self.finishOutputPlaybackStream()
        case "clear":
-            let marks = self.takePendingPlaybackMarks()
-            self.stopOutputPlayback()
-            self.acknowledgePlaybackMarks(marks)
+            self.handleOutputClear(payload)
        case "mark":
            self.handlePlaybackMark(payload)
        case "transcript":
@@ -613,6 +623,38 @@ public final class RealtimeTalkRelaySession {
        }
    }

+    private func handleOutputClear(_ payload: [String: AnyCodable]) {
+        let clearIdentity = OutputIdentity(payload)
+        if self.awaitingOutputClear,
+           let suppressed = self.suppressedOutputIdentity
+        {
+            let clearsSuppressed =
+                clearIdentity.isEmpty()
+                ? suppressed.isEmpty()
+                : suppressed.isEmpty() || suppressed.relation(to: clearIdentity) == .same
+            if clearsSuppressed {
+                if let generation = clearIdentity.outputGeneration ?? suppressed.outputGeneration {
+                    self.cancelledOutputGenerationWatermark = max(
+                        self.cancelledOutputGenerationWatermark ?? 0,
+                        generation)
+                }
+                self.retireOutputCancellation()
+            }
+        }
+        let currentMatches =
+            clearIdentity.isEmpty()
+            ? self.outputIdentity == nil
+            : self.outputIdentity?.relation(to: clearIdentity) == .same
+        guard currentMatches else { return }
+        let marks = self.takePendingPlaybackMarks()
+        // Cancellation already published the stopped state. A later clear with no
+        // active output only retires the fence; it must not emit a duplicate callback.
+        if self.isOutputPlaying || self.outputIdentity != nil {
+            self.stopOutputPlayback()
+        }
+        self.acknowledgePlaybackMarks(marks)
+    }
+
    private func waitForStartupResult(
        timeoutSeconds: Int,
        lifecycleGeneration: UInt64) async -> StartupWaitResult
@@ -975,7 +1017,7 @@ public final class RealtimeTalkRelaySession {
    }

    private func startPendingOutputPlaybackIfNeeded() {
-        guard !self.pendingOutputChunks.isEmpty else {
+        guard !self.isOutputPaused, !self.pendingOutputChunks.isEmpty else {
            self.pendingOutputDone = false
            return
        }
@@ -1022,6 +1064,7 @@ public final class RealtimeTalkRelaySession {
            self.outputIdleTask = nil
        }
        self.isOutputPlaying = false
+        self.outputIdentity = nil
        self.outputStartedAtMs = nil
        self.outputPlaybackExpectedEndMs = 0
        self.outputEnvelope?.cancel()
@@ -1079,6 +1122,7 @@ public final class RealtimeTalkRelaySession {
        self.pendingOutputDone = false
        _ = self.pcmPlayer.stop()
        self.isOutputPlaying = false
+        self.outputIdentity = nil
        self.outputStartedAtMs = nil
        self.outputPlaybackExpectedEndMs = 0
        self.outputEnvelope?.cancel()
@@ -1126,11 +1170,143 @@ public final class RealtimeTalkRelaySession {
    }
 }

+extension RealtimeTalkRelaySession {
+    private struct OutputIdentity {
+        enum Relation {
+            case same
+            case different
+            case unknown
+        }
+
+        let turnId: String?
+        let outputGeneration: Int?
+
+        init(_ payload: [String: AnyCodable]) {
+            let turnId = payload["talkEvent"]?.dictionaryValue?["turnId"]?.stringValue?
+                .trimmingCharacters(in: .whitespacesAndNewlines)
+            self.turnId = turnId?.isEmpty == false ? turnId : nil
+            self.outputGeneration = RealtimeTalkRelaySession.outputGeneration(payload["outputGeneration"])
+        }
+
+        func isEmpty() -> Bool {
+            self.turnId == nil && self.outputGeneration == nil
+        }
+
+        func relation(to other: OutputIdentity) -> Relation {
+            if self.isEmpty() || other.isEmpty() {
+                return self.isEmpty() == other.isEmpty() ? .same : .different
+            }
+            if let outputGeneration, let otherGeneration = other.outputGeneration {
+                return outputGeneration == otherGeneration ? .same : .different
+            }
+            if let turnId, let otherTurnId = other.turnId {
+                return turnId == otherTurnId ? .same : .different
+            }
+            return .unknown
+        }
+    }
+
+    @discardableResult
+    public func cancelOutputIfPlaying(reason: String = "user") -> Bool {
+        guard self.isOutputPlaying else { return false }
+        self.cancelOutput(reason: reason)
+        return true
+    }
+
+    public func cancelOutput(reason: String = "user") {
+        guard let relaySessionId else { return }
+        let outputIdentity = self.outputIdentity ?? OutputIdentity([:])
+        self.outputCancellationGeneration &+= 1
+        let cancellationGeneration = self.outputCancellationGeneration
+        self.outputCancellationTask?.cancel()
+        self.suppressedOutputIdentity = outputIdentity
+        self.awaitingOutputClear = true
+        self.stopOutputPlayback()
+        self.outputCancellationTask = Task { [weak self, transport] in
+            var payload: [String: AnyCodable] = [
+                "sessionId": AnyCodable(relaySessionId),
+                "reason": AnyCodable(reason),
+            ]
+            if let turnId = outputIdentity.turnId {
+                payload["turnId"] = AnyCodable(turnId)
+            }
+            if await transport.supportsOutputGeneration(),
+               let outputGeneration = outputIdentity.outputGeneration
+            {
+                payload["outputGeneration"] = AnyCodable(outputGeneration)
+            }
+            do {
+                _ = try await transport.request("talk.session.cancelOutput", payload, 8000)
+            } catch {
+                guard let self, self.isCurrentOutputCancellation(cancellationGeneration) else { return }
+                let issue = RealtimeTalkRelayIssue(
+                    code: "realtime_output_cancel_failed",
+                    message: error.localizedDescription,
+                    provider: self.options.provider,
+                    model: self.options.model,
+                    transport: "gateway-relay",
+                    phase: "output-cancel")
+                self.onIssue(issue)
+                self.onStatus(issue.message)
+                // A failed current cancellation leaves remote output ownership unknown.
+                // Keep the fence until terminal teardown makes late audio impossible.
+                self.close(sendClose: true)
+                self.onTermination(.outputCancellationFailed)
+            }
+        }
+    }
+
+    private func isCurrentOutputCancellation(_ generation: UInt64) -> Bool {
+        generation == self.outputCancellationGeneration && !self.isClosed
+    }
+
+    private func handleOutputAudio(_ payload: [String: AnyCodable]) {
+        guard !self.isOutputPaused else { return }
+        guard let base64 = payload["audioBase64"]?.stringValue,
+              let data = Data(base64Encoded: base64)
+        else { return }
+        let incomingIdentity = OutputIdentity(payload)
+        guard !self.awaitingOutputClear else { return }
+        if let watermark = self.cancelledOutputGenerationWatermark {
+            guard let generation = incomingIdentity.outputGeneration, generation > watermark else { return }
+        }
+        if !incomingIdentity.isEmpty() {
+            self.outputIdentity = incomingIdentity
+        }
+        self.recordOutputAudioChunk(byteCount: data.count)
+        self.markOutputAudioStarted(byteCount: data.count, nowMs: ProcessInfo.processInfo.systemUptime * 1000)
+        self.onSpeakingChanged(true)
+        if self.outputContinuation == nil, self.outputTask != nil {
+            self.pendingOutputChunks.append(data)
+            return
+        }
+        self.ensureOutputPlaybackStarted()
+        self.outputEnvelope?.append(data)
+        self.outputContinuation?.yield(data)
+    }
+
+    private func retireOutputCancellation() {
+        self.outputCancellationGeneration &+= 1
+        self.outputCancellationTask?.cancel()
+        self.outputCancellationTask = nil
+        self.suppressedOutputIdentity = nil
+        self.awaitingOutputClear = false
+    }
+
+    private nonisolated static func outputGeneration(_ value: AnyCodable?) -> Int? {
+        guard let raw = value?.doubleValue,
+              raw > 0,
+              raw <= Double(Int.max),
+              raw.rounded(.towardZero) == raw
+        else { return nil }
+        return Int(raw)
+    }
+}
+
 extension RealtimeTalkRelaySession {
    private func startMicrophonePump(lifecycleGeneration: UInt64) throws {
        self.stopMicrophonePump()
        guard !self.isInputPaused else { return }
-        self.audioCaptureGeneration &+= 1
        let audioCaptureGeneration = self.audioCaptureGeneration
        try self.audioCapture.start(
            targetSampleRate: self.inputSampleRateHz,
@@ -1206,9 +1382,15 @@ extension RealtimeTalkRelaySession {
        let task = Task { @MainActor [weak self, audioSender] in
            guard let self else { return }
            defer { self.audioSendTasks.removeValue(forKey: taskID) }
-            guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return }
+            guard self.isCurrentLifecycleLocally(lifecycleGeneration),
+                  self.audioCaptureGeneration == audioCaptureGeneration,
+                  !self.isInputPaused
+            else { return }
            guard let message = await audioSender.send(encoded, timestampMs: timestampMs) else { return }
-            guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return }
+            guard self.isCurrentLifecycleLocally(lifecycleGeneration),
+                  self.audioCaptureGeneration == audioCaptureGeneration,
+                  !self.isInputPaused
+            else { return }
            self.onStatus("Realtime audio failed: \(message)")
        }
        self.audioSendTasks[taskID] = task
@@ -1249,6 +1431,10 @@ extension RealtimeTalkRelaySession {

    private func stopMicrophonePump() {
        self.audioCaptureGeneration &+= 1
+        for task in self.audioSendTasks.values {
+            task.cancel()
+        }
+        self.audioSendTasks.removeAll()
        self.audioCapture.stop()
    }
 }
@@ -14,6 +14,20 @@ private final class UnusedPCMStreamingAudioPlayer: PCMStreamingAudioPlaying {
    }
 }

+@MainActor
+private final class DrainingPCMStreamingAudioPlayer: PCMStreamingAudioPlaying {
+    func play(stream: AsyncThrowingStream<Data, Error>, sampleRate _: Double) async -> StreamingPlaybackResult {
+        do {
+            for try await _ in stream {}
+        } catch {}
+        return StreamingPlaybackResult(finished: true, interruptedAt: nil)
+    }
+
+    func stop() -> Double? {
+        nil
+    }
+}
+
@MainActor
 private final class TestRealtimeTalkAudioCapture: RealtimeTalkAudioCapturing {
    var suppressesInputDuringOutput = false
@@ -97,14 +111,81 @@ private actor RealtimeRelayRouteFlag {
    }
 }

+private actor RealtimeRelayEventSource {
+    private var continuation: AsyncStream<EventFrame>.Continuation?
+
+    func stream() -> AsyncStream<EventFrame> {
+        AsyncStream { self.continuation = $0 }
+    }
+
+    func finish() {
+        self.continuation?.finish()
+    }
+}
+
 private func unusedRealtimeRelayTransport() -> RealtimeTalkRelayTransport {
    RealtimeTalkRelayTransport(
        subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
        request: { _, _, _ in throw CancellationError() })
 }

+private func outputAudioEvent(generation: Int) -> EventFrame {
+    EventFrame(
+        type: "event",
+        event: "talk.event",
+        payload: AnyCodable([
+            "relaySessionId": "relay-1",
+            "type": "audio",
+            "audioBase64": Data([0x01]).base64EncodedString(),
+            "outputGeneration": generation,
+        ]),
+        seq: nil,
+        stateversion: nil)
+}
+
@MainActor
 struct RealtimeTalkRelaySessionTests {
+    enum CancellationRetirement {
+        case clear
+        case close
+    }
+
+    private func makeIdleCancellationSession(
+        _ onSpeakingChanged: @escaping (Bool) -> Void) -> RealtimeTalkRelaySession
+    {
+        let transport = RealtimeTalkRelayTransport(
+            subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+            request: { _, _, _ in Data("{\"ok\":true}".utf8) })
+        let session = RealtimeTalkRelaySession(
+            transport: transport,
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onSpeakingChanged: onSpeakingChanged)
+        session._test_setRelaySessionId("relay-1")
+        return session
+    }
+
+    private func makeAudioSendSession() -> (RealtimeTalkRelaySession, RealtimeRelayStartupRequestLog) {
+        let requests = RealtimeRelayStartupRequestLog()
+        let transport = RealtimeTalkRelayTransport(
+            subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+            request: { method, params, _ in
+                await requests.record(method: method, params: params)
+                return Data("{\"ok\":true}".utf8)
+            })
+        let session = RealtimeTalkRelaySession(
+            transport: transport,
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: UnusedPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onSpeakingChanged: { _ in })
+        session._test_prepareAudioSender(relaySessionId: "relay-1")
+        return (session, requests)
+    }
+
    @Test func `transcript callback carries typed partial and final values`() async {
        var transcripts: [RealtimeTalkTranscript] = []
        let session = RealtimeTalkRelaySession(
@@ -226,6 +307,356 @@ struct RealtimeTalkRelaySessionTests {
        #expect(request.params?["sessionId"]?.stringValue == "relay-1")
        #expect(request.params?["markName"]?.stringValue == "audio-1")
    }
+}
+
+extension RealtimeTalkRelaySessionTests {
+    @Test func `output cancellation fences delayed audio and preserves exact identity`() async throws {
+        let requests = RealtimeRelayStartupRequestLog()
+        var speakingStates: [Bool] = []
+        let transport = RealtimeTalkRelayTransport(
+            subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+            request: { method, params, _ in
+                await requests.record(method: method, params: params)
+                return Data("{\"ok\":true}".utf8)
+            },
+            supportsOutputGeneration: { true })
+        let session = RealtimeTalkRelaySession(
+            transport: transport,
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onSpeakingChanged: { speakingStates.append($0) })
+        session._test_setRelaySessionId("relay-1")
+        let audio: (Int, String) -> EventFrame = { generation, turnId in
+            EventFrame(
+                type: "event",
+                event: "talk.event",
+                payload: AnyCodable([
+                    "relaySessionId": "relay-1",
+                    "type": "audio",
+                    "audioBase64": Data([0x01]).base64EncodedString(),
+                    "outputGeneration": generation,
+                    "talkEvent": ["turnId": turnId],
+                ]),
+                seq: nil,
+                stateversion: nil)
+        }
+        let clear: (Int) -> EventFrame = { generation in
+            EventFrame(
+                type: "event",
+                event: "talk.event",
+                payload: AnyCodable([
+                    "relaySessionId": "relay-1",
+                    "type": "clear",
+                    "outputGeneration": generation,
+                ]),
+                seq: nil,
+                stateversion: nil)
+        }
+
+        await session._test_handleGatewayEvent(audio(7, "turn-7"))
+        #expect(session.cancelOutputIfPlaying(reason: "barge-in"))
+        for _ in 0..<10 {
+            if await !requests.snapshot().isEmpty { break }
+            await Task.yield()
+        }
+        let request = try #require(await requests.snapshot().first)
+        #expect(request.method == "talk.session.cancelOutput")
+        #expect(request.params?["sessionId"]?.stringValue == "relay-1")
+        #expect(request.params?["turnId"]?.stringValue == "turn-7")
+        #expect(request.params?["outputGeneration"]?.doubleValue == 7)
+        #expect(request.params?["reason"]?.stringValue == "barge-in")
+
+        await session._test_handleGatewayEvent(audio(7, "turn-7"))
+        await session._test_handleGatewayEvent(audio(8, "turn-8"))
+        #expect(speakingStates.first == true)
+        #expect(!speakingStates.dropFirst().contains(true))
+        await session._test_handleGatewayEvent(clear(7))
+        await session._test_handleGatewayEvent(audio(7, "turn-7"))
+        #expect(speakingStates == [true, false])
+        await session._test_handleGatewayEvent(audio(8, "turn-8"))
+        #expect(speakingStates == [true, false, true])
+        await session._test_handleGatewayEvent(clear(7))
+        #expect(speakingStates == [true, false, true])
+        await session._test_handleGatewayEvent(clear(8))
+        #expect(speakingStates == [true, false, true, false])
+    }
+
+    @Test func `legacy cancellation keeps turn identity but omits generation`() async throws {
+        let requests = RealtimeRelayStartupRequestLog()
+        let transport = RealtimeTalkRelayTransport(
+            subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+            request: { method, params, _ in
+                await requests.record(method: method, params: params)
+                return Data("{\"ok\":true}".utf8)
+            })
+        let session = RealtimeTalkRelaySession(
+            transport: transport,
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onSpeakingChanged: { _ in })
+        session._test_setRelaySessionId("relay-1")
+        await session._test_handleGatewayEvent(EventFrame(
+            type: "event",
+            event: "talk.event",
+            payload: AnyCodable([
+                "relaySessionId": "relay-1",
+                "type": "audio",
+                "audioBase64": Data([0x01]).base64EncodedString(),
+                "outputGeneration": 7,
+                "talkEvent": ["turnId": "turn-7"],
+            ]),
+            seq: nil,
+            stateversion: nil))
+
+        #expect(session.cancelOutputIfPlaying(reason: "barge-in"))
+        for _ in 0..<10 {
+            if await !requests.snapshot().isEmpty { break }
+            await Task.yield()
+        }
+        let request = try #require(await requests.snapshot().first)
+        #expect(request.params?["turnId"]?.stringValue == "turn-7")
+        #expect(request.params?["outputGeneration"] == nil)
+    }
+
+    @Test func `idle cancellation and pause retain the relay without false interruption`() async {
+        let requests = RealtimeRelayStartupRequestLog()
+        var speakingStates: [Bool] = []
+        let transport = RealtimeTalkRelayTransport(
+            subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+            request: { method, params, _ in
+                await requests.record(method: method, params: params)
+                return Data("{\"ok\":true}".utf8)
+            },
+            supportsOutputGeneration: { true })
+        let session = RealtimeTalkRelaySession(
+            transport: transport,
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onSpeakingChanged: { speakingStates.append($0) })
+        session._test_setRelaySessionId("relay-1")
+
+        #expect(!session.cancelOutputIfPlaying())
+        session.setOutputPaused(true)
+        await session._test_handleGatewayEvent(outputAudioEvent(generation: 1))
+        #expect(speakingStates.isEmpty)
+        #expect(await requests.snapshot().isEmpty)
+        session.setOutputPaused(false)
+        await session._test_handleGatewayEvent(outputAudioEvent(generation: 2))
+        #expect(speakingStates == [true])
+        #expect(await requests.snapshot().isEmpty)
+    }
+
+    @Test func `idle cancellation waits for clear while cancellation without relay stays unfenced`() async {
+        var speakingStates: [Bool] = []
+        let session = self.makeIdleCancellationSession { speakingStates.append($0) }
+        session.cancelOutput(reason: "barge-in")
+        await session._test_handleGatewayEvent(outputAudioEvent(generation: 1))
+        #expect(speakingStates == [false])
+        await session._test_handleGatewayEvent(EventFrame(
+            type: "event",
+            event: "talk.event",
+            payload: AnyCodable([
+                "relaySessionId": "relay-1",
+                "type": "clear",
+            ]),
+            seq: nil,
+            stateversion: nil))
+        await session._test_handleGatewayEvent(outputAudioEvent(generation: 1))
+        #expect(speakingStates == [false, true])
+
+        var unfencedStates: [Bool] = []
+        let unfenced = RealtimeTalkRelaySession(
+            transport: unusedRealtimeRelayTransport(),
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onSpeakingChanged: { unfencedStates.append($0) })
+        unfenced.cancelOutput()
+        unfenced._test_setRelaySessionId("relay-1")
+        await unfenced._test_handleGatewayEvent(outputAudioEvent(generation: 1))
+        #expect(unfencedStates == [true])
+    }
+
+    @Test func `active output pause cancels the exact generation`() async throws {
+        let requests = RealtimeRelayStartupRequestLog()
+        let session = RealtimeTalkRelaySession(
+            transport: RealtimeTalkRelayTransport(
+                subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+                request: { method, params, _ in
+                    await requests.record(method: method, params: params)
+                    return Data("{\"ok\":true}".utf8)
+                },
+                supportsOutputGeneration: { true }),
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onSpeakingChanged: { _ in })
+        session._test_setRelaySessionId("relay-1")
+        await session._test_handleGatewayEvent(EventFrame(
+            type: "event",
+            event: "talk.event",
+            payload: AnyCodable([
+                "relaySessionId": "relay-1",
+                "type": "audio",
+                "audioBase64": Data([0x01]).base64EncodedString(),
+                "outputGeneration": 7,
+                "talkEvent": ["turnId": "turn-7"],
+            ]),
+            seq: nil,
+            stateversion: nil))
+
+        session.setOutputPaused(true)
+        for _ in 0..<10 {
+            if await !requests.snapshot().isEmpty { break }
+            await Task.yield()
+        }
+        let request = try #require(await requests.snapshot().first)
+        #expect(request.params?["turnId"]?.stringValue == "turn-7")
+        #expect(request.params?["outputGeneration"]?.doubleValue == 7)
+        #expect(request.params?["reason"]?.stringValue == "pause")
+    }
+
+    @Test func `current cancellation failure terminates and rejects late audio`() async {
+        let requests = RealtimeRelayStartupRequestLog()
+        var issues: [RealtimeTalkRelayIssue] = []
+        var terminations: [RealtimeTalkRelayTermination] = []
+        var speakingStates: [Bool] = []
+        let transport = RealtimeTalkRelayTransport(
+            subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+            request: { method, params, _ in
+                await requests.record(method: method, params: params)
+                if method == "talk.session.cancelOutput" {
+                    throw URLError(.cannotConnectToHost)
+                }
+                return Data("{\"ok\":true}".utf8)
+            })
+        let session = RealtimeTalkRelaySession(
+            transport: transport,
+            options: .init(sessionKey: "main", provider: "openai", model: "gpt-realtime-2", voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onIssue: { issues.append($0) },
+            onTermination: { terminations.append($0) },
+            onSpeakingChanged: { speakingStates.append($0) })
+        session._test_setRelaySessionId("relay-1")
+        await session._test_handleGatewayEvent(outputAudioEvent(generation: 1))
+
+        session.cancelOutput()
+        for _ in 0..<50 {
+            if !issues.isEmpty { break }
+            await Task.yield()
+        }
+        await session._test_handleGatewayEvent(outputAudioEvent(generation: 2))
+        for _ in 0..<50 {
+            if await requests.snapshot().contains(where: { $0.method == "talk.session.close" }) { break }
+            await Task.yield()
+        }
+
+        #expect(issues.map(\.code) == ["realtime_output_cancel_failed"])
+        #expect(issues.map(\.phase) == ["output-cancel"])
+        #expect(terminations == [.outputCancellationFailed])
+        #expect(await requests.snapshot().map(\.method) == [
+            "talk.session.cancelOutput",
+            "talk.session.close",
+        ])
+        #expect(speakingStates.first == true)
+        #expect(!speakingStates.dropFirst().contains(true))
+    }
+
+    @Test func `superseded cancellation failure leaves the active fence intact`() async {
+        let barrier = RealtimeRelayStartupBarrier()
+        let requests = RealtimeRelayStartupRequestLog()
+        var issues: [RealtimeTalkRelayIssue] = []
+        var speakingStates: [Bool] = []
+        let transport = RealtimeTalkRelayTransport(
+            subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+            request: { method, params, _ in
+                await requests.record(method: method, params: params)
+                if await requests.snapshot().count == 1 {
+                    await barrier.suspend()
+                    throw URLError(.cancelled)
+                }
+                return Data("{\"ok\":true}".utf8)
+            })
+        let session = RealtimeTalkRelaySession(
+            transport: transport,
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onIssue: { issues.append($0) },
+            onSpeakingChanged: { speakingStates.append($0) })
+        session._test_setRelaySessionId("relay-1")
+
+        session.cancelOutput()
+        await barrier.waitUntilEntered()
+        session.cancelOutput()
+        await barrier.release()
+        for _ in 0..<10 {
+            if await requests.snapshot().count == 2 { break }
+            await Task.yield()
+        }
+        await session._test_handleGatewayEvent(outputAudioEvent(generation: 1))
+
+        #expect(issues.isEmpty)
+        #expect(await requests.snapshot().count == 2)
+        #expect(!speakingStates.contains(true))
+    }
+
+    @Test(arguments: [CancellationRetirement.clear, .close])
+    func `clear and close retire in flight cancellation failures`(
+        retirement: CancellationRetirement) async
+    {
+        let barrier = RealtimeRelayStartupBarrier()
+        var issues: [RealtimeTalkRelayIssue] = []
+        let session = RealtimeTalkRelaySession(
+            transport: RealtimeTalkRelayTransport(
+                subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
+                request: { method, _, _ in
+                    if method == "talk.session.cancelOutput" {
+                        await barrier.suspend()
+                        throw URLError(.cancelled)
+                    }
+                    return Data("{\"ok\":true}".utf8)
+                }),
+            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
+            audioCapture: TestRealtimeTalkAudioCapture(),
+            pcmPlayer: DrainingPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onIssue: { issues.append($0) },
+            onSpeakingChanged: { _ in })
+        session._test_setRelaySessionId("relay-1")
+
+        session.cancelOutput()
+        await barrier.waitUntilEntered()
+        switch retirement {
+        case .clear:
+            await session._test_handleGatewayEvent(EventFrame(
+                type: "event",
+                event: "talk.event",
+                payload: AnyCodable([
+                    "relaySessionId": "relay-1",
+                    "type": "clear",
+                ]),
+                seq: nil,
+                stateversion: nil))
+        case .close:
+            session.stop()
+        }
+        await barrier.release()
+        await Task.yield()
+
+        #expect(issues.isEmpty)
+    }

    @Test func `close after classified error does not replace issue`() async {
        var issues: [RealtimeTalkRelayIssue] = []
@@ -376,6 +807,59 @@ struct RealtimeTalkRelaySessionTests {
        #expect(!audioCapture.isStarted)
    }

+    @Test func `event stream ending during relay creation closes the late relay`() async throws {
+        let barrier = RealtimeRelayStartupBarrier()
+        let events = RealtimeRelayEventSource()
+        let requests = RealtimeRelayStartupRequestLog()
+        let audioCapture = TestRealtimeTalkAudioCapture()
+        var issues: [RealtimeTalkRelayIssue] = []
+        let result = TalkSessionCreateResult(
+            sessionid: "talk-session",
+            mode: AnyCodable("realtime"),
+            transport: AnyCodable("gateway-relay"),
+            brain: AnyCodable("agent-consult"),
+            relaysessionid: "relay-1")
+        let resultData = try JSONEncoder().encode(result)
+        let session = RealtimeTalkRelaySession(
+            transport: RealtimeTalkRelayTransport(
+                subscribeServerEvents: { _ in await events.stream() },
+                request: { method, params, _ in
+                    await requests.record(method: method, params: params)
+                    if method == "talk.session.create" {
+                        await barrier.suspend()
+                        return resultData
+                    }
+                    return Data("{\"ok\":true}".utf8)
+                }),
+            options: .init(sessionKey: "main", provider: "openai", model: "gpt-realtime-2", voice: nil),
+            audioCapture: audioCapture,
+            pcmPlayer: UnusedPCMStreamingAudioPlayer(),
+            onStatus: { _ in },
+            onIssue: { issues.append($0) },
+            onSpeakingChanged: { _ in })
+        let start = Task { @MainActor in
+            do {
+                try await session.start()
+                return nil as String?
+            } catch {
+                return error.localizedDescription
+            }
+        }
+        await barrier.waitUntilEntered()
+
+        await events.finish()
+        while issues.isEmpty {
+            await Task.yield()
+        }
+        await barrier.release()
+
+        #expect(await start.value == "Realtime connection ended before it became ready.")
+        #expect(audioCapture.startCount == 0)
+        let recorded = await requests.snapshot()
+        #expect(recorded.map(\.method) == ["talk.session.create", "talk.session.close"])
+        #expect(recorded.last?.params?["sessionId"]?.stringValue == "relay-1")
+    }
+
    @Test func `microphone failure terminates relay and reports typed issue`() async throws {
        let requests = RealtimeRelayStartupRequestLog()
        let transport = RealtimeTalkRelayTransport(
@@ -636,28 +1120,23 @@ struct RealtimeTalkRelaySessionTests {
        #expect(statuses == ["Thinking…"])
    }

-    @Test func `stop cancels buffered microphone audio before dispatch`() async throws {
-        let requests = RealtimeRelayStartupRequestLog()
-        let transport = RealtimeTalkRelayTransport(
-            subscribeServerEvents: { _ in AsyncStream { $0.finish() } },
-            request: { method, params, _ in
-                await requests.record(method: method, params: params)
-                return Data("{\"ok\":true}".utf8)
-            })
-        let session = RealtimeTalkRelaySession(
-            transport: transport,
-            options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil),
-            audioCapture: TestRealtimeTalkAudioCapture(),
-            pcmPlayer: UnusedPCMStreamingAudioPlayer(),
-            onStatus: { _ in },
-            onSpeakingChanged: { _ in })
-        session._test_prepareAudioSender(relaySessionId: "relay-1")
-        let send = try #require(session._test_enqueueMicrophoneFrame(Data([0x01, 0x02])))
+    @Test func `stop and pause discard buffered microphone audio before dispatch`() async throws {
+        let (stoppedSession, stoppedRequests) = self.makeAudioSendSession()
+        let stoppedSend = try #require(stoppedSession._test_enqueueMicrophoneFrame(Data([0x01])))

-        session.stop()
-        await send.value
+        stoppedSession.stop()
+        await stoppedSend.value
+        #expect(await stoppedRequests.snapshot().isEmpty)

-        #expect(await requests.snapshot().isEmpty)
+        let (session, requests) = self.makeAudioSendSession()
+        _ = try #require(session._test_enqueueMicrophoneFrame(Data([0x01])))
+        try session.setInputPaused(true)
+        try session.setInputPaused(false)
+        try await (#require(session._test_enqueueMicrophoneFrame(Data([0x02])))).value
+
+        #expect(await requests.snapshot().compactMap { $0.params?["audioBase64"]?.stringValue } == [
+            Data([0x02]).base64EncodedString(),
+        ])
    }

    @Test func `gateway route lost during startup fails instead of reporting ready`() async throws {