Open History to see the last commit
1// @effect-diagnostics nodeBuiltinImport:off globalTimers:off - the eval owns a server subprocess and a scratch workspace.
2/**
3 * An isolated T3 server for one device run. Its only provider is `evalbot`, an Fx instance
4 * backed by `fake-fx.ts`, so every reply is scripted. State lives under the run's `--out`.
5 */
6import * as NodeChildProcess from "node:child_process";
7import * as NodeFS from "node:fs";
8import * as NodeNet from "node:net";
9import * as NodePath from "node:path";
10import * as NodeWS from "@effect/platform-node/NodeWS";
11import {
12 CommandId,
13 EnvironmentId,
14 ExecutionEnvironmentDescriptor,
15 MessageId,
16 ORCHESTRATION_PROTOCOL_HEADER,
17 ORCHESTRATION_PROTOCOL_VERSION,
18 OrchestrationV2ShellSnapshot,
19 ProjectId,
20 ProviderInstanceId,
21 ThreadId,
22 WsRpcGroup,
23 type OrchestrationV2ThreadProjection,
24} from "@t3tools/contracts";
25import * as SecureRandom from "effect/crypto/SecureRandom";
26import * as Effect from "effect/Effect";
27import * as Layer from "effect/Layer";
28import * as Schema from "effect/Schema";
29import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http";
30import { RpcClient, RpcSerialization } from "effect/rpc";
31import { Socket } from "effect/socket";
32
33const REPO_ROOT = NodePath.resolve(import.meta.dirname, "../..");
34export const EVAL_INSTANCE = ProviderInstanceId.make("evalbot");
35export const EVAL_MODEL = "eval/scripted";
36
37export const uuid = Effect.flatMap(SecureRandom.SecureRandom, (random) => random.uuidV4());
38
39export class EnvironmentError extends Schema.TaggedError<EnvironmentError>()("EnvironmentError", {
40 message: Schema.String,
41}) {}
42
43export const freePort = () =>
44 Effect.callback<number>((resume) => {
45 const server = NodeNet.createServer();
46 server.listen(0, () => {
47 const address = server.address();
48 server.close(() =>
49 resume(Effect.succeed(typeof address === "object" && address ? address.port : 0)),
50 );
51 });
52 });
53
54const waitForPort = Effect.fn("environment.waitForPort")(function* (port: number) {
55 for (let attempt = 0; attempt < 120; attempt++) {
56 const open = yield* Effect.callback<boolean>((resume) => {
57 const socket = NodeNet.connect(port, "127.0.0.1");
58 socket.once("connect", () => (socket.destroy(), resume(Effect.succeed(true))));
59 socket.once("error", () => resume(Effect.succeed(false)));
60 });
61 if (open) return;
62 yield* Effect.sleep("500 millis");
63 }
64 return yield* new EnvironmentError({ message: `The server never listened on ${port}.` });
65});
66
67/** Runs a server CLI command and returns both output streams. */
68const runCli = (args: ReadonlyArray<string>) =>
69 Effect.try({
70 try: () => {
71 const result = NodeChildProcess.spawnSync("node", ["apps/server/src/bin.ts", ...args], {
72 cwd: REPO_ROOT,
73 encoding: "utf8",
74 });
75 if (result.status !== 0) throw new Error(result.stderr);
76 return { stdout: result.stdout, stderr: result.stderr };
77 },
78 catch: (cause) =>
79 new EnvironmentError({ message: `bin.ts ${args.join(" ")} failed: ${cause}` }),
80 });
81
82function writeScratch(baseDir: string) {
83 const workspace = NodePath.join(baseDir, "workspace/acme-app");
84 const control = NodePath.join(baseDir, "control");
85 NodeFS.mkdirSync(NodePath.join(baseDir, "userdata"), { recursive: true });
86 NodeFS.mkdirSync(control, { recursive: true });
87 NodeFS.mkdirSync(workspace, { recursive: true });
88 NodeFS.writeFileSync(NodePath.join(workspace, "README.md"), "# acme app\n");
89 // The / menu's pickable row; without it, fx finds only skills in folders above `--out`.
90 const skill = NodePath.join(workspace, ".fx/skills/eval-notes");
91 NodeFS.mkdirSync(skill, { recursive: true });
92 NodeFS.writeFileSync(
93 NodePath.join(skill, "SKILL.md"),
94 "---\nname: eval-notes\ndescription: Takes notes for the eval.\n---\nTake notes.\n",
95 );
96 const git = (...args: string[]) =>
97 NodeChildProcess.execFileSync("git", ["-C", workspace, ...args], { stdio: "ignore" });
98 git("init", "-q");
99 git("add", "-A");
100 git("-c", "user.name=eval", "-c", "user.email=eval@example.com", "commit", "-qm", "init");
101 const launcher = NodePath.join(baseDir, "fx");
102 NodeFS.writeFileSync(
103 launcher,
104 `#!/bin/sh\nexec node ${JSON.stringify(NodePath.join(import.meta.dirname, "fake-fx.ts"))} "$@"\n`,
105 { mode: 0o755 },
106 );
107 NodeFS.writeFileSync(
108 NodePath.join(baseDir, "userdata/settings.json"),
109 JSON.stringify({
110 providerInstances: {
111 [EVAL_INSTANCE]: {
112 driver: "fx",
113 displayName: "Eval Bot",
114 environment: [{ name: "T3_EVAL_FX_CONTROL_DIR", value: control }],
115 config: { enabled: true, binaryPath: launcher },
116 },
117 },
118 }),
119 );
120 return { workspace, control };
121}
122
123/** Starts the server in the caller's scope; the scope's end stops it. */
124export const startEnvironment = Effect.fn("environment.start")(function* (input: {
125 readonly baseDir: string;
126 /** Origin the device uses to reach this machine, without a port. */
127 readonly deviceHost: string;
128}) {
129 NodeFS.rmSync(input.baseDir, { recursive: true, force: true });
130 const { workspace, control } = writeScratch(input.baseDir);
131 const home = NodePath.join(input.baseDir, "home");
132 NodeFS.mkdirSync(home, { recursive: true });
133 const port = yield* freePort();
134 const log = NodeFS.openSync(NodePath.join(input.baseDir, "server.log"), "w");
135 yield* Effect.acquireRelease(
136 Effect.sync(() =>
137 NodeChildProcess.spawn(
138 "node",
139 [
140 "apps/server/src/bin.ts",
141 "serve",
142 "--host",
143 "0.0.0.0",
144 "--port",
145 String(port),
146 "--base-dir",
147 input.baseDir,
148 "--no-browser",
149 "--log-level",
150 "warn",
151 workspace,
152 ],
153 // A scratch HOME keeps the user's own skills and config out of the app under test.
154 { cwd: REPO_ROOT, stdio: ["ignore", log, log], env: { ...process.env, HOME: home } },
155 ),
156 ),
157 (child) => Effect.sync(() => child.kill("SIGTERM")),
158 );
159 yield* waitForPort(port);
160 const origin = `http://127.0.0.1:${port}`;
161 const issued = yield* runCli([
162 "auth",
163 "session",
164 "issue",
165 "--base-dir",
166 input.baseDir,
167 "--token-only",
168 ]);
169 // The token is the last line of stdout; stderr is only a fallback.
170 const lastLine = (text: string) =>
171 text
172 .split("\n")
173 .filter((line) => line.trim())
174 .at(-1)
175 ?.trim();
176 const token = lastLine(issued.stdout) ?? lastLine(issued.stderr) ?? "";
177 const http = (yield* HttpClient.HttpClient).pipe(
178 HttpClient.mapRequest((request) =>
179 request.pipe(
180 HttpClientRequest.bearerToken(token),
181 HttpClientRequest.setHeader(
182 ORCHESTRATION_PROTOCOL_HEADER,
183 String(ORCHESTRATION_PROTOCOL_VERSION),
184 ),
185 ),
186 ),
187 HttpClient.filterStatusOk,
188 );
189 const descriptor = yield* http
190 .get(`${origin}/.well-known/t3/environment`)
191 .pipe(Effect.flatMap(HttpClientResponse.schemaBodyJson(ExecutionEnvironmentDescriptor)));
192 const shell = () =>
193 http
194 .get(`${origin}/api/orchestration/shell`)
195 .pipe(Effect.flatMap(HttpClientResponse.schemaBodyJson(OrchestrationV2ShellSnapshot)));
196 const createProject = Effect.fn("environment.createProject")(function* (
197 title: string,
198 workspaceRoot: string,
199 ) {
200 const projectId = ProjectId.make(yield* uuid);
201 yield* http.execute(
202 HttpClientRequest.post(`${origin}/api/projects/mutate`).pipe(
203 HttpClientRequest.bodyJsonUnsafe({
204 type: "project.create",
205 commandId: yield* uuid,
206 projectId,
207 title,
208 workspaceRoot,
209 }),
210 ),
211 );
212 return projectId;
213 });
214 const projectId = yield* createProject("acme-app", workspace);
215 const ticket = yield* http
216 .execute(HttpClientRequest.post(`${origin}/api/auth/websocket-ticket`))
217 .pipe(
218 Effect.flatMap(HttpClientResponse.schemaBodyJson(Schema.Struct({ ticket: Schema.String }))),
219 );
220 // Built into the caller's scope: `Effect.provide` would close the socket once `make` returns.
221 const protocol = yield* Layer.build(
222 RpcClient.layerProtocolSocket().pipe(
223 Layer.provide(
224 Socket.layerWebSocket(
225 `${origin.replace(/^http/, "ws")}/ws?orchestrationProtocol=${ORCHESTRATION_PROTOCOL_VERSION}&wsTicket=${encodeURIComponent(ticket.ticket)}`,
226 ),
227 ),
228 Layer.provide(RpcSerialization.layerJson),
229 Layer.provide(NodeWS.layerWebSocketConstructor),
230 ),
231 );
232 const client = yield* RpcClient.make(WsRpcGroup).pipe(Effect.provideContext(protocol));
233 const environmentId = EnvironmentId.make(descriptor.environmentId);
234
235 const command = Effect.map(uuid, CommandId.make);
236 const projection = (threadId: ThreadId) =>
237 client["orchestration.getThreadProjection"]({ threadId });
238
239 const create = Effect.fn("environment.create")(function* (
240 title: string,
241 prompt?: string,
242 project: ProjectId = projectId,
243 ) {
244 const threadId = ThreadId.make(yield* uuid);
245 yield* client["orchestration.launchThread"]({
246 commandId: yield* command,
247 threadId,
248 projectId: project,
249 title,
250 modelSelection: { instanceId: EVAL_INSTANCE, model: EVAL_MODEL },
251 runtimeMode: "full-access",
252 interactionMode: "default",
253 workspaceStrategy: { type: "root" },
254 ...(prompt === undefined ? {} : { initialMessage: { text: prompt, attachments: [] } }),
255 });
256 return threadId;
257 });
258
259 /** Starts a thread whose first reply follows `prompt`'s `[[eval:…]]` script. */
260 const launch = (title: string, prompt: string) => create(title, prompt);
261
262 const until = Effect.fn("environment.until")(function* <A>(
263 description: string,
264 threadId: ThreadId,
265 pick: (value: OrchestrationV2ThreadProjection) => A | undefined,
266 ) {
267 for (let attempt = 0; attempt < 150; attempt++) {
268 const found = pick(yield* projection(threadId));
269 if (found !== undefined) return found;
270 yield* Effect.sleep("200 millis");
271 }
272 return yield* new EnvironmentError({ message: `Thread ${threadId} never had ${description}.` });
273 });
274
275 const send = Effect.fn("environment.send")(function* (threadId: ThreadId, text: string) {
276 yield* client["orchestration.dispatchCommand"]({
277 type: "message.dispatch",
278 commandId: yield* command,
279 threadId,
280 messageId: MessageId.make(yield* uuid),
281 text,
282 attachments: [],
283 dispatchMode: { type: "queue_after_active" },
284 createdBy: "user",
285 creationSource: "server",
286 });
287 });
288
289 const isSettled = (value: OrchestrationV2ThreadProjection) =>
290 value.runs.length > 0 &&
291 value.runs.every((run) =>
292 ["completed", "failed", "interrupted", "cancelled"].includes(run.status),
293 );
294
295 /** Waits until every run of the thread has ended. */
296 const settled = Effect.fn("environment.settled")(function* (threadId: ThreadId) {
297 for (let attempt = 0; attempt < 600; attempt++) {
298 const value = yield* projection(threadId);
299 if (isSettled(value)) return value;
300 yield* Effect.sleep("200 millis");
301 }
302 return yield* new EnvironmentError({ message: `Thread ${threadId} never settled.` });
303 });
304
305 /** A settled thread of `turns` request and reply pairs, for scroll and jank cases. */
306 const longThread = Effect.fn("environment.longThread")(function* (title: string, turns: number) {
307 const threadId = yield* launch(title, "Summarize the module. [[eval:stream:400:0]]");
308 yield* settled(threadId);
309 for (let turn = 1; turn < turns; turn++) {
310 const text = `Next part ${turn}. [[eval:stream:${turn % 3 === 0 ? 900 : 300}:0]]`;
311 yield* send(threadId, text);
312 // The message and its run land in one command, so once it shows, `settled` sees the run.
313 yield* until("its next message", threadId, (value) =>
314 value.messages.some((message) => message.text === text) ? true : undefined,
315 );
316 yield* settled(threadId);
317 }
318 return threadId;
319 });
320
321 return {
322 /** An empty thread in its own project; `lock` then disables `evalbot` there, so sends are rejected. */
323 launchLockable: Effect.fn("environment.launchLockable")(function* (title: string) {
324 const root = NodePath.join(input.baseDir, `workspace/locked-${(yield* uuid).slice(0, 6)}`);
325 NodeFS.mkdirSync(root, { recursive: true });
326 const project = yield* createProject("locked", root);
327 const threadId = yield* create(title, undefined, project);
328 const lock = client["server.updateSettings"]({
329 patch: {
330 projectSettingsOverrides: {
331 [project]: { providerInstances: { [EVAL_INSTANCE]: { enabled: false } } },
332 },
333 },
334 });
335 return { threadId, lock };
336 }),
337 /** Archives every thread but `keep`, so earlier samples' rows never crowd Home. */
338 archiveAllExcept: Effect.fn("environment.archiveAllExcept")(function* (keep: ThreadId) {
339 for (const thread of (yield* shell()).threads) {
340 if (thread.id === keep) continue;
341 yield* client["orchestration.dispatchCommand"]({
342 type: "thread.archive",
343 commandId: yield* command,
344 threadId: thread.id,
345 });
346 }
347 }),
348 /** Stops every unfinished run, so one sample's held turns never leak into the next. */
349 interruptAll: Effect.fn("environment.interruptAll")(function* () {
350 const threads = (yield* shell()).threads;
351 for (const thread of threads)
352 for (const run of (yield* projection(thread.id)).runs) {
353 if (["completed", "failed", "interrupted", "cancelled"].includes(run.status)) continue;
354 yield* client["orchestration.dispatchCommand"]({
355 type: "run.interrupt",
356 commandId: yield* command,
357 threadId: thread.id,
358 runId: run.id,
359 });
360 }
361 for (const thread of threads)
362 if ((yield* projection(thread.id)).runs.length) yield* settled(thread.id);
363 }),
364 waitForPendingInput: (threadId: ThreadId) =>
365 until("a pending question", threadId, (value) =>
366 value.runtimeRequests.find((request) => request.status === "pending"),
367 ),
368 waitForSubagents: (threadId: ThreadId) =>
369 until("running subagents", threadId, (value) =>
370 value.subagents.length >= 2 ? value.subagents : undefined,
371 ),
372 origin,
373 environmentId,
374 archive: (threadId: ThreadId) =>
375 command.pipe(
376 Effect.flatMap((commandId) =>
377 client["orchestration.dispatchCommand"]({ type: "thread.archive", commandId, threadId }),
378 ),
379 Effect.asVoid,
380 ),
381 unpin: (threadId: ThreadId) =>
382 command.pipe(
383 Effect.flatMap((commandId) =>
384 client["orchestration.dispatchCommand"]({ type: "thread.unpin", commandId, threadId }),
385 ),
386 Effect.asVoid,
387 ),
388 projection,
389 launch,
390 send,
391 settled,
392 longThread,
393 shell,
394 release: (key: string) =>
395 Effect.sync(() => NodeFS.writeFileSync(NodePath.join(control, `release-${key}`), "")),
396 pairingUrl: () =>
397 runCli([
398 "auth",
399 "pairing",
400 "create",
401 "--base-dir",
402 input.baseDir,
403 "--base-url",
404 `http://${input.deviceHost}:${port}`,
405 "--ttl",
406 "15m",
407 "--label",
408 "mobile-behavior-eval",
409 ]).pipe(
410 Effect.flatMap((out) => {
411 const url = /Pair URL:\s*(\S+)/.exec(`${out.stdout}\n${out.stderr}`)?.[1];
412 return url
413 ? Effect.succeed(url)
414 : Effect.fail(new EnvironmentError({ message: "No pairing URL in CLI output." }));
415 }),
416 ),
417 };
418}, Effect.provide(FetchHttpClient.layer));
419
420export type EvalEnvironment = Effect.Success<ReturnType<typeof startEnvironment>>;