authorgravatar for 71498452+StiensWout@users.noreply.github.comWout Stiens <71498452+StiensWout@users.noreply.github.com> 2026-10-08 20:56:27+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-08 13:56:27-05:00
log0b6ec43c54165a87941b7a070849be8979646040
tree1650dd1e05240c392462a4bfe33a03de3ffc0cac
parenta0066d5cb7d18abd786c1c23bf835f898490fd37
signature Signed by PGP key B5690EEEBB952194

fix(server): Pi discovers workspace skills and commands (#17190)

3 files changed, 298 insertions(+), 18 deletions(-)

apps/server/src/provider/Drivers/PiDriver.test.ts created+219
...@@ -0,0 +1,219 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it } from "@effect/vitest";
3import { ProviderInstanceId, type PiSettings } from "@t3tools/contracts";
4import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";
6import * as Fiber from "effect/Fiber";
7import * as Layer from "effect/Layer";
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 TestClock from "effect/testing/TestClock";
13import { HttpClient } from "effect/http";
14import { ChildProcess, ChildProcessSpawner } from "effect/process";
15
16import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts";
17import * as ServerConfig from "../../config.ts";
18import * as IdAllocator from "../../orchestration-v2/IdAllocator.ts";
19import * as ServerSettings from "../../serverSettings.ts";
20import { PiDriver } from "./PiDriver.ts";
21
22const layerTest = Layer.mergeAll(
23 ServerConfig.layerTest("/machine", { prefix: "t3-pi-driver-" }),
24 IdAllocator.layer,
25 ServerSettings.layerTest({ enableProviderUpdateChecks: false }),
26 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
27 shouldRunScopeWork: () => Effect.succeed(false),
28 }),
29 Layer.succeed(
30 HttpClient.HttpClient,
31 HttpClient.make(() => Effect.die("Unexpected HTTP")),
32 ),
33).pipe(Layer.provideMerge(NodeServices.layer));
34
35const decodeRequest = Schema.decodeSync(
36 Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)),
37);
38const encoder = new TextEncoder();
39const personalSkill = {
40 name: "skill:personal",
41 source: "skill",
42 sourceInfo: { scope: "user", path: "/home/.pi/agent/skills/personal/SKILL.md" },
43};
44
45// Respond through the real stdio transport, with a distinct command catalog for each cwd.
46const makePiSpawner = Effect.gen(function* () {
47 const pendingCommand = yield* Deferred.make<void>();
48 const launches: Array<ChildProcess.StandardCommand> = [];
49 const spawner = ChildProcessSpawner.make((command) =>
50 Effect.gen(function* () {
51 assert.isTrue(ChildProcess.isStandardCommand(command));
52 if (!ChildProcess.isStandardCommand(command)) return yield* Effect.die("Unexpected pipeline");
53 launches.push(command);
54 const version = command.args.includes("--version");
55 const stdout = yield* Queue.unbounded<Uint8Array>();
56 const cwd = command.options.cwd;
57 return ChildProcessSpawner.makeHandle({
58 // Outside the valid PID range, so transport cleanup cannot signal a real process.
59 pid: ChildProcessSpawner.ProcessId(999_999_999),
60 exitCode: version ? Effect.succeed(ChildProcessSpawner.ExitCode(0)) : Effect.never,
61 isRunning: Effect.succeed(!version),
62 kill: () => Effect.void,
63 unref: Effect.succeed(Effect.void),
64 stdin: Sink.forEach((chunk: Uint8Array) => {
65 const request = decodeRequest(new TextDecoder().decode(chunk).trim());
66 if (cwd === "/pending" && request.type === "get_commands") {
67 return Deferred.succeed(pendingCommand, undefined).pipe(Effect.asVoid);
68 }
69 const failed = cwd === "/failed" && request.type === "get_commands";
70 const data =
71 request.type === "get_commands"
72 ? {
73 commands: [
74 personalSkill,
75 ...(cwd === "/machine"
76 ? []
77 : [
78 {
79 name: `skill:${cwd?.slice(1)}`,
80 source: "skill",
81 sourceInfo: {
82 scope: "project",
83 path: `${cwd}/.agents/skills/SKILL.md`,
84 },
85 },
86 { name: `prompt-${cwd?.slice(1)}`, source: "prompt" },
87 ]),
88 ],
89 }
90 : request.type === "get_available_models"
91 ? { models: [{ provider: "test", id: "model" }] }
92 : {};
93 return Queue.offer(
94 stdout,
95 encoder.encode(
96 `${JSON.stringify({ type: "response", id: request.id, success: !failed, data, ...(failed ? { error: "commands unavailable" } : {}) })}\n`,
97 ),
98 ).pipe(Effect.asVoid);
99 }),
100 stdout: version ? Stream.succeed(encoder.encode("pi 1.0.2\n")) : Stream.fromQueue(stdout),
101 stderr: Stream.empty,
102 all: Stream.empty,
103 getInputFd: () => Sink.drain,
104 getOutputFd: () => Stream.empty,
105 });
106 }),
107 );
108 return { spawner, launches, pendingCommand };
109});
110
111const create = (config: Partial<PiSettings> = {}, enabled = true) =>
112 PiDriver.create({
113 instanceId: ProviderInstanceId.make("pi-workspace-test"),
114 displayName: "My Pi",
115 accentColor: "#abcdef",
116 environment: [{ name: "PI_CODING_AGENT_DIR", value: "/isolated-pi", sensitive: false }],
117 enabled,
118 config: { ...PiDriver.defaultConfig(), binaryPath: "custom-pi", ...config },
119 });
120
121it.layer(layerTest)("PiDriver workspace discovery", (it) => {
122 it.effect("keeps each workspace's skills and commands separate from the machine catalog", () =>
123 Effect.gen(function* () {
124 const { spawner, launches } = yield* makePiSpawner;
125 const instance = yield* create({ launchArgs: '--approve --skill "extra skill"' }).pipe(
126 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
127 );
128 yield* instance.snapshot.refresh;
129 assert.isDefined(instance.snapshotForCwd);
130 const [first, second] = yield* Effect.all(
131 [instance.snapshotForCwd!("/first"), instance.snapshotForCwd!("/second")],
132 { concurrency: "unbounded" },
133 );
134 assert.deepEqual(
135 first.skills.map((skill) => skill.name),
136 ["personal", "first"],
137 );
138 assert.deepEqual(
139 second.skills.map((skill) => skill.name),
140 ["personal", "second"],
141 );
142 assert.deepEqual(
143 first.slashCommands.map((command) => command.name),
144 ["compact", "prompt-first"],
145 );
146 const machine = yield* instance.snapshot.getSnapshot;
147 assert.deepEqual(
148 machine.skills.map((skill) => skill.name),
149 ["personal"],
150 );
151 assert.deepEqual(
152 machine.slashCommands.map((command) => command.name),
153 ["compact"],
154 );
155 assert.equal(first.instanceId, instance.instanceId);
156 assert.equal(first.displayName, "My Pi");
157 assert.equal(first.accentColor, "#abcdef");
158 assert.deepEqual(first.models, machine.models);
159 const workspaceLaunch = launches.find((launch) => launch.options.cwd === "/first");
160 assert.isDefined(workspaceLaunch);
161 assert.equal(workspaceLaunch!.command, "custom-pi");
162 assert.includeMembers(
163 [...workspaceLaunch!.args],
164 ["--approve", "--skill", "extra skill", "--no-session"],
165 );
166 assert.equal(workspaceLaunch!.options.env?.PI_CODING_AGENT_DIR, "/isolated-pi");
167 }).pipe(Effect.scoped),
168 );
169
170 it.effect("does not run a disabled provider's workspace probe", () =>
171 Effect.gen(function* () {
172 const instance = yield* create({}, false).pipe(
173 Effect.provideService(
174 ChildProcessSpawner.ChildProcessSpawner,
175 ChildProcessSpawner.make(() => Effect.die("Disabled Pi must not spawn")),
176 ),
177 );
178 assert.isDefined(instance.snapshotForCwd);
179 const workspace = yield* instance.snapshotForCwd!("/first");
180 assert.isFalse(workspace.enabled);
181 assert.deepEqual(workspace.skills, []);
182 }).pipe(Effect.scoped),
183 );
184
185 it.effect(
186 "fails command discovery instead of returning an empty successful workspace catalog",
187 () =>
188 Effect.gen(function* () {
189 const { spawner } = yield* makePiSpawner;
190 const instance = yield* create().pipe(
191 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
192 );
193 yield* instance.snapshot.refresh;
194 assert.isDefined(instance.snapshotForCwd);
195 const error = yield* Effect.flip(instance.snapshotForCwd!("/failed"));
196 assert.equal(error._tag, "ProviderDriverError");
197 assert.equal(error.instanceId, instance.instanceId);
198 assert.deepEqual(
199 (yield* instance.snapshot.getSnapshot).skills.map((skill) => skill.name),
200 ["personal"],
201 );
202 }).pipe(Effect.scoped),
203 );
204
205 it.effect("times out workspace discovery that needs interactive input", () =>
206 Effect.gen(function* () {
207 const { spawner, pendingCommand } = yield* makePiSpawner;
208 const instance = yield* create().pipe(
209 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
210 );
211 const probe = yield* instance.snapshotForCwd!("/pending").pipe(Effect.flip, Effect.forkChild);
212 yield* Deferred.await(pendingCommand);
213 yield* TestClock.adjust("15 seconds");
214 const error = yield* Fiber.join(probe);
215 assert.equal(error._tag, "ProviderDriverError");
216 assert.include(error.detail, "workspace commands");
217 }).pipe(Effect.scoped),
218 );
219});
apps/server/src/provider/Drivers/PiDriver.ts+21
...@@ -26,6 +26,7 @@ import { ProviderDriverError } from "../Errors.ts";...@@ -26,6 +26,7 @@ import { ProviderDriverError } from "../Errors.ts";
26import {26import {
27 buildInitialPiProviderSnapshot,27 buildInitialPiProviderSnapshot,
28 checkPiProviderStatus,28 checkPiProviderStatus,
29 discoverPiCommandsForCwd,
29 enrichPiSnapshot,30 enrichPiSnapshot,
30} from "../PiProvider.ts";31} from "../PiProvider.ts";
31import { makeManagedServerProvider } from "../makeManagedServerProvider.ts";32import { makeManagedServerProvider } from "../makeManagedServerProvider.ts";
...@@ -188,6 +189,26 @@ export const PiDriver: ProviderDriver<PiSettings, PiDriverEnv> = {...@@ -188,6 +189,26 @@ export const PiDriver: ProviderDriver<PiSettings, PiDriverEnv> = {
188 accentColor,189 accentColor,
189 enabled,190 enabled,
190 snapshot,191 snapshot,
192 snapshotForCwd: (workspaceCwd) =>
193 !effectiveConfig.enabled
194 ? snapshot.getSnapshot
195 : Effect.all([
196 snapshot.getSnapshot,
197 discoverPiCommandsForCwd(effectiveConfig, processEnv, workspaceCwd).pipe(
198 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
199 Effect.mapError(
200 (cause) =>
201 new ProviderDriverError({
202 driver: DRIVER_KIND,
203 instanceId,
204 detail: "Failed to discover Pi workspace commands.",
205 cause,
206 }),
207 ),
208 ),
209 ]).pipe(
210 Effect.map(([machineSnapshot, commands]) => ({ ...machineSnapshot, ...commands })),
211 ),
191 orchestrationAdapter,212 orchestrationAdapter,
192 textGeneration,213 textGeneration,
193 } satisfies ProviderInstance;214 } satisfies ProviderInstance;
apps/server/src/provider/PiProvider.ts+58-18
...@@ -32,6 +32,8 @@ import {...@@ -32,6 +32,8 @@ import {
32} from "../orchestration-v2/Adapters/piT3McpInjection.ts";32} from "../orchestration-v2/Adapters/piT3McpInjection.ts";
33import {33import {
34 makePiRpcConnection,34 makePiRpcConnection,
35 PiRpcError,
36 PiRpcTimeoutError,
35 piRecordField as recordField,37 piRecordField as recordField,
36 piRecordString as recordString,38 piRecordString as recordString,
37} from "../orchestration-v2/Adapters/PiRpc.ts";39} from "../orchestration-v2/Adapters/PiRpc.ts";
...@@ -126,6 +128,33 @@ function parseDiscoveredModels(...@@ -126,6 +128,33 @@ function parseDiscoveredModels(
126 return parsed;128 return parsed;
127}129}
128130
131const makePiDiscoveryConnection = Effect.fnUntraced(function* (
132 piSettings: PiSettings,
133 environment: NodeJS.ProcessEnv,
134 launchArgs: ReadonlyArray<string>,
135 cwd?: string,
136) {
137 const launch = buildPiRpcLaunch({
138 launchArgs,
139 environment,
140 mcpSession: undefined,
141 extensionPath: undefined,
142 ephemeral: true,
143 });
144 const connection = yield* makePiRpcConnection({
145 command: piSettings.binaryPath || "pi",
146 args: launch.args,
147 cwd,
148 env: launch.env,
149 });
150 yield* Stream.fromQueue(connection.events).pipe(
151 Stream.runDrain,
152 Effect.ignore,
153 Effect.forkScoped,
154 );
155 return connection;
156});
157
129const discoverPiViaRpc = (158const discoverPiViaRpc = (
130 piSettings: PiSettings,159 piSettings: PiSettings,
131 environment: NodeJS.ProcessEnv,160 environment: NodeJS.ProcessEnv,
...@@ -133,24 +162,7 @@ const discoverPiViaRpc = (...@@ -133,24 +162,7 @@ const discoverPiViaRpc = (
133 cwd?: string,162 cwd?: string,
134) =>163) =>
135 Effect.gen(function* () {164 Effect.gen(function* () {
136 const launch = buildPiRpcLaunch({165 const connection = yield* makePiDiscoveryConnection(piSettings, environment, launchArgs, cwd);
137 launchArgs,
138 environment,
139 mcpSession: undefined,
140 extensionPath: undefined,
141 ephemeral: true,
142 });
143 const connection = yield* makePiRpcConnection({
144 command: piSettings.binaryPath || "pi",
145 args: launch.args,
146 cwd,
147 env: launch.env,
148 });
149 yield* Stream.fromQueue(connection.events).pipe(
150 Stream.runDrain,
151 Effect.ignore,
152 Effect.forkScoped,
153 );
154 const stateData = yield* connection.request({ type: "get_state" });166 const stateData = yield* connection.request({ type: "get_state" });
155 const modelsData = yield* connection.request({ type: "get_available_models" });167 const modelsData = yield* connection.request({ type: "get_available_models" });
156 const commandsData = yield* connection168 const commandsData = yield* connection
...@@ -169,6 +181,34 @@ const discoverPiViaRpc = (...@@ -169,6 +181,34 @@ const discoverPiViaRpc = (
169 } satisfies PiDiscovery;181 } satisfies PiDiscovery;
170 }).pipe(Effect.scoped);182 }).pipe(Effect.scoped);
171183
184/** Probe the full command catalog Pi exposes in a workspace without changing machine health. */
185export const discoverPiCommandsForCwd = Effect.fn("discoverPiCommandsForCwd")(
186 function* (piSettings: PiSettings, environment: NodeJS.ProcessEnv, cwd: string) {
187 const launchArgs = resolvePiLaunchArgs(piSettings.launchArgs);
188 if (!launchArgs.ok) {
189 return yield* new PiRpcError({ operation: "launch", detail: launchArgs.message });
190 }
191 const connection = yield* makePiDiscoveryConnection(
192 piSettings,
193 environment,
194 launchArgs.args,
195 cwd,
196 );
197 // A failed read must not replace a previously usable workspace catalog with an empty one.
198 const commandsData = yield* connection.request({ type: "get_commands" });
199 const { slashCommands, skills } = parsePiDiscoveredCommands(commandsData);
200 return { slashCommands: withPiBuiltinSlashCommands(slashCommands), skills };
201 },
202 Effect.scoped,
203 Effect.timeoutOrElse({
204 duration: PI_RPC_DISCOVERY_TIMEOUT_MS,
205 orElse: () =>
206 Effect.fail(
207 new PiRpcTimeoutError({ operation: "discovery", timeoutMs: PI_RPC_DISCOVERY_TIMEOUT_MS }),
208 ),
209 }),
210);
211
172const runPiVersionCommand = (piSettings: PiSettings, environment: NodeJS.ProcessEnv) =>212const runPiVersionCommand = (piSettings: PiSettings, environment: NodeJS.ProcessEnv) =>
173 Effect.gen(function* () {213 Effect.gen(function* () {
174 const command = piSettings.binaryPath || "pi";214 const command = piSettings.binaryPath || "pi";