1import {
2 defaultInstanceIdForDriver,
3 ProviderDriverKind,
4 ServerProviderUpdateError,
5 type ProviderInstanceId,
6 type ServerProvider,
7 type ServerProviderUpdatedPayload,
8 type ServerProviderUpdateState,
9} from "@t3tools/contracts";
10import { resolveSpawnCommand } from "@t3tools/shared/shell";
11import * as Cause from "effect/Cause";
12import * as Context from "effect/Context";
13import * as Data from "effect/Data";
14import * as DateTime from "effect/DateTime";
15import * as Duration from "effect/Duration";
16import * as Effect from "effect/Effect";
17import * as Layer from "effect/Layer";
18import * as Option from "effect/Option";
19import * as Ref from "effect/Ref";
20import * as Schedule from "effect/Schedule";
21import * as Schema from "effect/Schema";
22import * as Stream from "effect/Stream";
23import { HttpClient } from "effect/http";
24import { ChildProcess, ChildProcessSpawner } from "effect/process";
25import * as NodeUtil from "node:util";
26
27import * as ModelManifest from "./ModelManifest.ts";
28import { resolveProviderCompatibility } from "./providerCompatibility.ts";
29import * as ProviderRegistry from "./ProviderRegistry.ts";
30import { makeProviderMaintenanceCommandCoordinator } from "./providerMaintenanceCommandCoordinator.ts";
31import {
32 enrichProviderSnapshotWithVersionAdvisory,
33 makeTargetedProviderUpdateAction,
34 resolveLatestProviderVersion,
35 type ProviderMaintenanceCommandAction,
36} from "@t3tools/provider-core/server/maintenanceResolver";
37import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions";
38import type { ProviderMaintenanceCapabilities } from "@t3tools/provider-core/server/maintenanceResolver";
39import { collectUint8StreamText } from "@t3tools/provider-core/server/collectStreamText";
40const isServerProviderUpdateError = Schema.is(ServerProviderUpdateError);
41
42const UPDATE_TIMEOUT_MS = 5 * 60_000;
43const UPDATE_OUTPUT_MAX_BYTES = 10_000;
44// Every progress publish resends the provider list to each client, so the
45// installer's output is sampled rather than forwarded line by line.
46const UPDATE_PROGRESS_INTERVAL = Duration.seconds(1);
47const UPDATE_PROGRESS_MAX_LENGTH = 200;
48// An installer may write for minutes without a line ending; only its tail matters.
49const UPDATE_PARTIAL_LINE_MAX_LENGTH = 4_096;
50
51export interface ProviderMaintenanceCommandResult {
52 readonly stdout: string;
53 readonly stderr: string;
54 readonly exitCode: number | null;
55 readonly timedOut: boolean;
56 readonly stdoutTruncated: boolean;
57 readonly stderrTruncated: boolean;
58}
59
60export interface ProviderMaintenanceRunnerShape {
61 readonly updateProvider: (
62 target:
63 | ProviderDriverKind
64 | {
65 readonly provider: ProviderDriverKind;
66 readonly instanceId?: ProviderInstanceId | undefined;
67 readonly targetVersion?: string | undefined;
68 },
69 ) => Effect.Effect<ServerProviderUpdatedPayload, ServerProviderUpdateError>;
70}
71
72export class ProviderMaintenanceRunner extends Context.Service<
73 ProviderMaintenanceRunner,
74 ProviderMaintenanceRunnerShape
75>()("t3/provider/providerMaintenanceRunner") {}
76
77class ProviderMaintenanceCommandError extends Data.TaggedError("ProviderMaintenanceCommandError")<{
78 readonly message: string;
79 readonly cause?: unknown;
80}> {}
81
82interface VerifiedProviderRefresh {
83 readonly providers: ReadonlyArray<ServerProvider>;
84 readonly verifiedProviders: ReadonlyArray<ServerProvider>;
85}
86
87const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
88
89const runProviderMaintenanceCommandWithSpawner = Effect.fn("ProviderMaintenanceRunner.runCommand")(
90 function* (input: {
91 readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"];
92 readonly command: string;
93 readonly args: ReadonlyArray<string>;
94 readonly env?: NodeJS.ProcessEnv;
95 readonly onProgress?: (line: string) => Effect.Effect<void>;
96 }) {
97 const collectCommandResult = Effect.fn("ProviderMaintenanceRunner.collectCommandResult")(
98 function* () {
99 // Resolve the executable for the host platform before spawning. On
100 // Windows the update tools are batch shims (e.g. `npm` -> `npm.cmd`),
101 // which a bare ChildProcess.spawn cannot launch (spawn npm ENOENT);
102 // resolveSpawnCommand finds the real `.cmd` and routes it through the
103 // shell. On Linux/macOS (incl. the WSL backend) this is a no-op.
104 const resolved = yield* resolveSpawnCommand(input.command, input.args);
105 const child = yield* input.spawner
106 .spawn(
107 ChildProcess.make(resolved.command, resolved.args, {
108 shell: resolved.shell,
109 ...(input.env ? { env: input.env, extendEnv: true } : {}),
110 }),
111 )
112 .pipe(
113 Effect.mapError(
114 (cause) =>
115 new ProviderMaintenanceCommandError({
116 message: `Failed to run update command ${input.command}: ${cause.message}`,
117 cause,
118 }),
119 ),
120 );
121 yield* Effect.addFinalizer(() => child.kill().pipe(Effect.ignore));
122
123 // Holds the newest output line not yet reported; the sampler takes it.
124 const pendingProgress = yield* Ref.make<string | null>(null);
125 const onProgress = input.onProgress;
126 if (onProgress) {
127 yield* Ref.getAndSet(pendingProgress, null).pipe(
128 Effect.flatMap((line) => (line === null ? Effect.void : onProgress(line))),
129 Effect.repeat(Schedule.spaced(UPDATE_PROGRESS_INTERVAL)),
130 Effect.forkScoped,
131 );
132 }
133 const trackProgress = <E>(stream: Stream.Stream<Uint8Array, E>) => {
134 if (!onProgress) {
135 return stream;
136 }
137 const decoder = new TextDecoder();
138 let partialLine = "";
139 return stream.pipe(
140 Stream.tap((chunk) => {
141 const split = splitOutputLines(partialLine, decoder.decode(chunk, { stream: true }));
142 partialLine = split.partialLine;
143 // A progress bar that redraws with a leading `\r` keeps its newest
144 // frame unterminated, so that frame is the latest status.
145 const line =
146 toProgressLine(split.partialLine) ??
147 split.lines.map(toProgressLine).findLast((value) => value !== null);
148 return line ? Ref.set(pendingProgress, line) : Effect.void;
149 }),
150 );
151 };
152
153 const [stdout, stderr, exitCode] = yield* Effect.all(
154 [
155 collectUint8StreamText({
156 stream: trackProgress(child.stdout),
157 maxBytes: UPDATE_OUTPUT_MAX_BYTES,
158 }),
159 collectUint8StreamText({
160 stream: trackProgress(child.stderr),
161 maxBytes: UPDATE_OUTPUT_MAX_BYTES,
162 }),
163 child.exitCode,
164 ],
165 { concurrency: "unbounded" },
166 ).pipe(
167 Effect.mapError(
168 (cause) =>
169 new ProviderMaintenanceCommandError({
170 message: cause instanceof Error ? cause.message : "Update command failed to run.",
171 cause,
172 }),
173 ),
174 );
175
176 return {
177 stdout: stdout.text,
178 stderr: stderr.text,
179 exitCode: Number(exitCode),
180 timedOut: false,
181 stdoutTruncated: stdout.truncated,
182 stderrTruncated: stderr.truncated,
183 } satisfies ProviderMaintenanceCommandResult;
184 },
185 );
186
187 return yield* collectCommandResult().pipe(
188 Effect.scoped,
189 Effect.timeoutOption(Duration.millis(UPDATE_TIMEOUT_MS)),
190 Effect.map((result) =>
191 Option.match(result, {
192 onSome: (value) => value,
193 onNone: () =>
194 ({
195 stdout: "",
196 stderr: "",
197 exitCode: null,
198 timedOut: true,
199 stdoutTruncated: false,
200 stderrTruncated: false,
201 }) satisfies ProviderMaintenanceCommandResult,
202 }),
203 ),
204 );
205 },
206);
207
208/**
209 * Split a chunk of installer output into complete lines. `\r` ends a line too,
210 * so a redrawn progress bar reports its latest frame.
211 */
212export function splitOutputLines(
213 partialLine: string,
214 text: string,
215): { readonly lines: ReadonlyArray<string>; readonly partialLine: string } {
216 const parts = (partialLine + text).split(/\r\n|\r|\n/);
217 return { partialLine: (parts.pop() ?? "").slice(-UPDATE_PARTIAL_LINE_MAX_LENGTH), lines: parts };
218}
219
220/** Turn one raw output line into a short status message, or null if it has no text. */
221export function toProgressLine(line: string): string | null {
222 const text = NodeUtil.stripVTControlCharacters(line).replace(/\s+/g, " ").trim();
223 if (text.length === 0) {
224 return null;
225 }
226 return text.length <= UPDATE_PROGRESS_MAX_LENGTH
227 ? text
228 : `${text.slice(0, UPDATE_PROGRESS_MAX_LENGTH - 1)}…`;
229}
230
231/** `claude update` reads better in a status line than the full executable path. */
232function describeCommand(command: ProviderMaintenanceCommandAction): string {
233 const executable = command.executable.split(/[\\/]/).pop() || command.executable;
234 return [executable, ...command.args].join(" ");
235}
236
237function trimNullable(value: string): string | null {
238 const trimmed = value.trim();
239 return trimmed.length > 0 ? trimmed : null;
240}
241
242function truncateText(value: string, maxLength: number): string {
243 return value.length <= maxLength ? value : value.slice(0, maxLength);
244}
245
246function commandOutput(result: ProviderMaintenanceCommandResult): string | null {
247 const output = trimNullable([result.stderr, result.stdout].filter(Boolean).join("\n\n"));
248 if (!output) {
249 return null;
250 }
251 return truncateText(output, UPDATE_OUTPUT_MAX_BYTES);
252}
253
254function failureMessage(result: ProviderMaintenanceCommandResult): string {
255 if (result.timedOut) {
256 return "Update timed out.";
257 }
258 if (result.exitCode !== null && result.exitCode !== 0) {
259 return `Update command exited with code ${result.exitCode}.`;
260 }
261 return "Update command failed.";
262}
263
264function isOutdatedProvider(provider: ServerProvider | undefined): boolean {
265 return provider?.versionAdvisory?.status === "behind_latest";
266}
267
268function isStillInstalled(provider: ServerProvider): boolean {
269 return provider.installed;
270}
271
272function makeUpdateState(input: {
273 readonly status: ServerProviderUpdateState["status"];
274 readonly startedAt: string | null;
275 readonly finishedAt: string | null;
276 readonly message: string | null;
277 readonly output?: string | null;
278}): ServerProviderUpdateState {
279 return {
280 status: input.status,
281 startedAt: input.startedAt,
282 finishedAt: input.finishedAt,
283 message: input.message,
284 output: input.output ?? null,
285 };
286}
287
288/** @public Service construction is part of the canonical Effect module API. */
289export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () {
290 const providerRegistry = yield* ProviderRegistry.ProviderRegistry;
291 const manifestService = yield* ModelManifest.ModelManifest;
292 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
293 const httpClient = yield* HttpClient.HttpClient;
294 const latestVersions = yield* ProviderLatestVersions.ProviderLatestVersions;
295 const runMaintenanceCommand = (
296 update: ProviderMaintenanceCommandAction,
297 onProgress: (line: string) => Effect.Effect<void>,
298 ) =>
299 runProviderMaintenanceCommandWithSpawner({
300 spawner,
301 command: update.executable,
302 args: update.args,
303 onProgress,
304 ...(update.env ? { env: update.env } : {}),
305 });
306 const commandCoordinator = yield* makeProviderMaintenanceCommandCoordinator({
307 makeAlreadyRunningError: () =>
308 new ServerProviderUpdateError({
309 provider: ProviderDriverKind.make("unknown"),
310 reason: "An update is already running for this provider.",
311 }),
312 });
313
314 const verifyRefreshedProvider = (
315 provider: ProviderDriverKind,
316 maintenanceCapabilities: ProviderMaintenanceCapabilities,
317 instanceId: ProviderInstanceId,
318 ): Effect.Effect<VerifiedProviderRefresh> =>
319 providerRegistry.getProviders.pipe(
320 Effect.map((providers) => {
321 const instanceIds: Array<ProviderInstanceId> = [];
322 for (const candidate of providers) {
323 if (candidate.driver === provider && candidate.instanceId === instanceId) {
324 instanceIds.push(candidate.instanceId);
325 }
326 }
327 return instanceIds;
328 }),
329 Effect.flatMap((instanceIds) =>
330 instanceIds.length === 0
331 ? providerRegistry.refreshInstance(instanceId)
332 : Effect.forEach(
333 instanceIds,
334 (instanceId) => providerRegistry.refreshInstance(instanceId),
335 {
336 concurrency: "unbounded",
337 discard: true,
338 },
339 ).pipe(Effect.andThen(providerRegistry.getProviders)),
340 ),
341 Effect.flatMap((providers) => {
342 const refreshedProviders = providers.filter(
343 (candidate) => candidate.driver === provider && candidate.instanceId === instanceId,
344 );
345 if (refreshedProviders.length === 0) {
346 return Effect.succeed<VerifiedProviderRefresh>({
347 providers,
348 verifiedProviders: [],
349 });
350 }
351 return Effect.forEach(
352 refreshedProviders,
353 (refreshedProvider) =>
354 enrichProviderSnapshotWithVersionAdvisory(
355 refreshedProvider,
356 maintenanceCapabilities,
357 ).pipe(
358 Effect.provideService(HttpClient.HttpClient, httpClient),
359 Effect.provideService(ProviderLatestVersions.ProviderLatestVersions, latestVersions),
360 ),
361 {
362 concurrency: "unbounded",
363 },
364 ).pipe(
365 Effect.map((verifiedProviders): VerifiedProviderRefresh => ({
366 providers,
367 verifiedProviders,
368 })),
369 Effect.catchCause((cause) =>
370 Effect.logWarning("Provider post-update version verification failed", {
371 provider,
372 cause: Cause.pretty(cause),
373 }).pipe(
374 Effect.as<VerifiedProviderRefresh>({
375 providers,
376 verifiedProviders: refreshedProviders,
377 }),
378 ),
379 ),
380 );
381 }),
382 );
383
384 const updateProvider: ProviderMaintenanceRunnerShape["updateProvider"] = Effect.fn(
385 "ProviderMaintenanceRunner.updateProvider",
386 )(function* (target) {
387 const provider = typeof target === "string" ? target : target.provider;
388 const instanceId =
389 typeof target === "string"
390 ? defaultInstanceIdForDriver(provider)
391 : (target.instanceId ?? defaultInstanceIdForDriver(provider));
392 const targetVersion = typeof target === "string" ? undefined : target.targetVersion;
393 const targetKey = `instance:${instanceId}`;
394 const capabilities = yield* providerRegistry.getProviderMaintenanceCapabilitiesForInstance(
395 instanceId,
396 provider,
397 );
398 const update = capabilities.update;
399 if (!update) {
400 return yield* new ServerProviderUpdateError({
401 provider,
402 reason: "This provider does not support one-click updates.",
403 });
404 }
405
406 const setUpdateState = (state: ServerProviderUpdateState | null) =>
407 providerRegistry.setProviderMaintenanceActionState({
408 instanceId,
409 action: "update",
410 state,
411 });
412 const setQueuedState = setUpdateState(
413 makeUpdateState({
414 status: "queued",
415 startedAt: null,
416 finishedAt: null,
417 message: "Waiting for another provider update to finish.",
418 }),
419 ).pipe(Effect.asVoid);
420
421 const runProviderUpdate = Effect.fn("ProviderMaintenanceRunner.runProviderUpdate")(
422 function* () {
423 const finish = (state: ServerProviderUpdateState) =>
424 setUpdateState(state).pipe(Effect.map((providers) => ({ providers })));
425 const startedAtRef = yield* Ref.make<string | null>(null);
426
427 const runCommandAndVerify = Effect.fn("ProviderMaintenanceRunner.runCommandAndVerify")(
428 function* () {
429 const startedAt = yield* nowIso;
430 yield* Ref.set(startedAtRef, startedAt);
431 const setRunningMessage = (message: string) =>
432 setUpdateState(
433 makeUpdateState({
434 status: "running",
435 startedAt,
436 finishedAt: null,
437 message,
438 }),
439 ).pipe(Effect.asVoid);
440 yield* setRunningMessage("Checking for the latest version");
441
442 // The cached capabilities chose the lock; re-derive ownership
443 // now so the command that runs matches the executable as it is
444 // at click time, not as it was at the last health refresh.
445 const fresh = yield* providerRegistry.getProviderMaintenanceCapabilitiesForInstance(
446 instanceId,
447 provider,
448 { fresh: true },
449 );
450 if (!fresh.update || fresh.update.lockKey !== update.lockKey) {
451 return yield* finish(
452 makeUpdateState({
453 status: "failed",
454 startedAt,
455 finishedAt: yield* nowIso,
456 message: "Provider installation changed. Refresh and try again.",
457 }),
458 );
459 }
460
461 const manifest = yield* manifestService.current;
462 const candidateVersion =
463 targetVersion ??
464 (yield* resolveLatestProviderVersion(fresh).pipe(
465 Effect.provideService(HttpClient.HttpClient, httpClient),
466 Effect.provideService(
467 ProviderLatestVersions.ProviderLatestVersions,
468 latestVersions,
469 ),
470 ));
471 const advisory =
472 resolveProviderCompatibility(manifest.compatibility, provider, candidateVersion) ??
473 resolveProviderCompatibility(
474 ModelManifest.BUNDLED_MODEL_MANIFEST.compatibility,
475 provider,
476 candidateVersion,
477 );
478 const command =
479 targetVersion !== undefined
480 ? makeTargetedProviderUpdateAction(fresh, targetVersion)
481 : fresh.update;
482 const rejected =
483 targetVersion !== undefined
484 ? !command ||
485 advisory?.recommendedVersion !== targetVersion ||
486 advisory.status !== "supported"
487 : advisory?.status === "broken" || advisory?.status === "unsupported";
488 if (rejected || !command) {
489 return yield* finish(
490 makeUpdateState({
491 status: "failed",
492 startedAt,
493 finishedAt: yield* nowIso,
494 message:
495 targetVersion !== undefined
496 ? "This version is no longer recommended or this installer cannot install a specific version. Refresh provider settings."
497 : "The latest provider version is incompatible with this T3 Code release. Review provider settings.",
498 }),
499 );
500 }
501 yield* setRunningMessage(`Running ${describeCommand(command)}`);
502 const result = yield* runMaintenanceCommand(command, setRunningMessage);
503 const finishedAt = yield* nowIso;
504 if (result.timedOut || result.exitCode !== 0) {
505 return yield* finish(
506 makeUpdateState({
507 status: "failed",
508 startedAt,
509 finishedAt,
510 message: failureMessage(result),
511 output: commandOutput(result),
512 }),
513 );
514 }
515
516 yield* setRunningMessage("Verifying the installed version");
517 // Homebrew's "latest" moves once the upgrade lands; read it again.
518 const verified = yield* providerRegistry.getProviderMaintenanceCapabilitiesForInstance(
519 instanceId,
520 provider,
521 { fresh: true },
522 );
523 const { verifiedProviders } = yield* verifyRefreshedProvider(
524 provider,
525 verified,
526 instanceId,
527 );
528 // "Succeeded" needs the provider to still be installed: an
529 // installer that exits 0 and leaves the binary missing is not a
530 // success. A missing version alone is not held against it, since
531 // Cursor's `about` probe can fail transiently on a healthy binary.
532 const couldNotVerify =
533 verifiedProviders.length === 0 ||
534 verifiedProviders.some(
535 (verifiedProvider) =>
536 !isStillInstalled(verifiedProvider) ||
537 (targetVersion !== undefined &&
538 verifiedProvider.version?.replace(/^v/, "") !== targetVersion),
539 );
540 const stillOutdated =
541 targetVersion === undefined &&
542 verifiedProviders.some((verifiedProvider) => isOutdatedProvider(verifiedProvider));
543 return yield* finish(
544 makeUpdateState({
545 status: couldNotVerify || stillOutdated ? "unchanged" : "succeeded",
546 startedAt,
547 finishedAt,
548 message: couldNotVerify
549 ? "Update command completed, but T3 Code could not verify the provider version."
550 : stillOutdated
551 ? "Update command completed, but T3 Code still detects an outdated provider version."
552 : "Provider updated.",
553 output: commandOutput(result),
554 }),
555 );
556 },
557 );
558
559 const recordFailedUpdate = Effect.fn("ProviderMaintenanceRunner.recordFailedUpdate")(
560 function* (cause: Cause.Cause<unknown>) {
561 const failure = Cause.squash(cause);
562 const startedAt = yield* Ref.get(startedAtRef);
563 return yield* finish(
564 makeUpdateState({
565 status: "failed",
566 startedAt,
567 finishedAt: yield* nowIso,
568 message: failure instanceof Error ? failure.message : "Update command failed.",
569 output: null,
570 }),
571 );
572 },
573 );
574
575 return yield* runCommandAndVerify().pipe(Effect.catchCause(recordFailedUpdate));
576 },
577 );
578
579 return yield* commandCoordinator
580 .withCommandLock({
581 targetKey,
582 lockKey: update.lockKey,
583 onQueued: setQueuedState,
584 run: runProviderUpdate(),
585 })
586 .pipe(
587 Effect.mapError((error) =>
588 isServerProviderUpdateError(error)
589 ? new ServerProviderUpdateError({
590 provider,
591 reason: error.reason,
592 })
593 : error,
594 ),
595 );
596 });
597
598 return ProviderMaintenanceRunner.of({
599 updateProvider,
600 });
601});
602
603export const layer = Layer.effect(ProviderMaintenanceRunner, make());