From 115b10261df5ded56e7e484bda8def366302666b Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 21 Aug 2026 00:36:21 -0700 Subject: [PATCH] fix(talk): require accepted cancellation results --- .../openclaw/app/voice/TalkModeManagerTest.kt | 2 +- .../RealtimeTalkRelaySessionTests.swift | 78 ++++++++++++++++--- .../src/channels.schema.test.ts | 1 + .../gateway-protocol/src/schema/channels.ts | 2 +- scripts/protocol-gen-kotlin.ts | 4 +- 5 files changed, 74 insertions(+), 13 deletions(-) diff --git a/apps/android/app/src/test/java/ai/openclaw/app/voice/TalkModeManagerTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/voice/TalkModeManagerTest.kt index 3cf1e1ad32b5..ded216e79a15 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/voice/TalkModeManagerTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/voice/TalkModeManagerTest.kt @@ -860,7 +860,7 @@ class TalkModeManagerTest { @Test fun malformedOutputCancellationResultFailsClosed() { - for (response in listOf("""{"status":"stale"}""", """{"ok":true,"status":"unknown"}""", """{"ok":true,"extra":1}""")) { + for (response in listOf("""{"status":"stale"}""", """{"ok":false}""", """{"ok":true,"status":"unknown"}""", """{"ok":true,"extra":1}""")) { assertTrue(runCatching { shouldRetireRealtimeOutputCancellation(response) }.isFailure) } } diff --git a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift index b43a8ff72431..63897fefced5 100644 --- a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift +++ b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift @@ -549,6 +549,19 @@ private func outputAudioEvent( stateversion: nil) } +private func outputClearEvent(turnId: String) -> EventFrame { + EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "clear", + "talkEvent": ["turnId": turnId], + ]), + seq: nil, + stateversion: nil) +} + @MainActor struct RealtimeTalkRelaySessionTests { private func makeIdleCancellationSession( @@ -1257,16 +1270,7 @@ extension RealtimeTalkRelaySessionTests { #expect(session.cancelOutput()) try await barrier.waitUntilEntered() - await session._test_handleGatewayEvent(EventFrame( - type: "event", - event: "talk.event", - payload: AnyCodable([ - "relaySessionId": "relay-1", - "type": "clear", - "talkEvent": ["turnId": "turn-1"], - ]), - seq: nil, - stateversion: nil)) + await session._test_handleGatewayEvent(outputClearEvent(turnId: "turn-1")) await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-2")) #expect(session.cancelOutput()) await barrier.release() @@ -1324,6 +1328,60 @@ extension RealtimeTalkRelaySessionTests { ]) } + @Test(arguments: [ + #"{"ok":true}"#, + #"{"ok":true,"status":"applied","turnId":"turn-1"}"#, + ]) + func `accepted cancellation response keeps fence until matching clear`( + response: String) async throws + { + let barrier = RealtimeRelayStartupBarrier() + var speakingStates: [Bool] = [] + let session = RealtimeTalkRelaySession( + transport: RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, _, _ in + if method == "talk.session.cancelOutput" { + await barrier.suspend() + return Data(response.utf8) + } + return Data(#"{"ok":true}"#.utf8) + }), + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: { speakingStates.append($0) }) + defer { session.stop() } + session._test_setRelaySessionId("relay-1") + session._test_prepareAudioSender(relaySessionId: "relay-1") + await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-1")) + #expect(session.cancelOutput()) + var cancellationTask: Task? + do { + try await barrier.waitUntilEntered() + guard let exactCancellationTask = session._test_outputCancellationTask() else { + throw RealtimeRelayTestTimeout(operation: "cancellation task registration") + } + cancellationTask = exactCancellationTask + await barrier.release() + await exactCancellationTask.value + #expect(session._test_enqueueMicrophoneFrame(Data([0x01])) == nil) + await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-2")) + #expect(speakingStates == [true, false]) + + await session._test_handleGatewayEvent(outputClearEvent(turnId: "turn-1")) + let admittedTask = try #require(session._test_enqueueMicrophoneFrame(Data([0x02]))) + await admittedTask.value + await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-2")) + } catch { + await barrier.release() + await cancellationTask?.value + throw error + } + #expect(speakingStates == [true, false, true]) + } + @Test func `close retires in flight cancellation failure`() async throws { let barrier = RealtimeRelayStartupBarrier() var issues: [RealtimeTalkRelayIssue] = [] diff --git a/packages/gateway-protocol/src/channels.schema.test.ts b/packages/gateway-protocol/src/channels.schema.test.ts index 67aae8e1f758..0ca0f5cb9ca0 100644 --- a/packages/gateway-protocol/src/channels.schema.test.ts +++ b/packages/gateway-protocol/src/channels.schema.test.ts @@ -52,6 +52,7 @@ describe("TalkSessionCancelOutputResultSchema", () => { for (const value of [ {}, { status: "applied" }, + { ok: false }, { ok: true, status: "unknown" }, { ok: true, turnId: "" }, { ok: true, extra: true }, diff --git a/packages/gateway-protocol/src/schema/channels.ts b/packages/gateway-protocol/src/schema/channels.ts index a2dfd607c709..e0b2976e7008 100644 --- a/packages/gateway-protocol/src/schema/channels.ts +++ b/packages/gateway-protocol/src/schema/channels.ts @@ -319,7 +319,7 @@ export const TalkSessionCancelOutputParamsSchema = closedObject({ /** Reports whether a Talk output cancellation applied to the requested turn. */ export const TalkSessionCancelOutputResultSchema = closedObject({ - ok: Type.Boolean(), + ok: Type.Literal(true), status: Type.Optional( Type.Union([Type.Literal("applied"), Type.Literal("stale"), Type.Literal("idle")]), ), diff --git a/scripts/protocol-gen-kotlin.ts b/scripts/protocol-gen-kotlin.ts index 6e3e6dfc3799..ec62612fbb54 100644 --- a/scripts/protocol-gen-kotlin.ts +++ b/scripts/protocol-gen-kotlin.ts @@ -336,11 +336,13 @@ function emitWireModels(): string[] { const type = kotlinType(propertySchema, `${name}${upperCamel(wireName)}`); const literal = literalValue(propertySchema); const optional = !required.has(wireName); + const useLiteralDefault = + literal !== undefined && (optional || typeof literal !== "boolean"); return { annotation: propertyName === wireName ? [] : [` @SerialName(${JSON.stringify(wireName)})`], declaration: ` val ${propertyName}: ${type}${optional ? "?" : ""}${ - literal !== undefined ? ` = ${kotlinLiteral(literal)}` : optional ? " = null" : "" + useLiteralDefault ? ` = ${kotlinLiteral(literal)}` : optional ? " = null" : "" },`, }; });