diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/TalkModeWebSocketHandler.java b/mateclaw-server/src/main/java/vip/mate/channel/web/TalkModeWebSocketHandler.java index b487663d..f0a407d8 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/TalkModeWebSocketHandler.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/TalkModeWebSocketHandler.java @@ -119,8 +119,8 @@ public class TalkModeWebSocketHandler extends AbstractWebSocketHandler { // 2. STT: 音频转文字 // 前端用 WavRecorder（Web Audio API + 手写 PCM WAV 编码）— 见 - // mateclaw-ui/src/utils/wavEncoder.ts. WebM/Opus 被 DashScope - // Paraformer 拒收，WAV 是所有 STT provider 都接受的最大公约数。 + // mateclaw-ui/src/utils/wavEncoder.ts. WAV 是所有 STT provider + // 都接受的最大公约数，也让后端能对 PCM 做静音预检。 Map sttResult = sttService.transcribe(audioData, "audio.wav", "audio/wav", null); if (!Boolean.TRUE.equals(sttResult.get("success"))) { sendJson(session, Map.of("type", "error", "message", "Speech recognition failed: " + sttResult.get("error"))); diff --git a/mateclaw-server/src/main/java/vip/mate/stt/WavPcmExtractor.java b/mateclaw-server/src/main/java/vip/mate/stt/WavPcmExtractor.java index 1f634385..e357477a 100644 --- a/mateclaw-server/src/main/java/vip/mate/stt/WavPcmExtractor.java +++ b/mateclaw-server/src/main/java/vip/mate/stt/WavPcmExtractor.java @@ -6,21 +6,18 @@ import java.nio.ByteOrder; /** * Strip the RIFF/WAVE header off a WAV blob to expose raw PCM samples. * - *

DashScope's realtime ASR expects the {@code parameters.format = "pcm"} - * input as **bare 16-bit signed little-endian PCM**, not WAV. The frontend - * (see {@code mateclaw-ui/src/utils/wavEncoder.ts}) emits a 16 kHz mono - * 16-bit WAV with the canonical 44-byte header — this helper unwraps it. - * - *

Why not just send the WAV: DashScope rejects with "format mismatch" - * because the first 44 bytes look like garbage when interpreted as PCM - * samples — they're the RIFF magic + format chunk metadata. + *

Used for pre-flight audio diagnostics: the web recorder (see + * {@code mateclaw-ui/src/utils/wavEncoder.ts}) emits a 16 kHz mono 16-bit + * WAV with the canonical 44-byte header, and unwrapping it lets STT + * providers run a peak/RMS silence check on the raw samples before paying + * for a recognition call — "mic captured nothing" then surfaces as a + * precise local error instead of an empty transcript. * *

Limitations: handles only the canonical 44-byte WAV layout produced by * MateClaw's WavRecorder. WAVs with extra chunks (LIST, JUNK, …) before the - * data chunk would need a chunk-walking parser. We don't currently accept - * arbitrary uploads, so the tighter scope is fine; if this changes, - * extend {@link #extract} to scan for the {@code "data"} chunk header - * instead of assuming offset 36. + * data chunk would need a chunk-walking parser. Callers should gate on + * {@link #isCanonicalWav} and skip the diagnostics for anything else, + * rather than treating non-WAV input as an error. */ public final class WavPcmExtractor { @@ -32,6 +29,17 @@ public final class WavPcmExtractor { private WavPcmExtractor() {} + /** + * True when the bytes carry the RIFF/WAVE magic and are long enough to + * hold the canonical 44-byte header. Cheap gate for callers that only + * want PCM diagnostics on inputs {@link #extract} can actually handle. + */ + public static boolean isCanonicalWav(byte[] bytes) { + return bytes != null && bytes.length >= CANONICAL_HEADER_BYTES + && bytes[0] == 'R' && bytes[1] == 'I' && bytes[2] == 'F' && bytes[3] == 'F' + && bytes[8] == 'W' && bytes[9] == 'A' && bytes[10] == 'V' && bytes[11] == 'E'; + } + /** * Extract raw PCM bytes from a WAV blob. Throws when the input is too short * or the magic header bytes don't look like RIFF/WAVE — better to fail loud diff --git a/mateclaw-server/src/main/java/vip/mate/stt/provider/DashScopeSttProvider.java b/mateclaw-server/src/main/java/vip/mate/stt/provider/DashScopeSttProvider.java index 8c4a8415..448895d4 100644 --- a/mateclaw-server/src/main/java/vip/mate/stt/provider/DashScopeSttProvider.java +++ b/mateclaw-server/src/main/java/vip/mate/stt/provider/DashScopeSttProvider.java @@ -1,114 +1,96 @@ package vip.mate.stt.provider; +import cn.hutool.http.HttpRequest; +import cn.hutool.http.HttpResponse; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import vip.mate.llm.service.ModelProviderService; +import vip.mate.stt.AudioMimeTypes; import vip.mate.stt.SttProvider; import vip.mate.stt.SttRequest; import vip.mate.stt.SttResult; import vip.mate.stt.WavPcmExtractor; import vip.mate.system.model.SystemSettingsDTO; -import java.net.URI; -import java.net.http.HttpClient; -import java.net.http.WebSocket; -import java.nio.ByteBuffer; -import java.time.Duration; +import java.util.Base64; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; -import java.util.UUID; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicReference; /** - * DashScope STT Provider — Paraformer Realtime via WebSocket. + * DashScope STT Provider — synchronous HTTP recognition via Qwen3-ASR. * - *

DashScope's only sync-callable STT path is the realtime WebSocket API - * — there is no /audio/transcriptions endpoint on either the - * native or OpenAI-compatible HTTP surface (verified empirically, returns - * 404). The earlier sync-HTTP version of this provider was speculative and - * has been replaced by this one. + *

Recognizes a complete recorded clip with a single + * {@code POST /compatible-mode/v1/chat/completions} call: the audio travels + * as a base64 {@code input_audio} content part and the transcript comes back + * as the assistant message content. One request, one response — no session + * protocol, no timing constraints. * - *

Wire protocol

- * Documented at Aliyun DashScope Realtime ASR. Message exchange: - *
    - *
  1. Open WS to {@value #WS_URL} with {@code Authorization: bearer - * } header.
    2. - *
  2. Client sends a {@code run-task} text frame with task_id + - * paraformer-realtime-v2 model + format/sample-rate parameters.
    3. - *
  3. Server replies with {@code task-started} text frame.
    4. - *
  4. Client streams raw 16-bit PCM bytes as binary frames (chunked at - * ~100ms each = {@value #CHUNK_BYTES} bytes for 16 kHz mono).
    5. - *
  5. Server emits {@code result-generated} events as transcripts come - * in. Each event carries a sentence keyed by {@code begin_time}; - * later events with the same {@code begin_time} update the same - * sentence (interim → final).
    6. - *
  6. Client sends {@code finish-task} text frame; server replies with - * {@code task-finished}; both sides close.
    7. - *
+ *

Why HTTP recognition instead of the realtime WebSocket

+ * An earlier version of this provider replayed the recorded clip through the + * {@code paraformer-realtime-v2} WebSocket. That endpoint is built for live + * microphone streams: its server-side VAD assumes audio arrives at wall-clock + * pace, and a replayed clip that falls outside that contract is silently + * discarded — the protocol completes cleanly ({@code task-started} → + * {@code task-finished}) with zero {@code result-generated} events + * (see issue #580). Pacing the replay with 100ms sleeps per chunk made it + * work in some environments, but: + * + * The synchronous recognition endpoint is the purpose-built API for + * "recorded clip in, text out": latency is a single round trip regardless of + * clip length, and it accepts wav/mp3/ogg/opus/m4a/amr/webm and more, which + * also makes IM-channel voice notes first-class here. * - *

The {@link SttProvider} interface is sync — we bridge the async WS - * conversation to a blocking call via {@link CountDownLatch} (run-task ack - * + task-finished ack) plus an overall hard timeout. The whole transcribe - * call returns either a full transcript or a domain-typed - * {@link SttResult#failure} after at most {@value #OVERALL_TIMEOUT_MS}ms. + *

Wire format

+ * OpenAI-compatible chat completion with an audio content part: + * 
{@code
+ * {"model":"qwen3-asr-flash",
+ *  "messages":[{"role":"user","content":[
+ *      {"type":"input_audio","input_audio":{"data":"data:;base64,...","format":"wav"}}]}],
+ *  "stream":false,
+ *  "asr_options":{"language":"zh"}}          // omitted → auto language detection
+ * }
+ * Response: standard chat completion; transcript at + * {@code choices[0].message.content}. Errors arrive as HTTP 4xx/5xx with an + * {@code error.code} / {@code error.message} body. */ @Slf4j @Component @RequiredArgsConstructor public class DashScopeSttProvider implements SttProvider { - /** DashScope WS endpoint for realtime inference (audio/text/multimodal). */ - static final URI WS_URL = URI.create("wss://dashscope.aliyuncs.com/api-ws/v1/inference/"); + /** OpenAI-compatible chat completions endpoint carrying ASR requests. */ + static final String ASR_ENDPOINT = + "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"; - /** Default model — paraformer-realtime-v2 is the canonical 2024+ realtime ASR. */ - static final String DEFAULT_MODEL = "paraformer-realtime-v2"; + /** Default recognition model — multilingual, auto language detection. */ + static final String DEFAULT_MODEL = "qwen3-asr-flash"; - /** Default sample rate in Hz. Must match the actual WAV — the helper reads it. */ - static final int DEFAULT_SAMPLE_RATE_HZ = 16_000; - - /** ~100ms of 16 kHz / 16-bit / mono PCM. DashScope recommends 100-300ms chunks. */ - static final int CHUNK_BYTES = 3200; - - /** - * How long to sleep between chunks. Paraformer-Realtime expects audio to - * arrive at roughly the natural recording rate; if we dump the whole clip - * in tens of milliseconds the server discards the stream and replies with - * task-finished + zero result-generated events. The official Python SDK - * does the same with {@code time.sleep(0.1)} between chunks. Matches - * {@link #CHUNK_BYTES} (100ms of audio → 100ms wall sleep). - */ - static final long CHUNK_PACING_MS = 100L; - - /** How long to wait for the WS handshake + task-started ack before giving up. */ - static final long TASK_STARTED_TIMEOUT_MS = 10_000L; - - /** Overall budget for a single transcribe — beyond this we abort the WS. */ - static final long OVERALL_TIMEOUT_MS = 60_000L; + /** Overall budget for the single HTTP round trip. */ + static final int HTTP_TIMEOUT_MS = 60_000; private final ModelProviderService modelProviderService; private final ObjectMapper objectMapper; - /** Shared HttpClient — JDK's WebSocket builder doesn't reuse the underlying - * connection pool when you allocate a fresh client per call, so making - * this a field saves a connection-pool spin-up on every transcribe. */ - private final HttpClient httpClient = HttpClient.newHttpClient(); @Override public String id() { return "dashscope"; } - @Override public String label() { return "DashScope (Paraformer Realtime)"; } + @Override public String label() { return "DashScope (Qwen3 ASR)"; } @Override public boolean requiresCredential() { return true; } @Override public int autoDetectOrder() { return 150; } /** - * Per-language priority. Paraformer is the strongest mainstream Chinese - * STT, so push it ahead of Whisper on zh — see {@link SttProvider} javadoc - * for the routing rationale. + * Per-language priority. DashScope's ASR family is the strongest + * mainstream Chinese STT, so push it ahead of Whisper on zh — see + * {@link SttProvider} javadoc for the routing rationale. */ @Override public int autoDetectOrder(String language) { @@ -136,124 +118,51 @@ public class DashScopeSttProvider implements SttProvider { return SttResult.failure("DashScope API Key 未配置"); } byte[] audio = request.getAudioData(); - if (audio == null || audio.length < WavPcmExtractor.CANONICAL_HEADER_BYTES) { - return SttResult.failure("音频为空或过短"); - } - byte[] pcm = WavPcmExtractor.extract(audio); - int sampleRate = WavPcmExtractor.sampleRate(audio); - String model = request.getModel() != null ? request.getModel() : DEFAULT_MODEL; - String taskId = UUID.randomUUID().toString().replace("-", ""); - - // Peak/RMS check — the silence path is a failure mode worth its - // own log line so users can tell "mic captured nothing" from - // "DashScope rejected real audio". Successful calls log peak/rms - // at DEBUG only; a healthy call shouldn't produce a per-request - // INFO log every time the user holds the talk button. - int[] peakRms = computePcmPeakRms(pcm); - if (peakRms[0] == 0) { - log.warn("[DashScope STT] PCM is silent (peak=0, bytes={}) — check mic permission / frontend recording", - pcm.length); - return SttResult.failure( - "音频为静音（PCM peak=0）— 检查麦克风权限或前端录制实现"); - } - log.debug("[DashScope STT] PCM stats — bytes={} samples={} peak={} rms={} sampleRate={}", - pcm.length, pcm.length / 2, peakRms[0], peakRms[1], sampleRate); - - DashScopeSession session = new DashScopeSession(taskId, objectMapper); - WebSocket ws; - try { - ws = httpClient.newWebSocketBuilder() - .header("Authorization", "bearer " + apiKey) - .connectTimeout(Duration.ofMillis(TASK_STARTED_TIMEOUT_MS)) - .buildAsync(WS_URL, session) - .get(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS); - } catch (TimeoutException e) { - return SttResult.failure("DashScope WS 握手超时"); + if (audio == null || audio.length == 0) { + return SttResult.failure("音频为空"); } - try { - // 1. run-task. Envelope dumped at DEBUG only — the JSON is - // identical across calls modulo task_id + language hint, so - // logging it on every transcribe just clutters logs. - String runTask = buildRunTask(taskId, model, sampleRate, request.getLanguage()); - log.debug("[DashScope STT] run-task envelope: {}", runTask); - ws.sendText(runTask, true).get(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS); - - // 2. wait for task-started ack - if (!session.awaitTaskStarted(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { - return SttResult.failure("DashScope task-started 超时"); - } - if (session.failed()) { - return SttResult.failure("DashScope: " + session.errorMessage()); - } - - // 3. stream PCM chunks at real-time pace. Paraformer-Realtime - // is built for live mic input and silently drops audio when it - // arrives faster than wall-clock — symptom is 0 chars - // transcribed even though the protocol completes successfully - // (no task-failed). Sleep 100ms between 100ms chunks so total - // send time ≈ audio duration, matching what DashScope's own - // SDK examples do (time.sleep(0.1) per chunk). - int chunksSent = 0; - long sendStart = System.currentTimeMillis(); - for (int offset = 0; offset < pcm.length; offset += CHUNK_BYTES) { - int len = Math.min(CHUNK_BYTES, pcm.length - offset); - ByteBuffer chunk = ByteBuffer.wrap(pcm, offset, len); - ws.sendBinary(chunk, true).get(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS); - chunksSent++; - Thread.sleep(CHUNK_PACING_MS); - // Cheap fail-fast: if the server already said we're done / - // failed mid-stream, stop sending so we don't waste seconds - // sleeping on a dead connection. - if (session.failed() || session.taskFinishedRaised()) break; - } - long sendDuration = System.currentTimeMillis() - sendStart; - log.debug("[DashScope STT] streamed {} chunks ({} bytes) in {} ms", - chunksSent, pcm.length, sendDuration); - - // 4. finish-task - ws.sendText(buildFinishTask(taskId), true).get(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS); - - // 5. wait for task-finished - if (!session.awaitTaskFinished(OVERALL_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { - return SttResult.failure("DashScope task-finished 超时"); - } - if (session.failed()) { - return SttResult.failure("DashScope: " + session.errorMessage()); - } - - String text = session.aggregatedText(); - log.info("[DashScope STT] Transcribed {} chars from {} result-events " - + "(model={}, sampleRate={}, pcmBytes={})", - text.length(), session.resultEventCount(), model, sampleRate, pcm.length); - if (text.isEmpty() && session.resultEventCount() == 0) { - // Distinct failure mode: protocol completed cleanly but - // server never sent a single result-generated event. - // Almost always means the audio was discarded for - // pacing/format reasons. Surface as a typed failure so - // the fallback chain (Whisper) can still try. + // Silence gate — only for WAV, where we can read PCM directly. + // "Mic captured nothing" is by far the most common voice-input + // failure; catching it here yields a precise error instead of an + // empty transcript from the model. Non-WAV inputs (IM voice + // notes) skip the gate and go straight to the API. + if (WavPcmExtractor.isCanonicalWav(audio)) { + byte[] pcm = WavPcmExtractor.extract(audio); + int[] peakRms = computePcmPeakRms(pcm); + if (peakRms[0] == 0) { + log.warn("[DashScope STT] PCM is silent (peak=0, bytes={}) — check mic permission / frontend recording", + pcm.length); return SttResult.failure( - "DashScope 收到 0 个识别事件——可能是音频格式或节奏问题"); - } - return SttResult.success(text); - } finally { - // Best-effort close. abort() is fire-and-forget; we don't need to wait. - try { - ws.sendClose(WebSocket.NORMAL_CLOSURE, "done"); - } catch (Exception ignored) { - ws.abort(); + "音频为静音（PCM peak=0）— 检查麦克风权限或前端录制实现"); } + log.debug("[DashScope STT] PCM stats — bytes={} peak={} rms={}", + pcm.length, peakRms[0], peakRms[1]); } - } catch (TimeoutException e) { - log.warn("[DashScope STT] timeout: {}", e.getMessage()); - return SttResult.failure("DashScope STT 超时: " + e.getMessage()); - } catch (ExecutionException e) { - Throwable cause = e.getCause() != null ? e.getCause() : e; - log.error("[DashScope STT] WS error: {}", cause.getMessage(), cause); - return SttResult.failure("DashScope STT WS 错误: " + cause.getMessage()); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return SttResult.failure("DashScope STT 被中断"); + + String model = (request.getModel() != null && !request.getModel().isBlank()) + ? request.getModel() : DEFAULT_MODEL; + String format = resolveFormat(request.getFileName(), request.getContentType()); + String body = buildRequestBody(model, audio, format, request.getLanguage()); + + HttpResponse response = HttpRequest.post(ASR_ENDPOINT) + .header("Authorization", "Bearer " + apiKey.trim()) + .header("Content-Type", "application/json") + .body(body) + .timeout(HTTP_TIMEOUT_MS) + .execute(); + + if (response.getStatus() != 200) { + String error = parseErrorMessage(response.body()); + log.warn("[DashScope STT] HTTP {} — {}", response.getStatus(), error); + return SttResult.failure("DashScope STT 失败: HTTP " + response.getStatus() + + (error.isEmpty() ? "" : " — " + error)); + } + + String text = parseTranscript(response.body()); + log.info("[DashScope STT] Transcribed {} chars (model={}, format={}, audioBytes={})", + text.length(), model, format, audio.length); + return SttResult.success(text); } catch (Exception e) { log.error("[DashScope STT] Error: {}", e.getMessage(), e); return SttResult.failure("DashScope STT 异常: " + e.getMessage()); @@ -264,35 +173,93 @@ public class DashScopeSttProvider implements SttProvider { /* Wire-format helpers (package-private for unit testing). */ /* ====================================================================== */ - String buildRunTask(String taskId, String model, int sampleRate, String language) throws Exception { - Map parameters = new LinkedHashMap<>(); - parameters.put("format", "pcm"); - parameters.put("sample_rate", sampleRate); - // Language hint when supplied — paraformer-realtime-v2 supports - // "zh", "en", "ja", "ko" via language_hints. Skip when null/blank - // to let the model auto-detect. - if (language != null && !language.isBlank()) { - // Strip locale suffix (zh-CN → zh). - String hint = language.toLowerCase(); - int dash = hint.indexOf('-'); - if (dash > 0) hint = hint.substring(0, dash); - parameters.put("language_hints", new String[]{hint}); - } + /** + * Build the recognition request. The audio rides in a + * {@code data:;base64,} URI — the separate {@code format} field tells the + * service how to decode it, so the URI needs no media type. + */ + String buildRequestBody(String model, byte[] audio, String format, String language) throws Exception { + Map inputAudio = new LinkedHashMap<>(); + inputAudio.put("data", "data:;base64," + Base64.getEncoder().encodeToString(audio)); + inputAudio.put("format", format); - Map payload = Map.of( - "task_group", "audio", - "task", "asr", - "function", "recognition", - "model", model, - "parameters", parameters, - "input", Map.of()); - Map message = Map.of( - "header", Map.of( - "action", "run-task", - "task_id", taskId, - "streaming", "duplex"), - "payload", payload); - return objectMapper.writeValueAsString(message); + Map payload = new LinkedHashMap<>(); + payload.put("model", model); + payload.put("messages", List.of(Map.of( + "role", "user", + "content", List.of(Map.of( + "type", "input_audio", + "input_audio", inputAudio))))); + payload.put("stream", false); + + // Language hint when supplied ("zh", "en", "ja", ...). Strip the + // locale suffix (zh-CN → zh); omit entirely to let the model + // auto-detect among its supported languages. + String hint = stripLocale(language); + if (hint != null) { + payload.put("asr_options", Map.of("language", hint)); + } + return objectMapper.writeValueAsString(payload); + } + + /** {@code zh-CN → zh}; null/blank → null (auto-detect). */ + static String stripLocale(String language) { + if (language == null || language.isBlank()) return null; + String hint = language.toLowerCase(); + int dash = hint.indexOf('-'); + return dash > 0 ? hint.substring(0, dash) : hint; + } + + /** + * Derive the {@code input_audio.format} value ("wav", "mp3", ...) from + * the upload's filename/content-type. Falls back to "wav", matching the + * web recorder's output. + */ + static String resolveFormat(String fileName, String contentType) { + String resolved = AudioMimeTypes.resolveFileName(fileName, contentType); + int dot = resolved.lastIndexOf('.'); + return dot >= 0 ? resolved.substring(dot + 1) : "wav"; + } + + /** + * Extract the transcript from a chat-completion response. Content is + * normally a plain string; tolerate the content-part array form + * ({@code [{"text": "..."}]}) that multimodal-capable endpoints may emit. + */ + String parseTranscript(String json) throws Exception { + JsonNode content = objectMapper.readTree(json) + .path("choices").path(0).path("message").path("content"); + if (content.isTextual()) { + return content.asText(); + } + if (content.isArray()) { + StringBuilder sb = new StringBuilder(); + for (JsonNode part : content) { + sb.append(part.path("text").asText("")); + } + return sb.toString(); + } + return ""; + } + + /** + * Pull a human-readable message out of an error body. DashScope's + * compatible mode wraps errors as {@code {"error":{"code","message"}}}; + * the native surface uses top-level {@code code}/{@code message}. + * Returns "" when the body isn't parseable JSON. + */ + String parseErrorMessage(String body) { + if (body == null || body.isBlank()) return ""; + try { + JsonNode root = objectMapper.readTree(body); + JsonNode error = root.has("error") ? root.path("error") : root; + String code = error.path("code").asText(""); + String message = error.path("message").asText(""); + if (code.isEmpty()) return message; + return message.isEmpty() ? code : code + " — " + message; + } catch (Exception e) { + return ""; + } } /** @@ -327,168 +294,4 @@ public class DashScopeSttProvider implements SttProvider { int rms = (int) Math.sqrt((double) sumSq / sampleCount); return new int[]{peak, rms}; } - - String buildFinishTask(String taskId) throws Exception { - Map message = Map.of( - "header", Map.of( - "action", "finish-task", - "task_id", taskId, - "streaming", "duplex"), - "payload", Map.of("input", Map.of())); - return objectMapper.writeValueAsString(message); - } - - /* ====================================================================== */ - /* WebSocket.Listener: collects events and signals task-started/finished. */ - /* ====================================================================== */ - - /** - * State machine for one DashScope ASR conversation. Package-private so - * unit tests can drive it with synthetic JSON without hitting the network. - */ - static class DashScopeSession implements WebSocket.Listener { - private final String taskId; - private final ObjectMapper mapper; - private final CountDownLatch taskStarted = new CountDownLatch(1); - private final CountDownLatch taskFinished = new CountDownLatch(1); - - /** - * Sentence buffer keyed by begin_time. DashScope emits multiple - * {@code result-generated} events for the same sentence as it gets - * refined (interim → final); each new event for a given begin_time - * supersedes the previous text. LinkedHashMap preserves arrival - * order, which roughly matches speech order, for the final concat. - */ - private final Map sentencesByBeginTime = new LinkedHashMap<>(); - - /** Buffer for fragmented text frames (WS allows partial messages). */ - private final StringBuilder textFrameBuf = new StringBuilder(); - - private final AtomicReference errorMessage = new AtomicReference<>(); - - /** Counts result-generated events — distinguishes "server got our audio - * but recognised nothing" (>0 events, all empty text) from "server - * saw zero audio frames" (0 events). Helps diagnose pacing / - * format issues. */ - private int resultEventCount; - - DashScopeSession(String taskId, ObjectMapper mapper) { - this.taskId = taskId; - this.mapper = mapper; - } - - @Override - public CompletionStage onText(WebSocket webSocket, CharSequence data, boolean last) { - textFrameBuf.append(data); - if (last) { - handleMessage(textFrameBuf.toString()); - textFrameBuf.setLength(0); - } - webSocket.request(1); - return null; - } - - @Override - public void onError(WebSocket webSocket, Throwable error) { - errorMessage.compareAndSet(null, "WS error: " + error.getMessage()); - taskStarted.countDown(); - taskFinished.countDown(); - } - - @Override - public CompletionStage onClose(WebSocket webSocket, int statusCode, String reason) { - // If the server closes before task-finished, unblock waiters. - if (taskFinished.getCount() > 0) { - errorMessage.compareAndSet(null, - "WS closed before task-finished (status=" + statusCode + ", reason=" + reason + ")"); - } - taskStarted.countDown(); - taskFinished.countDown(); - return null; - } - - /** Package-private hook so unit tests can drive {@link DashScopeSession} without a real WebSocket. */ - void handleMessage(String json) { - // Always trace the raw frame at DEBUG — this is invaluable when - // the protocol completes "successfully" but produces no - // transcripts. Without seeing every frame it's impossible to - // tell whether DashScope sent us a status-update / warning we - // ignored, or just stayed silent between task-started and - // task-finished. - log.debug("[DashScope STT] frame: {}", json); - try { - JsonNode node = mapper.readTree(json); - String event = node.path("header").path("event").asText(); - switch (event) { - case "task-started" -> taskStarted.countDown(); - case "result-generated" -> { - resultEventCount++; - JsonNode sentence = node.path("payload").path("output").path("sentence"); - if (sentence.isObject()) { - long beginTime = sentence.path("begin_time").asLong(0L); - String text = sentence.path("text").asText(""); - // Always overwrite — later events for the same begin_time - // carry the more-final transcript. - sentencesByBeginTime.put(beginTime, text); - } - } - case "task-finished" -> taskFinished.countDown(); - case "task-failed" -> { - String msg = node.path("header").path("error_message").asText("unknown"); - String code = node.path("header").path("error_code").asText(""); - errorMessage.compareAndSet(null, - code.isEmpty() ? msg : (code + " — " + msg)); - // Wake both latches so the caller can return the typed - // failure instead of timing out for the full budget. - taskStarted.countDown(); - taskFinished.countDown(); - } - // Anything else (status updates, model warnings, beta - // events) gets surfaced at INFO so it shows up without - // turning DEBUG on. If DashScope rolls out a new event - // type we should know about, this catches it. - default -> log.info("[DashScope STT] unhandled event '{}' frame={}", event, json); - } - } catch (Exception e) { - log.warn("[DashScope STT] failed to parse WS message: {}", e.getMessage()); - } - } - - boolean awaitTaskStarted(long timeout, TimeUnit unit) throws InterruptedException { - return taskStarted.await(timeout, unit); - } - - boolean awaitTaskFinished(long timeout, TimeUnit unit) throws InterruptedException { - return taskFinished.await(timeout, unit); - } - - boolean failed() { - return errorMessage.get() != null; - } - - String errorMessage() { - return errorMessage.get(); - } - - /** True once task-finished has been observed — used by the sender - * loop to bail out early instead of pacing through dead-WS sleeps. */ - boolean taskFinishedRaised() { - return taskFinished.getCount() == 0; - } - - int resultEventCount() { - return resultEventCount; - } - - String aggregatedText() { - // Concat in begin_time order. Different sentences typically don't - // need separator characters because Chinese text streams are - // already glued; for safety against missed punctuation we leave - // a soft join ("") rather than space — Whisper-style space - // joining produces odd-looking Chinese transcripts. - StringBuilder sb = new StringBuilder(); - sentencesByBeginTime.values().forEach(sb::append); - return sb.toString(); - } - } } diff --git a/mateclaw-server/src/test/java/vip/mate/stt/WavPcmExtractorTest.java b/mateclaw-server/src/test/java/vip/mate/stt/WavPcmExtractorTest.java index 657788e6..888ee2ed 100644 --- a/mateclaw-server/src/test/java/vip/mate/stt/WavPcmExtractorTest.java +++ b/mateclaw-server/src/test/java/vip/mate/stt/WavPcmExtractorTest.java @@ -8,17 +8,18 @@ import java.nio.ByteOrder; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * Pinned behaviour for the WAV → raw-PCM helper. * - *

Why this matters: DashScope's realtime ASR rejects bare WAV with - * "format mismatch" because the first 44 bytes look like garbage when - * interpreted as PCM. {@link WavPcmExtractor} is the chokepoint that - * converts the frontend's WAV blob to the bytes DashScope actually wants. - * Wrong header offset → silent garbage transcripts; wrong sample-rate read - * → audibly distorted. + *

Why this matters: the peak/RMS silence pre-check reads raw samples off + * the frontend's WAV blob before any recognition call is made. A wrong + * header offset would feed header bytes into the PCM math and misreport + * silence vs signal; a wrong sample-rate read would break any consumer + * that needs the true capture rate. */ class WavPcmExtractorTest { @@ -49,6 +50,15 @@ class WavPcmExtractorTest { assertThrows(IllegalArgumentException.class, () -> WavPcmExtractor.extract(null)); } + @Test + @DisplayName("isCanonicalWav: true for RIFF/WAVE, false for junk / short / null") + void isCanonicalWav_gates() { + assertTrue(WavPcmExtractor.isCanonicalWav(buildWav(16_000, 16, new byte[8]))); + assertFalse(WavPcmExtractor.isCanonicalWav(new byte[64])); // no magic + assertFalse(WavPcmExtractor.isCanonicalWav(new byte[10])); // too short + assertFalse(WavPcmExtractor.isCanonicalWav(null)); + } + @Test @DisplayName("sampleRate: reads 16 kHz from the canonical header offset") void sampleRate_reads16kHz() { diff --git a/mateclaw-server/src/test/java/vip/mate/stt/provider/DashScopeSttProviderTest.java b/mateclaw-server/src/test/java/vip/mate/stt/provider/DashScopeSttProviderTest.java index 81dfb888..62295d02 100644 --- a/mateclaw-server/src/test/java/vip/mate/stt/provider/DashScopeSttProviderTest.java +++ b/mateclaw-server/src/test/java/vip/mate/stt/provider/DashScopeSttProviderTest.java @@ -5,223 +5,151 @@ import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; -import vip.mate.stt.provider.DashScopeSttProvider.DashScopeSession; -import java.util.concurrent.TimeUnit; +import java.util.Base64; +import java.nio.charset.StandardCharsets; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; /** - * Unit tests for the message-handling state machine of - * {@link DashScopeSttProvider}. The end-to-end WebSocket flow can't be - * exercised without a mock WS server, but the JSON parsing + transcript - * aggregation + latch transitions are fully testable in isolation by - * driving {@link DashScopeSession#handleMessage(String)} directly. + * Unit tests for {@link DashScopeSttProvider}'s wire-format helpers. The + * HTTP round trip itself isn't exercised (no mock server); request-body + * construction, response/transcript parsing, and error extraction are the + * parts with encoding rules worth pinning: * - *

What these tests guard against: *

    - *
  • "Two events for the same begin_time" — the second event must - * overwrite the first (interim → final), not append. - * Otherwise you get duplicated text in the final transcript.
    • - *
  • Sentence ordering — multi-sentence speech must come out in - * arrival order regardless of begin_time int values.
    • - *
  • task-failed must surface the error message on both latches so - * the caller doesn't time out for the full 60s budget.
    • + *
  • The audio must ride as a {@code data:;base64,} URI plus an explicit + * {@code format} field — dropping either breaks decoding server-side.
    • + *
  • {@code asr_options} must be omitted entirely when no language hint + * is supplied, so the service auto-detects.
    • + *
  • Transcript extraction must tolerate both plain-string and + * content-part-array response shapes.
    • *
*/ class DashScopeSttProviderTest { - private DashScopeSession session; private DashScopeSttProvider provider; private ObjectMapper mapper; @BeforeEach void setUp() { mapper = new ObjectMapper(); - session = new DashScopeSession("test-task-id", mapper); provider = new DashScopeSttProvider(null, mapper); } @Test - @DisplayName("task-started event releases the start latch") - void taskStarted_releasesLatch() throws Exception { - session.handleMessage(""" - {"header":{"task_id":"test-task-id","event":"task-started"},"payload":{}} - """); - assertTrue(session.awaitTaskStarted(100, TimeUnit.MILLISECONDS)); - assertFalse(session.failed()); - } - - @Test - @DisplayName("result-generated builds transcript text") - void resultGenerated_appendsToTranscript() { - session.handleMessage(""" - {"header":{"task_id":"test-task-id","event":"result-generated"}, - "payload":{"output":{"sentence":{"begin_time":0,"end_time":1500,"text":"你好"}}}} - """); - assertEquals("你好", session.aggregatedText()); - } - - @Test - @DisplayName("interim updates for the same begin_time overwrite (not append)") - void resultGenerated_overwritesSameBeginTime() { - // Real DashScope behaviour: each sentence starts as a partial - // transcript and gets refined on subsequent events. Both events - // share the same begin_time. If we appended instead of overwriting - // we'd produce "你你好" instead of "你好". - session.handleMessage(""" - {"header":{"event":"result-generated"}, - "payload":{"output":{"sentence":{"begin_time":0,"end_time":500,"text":"你"}}}} - """); - session.handleMessage(""" - {"header":{"event":"result-generated"}, - "payload":{"output":{"sentence":{"begin_time":0,"end_time":1500,"text":"你好"}}}} - """); - assertEquals("你好", session.aggregatedText()); - } - - @Test - @DisplayName("multiple sentences concatenate in arrival order") - void resultGenerated_concatenatesSentencesInOrder() { - // Different begin_time → different sentences. Final transcript is - // the concat of all sentences in arrival order (LinkedHashMap). - session.handleMessage(""" - {"header":{"event":"result-generated"}, - "payload":{"output":{"sentence":{"begin_time":0,"end_time":1500,"text":"你好"}}}} - """); - session.handleMessage(""" - {"header":{"event":"result-generated"}, - "payload":{"output":{"sentence":{"begin_time":1500,"end_time":3000,"text":"世界"}}}} - """); - assertEquals("你好世界", session.aggregatedText()); - } - - @Test - @DisplayName("task-finished releases the finish latch") - void taskFinished_releasesLatch() throws Exception { - session.handleMessage(""" - {"header":{"event":"task-finished"},"payload":{}} - """); - assertTrue(session.awaitTaskFinished(100, TimeUnit.MILLISECONDS)); - assertFalse(session.failed()); - } - - @Test - @DisplayName("task-failed surfaces error message and unblocks both latches") - void taskFailed_surfacesErrorAndUnblocks() throws Exception { - // Critical for fail-fast behaviour: without this the caller would - // time out after the full 60s OVERALL_TIMEOUT_MS instead of seeing - // the typed error within milliseconds. - session.handleMessage(""" - {"header":{"event":"task-failed", - "error_code":"InvalidParameter.SampleRate", - "error_message":"sample rate not supported"}, - "payload":{}} - """); - assertTrue(session.awaitTaskStarted(100, TimeUnit.MILLISECONDS)); - assertTrue(session.awaitTaskFinished(100, TimeUnit.MILLISECONDS)); - assertTrue(session.failed()); - assertTrue(session.errorMessage().contains("InvalidParameter.SampleRate")); - assertTrue(session.errorMessage().contains("sample rate not supported")); - } - - @Test - @DisplayName("resultEventCount tracks every result-generated event (regardless of text)") - void resultEventCount_isIncrementedPerEvent() { - // Distinguishing "server got our audio but didn't recognise anything" - // (>0 events with empty text) from "server saw 0 audio frames" - // (0 events) is the diagnostic that fingered the chunk-pacing bug. - // Pin the counter behaviour so it doesn't regress. - assertEquals(0, session.resultEventCount()); - session.handleMessage(""" - {"header":{"event":"result-generated"}, - "payload":{"output":{"sentence":{"begin_time":0,"text":"hi"}}}} - """); - session.handleMessage(""" - {"header":{"event":"result-generated"}, - "payload":{"output":{"sentence":{"begin_time":1000,"text":""}}}} - """); - assertEquals(2, session.resultEventCount()); - } - - @Test - @DisplayName("taskFinishedRaised flips once task-finished arrives — sender uses it to bail out early") - void taskFinishedRaised_signalsSender() { - // The sender loop polls this between paced chunks so a server that - // closes the stream early doesn't make us sleep through the rest of - // the audio for nothing. - assertFalse(session.taskFinishedRaised()); - session.handleMessage(""" - {"header":{"event":"task-finished"},"payload":{}} - """); - assertTrue(session.taskFinishedRaised()); - } - - @Test - @DisplayName("malformed JSON doesn't crash the session") - void malformedJson_isLoggedNotThrown() { - // The session is fed straight from WS frames — corrupt input must - // not bubble up into the WebSocket.Listener and tear down the - // connection. - session.handleMessage("not valid json"); - session.handleMessage("{\"missing_header\":true}"); - // No event released either latch; session is still waiting. - assertFalse(session.failed()); - } - - @Test - @DisplayName("buildRunTask serialises the documented run-task envelope") - void buildRunTask_envelopeShape() throws Exception { - // The wire format is documented by Aliyun — pin it so future - // refactors don't accidentally drop a required field. - String json = provider.buildRunTask( - "abcd1234efgh5678", "paraformer-realtime-v2", 16_000, "zh-CN"); + @DisplayName("buildRequestBody serialises model, base64 audio, format and stream=false") + void buildRequestBody_coreShape() throws Exception { + byte[] audio = "fake-wav-bytes".getBytes(StandardCharsets.UTF_8); + String json = provider.buildRequestBody("qwen3-asr-flash", audio, "wav", null); JsonNode node = mapper.readTree(json); - assertEquals("run-task", node.path("header").path("action").asText()); - assertEquals("abcd1234efgh5678", node.path("header").path("task_id").asText()); - assertEquals("duplex", node.path("header").path("streaming").asText()); - assertEquals("audio", node.path("payload").path("task_group").asText()); - assertEquals("asr", node.path("payload").path("task").asText()); - assertEquals("recognition", node.path("payload").path("function").asText()); - assertEquals("paraformer-realtime-v2", node.path("payload").path("model").asText()); - assertEquals("pcm", node.path("payload").path("parameters").path("format").asText()); - assertEquals(16_000, node.path("payload").path("parameters").path("sample_rate").asInt()); - // language_hints strips the locale: zh-CN → zh - assertEquals("zh", node.path("payload").path("parameters").path("language_hints").get(0).asText()); + + assertEquals("qwen3-asr-flash", node.path("model").asText()); + assertEquals(false, node.path("stream").asBoolean(true)); + + JsonNode content = node.path("messages").path(0).path("content").path(0); + assertEquals("user", node.path("messages").path(0).path("role").asText()); + assertEquals("input_audio", content.path("type").asText()); + assertEquals("wav", content.path("input_audio").path("format").asText()); + + String data = content.path("input_audio").path("data").asText(); + assertTrue(data.startsWith("data:;base64,"), "audio must be a base64 data URI"); + assertEquals(Base64.getEncoder().encodeToString(audio), + data.substring("data:;base64,".length())); } @Test - @DisplayName("buildRunTask omits language_hints when language is null") - void buildRunTask_skipsLanguageHintsWhenNull() throws Exception { - // Null language means "let DashScope auto-detect" — sending an - // empty array would flag as a parameter error on some accounts. - String json = provider.buildRunTask("task1", "paraformer-realtime-v2", 16_000, null); + @DisplayName("buildRequestBody adds asr_options.language with locale stripped") + void buildRequestBody_languageHint() throws Exception { + String json = provider.buildRequestBody("qwen3-asr-flash", new byte[]{1}, "wav", "zh-CN"); JsonNode node = mapper.readTree(json); - assertTrue(node.path("payload").path("parameters").path("language_hints").isMissingNode(), - "language_hints should be omitted when language is null"); + assertEquals("zh", node.path("asr_options").path("language").asText()); } @Test - @DisplayName("buildFinishTask serialises the documented finish-task envelope") - void buildFinishTask_envelopeShape() throws Exception { - String json = provider.buildFinishTask("abcd1234"); - JsonNode node = mapper.readTree(json); - assertEquals("finish-task", node.path("header").path("action").asText()); - assertEquals("abcd1234", node.path("header").path("task_id").asText()); - assertEquals("duplex", node.path("header").path("streaming").asText()); - // payload.input is required to be an empty object — DashScope - // rejects requests where it's missing or null. - assertTrue(node.path("payload").path("input").isObject()); + @DisplayName("buildRequestBody omits asr_options when language is null — auto-detect") + void buildRequestBody_omitsAsrOptionsWhenNoLanguage() throws Exception { + // An empty or null language means "let the service detect the + // language"; sending asr_options with a null/blank language field + // would be rejected as a parameter error. + String json = provider.buildRequestBody("qwen3-asr-flash", new byte[]{1}, "wav", null); + assertTrue(mapper.readTree(json).path("asr_options").isMissingNode()); + String jsonBlank = provider.buildRequestBody("qwen3-asr-flash", new byte[]{1}, "wav", " "); + assertTrue(mapper.readTree(jsonBlank).path("asr_options").isMissingNode()); + } + + @Test + @DisplayName("stripLocale: zh-CN → zh, en → en, null/blank → null") + void stripLocale_variants() { + assertEquals("zh", DashScopeSttProvider.stripLocale("zh-CN")); + assertEquals("zh", DashScopeSttProvider.stripLocale("ZH-Hant")); + assertEquals("en", DashScopeSttProvider.stripLocale("en")); + assertNull(DashScopeSttProvider.stripLocale(null)); + assertNull(DashScopeSttProvider.stripLocale(" ")); + } + + @Test + @DisplayName("resolveFormat maps filename/content-type to the input_audio format value") + void resolveFormat_variants() { + assertEquals("wav", DashScopeSttProvider.resolveFormat("clip.wav", null)); + assertEquals("mp3", DashScopeSttProvider.resolveFormat(null, "audio/mpeg")); + assertEquals("ogg", DashScopeSttProvider.resolveFormat("note.ogg", "audio/ogg")); + assertEquals("webm", DashScopeSttProvider.resolveFormat(null, "audio/webm; codecs=opus")); + // Unknown everything → wav (matches the web recorder's output). + assertEquals("wav", DashScopeSttProvider.resolveFormat(null, null)); + } + + @Test + @DisplayName("parseTranscript reads plain-string message content") + void parseTranscript_stringContent() throws Exception { + String response = """ + {"choices":[{"message":{"role":"assistant","content":"你好世界"}, + "finish_reason":"stop"}],"usage":{}} + """; + assertEquals("你好世界", provider.parseTranscript(response)); + } + + @Test + @DisplayName("parseTranscript concatenates content-part arrays") + void parseTranscript_arrayContent() throws Exception { + String response = """ + {"choices":[{"message":{"content":[{"text":"你好"},{"text":"世界"}]}}]} + """; + assertEquals("你好世界", provider.parseTranscript(response)); + } + + @Test + @DisplayName("parseTranscript returns empty string on missing/odd shapes instead of throwing") + void parseTranscript_missingContent() throws Exception { + assertEquals("", provider.parseTranscript("{}")); + assertEquals("", provider.parseTranscript("{\"choices\":[]}")); + } + + @Test + @DisplayName("parseErrorMessage handles compatible-mode and native error bodies") + void parseErrorMessage_variants() { + assertEquals("InvalidApiKey — Invalid API-key provided.", + provider.parseErrorMessage(""" + {"error":{"code":"InvalidApiKey","message":"Invalid API-key provided."}} + """)); + assertEquals("Throttling — Requests throttled.", + provider.parseErrorMessage(""" + {"code":"Throttling","message":"Requests throttled."} + """)); + assertEquals("just a message", + provider.parseErrorMessage("{\"message\":\"just a message\"}")); + assertEquals("", provider.parseErrorMessage("not json")); + assertEquals("", provider.parseErrorMessage(null)); } @Test @DisplayName("computePcmPeakRms returns 0,0 on silence; non-zero on synthetic tone") void computePcmPeakRms_distinguishesSilenceFromSignal() { // The diagnostic distinguishing "mic captured silence" (peak=0) from - // "DashScope rejected non-empty audio" (peak>0 but 0 events) is a - // critical user-visible signal — pin its math. + // "provider rejected non-empty audio" is a critical user-visible + // signal — pin its math. byte[] silent = new byte[1000]; // all zeros int[] silentStats = DashScopeSttProvider.computePcmPeakRms(silent); assertEquals(0, silentStats[0]);