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()