1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import * as Cause from "effect/Cause";
4import * as Effect from "effect/Effect";
5import * as Deferred from "effect/Deferred";
6import * as Fiber from "effect/Fiber";
7import * as TestClock from "effect/testing/TestClock";
8import * as Queue from "effect/Queue";
9import * as Schema from "effect/Schema";
10import * as Sink from "effect/Sink";
11import * as Stream from "effect/Stream";
12import * as ChildProcess from "effect/process/ChildProcess";
13import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
14
15import { checkPiProviderStatus, MINIMUM_PI_VERSION } from "./status.ts";
16
17const encoder = new TextEncoder();
18const decodeRpcRequest = Schema.decodeSync(
19 Schema.fromJsonString(Schema.Struct({ id: Schema.String, type: Schema.String })),
20);
21const encodeJsonLine = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
22
23function processHandle(input: {
24 readonly stdout?: string;
25 readonly stderr?: string;
26 readonly exitCode?: number;
27 /** Holds stdout and the exit back until it completes, like a CLI slow to answer. */
28 readonly answered?: Effect.Effect<void>;
29}) {
30 const answered = input.answered ?? Effect.void;
31 const bytes = (value: string | undefined) =>
32 value === undefined || value.length === 0
33 ? Stream.empty
34 : Stream.succeed(encoder.encode(value));
35 return ChildProcessSpawner.makeHandle({
36 pid: ChildProcessSpawner.ProcessId(900_000_001),
37 exitCode: Effect.as(answered, ChildProcessSpawner.ExitCode(input.exitCode ?? 0)),
38 isRunning: Effect.succeed(false),
39 kill: () => Effect.void,
40 unref: Effect.succeed(Effect.void),
41 stdin: Sink.drain,
42 stdout: Stream.unwrap(Effect.as(answered, bytes(input.stdout))),
43 stderr: bytes(input.stderr),
44 all: Stream.empty,
45 getInputFd: () => Sink.drain,
46 getOutputFd: () => Stream.empty,
47 });
48}
49
50function piProbeSpawner(version: string) {
51 return ChildProcessSpawner.make((command) => {
52 const args = ChildProcess.isStandardCommand(command) ? command.args : [];
53 return Effect.succeed(
54 args.includes("--version")
55 ? processHandle({ stdout: `pi ${version}\n` })
56 : processHandle({ stderr: "RPC startup failed", exitCode: 1 }),
57 );
58 });
59}
60
61const piRpcProcess = (
62 inventory: unknown,
63 onModelRequest: Effect.Effect<void> = Effect.void,
64 state: unknown = {},
65) =>
66 Effect.gen(function* () {
67 const stdout = yield* Queue.unbounded<Uint8Array, Cause.Done>();
68 let stdinBuffer = "";
69 const stdin = Sink.forEach((chunk: Uint8Array) =>
70 Effect.gen(function* () {
71 stdinBuffer += new TextDecoder().decode(chunk);
72 while (true) {
73 const newline = stdinBuffer.indexOf("\n");
74 if (newline === -1) return;
75 const line = stdinBuffer.slice(0, newline);
76 stdinBuffer = stdinBuffer.slice(newline + 1);
77 if (line.length === 0) continue;
78 const request = decodeRpcRequest(line);
79 if (request.type === "get_available_models") {
80 yield* onModelRequest;
81 if (inventory === null) continue;
82 }
83 const data =
84 request.type === "get_available_models"
85 ? inventory
86 : request.type === "get_commands"
87 ? { commands: [] }
88 : request.type === "get_state"
89 ? state
90 : {};
91 yield* Queue.offer(
92 stdout,
93 encoder.encode(
94 `${encodeJsonLine({ type: "response", id: request.id, command: request.type, success: true, data })}\n`,
95 ),
96 );
97 }
98 }),
99 );
100 return ChildProcessSpawner.makeHandle({
101 ...processHandle({}),
102 exitCode: Effect.never,
103 isRunning: Effect.succeed(true),
104 stdin,
105 stdout: Stream.fromQueue(stdout),
106 });
107 });
108
109/** Makes the extension-free launch exit before answering, like launch args naming an extension provider. */
110const exitsWithoutExtensions = Symbol("exitsWithoutExtensions");
111
112// Text generation lists models with --no-extensions, so its inventory can differ.
113const piDiscoverySpawner = (
114 inventory: unknown,
115 extensionFreeInventory: unknown = inventory,
116 onExtensionFreeModelRequest: Effect.Effect<void> = Effect.void,
117 options: {
118 readonly version?: string;
119 readonly state?: unknown;
120 readonly authCheck?: string | undefined;
121 } = {},
122) =>
123 Effect.succeed(
124 ChildProcessSpawner.make((command) => {
125 const args = ChildProcess.isStandardCommand(command) ? command.args : [];
126 if (args.includes("--version")) {
127 return Effect.succeed(processHandle({ stdout: `pi ${options.version ?? "0.84.3"}\n` }));
128 }
129 if (args[0] === "auth") {
130 const expected = ["auth", "check", "--provider", "openai", "--json", "--no-refresh"];
131 return options.authCheck === undefined || args.join(" ") !== expected.join(" ")
132 ? Effect.die(`Unexpected pi ${args.join(" ")}`)
133 : Effect.succeed(processHandle({ stdout: options.authCheck, exitCode: 1 }));
134 }
135 if (args.includes("--no-extensions") && extensionFreeInventory === exitsWithoutExtensions) {
136 return Effect.succeed(
137 processHandle({ stderr: 'Error: Unknown provider "demo".', exitCode: 1 }),
138 );
139 }
140 return piRpcProcess(
141 args.includes("--no-extensions") ? extensionFreeInventory : inventory,
142 args.includes("--no-extensions") ? onExtensionFreeModelRequest : Effect.void,
143 options.state,
144 );
145 }),
146 );
147
148const settings = {
149 enabled: true,
150 binaryPath: "pi",
151 launchArgs: "",
152 customModels: [],
153} as const;
154
155describe("PiProvider", () => {
156 it.effect("requires the first published Pi version with entries and settlement hooks", () =>
157 Effect.gen(function* () {
158 const snapshot = yield* checkPiProviderStatus(settings).pipe(
159 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, piProbeSpawner("0.80.3")),
160 );
161 assert.equal(snapshot.status, "error");
162 assert.equal(snapshot.version, "0.80.3");
163 assert.include(snapshot.message ?? "", `Pi ${MINIMUM_PI_VERSION} or newer`);
164 }).pipe(Effect.provide(NodeServices.layer)),
165 );
166
167 it.effect("keeps compatible Pi selectable when optional discovery fails", () =>
168 Effect.gen(function* () {
169 const snapshot = yield* checkPiProviderStatus(settings).pipe(
170 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, piProbeSpawner("0.84.3")),
171 );
172 assert.equal(snapshot.status, "ready");
173 assert.equal(snapshot.auth.status, "unknown");
174 assert.deepEqual(
175 snapshot.models.map((model) => model.slug),
176 ["default"],
177 );
178 assert.include(snapshot.message ?? "", "could not refresh its models and commands");
179 }).pipe(Effect.provide(NodeServices.layer)),
180 );
181
182 it.effect.each([
183 {
184 label: "answers after 6 s",
185 versionDelay: "6 seconds",
186 status: "ready",
187 version: "0.84.3",
188 message: "could not refresh its models and commands",
189 probesAfterVersion: true,
190 },
191 {
192 label: "never answers",
193 versionDelay: "Infinity",
194 status: "error",
195 version: null,
196 message: "Pi CLI is installed but timed out while running `pi --version`.",
197 probesAfterVersion: false,
198 },
199 ] as const)("gives `pi --version` 10 s when it $label", (row) =>
200 Effect.gen(function* () {
201 const spawned: Array<string> = [];
202 const spawner = ChildProcessSpawner.make((command) => {
203 const args = ChildProcess.isStandardCommand(command) ? command.args : [];
204 spawned.push(args.join(" "));
205 return Effect.succeed(
206 args.includes("--version")
207 ? processHandle({ stdout: "pi 0.84.3\n", answered: Effect.sleep(row.versionDelay) })
208 : processHandle({ stderr: "RPC startup failed", exitCode: 1 }),
209 );
210 });
211 const fiber = yield* checkPiProviderStatus(settings).pipe(
212 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
213 Effect.forkChild,
214 );
215 yield* TestClock.adjust("10 seconds");
216 const snapshot = yield* Fiber.join(fiber);
217 assert.equal(snapshot.status, row.status);
218 assert.equal(snapshot.version, row.version);
219 assert.include(snapshot.message ?? "", row.message);
220 assert.equal(spawned.length > 1, row.probesAfterVersion);
221 }).pipe(Effect.provide(NodeServices.layer)),
222 );
223
224 it.effect("reports a known unauthenticated inventory when Pi returns no usable models", () =>
225 Effect.gen(function* () {
226 const spawner = yield* piDiscoverySpawner({ models: [] });
227 const snapshot = yield* checkPiProviderStatus(settings).pipe(
228 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
229 );
230 assert.equal(snapshot.status, "warning");
231 assert.equal(snapshot.auth.status, "unauthenticated");
232 assert.deepEqual(
233 snapshot.models.map((model) => model.slug),
234 ["default"],
235 );
236 }).pipe(Effect.provide(NodeServices.layer)),
237 );
238
239 it.effect.each([
240 {
241 label: "a pinned model without credentials",
242 version: "1.4.0",
243 activeModel: { provider: "openai", id: "gpt-5" },
244 authCheck: '{"status":"not_ready","provider":"openai","reason":"credentials_not_configured"}',
245 message: "Pi default (openai/gpt-5) has no credentials configured for openai.",
246 },
247 {
248 label: "a pinned model with a runtime --api-key",
249 version: "1.4.0",
250 launchArgs: "--provider openai --model gpt-5 --api-key fake-key",
251 activeModel: { provider: "openai", id: "gpt-5" },
252 authCheck: undefined,
253 message: "Pi default (openai/gpt-5) uses openai, which has no available models in Pi.",
254 },
255 {
256 label: "a pinned model whose provider auth check cannot find",
257 version: "1.4.0",
258 activeModel: { provider: "openai", id: "gpt-5" },
259 authCheck: '{"status":"not_ready","provider":"openai","reason":"provider_not_found"}',
260 message: "Pi default (openai/gpt-5) uses openai, which has no available models in Pi.",
261 },
262 {
263 label: "a pinned model on Pi without auth check",
264 version: MINIMUM_PI_VERSION,
265 activeModel: { provider: "openai", id: "gpt-5" },
266 authCheck: undefined,
267 message: "Pi default (openai/gpt-5) uses openai, which has no available models in Pi.",
268 },
269 {
270 label: "an uncatalogued model id of a provider with credentials",
271 version: "1.4.0",
272 activeModel: { provider: "anthropic", id: "claude-next" },
273 authCheck: undefined,
274 message: undefined,
275 },
276 {
277 label: "an extension model Pi lists as available",
278 version: "1.4.0",
279 activeModel: { provider: "router", id: "auto" },
280 authCheck: undefined,
281 message: undefined,
282 },
283 ])("diagnoses Pi's active model readiness for $label", (row) =>
284 Effect.gen(function* () {
285 const spawner = yield* piDiscoverySpawner(
286 {
287 models: [
288 { provider: "anthropic", id: "claude-sonnet-4" },
289 { provider: "router", id: "auto" },
290 ],
291 },
292 undefined,
293 Effect.void,
294 { version: row.version, state: { model: row.activeModel }, authCheck: row.authCheck },
295 );
296 const snapshot = yield* checkPiProviderStatus({
297 ...settings,
298 launchArgs: row.launchArgs ?? "",
299 }).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
300 assert.equal(snapshot.status, "ready");
301 assert.equal(snapshot.auth.status, "authenticated");
302 if (row.message === undefined) assert.isUndefined(snapshot.message);
303 else assert.include(snapshot.message ?? "", row.message);
304 }).pipe(Effect.provide(NodeServices.layer)),
305 );
306
307 it.effect("marks models that only load with Pi extensions as unusable for text generation", () =>
308 Effect.gen(function* () {
309 const spawner = yield* piDiscoverySpawner(
310 {
311 models: [
312 { provider: "anthropic", id: "claude-sonnet-4" },
313 { provider: "demo", id: "demo-model" },
314 ],
315 },
316 { models: [{ provider: "anthropic", id: "claude-sonnet-4" }] },
317 );
318 const snapshot = yield* checkPiProviderStatus(settings).pipe(
319 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
320 );
321 assert.deepEqual(
322 snapshot.models.map((model) => [model.slug, model.supportsTextGeneration]),
323 [
324 ["default", undefined],
325 ["anthropic/claude-sonnet-4", undefined],
326 ["demo/demo-model", false],
327 ],
328 );
329 assert.equal(snapshot.supportsTextGeneration, undefined);
330 }).pipe(Effect.provide(NodeServices.layer)),
331 );
332
333 it.effect.each([
334 { label: "exits before answering", extensionFreeInventory: exitsWithoutExtensions },
335 { label: "lists no models", extensionFreeInventory: { models: [] } },
336 ])("makes Pi unusable for text generation when Pi without extensions $label", (row) =>
337 Effect.gen(function* () {
338 const spawner = yield* piDiscoverySpawner(
339 { models: [{ provider: "demo", id: "extension-model" }] },
340 row.extensionFreeInventory,
341 );
342 const snapshot = yield* checkPiProviderStatus({
343 ...settings,
344 launchArgs: "--provider demo --model extension-model",
345 customModels: ["local/custom-alias"],
346 }).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
347 assert.equal(snapshot.status, "ready");
348 assert.equal(snapshot.auth.status, "authenticated");
349 // Pi default and custom models still run in chat, so the picker keeps them.
350 assert.deepEqual(
351 snapshot.models.map((model) => model.slug),
352 ["default", "demo/extension-model", "local/custom-alias"],
353 );
354 assert.equal(snapshot.supportsTextGeneration, false);
355 }).pipe(Effect.provide(NodeServices.layer)),
356 );
357
358 it.effect("preserves normal discovery when the extension-free model listing times out", () =>
359 Effect.gen(function* () {
360 const modelRequested = yield* Deferred.make<void>();
361 const spawner = yield* piDiscoverySpawner(
362 { models: [{ provider: "demo", id: "demo-model" }] },
363 null,
364 Deferred.succeed(modelRequested, undefined).pipe(Effect.asVoid),
365 );
366 const fiber = yield* checkPiProviderStatus(settings).pipe(
367 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
368 Effect.forkChild,
369 );
370 yield* Deferred.await(modelRequested);
371 yield* TestClock.adjust("5 seconds");
372 const snapshot = yield* Fiber.join(fiber);
373 assert.equal(snapshot.status, "ready");
374 assert.equal(snapshot.auth.status, "authenticated");
375 assert.deepEqual(
376 snapshot.models.map((model) => [model.slug, model.supportsTextGeneration]),
377 [
378 ["default", undefined],
379 ["demo/demo-model", undefined],
380 ],
381 );
382 assert.equal(snapshot.supportsTextGeneration, undefined);
383 assert.include(
384 snapshot.slashCommands.map((command) => command.name),
385 "compact",
386 );
387 }).pipe(Effect.provide(NodeServices.layer)),
388 );
389
390 it.effect("advertises a model's service tiers beside its thinking levels", () =>
391 Effect.gen(function* () {
392 const spawner = yield* piDiscoverySpawner({
393 models: [
394 {
395 provider: "columbia-codex",
396 id: "gpt-6.1-sol",
397 reasoning: true,
398 serviceTiers: [
399 { id: "priority", name: "Fast", description: "2x speed, increased usage" },
400 { id: "ultrafast", name: "Ultrafast" },
401 ],
402 },
403 { provider: "anthropic", id: "claude-sonnet-4", reasoning: true },
404 ],
405 });
406 const snapshot = yield* checkPiProviderStatus(settings).pipe(
407 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
408 );
409 const descriptors = (slug: string) =>
410 snapshot.models.find((model) => model.slug === slug)?.capabilities?.optionDescriptors ?? [];
411 assert.deepEqual(
412 descriptors("columbia-codex/gpt-6.1-sol").map((descriptor) => descriptor.id),
413 ["thinking", "serviceTier"],
414 );
415 assert.deepEqual(descriptors("columbia-codex/gpt-6.1-sol")[1], {
416 id: "serviceTier",
417 label: "Service Tier",
418 type: "select",
419 options: [
420 { id: "default", label: "Standard", isDefault: true },
421 { id: "priority", label: "Fast", description: "2x speed, increased usage" },
422 { id: "ultrafast", label: "Ultrafast" },
423 ],
424 currentValue: "default",
425 });
426 assert.deepEqual(
427 descriptors("anthropic/claude-sonnet-4").map((descriptor) => descriptor.id),
428 ["thinking"],
429 );
430 }).pipe(Effect.provide(NodeServices.layer)),
431 );
432
433 // regression: #18498 a malformed model inventory must not restore stale Pi models
434 it.effect.each([
435 ["missing models", {}],
436 ["missing model id", { models: [{ provider: "anthropic" }] }],
437 [
438 "mixed valid and invalid models",
439 { models: [{ provider: "anthropic", id: "claude-sonnet-4" }, { provider: "openrouter" }] },
440 ],
441 ] as const)("reports unknown authentication when Pi discovery returns %s", (_label, inventory) =>
442 Effect.gen(function* () {
443 const spawner = yield* piDiscoverySpawner(inventory);
444 const snapshot = yield* checkPiProviderStatus(settings).pipe(
445 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
446 );
447 assert.equal(snapshot.status, "ready");
448 assert.equal(snapshot.auth.status, "unknown");
449 }).pipe(Effect.provide(NodeServices.layer)),
450 );
451});