1// @effect-diagnostics nodeBuiltinImport:off - Effect has no incremental digest.
2import * as NodeCrypto from "node:crypto";
3import * as Context from "effect/Context";
4import * as SecureRandom from "effect/crypto/SecureRandom";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as FileSystem from "effect/FileSystem";
8import type { HttpClientError } from "effect/http";
9import { HttpClient, HttpClientRequest } from "effect/http";
10import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
12import * as Path from "effect/Path";
13import type { PlatformError } from "effect/PlatformError";
14import * as Schema from "effect/Schema";
15import * as Stream from "effect/Stream";
16
17import * as HostProcess from "@t3tools/shared/HostProcess";
18
19import * as ServerConfig from "../config.ts";
20import { expandHomePath } from "../os-jank.ts";
21import type { SpeechModel } from "./model.ts";
22
23type ModelFile = Pick<SpeechModel, "id" | "filename" | "revision" | "size" | "sha256">;
24
25class SpeechModelDownloadError extends Schema.TaggedError<SpeechModelDownloadError>()(
26 "SpeechModelDownloadError",
27 {
28 modelId: Schema.String,
29 reason: Schema.Literals(["status", "range", "size", "verification"]),
30 status: Schema.optional(Schema.Number),
31 },
32) {
33 override get message(): string {
34 return `Speech model download failed: ${this.reason}${this.status === undefined ? "" : ` ${this.status}`}.`;
35 }
36}
37
38type SpeechModelDownloadProgress = {
39 readonly onProgress?: (downloaded: number) => void;
40 readonly onVerificationStart?: () => void;
41};
42
43export class SpeechModelStore extends Context.Service<
44 SpeechModelStore,
45 {
46 /** Cheap size check for status polls; never hashes the file. */
47 readonly isReady: (model: ModelFile) => Effect.Effect<boolean>;
48 /** The installed file once its SHA-256 has been verified, hashing it on first use. */
49 readonly installedPath: (model: ModelFile) => Effect.Effect<string | undefined>;
50 /** Links a copy the Hugging Face cache or Handy already holds; fetches only without one. */
51 readonly download: (
52 model: ModelFile,
53 progress?: SpeechModelDownloadProgress,
54 ) => Effect.Effect<
55 string,
56 SpeechModelDownloadError | PlatformError | HttpClientError.HttpClientError
57 >;
58 /**
59 * Installs catalog files already on this machine as symlinks, never copies.
60 * `source` is a file or a folder to search; returns the ids it installed.
61 */
62 readonly link: (
63 models: ReadonlyArray<ModelFile>,
64 source: string,
65 ) => Effect.Effect<ReadonlyArray<string>, PlatformError>;
66 /** The outside file a linked model points at. */
67 readonly linkedFrom: (model: ModelFile) => Effect.Effect<string | undefined>;
68 /** Removes T3 Code's file or link; a linked file itself is left in place. */
69 readonly remove: (model: ModelFile) => Effect.Effect<void, PlatformError>;
70 }
71>()("t3/speech/SpeechModelStore") {}
72
73const DOWNLOAD_ATTEMPTS = 3;
74const PARTIAL_FILENAME =
75 /\.gguf\.[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\.part$/i;
76
77const LINK_SEARCH_DEPTH = 5;
78const LINK_SEARCH_ENTRIES = 5_000;
79
80const make = Effect.gen(function* () {
81 const fs = yield* FileSystem.FileSystem;
82 const path = yield* Path.Path;
83 const home = yield* HostProcess.HomeDirectory;
84 const platform = yield* HostProcess.Platform;
85 const http = yield* HttpClient.HttpClient;
86 const random = yield* SecureRandom.SecureRandom;
87 const config = yield* ServerConfig.ServerConfig;
88 const directory = path.join(config.stateDir, "speech", "models");
89 // The server already owns this state directory; no download can be active before the store opens.
90 for (const name of yield* fs.readDirectory(directory).pipe(Effect.orElseSucceed(() => []))) {
91 if (PARTIAL_FILENAME.test(name)) {
92 yield* fs.remove(path.join(directory, name), { force: true }).pipe(Effect.ignore);
93 }
94 }
95 const modelPath = (model: ModelFile) => path.join(directory, model.filename);
96 const appDataDirectories =
97 platform === "darwin"
98 ? [path.join(home, "Library", "Application Support")]
99 : platform === "win32"
100 ? [path.join(home, "AppData", "Roaming")]
101 : [path.join(home, ".local", "share"), path.join(home, ".config")];
102 // Where the Hugging Face cache and Handy keep these same pinned files.
103 const knownCopies = (model: ModelFile) => [
104 path.join(
105 home,
106 ".cache",
107 "huggingface",
108 "hub",
109 `models--${model.id.replace("/", "--")}`,
110 "snapshots",
111 model.revision,
112 model.filename,
113 ),
114 ...appDataDirectories.map((appData) =>
115 path.join(appData, "com.pais.handy", "models", model.filename),
116 ),
117 ];
118 // Files whose digest failed, by identity, so status polls stop calling them ready.
119 const corrupt = new Map<string, string>();
120
121 const identity = (model: ModelFile, info: FileSystem.File.Info) =>
122 [
123 model.sha256,
124 info.size,
125 Option.getOrUndefined(info.mtime)?.getTime(),
126 info.dev,
127 Option.getOrUndefined(info.ino),
128 ].join(":");
129
130 const statFile = (file: string) => fs.stat(file).pipe(Effect.orElseSucceed(() => undefined));
131
132 const expectedFile = (model: ModelFile, file: string) =>
133 statFile(file).pipe(
134 Effect.map((info) =>
135 info?.type === "File" && Number(info.size) === model.size ? info : undefined,
136 ),
137 );
138
139 const digestMatches = Effect.fn("SpeechModelStore.digest")(
140 function* (model: ModelFile, file: string) {
141 const hash = NodeCrypto.createHash("sha256");
142 yield* fs
143 .stream(file)
144 .pipe(Stream.runForEach((chunk) => Effect.sync(() => void hash.update(chunk))));
145 return hash.digest("hex") === model.sha256;
146 },
147 Effect.orElseSucceed(() => false),
148 );
149
150 // A verified file is trusted while its size, mtime, device and inode are unchanged.
151 const verifiedPath = Effect.fn("SpeechModelStore.verifiedPath")(function* (
152 model: ModelFile,
153 file: string,
154 onVerificationStart?: () => void,
155 ) {
156 const info = yield* expectedFile(model, file);
157 if (!info) return undefined;
158 const marker = identity(model, info);
159 if (corrupt.get(file) === marker) return undefined;
160 const recorded = yield* fs
161 .readFileString(`${file}.verified`)
162 .pipe(Effect.orElseSucceed(() => undefined));
163 if (recorded === marker) return file;
164 onVerificationStart?.();
165 const valid = yield* digestMatches(model, file);
166 if (!valid) corrupt.set(file, marker);
167 const current = yield* expectedFile(model, file);
168 if (!valid || !current || identity(model, current) !== marker) return undefined;
169 yield* fs.writeFileString(`${file}.verified`, marker, { mode: 0o600 }).pipe(Effect.ignore);
170 return file;
171 });
172
173 const fetchInto = Effect.fn("SpeechModelStore.fetch")(function* (
174 model: ModelFile,
175 url: string,
176 partialPath: string,
177 offset: number,
178 onProgress?: (downloaded: number) => void,
179 ) {
180 const request = HttpClientRequest.get(url).pipe(
181 offset > 0 ? HttpClientRequest.setHeader("Range", `bytes=${offset}-`) : (r) => r,
182 );
183 const response = yield* http.execute(request);
184 const resumed = offset > 0 && response.status === 206;
185 if (!resumed && response.status !== 200)
186 return yield* new SpeechModelDownloadError({
187 modelId: model.id,
188 reason: "status",
189 status: response.status,
190 });
191 if (resumed) {
192 const range = /^bytes (\d+)-(\d+)\/(\d+)$/.exec(response.headers["content-range"] ?? "");
193 if (
194 !range ||
195 Number(range[1]) !== offset ||
196 Number(range[2]) !== model.size - 1 ||
197 Number(range[3]) !== model.size
198 )
199 return yield* new SpeechModelDownloadError({ modelId: model.id, reason: "range" });
200 }
201 const start = resumed ? offset : 0;
202 const contentLength = response.headers["content-length"];
203 if (
204 contentLength !== undefined &&
205 Number.isFinite(Number(contentLength)) &&
206 Number(contentLength) !== model.size - start
207 )
208 return yield* new SpeechModelDownloadError({ modelId: model.id, reason: "size" });
209 let downloaded = start;
210 onProgress?.(downloaded);
211 yield* response.stream.pipe(
212 Stream.tap((chunk) => {
213 downloaded += chunk.byteLength;
214 if (downloaded > model.size)
215 return Effect.fail(new SpeechModelDownloadError({ modelId: model.id, reason: "size" }));
216 onProgress?.(downloaded);
217 return Effect.void;
218 }),
219 Stream.run(fs.sink(partialPath, { flag: resumed ? "a" : "w", mode: 0o600 })),
220 );
221 return downloaded;
222 });
223
224 const filesUnder = Effect.fn("SpeechModelStore.filesUnder")(function* (source: string) {
225 const files: Array<{ readonly file: string; readonly size: number }> = [];
226 const pending = [{ entry: source, depth: LINK_SEARCH_DEPTH }];
227 for (let visited = 0; pending.length > 0 && visited < LINK_SEARCH_ENTRIES; visited += 1) {
228 const { entry, depth } = pending.pop()!;
229 const info = yield* statFile(entry);
230 if (info?.type === "File") files.push({ file: entry, size: Number(info.size) });
231 if (info?.type !== "Directory" || depth === 0) continue;
232 for (const name of yield* fs.readDirectory(entry).pipe(Effect.orElseSucceed(() => [])))
233 pending.push({ entry: path.join(entry, name), depth: depth - 1 });
234 }
235 return files;
236 });
237
238 const realPath = (file: string) => fs.realPath(file).pipe(Effect.orElseSucceed(() => undefined));
239
240 const link: SpeechModelStore["Service"]["link"] = Effect.fn("SpeechModelStore.link")(
241 function* (models, source) {
242 const root = path.resolve(
243 yield* expandHomePath(source.trim()).pipe(
244 Effect.provideService(Path.Path, path),
245 Effect.provideService(HostProcess.HomeDirectory, home),
246 ),
247 );
248 yield* fs.makeDirectory(directory, { recursive: true });
249 const linked: string[] = [];
250 for (const { file, size } of yield* filesUnder(root)) {
251 for (const model of models) {
252 if (model.size !== size || linked.includes(model.id)) continue;
253 const finalPath = modelPath(model);
254 // Pointing at T3 Code's own model folder must not replace a file with a link to itself.
255 if ((yield* realPath(file)) === (yield* realPath(finalPath))) continue;
256 if (yield* verifiedPath(model, finalPath)) continue;
257 const hashed = yield* expectedFile(model, file);
258 if (!hashed || !(yield* digestMatches(model, file))) continue;
259 // Staged then renamed, so a failed link never removes the file already installed.
260 const staged = `${finalPath}.${yield* random.uuidV4()}.part`;
261 // Windows grants symlinks only to privileged accounts; a hard link also shares the bytes.
262 const hardLinked = yield* fs.symlink(file, staged).pipe(
263 Effect.as(false),
264 Effect.catch(() => fs.link(file, staged).pipe(Effect.as(true))),
265 );
266 const current = yield* expectedFile(model, staged);
267 const marker = identity(model, hashed);
268 if (!current || identity(model, current) !== marker) {
269 yield* fs.remove(staged, { force: true });
270 continue;
271 }
272 yield* fs.rename(staged, finalPath);
273 // A hard link cannot be read back, so its origin is recorded beside it.
274 yield* (
275 hardLinked
276 ? fs.writeFileString(`${finalPath}.source`, file, { mode: 0o600 })
277 : fs.remove(`${finalPath}.source`, { force: true })
278 ).pipe(Effect.ignore);
279 yield* fs
280 .writeFileString(`${finalPath}.verified`, marker, { mode: 0o600 })
281 .pipe(Effect.ignore);
282 linked.push(model.id);
283 }
284 }
285 return linked;
286 },
287 );
288
289 const download: SpeechModelStore["Service"]["download"] = Effect.fn("SpeechModelStore.download")(
290 function* (model, progress = {}) {
291 yield* Effect.annotateCurrentSpan({ modelId: model.id });
292 const finalPath = modelPath(model);
293 yield* fs.makeDirectory(directory, { recursive: true });
294 if (yield* verifiedPath(model, finalPath, progress.onVerificationStart)) return finalPath;
295 for (const copy of knownCopies(model)) {
296 const linked = yield* link([model], copy).pipe(Effect.orElseSucceed(() => []));
297 if (linked.length > 0) return finalPath;
298 }
299 const partialPath = `${finalPath}.${yield* random.uuidV4()}.part`;
300 const url = `https://huggingface.co/${model.id}/resolve/${model.revision}/${model.filename}`;
301 return yield* Effect.gen(function* () {
302 for (let attempt = 1; attempt <= DOWNLOAD_ATTEMPTS; attempt += 1) {
303 const offset = Number((yield* statFile(partialPath))?.size ?? 0);
304 if (offset === model.size) break;
305 // Transport and disk failures resume from the partial file; a bad response does not.
306 const downloaded = yield* fetchInto(
307 model,
308 url,
309 partialPath,
310 offset,
311 progress.onProgress,
312 ).pipe(
313 Effect.catchIf(
314 (error) => attempt < DOWNLOAD_ATTEMPTS && error._tag !== "SpeechModelDownloadError",
315 () => Effect.succeed(-1),
316 ),
317 );
318 if (downloaded === model.size) break;
319 }
320 progress.onVerificationStart?.();
321 const complete = yield* expectedFile(model, partialPath);
322 if (!complete || !(yield* digestMatches(model, partialPath)))
323 return yield* new SpeechModelDownloadError({ modelId: model.id, reason: "verification" });
324 yield* fs.rename(partialPath, finalPath);
325 yield* fs.remove(`${finalPath}.source`, { force: true }).pipe(Effect.ignore);
326 // Rename changes file metadata, so record the verified file's final identity.
327 const info = yield* fs.stat(finalPath);
328 yield* fs
329 .writeFileString(`${finalPath}.verified`, identity(model, info), { mode: 0o600 })
330 .pipe(Effect.ignore);
331 return finalPath;
332 }).pipe(
333 Effect.onExit((exit) =>
334 Exit.isSuccess(exit)
335 ? Effect.void
336 : fs.remove(partialPath, { force: true }).pipe(Effect.ignore),
337 ),
338 );
339 },
340 );
341
342 return SpeechModelStore.of({
343 link,
344 linkedFrom: Effect.fn("SpeechModelStore.linkedFrom")((model) =>
345 fs.readLink(modelPath(model)).pipe(
346 Effect.catch(() => fs.readFileString(`${modelPath(model)}.source`)),
347 Effect.orElseSucceed(() => undefined),
348 ),
349 ),
350 isReady: Effect.fn("SpeechModelStore.isReady")((model) =>
351 expectedFile(model, modelPath(model)).pipe(
352 Effect.map(
353 (info) => info !== undefined && corrupt.get(modelPath(model)) !== identity(model, info),
354 ),
355 ),
356 ),
357 installedPath: Effect.fn("SpeechModelStore.installedPath")((model) =>
358 verifiedPath(model, modelPath(model)),
359 ),
360 download,
361 remove: Effect.fn("SpeechModelStore.remove")(function* (model) {
362 const file = modelPath(model);
363 yield* fs.remove(file, { force: true });
364 yield* fs.remove(`${file}.verified`, { force: true });
365 yield* fs.remove(`${file}.source`, { force: true });
366 }),
367 });
368});
369
370export const layer = Layer.effect(SpeechModelStore, make);