feat(rime): support WebSocket v1 streaming - #2450
Conversation
🦋 Changeset detectedLatest commit: e5584b1 The changes in this PR will be included in the next version bump. This PR includes changesets to release 38 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
|
|
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
There was a problem hiding this comment.
Devin Review found 3 potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| const pool = connectionPools.get(parent); | ||
| if (!pool) throw new Error('Rime connection pool is not initialized'); | ||
| this.pool = pool; | ||
| this.pool.retain(); |
There was a problem hiding this comment.
🔴 Updates redirect existing streams
Calling updateOptions() immediately after stream() makes the existing stream retain the replacement pool. Its copied options still describe the old endpoint, so synthesis reaches the new endpoint with stale configuration.
Prompt for agents
Capture and retain the exact RimePool associated with a stream synchronously when TTS.stream() is called. Do not look it up later through the mutable connectionPools WeakMap in SynthesizeStream's deferred construction path. Ensure pool retirement waits for every stream created against that pool, including a stream followed immediately by updateOptions(). Add a regression test that calls stream(), updates the endpoint before the next event-loop turn, then verifies the old stream connects to the old endpoint and the new stream connects to the new endpoint.
Was this helpful? React with 👍 or 👎 to provide feedback.
| } | ||
| reader = response.body.getReader(); | ||
| while (true) { | ||
| const result = await bounded(reader.read(), signal, this.requestOptions.timeoutMs); |
There was a problem hiding this comment.
🔴 HTTP retries overlap timed-out requests
When bounded() times out, the pending body read continues while the retry loop starts another request. Cleanup waits on reader.cancel(), so retries can overlap or stall behind the abandoned response.
Prompt for agents
Make each HTTP body-read timeout actively abort and dispose that attempt before returning a retryable error. Ensure run() does not resolve or reject until the pending read and response body are settled, and avoid awaiting cancellation indefinitely. Add a test with a body whose read never settles, verify timeout closes the first request, and verify only then a retry can begin.
Was this helpful? React with 👍 or 👎 to provide feedback.
| statusCode: known ? statuses[error.kind] : 500, | ||
| requestId: error.requestId ?? requestId, | ||
| body: { kind: known ? error.kind : 'unknown' }, | ||
| retryable: error.kind !== 'unimplemented', |
There was a problem hiding this comment.
🟡 Unknown provider errors trigger retries
An unrecognized provider error maps to status 500 and remains retryable. Deterministic unsupported or malformed requests then repeat until the retry budget expires.
| retryable: error.kind !== 'unimplemented', | |
| retryable: known && !['invalid_input', 'unauthenticated', 'permission_denied', 'not_found', 'unimplemented'].includes(error.kind), |
Was this helpful? React with 👍 or 👎 to provide feedback.
2 participants
Summary
websocketURL, with binary protobuf and canonical protobuf JSON from@rimelabs/api@0.0.1.updateOptions(). Handle terminal HTTP TTS errors through the existing error event without an unhandled background rejection.Python counterpart: livekit/agents#6978
Interface and data flow
flowchart TB subgraph MAIN["Before this PR"] M_TTS["rime.TTS"] M_MODE{"useWebsocket / baseURL"} M_HTTP["HTTP synthesis"] M_WS3["Legacy WS3<br/>JSON messages + aligned timestamps"] M_TTS --> M_MODE M_MODE -->|"HTTP mode"| M_HTTP M_MODE -->|"WebSocket mode"| M_WS3 end subgraph BRANCH["This PR"] TTS["rime.TTS"] SELECT{"Endpoint configuration"} TTS --> SELECT SELECT -->|"No websocketURL"| LEGACY{"Legacy mode"} LEGACY -->|"HTTP URL"| HTTP["ChunkedStream<br/>HTTP synthesis"] LEGACY -->|"WebSocket baseURL or<br/>useWebsocket=true"| WS3["SynthesizeStream + RimePool<br/>WS3 JSON + aligned timestamps"] SELECT -->|"websocketURL"| MODEL{"resolveOptions<br/>model resolution"} MODEL -->|"/coda/ws"| CODA["modelId = coda"] MODEL -->|"/mist/ws"| MIST["modelId = mistv3"] MODEL -->|"/ws"| DEDICATED["Dedicated endpoint<br/>explicit modelId required"] CODA --> V1 MIST --> V1 DEDICATED --> V1 V1["SynthesizeStream + RimePool<br/>sentence tokenizer + connection reuse"] V1 --> INPUT["pushText: send complete sentences<br/>flush: release buffered text locally<br/>endInput: send end<br/>close: cancel active context"] INPUT --> PROTOCOL{"websocketProtocol"} PROTOCOL -->|"binary, default"| BINARY["rime.v1.binary<br/>protobuf frames"] PROTOCOL -->|"json"| JSON["rime.v1.json<br/>canonical protobuf JSON"] BINARY --> WIRE JSON --> WIRE WIRE["RimeConnection<br/>ready, start, text*, end or cancel<br/>started, audio*, done"] WIRE --> AUDIO["RimeAudio<br/>PCM, PCMU, WAV, MP3, Ogg Opus, WebM Opus<br/>mono PCM frames at samplingRate"] end subgraph MODEL_STATE["Endpoint and model safety"] UPDATE["updateOptions"] UPDATE --> IDENTITY{"Model changed on the same<br/>normalized model endpoint?"} IDENTITY -->|"yes"| REJECT["Reject update"] IDENTITY -->|"no"| ALLOW["Accept valid options<br/>transport mode stays fixed"] ALLOW --> CONNECTION{"Full URL, API key,<br/>or protocol changed?"} CONNECTION -->|"yes"| POOL["Replace RimePool<br/>retire old pool after its streams finish"] CONNECTION -->|"no"| KEEP["Keep RimePool"] NORMAL["Model endpoint identity ignores query and trailing slash<br/>Normalizes scheme, host, and effective port<br/>Endpoint validation rejects fragments and user information"] NORMAL -.-> IDENTITY end subgraph STREAM_STATE["Per-stream state"] CURRENT["Current TTS options and RimePool"] CURRENT --> NEW["Create SynthesizeStream"] NEW --> SNAPSHOT["Copy options and retain pool<br/>metricsModel reads the copied modelId"] SNAPSHOT --> METRICS["Existing streams keep their model and endpoint<br/>after later TTS updates"] BASE["Base SynthesizeStream.metricsModel<br/>reads the current TTS model by default"] BASE --> INFERENCE["Inference stream metrics<br/>continue to follow model updates"] end TTS -.-> UPDATE TTS -.-> CURRENTThe JS implementation uses one
SynthesizeStreamfor v1 and WS3, withRimeConnectionfor transport andRimeAudiofor decoding. V1 always uses sentence input. Itsflush()keeps the synthesis context open. V1 does not provide aligned timestamps.Testing
pnpm exec vitest run plugins/rime/src agents/src/tts/tts.test.ts agents/src/inference/tts.test.ts: 206 passed, 2 skipped.pnpm --filter @livekit/agents-plugin-rime exec tsc --noEmit: passed.pnpm --filter @livekit/agents-plugin-rime lint: passed.