1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it } from "@effect/vitest";
3import { DEFAULT_SERVER_SETTINGS, ProviderInstanceId } from "@t3tools/contracts";
4import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Queue from "effect/Queue";
8import * as Schema from "effect/Schema";
9import * as Sink from "effect/Sink";
10import * as Stream from "effect/Stream";
11import * as HttpClient from "effect/http/HttpClient";
12import * as ChildProcess from "effect/process/ChildProcess";
13import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
14
15import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions";
16import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
17import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
18import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
19import type { PiSettings } from "../settings.ts";
20import { PiDriver } from "./driver.ts";
21
22const layerTest = Layer.mergeAll(
23 TestProviderHost.layer({
24 cwd: "/machine",
25 settings: { ...DEFAULT_SERVER_SETTINGS, enableProviderUpdateChecks: false },
26 runBackgroundWork: false,
27 }),
28 IdAllocator.layer,
29 McpProviderSessions.layer,
30 ProviderLatestVersions.layer,
31 Layer.succeed(
32 HttpClient.HttpClient,
33 HttpClient.make(() => Effect.die("Unexpected HTTP")),
34 ),
35).pipe(Layer.provideMerge(NodeServices.layer));
36
37const decodeRequest = Schema.decodeSync(
38 Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)),
39);
40const encoder = new TextEncoder();
41const personalSkill = {
42 name: "skill:personal",
43 source: "skill",
44 sourceInfo: { scope: "user", path: "/home/.pi/agent/skills/personal/SKILL.md" },
45};
46
47const makePiSpawner = Effect.fnUntraced(function* (
48 discovery: { readonly state?: unknown; readonly models?: ReadonlyArray<unknown> } = {},
49) {
50 const pendingCommand = yield* Deferred.make<void>();
51 const launches: Array<ChildProcess.StandardCommand> = [];
52 const spawner = ChildProcessSpawner.make((command) =>
53 Effect.gen(function* () {
54 assert.isTrue(ChildProcess.isStandardCommand(command));
55 if (!ChildProcess.isStandardCommand(command)) return yield* Effect.die("Unexpected pipeline");
56 launches.push(command);
57 const version = command.args.includes("--version");
58 const stdout = yield* Queue.unbounded<Uint8Array>();
59 const cwd = command.options.cwd;
60 return ChildProcessSpawner.makeHandle({
61 // Outside the valid PID range, so transport cleanup cannot signal a real process.
62 pid: ChildProcessSpawner.ProcessId(999_999_999),
63 exitCode: version ? Effect.succeed(ChildProcessSpawner.ExitCode(0)) : Effect.never,
64 isRunning: Effect.succeed(!version),
65 kill: () => Effect.void,
66 unref: Effect.succeed(Effect.void),
67 stdin: Sink.forEach((chunk: Uint8Array) => {
68 const request = decodeRequest(new TextDecoder().decode(chunk).trim());
69 if (cwd === "/pending" && request.type === "get_commands") {
70 return Deferred.succeed(pendingCommand, undefined).pipe(Effect.asVoid);
71 }
72 const failed = cwd === "/failed" && request.type === "get_commands";
73 const data =
74 request.type === "get_commands"
75 ? {
76 commands: [
77 personalSkill,
78 ...(cwd === "/machine"
79 ? []
80 : [
81 {
82 name: `skill:${cwd?.slice(1)}`,
83 source: "skill",
84 sourceInfo: {
85 scope: "project",
86 path: `${cwd}/.agents/skills/SKILL.md`,
87 },
88 },
89 { name: `prompt-${cwd?.slice(1)}`, source: "prompt" },
90 ]),
91 ],
92 }
93 : request.type === "get_available_models"
94 ? { models: discovery.models ?? [{ provider: "test", id: "model" }] }
95 : request.type === "get_state"
96 ? (discovery.state ?? {})
97 : {};
98 return Queue.offer(
99 stdout,
100 encoder.encode(
101 `${JSON.stringify({ type: "response", id: request.id, success: !failed, data, ...(failed ? { error: "commands unavailable" } : {}) })}\n`,
102 ),
103 ).pipe(Effect.asVoid);
104 }),
105 stdout: version ? Stream.succeed(encoder.encode("pi 1.0.2\n")) : Stream.fromQueue(stdout),
106 stderr: Stream.empty,
107 all: Stream.empty,
108 getInputFd: () => Sink.drain,
109 getOutputFd: () => Stream.empty,
110 });
111 }),
112 );
113 return { spawner, launches, pendingCommand };
114});
115
116const create = (config: Partial<PiSettings> = {}, enabled = true) =>
117 PiDriver.create({
118 instanceId: ProviderInstanceId.make("pi-workspace-test"),
119 displayName: "My Pi",
120 accentColor: "#abcdef",
121 environment: [{ name: "PI_CODING_AGENT_DIR", value: "/isolated-pi", sensitive: false }],
122 enabled,
123 config: { ...PiDriver.defaultConfig(), binaryPath: "custom-pi", ...config },
124 });
125
126it.layer(layerTest)("PiDriver workspace discovery", (it) => {
127 it.effect("keeps each workspace's skills and commands separate from the machine catalog", () =>
128 Effect.gen(function* () {
129 const { spawner, launches } = yield* makePiSpawner();
130 const instance = yield* create({ launchArgs: '--approve --skill "extra skill"' }).pipe(
131 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
132 );
133 yield* instance.snapshot.refresh;
134 assert.isDefined(instance.snapshotForCwd);
135 const [first, second] = yield* Effect.all(
136 [instance.snapshotForCwd!("/first"), instance.snapshotForCwd!("/second")],
137 { concurrency: "unbounded" },
138 );
139 assert.deepEqual(
140 first.skills.map((skill) => skill.name),
141 ["personal", "first"],
142 );
143 assert.deepEqual(
144 second.skills.map((skill) => skill.name),
145 ["personal", "second"],
146 );
147 assert.deepEqual(
148 first.slashCommands.map((command) => command.name),
149 ["compact", "prompt-first"],
150 );
151 const machine = yield* instance.snapshot.getSnapshot;
152 assert.deepEqual(
153 machine.skills.map((skill) => skill.name),
154 ["personal"],
155 );
156 assert.deepEqual(
157 machine.slashCommands.map((command) => command.name),
158 ["compact"],
159 );
160 assert.equal(first.instanceId, instance.instanceId);
161 assert.equal(first.displayName, "My Pi");
162 assert.equal(first.accentColor, "#abcdef");
163 assert.deepEqual(first.models, machine.models);
164 const workspaceLaunch = launches.find((launch) => launch.options.cwd === "/first");
165 assert.isDefined(workspaceLaunch);
166 assert.equal(workspaceLaunch!.command, "custom-pi");
167 assert.includeMembers(
168 [...workspaceLaunch!.args],
169 ["--approve", "--skill", "extra skill", "--no-session"],
170 );
171 assert.equal(workspaceLaunch!.options.env?.PI_CODING_AGENT_DIR, "/isolated-pi");
172 // Discovery processes are stopped as soon as they answer, so neither may start Pi's
173 // network refresh at launch.
174 for (const cwd of ["/machine", "/first"]) {
175 const discovery = launches.find(
176 (launch) => launch.options.cwd === cwd && launch.args.includes("--mode"),
177 );
178 assert.deepEqual(
179 discovery?.args.filter((arg) => arg === "--offline"),
180 ["--offline"],
181 );
182 }
183 }),
184 );
185
186 it.effect("does not run a disabled provider's workspace probe", () =>
187 Effect.gen(function* () {
188 const instance = yield* create({}, false).pipe(
189 Effect.provideService(
190 ChildProcessSpawner.ChildProcessSpawner,
191 ChildProcessSpawner.make(() => Effect.die("Disabled Pi must not spawn")),
192 ),
193 );
194 assert.isDefined(instance.snapshotForCwd);
195 const workspace = yield* instance.snapshotForCwd!("/first");
196 assert.isFalse(workspace.enabled);
197 assert.deepEqual(workspace.skills, []);
198 }),
199 );
200
201 it.effect.each([true, false])(
202 "limits discovered thinking defaults to the known active model, known=%s",
203 (known) =>
204 Effect.gen(function* () {
205 const { spawner } = yield* makePiSpawner({
206 state: {
207 thinkingLevel: "high",
208 ...(known ? { model: { provider: "test", id: "model" } } : {}),
209 },
210 models: [
211 { provider: "test", id: "model", reasoning: true },
212 { provider: "other", id: "model", reasoning: true },
213 { provider: "test", id: "other", reasoning: true },
214 ],
215 });
216 const instance = yield* create().pipe(
217 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
218 );
219 yield* instance.snapshot.refresh;
220 const snapshot = yield* instance.snapshot.getSnapshot;
221 assert.deepEqual(
222 snapshot.models.map((model) => [
223 model.slug,
224 (model.capabilities?.optionDescriptors ?? []).flatMap((descriptor) =>
225 descriptor.type === "select"
226 ? descriptor.options.filter((option) => option.isDefault).map((option) => option.id)
227 : [],
228 ),
229 ]),
230 [
231 ["default", []],
232 ["test/model", known ? ["high"] : []],
233 ["other/model", []],
234 ["test/other", []],
235 ],
236 );
237 }),
238 );
239
240 it.effect(
241 "fails command discovery instead of returning an empty successful workspace catalog",
242 () =>
243 Effect.gen(function* () {
244 const { spawner } = yield* makePiSpawner();
245 const instance = yield* create().pipe(
246 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
247 );
248 yield* instance.snapshot.refresh;
249 assert.isDefined(instance.snapshotForCwd);
250 const error = yield* Effect.flip(instance.snapshotForCwd!("/failed"));
251 assert.equal(error._tag, "ProviderDriverError");
252 assert.equal(error.instanceId, instance.instanceId);
253 assert.deepEqual(
254 (yield* instance.snapshot.getSnapshot).skills.map((skill) => skill.name),
255 ["personal"],
256 );
257 }),
258 );
259});