1import { encodeSpeechPcmRequest } from "@t3tools/shared/speech";
2import * as Effect from "effect/Effect";
3import * as Redacted from "effect/Redacted";
4import * as Schema from "effect/Schema";
5import { CloudSpeechError, SPEECH_STREAM_PATH } from "@t3tools/contracts";
6import type {
7 CloudSpeechModelId,
8 CloudSpeechProvider,
9 SettingsSecretValue,
10 SpeechAcceleration,
11 SpeechModelUnloadTimeout,
12 SpeechTranscriptionOptions,
13 SpeechPostProcessingOptions,
14} from "@t3tools/contracts";
15
16import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
17import type { PreparedConnection } from "../connection/model.ts";
18import * as ManagedRelay from "../relay/managedRelay.ts";
19import {
20 makeEnvironmentHttpApiGroupClient,
21 makeEnvironmentHttpApiUrlBuilder,
22} from "../rpc/http.ts";
23import { executeAuthenticatedEnvironmentHttpRequest } from "../state/environmentHttpAuth.ts";
24import { VoiceTranscriptionError } from "./transcription.ts";
25
26const INFERENCE_OR_DOWNLOAD_TIMEOUT_MS = 30 * 60_000;
27const isCloudSpeechError = Schema.is(CloudSpeechError);
28type EnvironmentApiClient<G extends "voice" | "auth"> = Effect.Success<
29 ReturnType<typeof makeEnvironmentHttpApiGroupClient<G>>
30>;
31
32const request = Effect.fn("clientRuntime.voiceInput.environmentRequest")(function* <
33 A,
34 E,
35 G extends "voice" | "auth",
36>(input: {
37 readonly group: G;
38 readonly prepared: PreparedConnection;
39 readonly method: "GET" | "POST" | "DELETE";
40 readonly timeoutMs?: number;
41 readonly canSend?: () => boolean;
42 readonly path: (urls: ReturnType<typeof makeEnvironmentHttpApiUrlBuilder>[G]) => string;
43 readonly run: (input: {
44 readonly client: EnvironmentApiClient<G>;
45 readonly headers: { readonly authorization?: string; readonly dpop?: string };
46 }) => Effect.Effect<A, E>;
47}) {
48 return yield* executeAuthenticatedEnvironmentHttpRequest({
49 group: input.group,
50 prepared: input.prepared,
51 signer: yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner),
52 remoteAuthorization: yield* Effect.serviceOption(
53 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization,
54 ),
55 method: input.method,
56 url: input.path,
57 timeoutMs: input.timeoutMs ?? 30_000,
58 // Audio and credentials never follow a redirect to another host.
59 requestInit: { redirect: "error" },
60 request: (args) =>
61 Effect.suspend(() => (input.canSend?.() === false ? Effect.interrupt : input.run(args))),
62 });
63});
64
65export const getEnvironmentSpeechStatus = (prepared: PreparedConnection) =>
66 request({
67 group: "voice",
68 prepared,
69 method: "GET",
70 path: (urls) => urls.status(),
71 run: ({ client, headers }) => client.status({ headers }),
72 });
73
74export const getCloudSpeechKeys = (prepared: PreparedConnection) =>
75 request({
76 group: "voice",
77 prepared,
78 method: "GET",
79 path: (urls) => urls.cloudKeys(),
80 run: ({ client, headers }) => client.cloudKeys({ headers }),
81 });
82
83export const getCloudSpeechModels = (prepared: PreparedConnection) =>
84 request({
85 group: "voice",
86 prepared,
87 method: "GET",
88 path: (urls) => urls.cloudModels(),
89 run: ({ client, headers }) => client.cloudModels({ headers }),
90 });
91
92export const requireCloudSpeechKey = (
93 prepared: PreparedConnection,
94 provider: CloudSpeechProvider,
95) =>
96 getCloudSpeechKeys(prepared).pipe(
97 Effect.flatMap((keys) =>
98 keys[provider]
99 ? Effect.void
100 : Effect.fail(
101 new VoiceTranscriptionError({
102 code: "unavailable",
103 message: new CloudSpeechError({ provider, reason: "missing-key" }).message,
104 }),
105 ),
106 ),
107 );
108
109export const setCloudSpeechKey = (
110 prepared: PreparedConnection,
111 provider: CloudSpeechProvider,
112 key: SettingsSecretValue | null,
113) =>
114 request({
115 group: "voice",
116 prepared,
117 method: "POST",
118 path: (urls) => urls.setCloudKey({ params: { provider } }),
119 run: ({ client, headers }) =>
120 client.setCloudKey({
121 headers,
122 params: { provider },
123 payload: { key: typeof key === "string" ? Redacted.make(key) : key },
124 }),
125 });
126
127export const transcribeCloudPcm = (
128 prepared: PreparedConnection,
129 provider: CloudSpeechProvider,
130 pcm: Uint8Array,
131 options: SpeechTranscriptionOptions,
132 canSend: () => boolean,
133 model?: CloudSpeechModelId,
134) => {
135 const query = model === undefined ? {} : { model };
136 return request({
137 group: "voice",
138 prepared,
139 method: "POST",
140 canSend,
141 timeoutMs: 130_000,
142 path: (urls) => urls.transcribeCloud({ params: { provider }, query }),
143 run: ({ client, headers }) =>
144 client.transcribeCloud({
145 headers,
146 params: { provider },
147 query,
148 payload: encodeSpeechPcmRequest(pcm, options),
149 }),
150 }).pipe(
151 Effect.mapError((error) =>
152 error._tag === "RemoteEnvironmentAuthFetchError" && isCloudSpeechError(error.cause)
153 ? new VoiceTranscriptionError({
154 code: "transcription-failed",
155 message: error.cause.message,
156 })
157 : error,
158 ),
159 );
160};
161
162export const getEnvironmentSpeechStreamUrl = (prepared: PreparedConnection) => {
163 let endpoint = prepared.httpBaseUrl;
164 return request({
165 group: "auth",
166 prepared,
167 method: "POST",
168 path: (urls) => {
169 endpoint = urls.webSocketTicketFor({ params: { use: "media" } });
170 return endpoint;
171 },
172 run: ({ client, headers }) => client.webSocketTicketFor({ headers, params: { use: "media" } }),
173 }).pipe(
174 Effect.map(({ ticket }) => {
175 const url = new URL(SPEECH_STREAM_PATH, endpoint);
176 url.protocol = url.protocol === "https:" ? "wss:" : "ws:";
177 url.searchParams.set("wsTicket", Redacted.value(ticket));
178 return url.toString();
179 }),
180 );
181};
182
183export const getEnvironmentSpeechModels = (
184 prepared: PreparedConnection,
185 includeBenchmarks = false,
186) => {
187 const query = includeBenchmarks ? { benchmarks: "1" as const } : {};
188 return request({
189 group: "voice",
190 prepared,
191 method: "GET",
192 path: (urls) => urls.models({ query }),
193 run: ({ client, headers }) => client.models({ headers, query }),
194 });
195};
196
197export const downloadEnvironmentSpeechModel = (prepared: PreparedConnection, modelId: string) =>
198 request({
199 group: "voice",
200 prepared,
201 method: "POST",
202 path: (urls) => urls.downloadModel(),
203 timeoutMs: INFERENCE_OR_DOWNLOAD_TIMEOUT_MS,
204 run: ({ client, headers }) => client.downloadModel({ headers, payload: { modelId } }),
205 });
206
207export const selectEnvironmentSpeechModel = (prepared: PreparedConnection, modelId: string) =>
208 request({
209 group: "voice",
210 prepared,
211 method: "POST",
212 path: (urls) => urls.selectModel(),
213 run: ({ client, headers }) => client.selectModel({ headers, payload: { modelId } }),
214 });
215
216export const cancelEnvironmentSpeechModelDownload = (
217 prepared: PreparedConnection,
218 modelId: string,
219) =>
220 request({
221 group: "voice",
222 prepared,
223 method: "POST",
224 path: (urls) => urls.cancelModelDownload(),
225 run: ({ client, headers }) => client.cancelModelDownload({ headers, payload: { modelId } }),
226 });
227
228export const transcribeEnvironmentPcm = (
229 prepared: PreparedConnection,
230 pcm: Uint8Array,
231 options: SpeechTranscriptionOptions,
232 canSend: () => boolean,
233) =>
234 request({
235 group: "voice",
236 prepared,
237 method: "POST",
238 path: (urls) => urls.transcribe(),
239 canSend,
240 timeoutMs: INFERENCE_OR_DOWNLOAD_TIMEOUT_MS,
241 run: ({ client, headers }) =>
242 client.transcribe({
243 headers,
244 payload: encodeSpeechPcmRequest(pcm, options),
245 }),
246 });
247
248/** Sends the transcript and draft context to a provider on `prepared`; a separate opt-in. */
249export const postProcessEnvironmentTranscript = (
250 prepared: PreparedConnection,
251 transcript: string,
252 draft: {
253 readonly text: string;
254 readonly selection: { readonly start: number; readonly end: number };
255 },
256 options: SpeechPostProcessingOptions,
257 canSend: () => boolean,
258) =>
259 request({
260 group: "voice",
261 prepared,
262 method: "POST",
263 path: (urls) => urls.postProcess(),
264 canSend,
265 timeoutMs: INFERENCE_OR_DOWNLOAD_TIMEOUT_MS,
266 run: ({ client, headers }) =>
267 client.postProcess({ headers, payload: { transcript, draft, options } }),
268 });
269
270export const updateEnvironmentSpeechAcceleration = (
271 prepared: PreparedConnection,
272 acceleration: SpeechAcceleration,
273) =>
274 request({
275 group: "voice",
276 prepared,
277 method: "POST",
278 path: (urls) => urls.updateAcceleration(),
279 run: ({ client, headers }) => client.updateAcceleration({ headers, payload: { acceleration } }),
280 });
281
282export const updateEnvironmentSpeechModelUnloadTimeout = (
283 prepared: PreparedConnection,
284 timeout: SpeechModelUnloadTimeout,
285) =>
286 request({
287 group: "voice",
288 prepared,
289 method: "POST",
290 path: (urls) => urls.updateModelUnloadTimeout(),
291 run: ({ client, headers }) =>
292 client.updateModelUnloadTimeout({ headers, payload: { timeout } }),
293 });
294
295export const linkEnvironmentSpeechModels = (prepared: PreparedConnection, path: string) =>
296 request({
297 group: "voice",
298 prepared,
299 method: "POST",
300 path: (urls) => urls.linkModels(),
301 // Hashing several multi-gigabyte files takes longer than an ordinary request.
302 timeoutMs: INFERENCE_OR_DOWNLOAD_TIMEOUT_MS,
303 run: ({ client, headers }) => client.linkModels({ headers, payload: { path } }),
304 });
305
306export const removeEnvironmentSpeechModel = (prepared: PreparedConnection, modelId: string) =>
307 request({
308 group: "voice",
309 prepared,
310 method: "POST",
311 path: (urls) => urls.removeModel(),
312 run: ({ client, headers }) => client.removeModel({ headers, payload: { modelId } }),
313 });