From e1fd35b1ffdce82c3ffd518091f13bb65915d996 Mon Sep 17 00:00:00 2001 From: Rocky <72559939+hkgood@users.noreply.github.com> Date: Wed, 19 Aug 2026 19:16:05 +0800 Subject: [PATCH] Fix managed ASR settlement and empty polish recovery Accept the final sequence format returned by Volcengine without weakening ordering checks, and retry one safe buffered DeepSeek empty response under the same credit reservation. --- .../providers/deepseek/DeepSeekProvider.kt | 120 ++++++++++++++++-- .../providers/volcengine/SaucV3Protocol.kt | 6 +- .../providers/deepseek/DeepSeekClientTest.kt | 90 +++++++++++++ .../volcengine/SaucV3ProtocolTest.kt | 27 +++- .../services/GatewayServiceBillingTest.kt | 44 +++++++ 5 files changed, 273 insertions(+), 14 deletions(-) diff --git a/src/main/kotlin/com/osglab/account/features/gateway/providers/deepseek/DeepSeekProvider.kt b/src/main/kotlin/com/osglab/account/features/gateway/providers/deepseek/DeepSeekProvider.kt index 1b0073e..e8bbf1c 100644 --- a/src/main/kotlin/com/osglab/account/features/gateway/providers/deepseek/DeepSeekProvider.kt +++ b/src/main/kotlin/com/osglab/account/features/gateway/providers/deepseek/DeepSeekProvider.kt @@ -34,6 +34,7 @@ import kotlinx.serialization.json.JsonNull import kotlinx.serialization.json.jsonArray import kotlinx.serialization.json.jsonObject import kotlinx.serialization.json.jsonPrimitive +import org.slf4j.LoggerFactory data class DeepSeekConfig( val endpoint: String, @@ -91,7 +92,25 @@ class DeepSeekProvider( ): ProviderUsage { require(request is TextProviderRequest) { "DeepSeek only accepts text requests" } validate(request) - return upstream.complete(request, output) + var attempt = 1 + while (true) { + try { + return upstream.complete(request, output) + } catch (failure: DeepSeekEmptyResultException) { + LOG.warn( + "DeepSeek returned empty content requestId={} capability={} attempt={} " + + "finishReason={} reasoningContentPresent={} usagePresent={}", + request.requestId, + request.capability.name, + attempt, + failure.finishReason, + failure.reasoningContentPresent, + failure.usagePresent, + ) + if (request.stream || attempt >= MAX_BUFFERED_ATTEMPTS) throw failure + attempt += 1 + } + } } private fun validate(request: TextProviderRequest) { @@ -110,6 +129,11 @@ class DeepSeekProvider( "agent requests must be non-streaming" } } + + private companion object { + const val MAX_BUFFERED_ATTEMPTS = 2 + val LOG = LoggerFactory.getLogger(DeepSeekProvider::class.java) + } } /** @@ -182,8 +206,14 @@ class KtorDeepSeekClient( ): ProviderUsage { val payload = bytes.decodeToString() val content = extractAssistantContent(payload) - ?: throw DeepSeekEmptyResultException() - if (content.isBlank()) throw DeepSeekEmptyResultException() + if (content.isNullOrBlank()) { + val metadata = emptyResultMetadata(payload) + throw DeepSeekEmptyResultException( + finishReason = metadata.finishReason, + reasoningContentPresent = metadata.reasoningContentPresent, + usagePresent = metadata.usagePresent, + ) + } if (request.capability == GatewayCapability.AGENT) validateAgentContent(content) val usage = extractUsage(payload)?.toProviderUsageOrNull() ?: throw DeepSeekProviderException("DeepSeek response omitted token usage") @@ -203,6 +233,8 @@ class KtorDeepSeekClient( var contentBytes = 0L var terminalChoiceSeen = false var doneSeen = false + var finishReason = "missing" + var reasoningContentPresent = false val assistantContent = StringBuilder() while (true) { val line = channel.readLineStrict( @@ -237,6 +269,7 @@ class KtorDeepSeekClient( } val eventUsage = extractUsage(data) eventUsage?.let { usage = it } + if (hasStreamReasoningContent(data)) reasoningContentPresent = true val choices = runCatching { event["choices"]?.jsonArray ?: throw IllegalArgumentException("choices is missing") @@ -250,13 +283,14 @@ class KtorDeepSeekClient( if (choice == null && eventUsage == null) { throw DeepSeekProviderException("DeepSeek returned an empty non-usage event") } - val finishReason = choice?.get("finish_reason") - if (finishReason != null && finishReason !is JsonNull) { - if (!finishReason.jsonPrimitive.isString || - finishReason.jsonPrimitive.content.isBlank() + val finishReasonElement = choice?.get("finish_reason") + if (finishReasonElement != null && finishReasonElement !is JsonNull) { + if (!finishReasonElement.jsonPrimitive.isString || + finishReasonElement.jsonPrimitive.content.isBlank() ) { throw DeepSeekProviderException("DeepSeek returned an invalid finish reason") } + finishReason = normalizeFinishReason(finishReasonElement.jsonPrimitive.content) terminalChoiceSeen = true } extractStreamContent(data)?.let { chunk -> @@ -270,7 +304,13 @@ class KtorDeepSeekClient( output.emit(encoded) } if (!doneSeen) throw DeepSeekProviderException("DeepSeek stream closed before [DONE]") - if (assistantContent.isBlank()) throw DeepSeekEmptyResultException() + if (assistantContent.isBlank()) { + throw DeepSeekEmptyResultException( + finishReason = finishReason, + reasoningContentPresent = reasoningContentPresent, + usagePresent = usage != null, + ) + } if (request.capability == GatewayCapability.AGENT) { validateAgentContent(assistantContent.toString()) } @@ -354,6 +394,58 @@ class KtorDeepSeekClient( ?.content }.getOrNull() + private fun hasStreamReasoningContent(payload: String): Boolean = runCatching { + json.parseToJsonElement(payload) + .jsonObject["choices"] + ?.let { choices -> choices.jsonArray.firstOrNull() } + ?.jsonObject + ?.get("delta") + ?.jsonObject + ?.get("reasoning_content") + ?.jsonPrimitive + ?.takeIf { it.isString } + ?.content + ?.isNotBlank() == true + }.getOrDefault(false) + + private fun emptyResultMetadata(payload: String): DeepSeekEmptyResultMetadata { + val root = runCatching { json.parseToJsonElement(payload).jsonObject }.getOrNull() + ?: return DeepSeekEmptyResultMetadata() + val choice = runCatching { + root["choices"]?.jsonArray?.firstOrNull()?.jsonObject + }.getOrNull() + val message = runCatching { choice?.get("message")?.jsonObject }.getOrNull() + val rawFinishReason = runCatching { + choice?.get("finish_reason") + ?.takeUnless { it is JsonNull } + ?.jsonPrimitive + ?.takeIf { it.isString } + ?.content + }.getOrNull() + val reasoningContentPresent = runCatching { + message?.get("reasoning_content") + ?.takeUnless { it is JsonNull } + ?.jsonPrimitive + ?.takeIf { it.isString } + ?.content + ?.isNotBlank() == true + }.getOrDefault(false) + return DeepSeekEmptyResultMetadata( + finishReason = normalizeFinishReason(rawFinishReason), + reasoningContentPresent = reasoningContentPresent, + usagePresent = root["usage"] != null && root["usage"] !is JsonNull, + ) + } + + private fun normalizeFinishReason(value: String?): String = + when (value?.trim()?.lowercase()) { + "stop", "length", "content_filter", "tool_calls", "insufficient_system_resource" -> + value.trim().lowercase() + + null, "" -> "missing" + else -> "other" + } + private suspend fun ByteReadChannel.readBounded(): ByteArray { val bytes = readRemaining(GatewayLimits.MAX_UPSTREAM_RESPONSE_BYTES.toLong() + 1L) .readByteArray() @@ -439,10 +531,20 @@ class DeepSeekConfigurationException(message: String) : IllegalStateException(me class DeepSeekProviderException(message: String) : RuntimeException(message) class DeepSeekUsageException(message: String) : ProviderCompletionException(message) -class DeepSeekEmptyResultException : RuntimeException("DeepSeek returned an empty result") +class DeepSeekEmptyResultException( + val finishReason: String = "missing", + val reasoningContentPresent: Boolean = false, + val usagePresent: Boolean = false, +) : RuntimeException("DeepSeek returned an empty result") private data class DeepSeekUsage( val total: Long, val input: Long?, val output: Long?, ) + +private data class DeepSeekEmptyResultMetadata( + val finishReason: String = "missing", + val reasoningContentPresent: Boolean = false, + val usagePresent: Boolean = false, +) diff --git a/src/main/kotlin/com/osglab/account/features/gateway/providers/volcengine/SaucV3Protocol.kt b/src/main/kotlin/com/osglab/account/features/gateway/providers/volcengine/SaucV3Protocol.kt index 54aa3e9..75e2e2c 100644 --- a/src/main/kotlin/com/osglab/account/features/gateway/providers/volcengine/SaucV3Protocol.kt +++ b/src/main/kotlin/com/osglab/account/features/gateway/providers/volcengine/SaucV3Protocol.kt @@ -231,7 +231,11 @@ class SaucSequenceValidator { ?: throw SaucProtocolException("SAUC server frame omitted its sequence") val expected = Math.addExact(lastSequence, 1) if (frame.isLast) { - if (sequence >= 0 || Math.abs(sequence.toLong()) != expected.toLong()) { + // Volcengine's protocol table documents a negative final sequence, + // while its examples and production service return the same + // strictly increasing sequence as a positive value. Accept both + // representations without weakening continuity validation. + if (Math.abs(sequence.toLong()) != expected.toLong()) { throw SaucProtocolException("SAUC final sequence is invalid") } finalSeen = true diff --git a/src/test/kotlin/com/osglab/account/features/gateway/providers/deepseek/DeepSeekClientTest.kt b/src/test/kotlin/com/osglab/account/features/gateway/providers/deepseek/DeepSeekClientTest.kt index a098786..86e644e 100644 --- a/src/test/kotlin/com/osglab/account/features/gateway/providers/deepseek/DeepSeekClientTest.kt +++ b/src/test/kotlin/com/osglab/account/features/gateway/providers/deepseek/DeepSeekClientTest.kt @@ -49,6 +49,90 @@ class DeepSeekClientTest : StringSpec({ } } + "captures only safe metadata for an empty buffered result" { + val client = client( + """ + { + "choices":[{ + "message":{"content":"","reasoning_content":"internal reasoning"}, + "finish_reason":"length" + }], + "usage":{"prompt_tokens":10,"completion_tokens":32,"total_tokens":42} + } + """.trimIndent(), + ) + var emitted = false + try { + val failure = shouldThrow { + KtorDeepSeekClient(client, CONFIG).complete( + request(), + ProviderOutput { emitted = true }, + ) + } + + failure.finishReason shouldBe "length" + failure.reasoningContentPresent shouldBe true + failure.usagePresent shouldBe true + emitted shouldBe false + } finally { + client.close() + } + } + + "retries one buffered empty result and returns the successful retry" { + var attempts = 0 + val provider = DeepSeekProvider( + DeepSeekClient { _, _ -> + attempts += 1 + if (attempts == 1) { + throw DeepSeekEmptyResultException( + finishReason = "length", + reasoningContentPresent = true, + usagePresent = true, + ) + } + SUCCESS_USAGE + }, + ) + + val usage = provider.execute(request(), DISCARD_OUTPUT) + + attempts shouldBe 2 + usage shouldBe SUCCESS_USAGE + } + + "stops after one buffered empty-result retry" { + var attempts = 0 + val provider = DeepSeekProvider( + DeepSeekClient { _, _ -> + attempts += 1 + throw DeepSeekEmptyResultException() + }, + ) + + shouldThrow { + provider.execute(request(), DISCARD_OUTPUT) + } + + attempts shouldBe 2 + } + + "does not retry an empty streaming result after output may have started" { + var attempts = 0 + val provider = DeepSeekProvider( + DeepSeekClient { _, _ -> + attempts += 1 + throw DeepSeekEmptyResultException() + }, + ) + + shouldThrow { + provider.execute(request().copy(stream = true), DISCARD_OUTPUT) + } + + attempts shouldBe 1 + } + "meters provider input and output tokens from a streamed response" { val response = """ data: {"choices":[{"delta":{"content":"ok"}}]} @@ -180,4 +264,10 @@ private val CONFIG = DeepSeekConfig( apiKey = "test-key", model = "configured-model", ) +private val SUCCESS_USAGE = com.osglab.account.features.gateway.models.ProviderUsage( + meter = com.osglab.account.features.gateway.models.UsageMeter.LLM_TOKEN, + units = 13, + inputUnits = 10, + outputUnits = 3, +) private val DISCARD_OUTPUT = ProviderOutput { } diff --git a/src/test/kotlin/com/osglab/account/features/gateway/providers/volcengine/SaucV3ProtocolTest.kt b/src/test/kotlin/com/osglab/account/features/gateway/providers/volcengine/SaucV3ProtocolTest.kt index 2f758fc..166b631 100644 --- a/src/test/kotlin/com/osglab/account/features/gateway/providers/volcengine/SaucV3ProtocolTest.kt +++ b/src/test/kotlin/com/osglab/account/features/gateway/providers/volcengine/SaucV3ProtocolTest.kt @@ -41,13 +41,32 @@ class SaucV3ProtocolTest : StringSpec({ extractFinalDuration(final) shouldBe 1_234L } - "rejects a positive final sequence" { - val frame = SaucV3Codec().decodeServerFrame( - serverFrame(1, true, """{"audio_info":{"duration":100}}"""), + "accepts the positive final sequence returned by Volcengine production" { + val codec = SaucV3Codec() + val validator = SaucSequenceValidator() + validator.accept(codec.decodeServerFrame(serverFrame(1, false, "{}"))) + val final = codec.decodeServerFrame( + serverFrame(2, true, """{"audio_info":{"duration":100}}"""), + ) + + validator.accept(final) + extractFinalDuration(final) shouldBe 100L + } + + "rejects a non-contiguous final sequence regardless of its sign" { + val codec = SaucV3Codec() + val positive = codec.decodeServerFrame( + serverFrame(2, true, """{"audio_info":{"duration":100}}"""), + ) + val negative = codec.decodeServerFrame( + serverFrame(-2, true, """{"audio_info":{"duration":100}}"""), ) shouldThrow { - SaucSequenceValidator().accept(frame) + SaucSequenceValidator().accept(positive) + } + shouldThrow { + SaucSequenceValidator().accept(negative) } } diff --git a/src/test/kotlin/com/osglab/account/features/gateway/services/GatewayServiceBillingTest.kt b/src/test/kotlin/com/osglab/account/features/gateway/services/GatewayServiceBillingTest.kt index 95650cc..05cca29 100644 --- a/src/test/kotlin/com/osglab/account/features/gateway/services/GatewayServiceBillingTest.kt +++ b/src/test/kotlin/com/osglab/account/features/gateway/services/GatewayServiceBillingTest.kt @@ -16,6 +16,9 @@ import com.osglab.account.features.gateway.ports.ProviderRequestMetadata import com.osglab.account.features.gateway.ports.ProviderUsageEstimate import com.osglab.account.features.gateway.providers.GatewayProvider import com.osglab.account.features.gateway.providers.ProviderCatalog +import com.osglab.account.features.gateway.providers.deepseek.DeepSeekClient +import com.osglab.account.features.gateway.providers.deepseek.DeepSeekEmptyResultException +import com.osglab.account.features.gateway.providers.deepseek.DeepSeekProvider import io.kotest.assertions.throwables.shouldThrow import io.kotest.core.spec.style.StringSpec import io.kotest.matchers.collections.shouldContainExactly @@ -63,6 +66,47 @@ class GatewayServiceBillingTest : StringSpec({ credits.lastEstimate?.outputUnits shouldBe 32L } + "uses one reservation when a buffered DeepSeek empty result succeeds on retry" { + val credits = FakeCredits() + var attempts = 0 + val provider = DeepSeekProvider( + DeepSeekClient { _, _ -> + attempts += 1 + if (attempts == 1) throw DeepSeekEmptyResultException() + TOKEN_USAGE + }, + ) + val service = service(credits, provider) + + service.execute(PRINCIPAL, request(), DISCARD_OUTPUT) + + attempts shouldBe 2 + credits.reserveCalls shouldBe 1 + credits.settled.shouldContainExactly(RESERVATION_ID to TOKEN_USAGE.units) + credits.released shouldBe emptyList() + } + + "releases one reservation after both buffered DeepSeek attempts are empty" { + val credits = FakeCredits() + var attempts = 0 + val provider = DeepSeekProvider( + DeepSeekClient { _, _ -> + attempts += 1 + throw DeepSeekEmptyResultException() + }, + ) + val service = service(credits, provider) + + shouldThrow { + service.execute(PRINCIPAL, request(), DISCARD_OUTPUT) + } + + attempts shouldBe 2 + credits.reserveCalls shouldBe 1 + credits.settled shouldBe emptyList() + credits.released.shouldContainExactly(RESERVATION_ID) + } + "releases a reservation when a provider call is cancelled" { val credits = FakeCredits() val started = CompletableDeferred()