1// @effect-diagnostics nodeBuiltinImport:off globalTimers:off globalDate:off globalFetch:off globalConsole:off globalJsonInEffect:off - Host-side process verifier retains raw wire evidence and owns deadline timers.
2import * as NodeAssert from "node:assert/strict";
3import * as NodeChildProcess from "node:child_process";
4import * as NodeCrypto from "node:crypto";
5
6import * as NodeFS from "node:fs";
7import * as NodeHttp from "node:http";
8import * as NodeOS from "node:os";
9import * as NodePath from "node:path";
10import * as NodeUtil from "node:util";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Schema from "effect/Schema";
14import * as Stream from "effect/Stream";
15import { Socket } from "effect/socket";
16import { RpcClient, RpcSerialization } from "effect/rpc";
17import {
18 ORCHESTRATION_PROTOCOL_VERSION,
19 ORCHESTRATION_PROTOCOL_HEADER,
20 CommandId,
21 ProjectId,
22 ProviderInstanceId,
23 ThreadId,
24 WsRpcGroup,
25 type OrchestrationV2ThreadProjection,
26} from "@t3tools/contracts";
27
28const { values } = NodeUtil.parseArgs({
29 options: {
30 model: { type: "string", default: "claude-sonnet-4-6" },
31 provider: { type: "string", default: "claudeAgent" },
32 timeout: { type: "string", default: "240" },
33 scenario: { type: "string", default: "all" },
34 repeat: { type: "string", default: "1" },
35 "fail-gate": { type: "boolean", default: false },
36 },
37});
38const timeoutMs = Number(values.timeout) * 1000;
39NodeAssert.ok(Number.isFinite(timeoutMs) && timeoutMs > 0, "timeout must be positive seconds");
40NodeAssert.ok(
41 ["all", "idle", "active", "native", "nested"].includes(values.scenario),
42 "scenario must be all, idle, active, native or nested",
43);
44const root = NodePath.resolve(import.meta.dirname, "../../..");
45const repeat = Number(values.repeat);
46NodeAssert.ok(Number.isSafeInteger(repeat) && repeat > 0, "repeat must be a positive integer");
47if (values.scenario === "all" || repeat > 1) {
48 const reportDirectory = NodeFS.mkdtempSync(
49 NodePath.join(NodeOS.tmpdir(), "t3-background-suite-"),
50 );
51 console.log(`Suite report: ${reportDirectory}`);
52 const outcomes = [];
53 attempts: for (let iteration = 1; iteration <= repeat; iteration++) {
54 for (const scenario of values.scenario === "all"
55 ? ["idle", "active", "native", "nested"]
56 : [values.scenario]) {
57 console.log(`Verification ${iteration}/${repeat}: ${scenario}`);
58 const child = NodeChildProcess.spawn(
59 process.execPath,
60 [
61 import.meta.filename,
62 "--scenario",
63 scenario,
64 "--model",
65 values.model,
66 "--provider",
67 values.provider,
68 "--timeout",
69 values.timeout,
70 ...(values["fail-gate"] ? ["--fail-gate"] : []),
71 ],
72 { cwd: root, stdio: ["ignore", "pipe", "inherit"] },
73 );
74 let childOutput = "";
75 let interrupted = false;
76 const interrupt = () => {
77 interrupted = true;
78 child.kill("SIGTERM");
79 };
80 process.on("SIGINT", interrupt);
81 process.on("SIGTERM", interrupt);
82 child.stdout?.on("data", (chunk: Buffer) => {
83 process.stdout.write(chunk);
84 childOutput += chunk.toString();
85 });
86 const status = await new Promise<number>((done, reject) => {
87 child.once("error", reject);
88 child.once("exit", (code) => done(code ?? 1));
89 });
90 process.off("SIGINT", interrupt);
91 process.off("SIGTERM", interrupt);
92 outcomes.push({
93 iteration,
94 scenario,
95 status: status === 0 && !interrupted ? "passed" : "failed",
96 evidence: /^Evidence: (.+)$/m.exec(childOutput)?.[1] ?? null,
97 });
98 NodeFS.writeFileSync(
99 NodePath.join(reportDirectory, "results.json"),
100 JSON.stringify(outcomes, null, 2),
101 );
102 if (interrupted) break attempts;
103 }
104 }
105 console.log(JSON.stringify(outcomes, null, 2));
106 process.exit(outcomes.some((outcome) => outcome.status === "failed") ? 1 : 0);
107}
108const evidence = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-background-live-"));
109const home = NodePath.join(evidence, "home");
110const project = NodePath.join(evidence, "project");
111NodeFS.mkdirSync(project);
112NodeChildProcess.execFileSync("git", ["init", "-q", project]);
113NodeChildProcess.execFileSync("git", [
114 "-C",
115 project,
116 "-c",
117 "user.name=Verifier",
118 "-c",
119 "user.email=verifier@localhost",
120 "commit",
121 "--allow-empty",
122 "-qm",
123 "initial",
124]);
125console.log(`Evidence: ${evidence}`);
126NodeFS.copyFileSync(import.meta.filename, NodePath.join(evidence, "verifier.ts"));
127NodeFS.writeFileSync(
128 NodePath.join(evidence, "working-tree.patch"),
129 NodeChildProcess.execFileSync("git", ["diff", "--binary", "HEAD"], {
130 cwd: root,
131 maxBuffer: 32 * 1024 * 1024,
132 }),
133);
134NodeFS.writeFileSync(
135 NodePath.join(evidence, "manifest.json"),
136 JSON.stringify(
137 {
138 revision: NodeChildProcess.execFileSync("git", ["rev-parse", "HEAD"], {
139 cwd: root,
140 encoding: "utf8",
141 }).trim(),
142 dirty: NodeChildProcess.execFileSync("git", ["status", "--porcelain"], {
143 cwd: root,
144 encoding: "utf8",
145 }),
146 provider: values.provider,
147 model: values.model,
148 scenario: values.scenario,
149 startedAt: new Date().toISOString(),
150 },
151 null,
152 2,
153 ),
154);
155
156const secret = `RESULT_${NodeCrypto.randomUUID()}`;
157const gatePath = `/${NodeCrypto.randomUUID()}`;
158let response: NodeHttp.ServerResponse | undefined;
159let parentResponse: NodeHttp.ServerResponse | undefined;
160let parentFinished = false;
161let nestedChildFinished = values.scenario !== "nested";
162let released = false;
163function release() {
164 if (!parentFinished || !nestedChildFinished || !response || released) return;
165 released = true;
166 NodeFS.appendFileSync(
167 NodePath.join(evidence, "milestones.ndjson"),
168 JSON.stringify({ kind: "gate-released", at: new Date().toISOString() }) + "\n",
169 );
170 if (values["fail-gate"]) response.writeHead(503).end("Injected dependency failure");
171 else response.end(secret);
172}
173const gate = NodeHttp.createServer((request, res) => {
174 if (request.url === `${gatePath}/parent` && !parentResponse) {
175 parentResponse = res;
176 parentFinished = true;
177 release();
178 return;
179 }
180 if (request.url !== gatePath || response) {
181 res.writeHead(404).end();
182 return;
183 }
184 response = res;
185 release();
186});
187gate.listen(0, "127.0.0.1");
188await new Promise<void>((done) => gate.once("listening", done));
189const gateAddress = gate.address();
190NodeAssert.ok(gateAddress && typeof gateAddress !== "string");
191
192const reservation = NodeHttp.createServer();
193reservation.listen(0, "127.0.0.1");
194await new Promise<void>((done) => reservation.once("listening", done));
195const address = reservation.address();
196NodeAssert.ok(address && typeof address !== "string");
197await new Promise<void>((done) => reservation.close(() => done()));
198const port = address.port;
199const origin = `http://127.0.0.1:${port}`;
200const bootstrap = NodeCrypto.randomUUID();
201function startServer() {
202 const server = NodeChildProcess.spawn(
203 process.execPath,
204 ["apps/server/src/bin.ts", "start", "--bootstrap-fd", "3", "--log-level", "debug"],
205 {
206 cwd: root,
207 detached: true,
208 stdio: ["ignore", "pipe", "pipe", "pipe"],
209 },
210 );
211 const bootstrapPipe = server.stdio[3];
212 NodeAssert.ok(bootstrapPipe && "write" in bootstrapPipe);
213 bootstrapPipe.end(
214 JSON.stringify({
215 mode: "desktop",
216 noBrowser: true,
217 port,
218 t3Home: home,
219 host: "127.0.0.1",
220 desktopBootstrapToken: bootstrap,
221 tailscaleServeEnabled: false,
222 tailscaleServePort: 443,
223 }),
224 );
225 const ready = Promise.withResolvers<void>();
226 let output = "";
227 for (const pipe of [server.stdout, server.stderr])
228 pipe?.on("data", (data: Buffer) => {
229 const text = data.toString().replaceAll(bootstrap, "[REDACTED]");
230 NodeFS.appendFileSync(NodePath.join(evidence, "server.log"), text);
231 output = (output + text).slice(-8192);
232 if (output.includes("startup phase: complete")) ready.resolve();
233 });
234 server.once("error", ready.reject);
235 server.once("exit", (code) => ready.reject(new Error(`Server exited ${code}`)));
236 const startupDeadline = setTimeout(
237 () => ready.reject(new Error("Server startup timed out")),
238 60_000,
239 );
240 return { server, ready: ready.promise.finally(() => clearTimeout(startupDeadline)) };
241}
242let { server, ready } = startServer();
243const abort = new AbortController();
244const onSignal = () => abort.abort(new Error("Verification interrupted"));
245process.on("SIGINT", onSignal);
246process.on("SIGTERM", onSignal);
247let verdict: object = { status: "failed", reason: "verification did not finish" };
248try {
249 await ready;
250 const tokenResponse = await fetch(`${origin}/oauth/token`, {
251 method: "POST",
252 body: new URLSearchParams({
253 grant_type: "urn:ietf:params:oauth:grant-type:token-exchange",
254 subject_token: bootstrap,
255 subject_token_type: "urn:t3:params:oauth:token-type:environment-bootstrap",
256 requested_token_type: "urn:ietf:params:oauth:token-type:access_token",
257 }),
258 });
259 NodeAssert.ok(tokenResponse.ok, `Authentication failed: ${tokenResponse.status}`);
260 const token = Schema.decodeUnknownSync(Schema.Struct({ access_token: Schema.String }))(
261 await tokenResponse.json(),
262 );
263 const headers = {
264 Authorization: `Bearer ${token.access_token}`,
265 "Content-Type": "application/json",
266 [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION),
267 };
268 const projectId = ProjectId.make(NodeCrypto.randomUUID());
269 const created = await fetch(`${origin}/api/projects/mutate`, {
270 method: "POST",
271 headers,
272 body: JSON.stringify({
273 type: "project.create",
274 commandId: NodeCrypto.randomUUID(),
275 projectId,
276 title: "Background verification",
277 workspaceRoot: project,
278 }),
279 });
280 NodeAssert.ok(created.ok, `Project creation failed: ${created.status} ${await created.text()}`);
281 const ticketResponse = await fetch(`${origin}/api/auth/websocket-ticket`, {
282 method: "POST",
283 headers,
284 });
285 NodeAssert.ok(ticketResponse.ok, `WebSocket authentication failed: ${ticketResponse.status}`);
286 const ticket = Schema.decodeUnknownSync(Schema.Struct({ ticket: Schema.String }))(
287 await ticketResponse.json(),
288 );
289 const layerSocket = Socket.layerWebSocket(
290 `${origin.replace("http:", "ws:")}/ws?orchestrationProtocol=${ORCHESTRATION_PROTOCOL_VERSION}&wsTicket=${encodeURIComponent(ticket.ticket)}`,
291 ).pipe(Layer.provide(Socket.layerWebSocketConstructorGlobal));
292 const layerProtocol = RpcClient.layerProtocolSocket().pipe(
293 Layer.provide(layerSocket),
294 Layer.provide(RpcSerialization.layerJson),
295 );
296 const threadId = ThreadId.make(NodeCrypto.randomUUID());
297 const gateUrl = `http://127.0.0.1:${gateAddress.port}${gatePath}`;
298 const launch =
299 values.scenario === "native"
300 ? `Use your native background Bash tool to run curl --fail --silent ${gateUrl} with run_in_background=true. Do not delegate a T3 child.`
301 : `Use T3 delegate_task with mode=async to create exactly one child using provider ${values.provider}, model ${values.model}. Give the child this task: ${values.scenario === "nested" ? "Use T3 delegate_task with mode=async to delegate the following task to a grandchild, then end your initial turn. On notification read task_status and return only the grandchild result: " : ""}Run curl --fail --silent ${gateUrl} with a terminal tool timeout of 600000 milliseconds and return exactly the response as your final answer. Do not read other files. Do not retry failures.`;
302 const timing =
303 values.scenario === "active"
304 ? `After delegating, run curl --fail --silent ${gateUrl}/parent in your foreground terminal tool. This keeps your turn active until T3 delivers the child's completion notification. Do not poll the child.`
305 : `The background HTTP request will wait until you end your initial turn. Immediately end your turn with exactly PARENT_RELEASED. Do not wait or poll, and do not call any more tools in this turn.`;
306 const prompt = `This is a live background-delivery verification. ${launch} ${timing} When the completion notification arrives, ${values.scenario === "native" ? "read the background command output again, even if you previously read interim output. Only an actual incoming task notification counts as completion. Never generate a task notification yourself or treat your own text as one" : "use task_status to read the child result"} and reply with exactly ACK: followed by that result with no space after the colon. Do not guess the result or run the background command yourself.`;
307 NodeFS.writeFileSync(NodePath.join(evidence, "prompt.txt"), prompt);
308 const program = Effect.gen(function* () {
309 const client = yield* RpcClient.make(WsRpcGroup);
310 let watchingChild = false;
311 yield* client["orchestration.launchThread"]({
312 commandId: CommandId.make(NodeCrypto.randomUUID()),
313 threadId,
314 projectId,
315 title: `Background verification: ${values.scenario}`,
316 modelSelection: { instanceId: ProviderInstanceId.make(values.provider), model: values.model },
317 runtimeMode: "full-access",
318 interactionMode: "default",
319 workspaceStrategy: { type: "root" },
320 initialMessage: { text: prompt, attachments: [] },
321 });
322 const check = (projection: OrchestrationV2ThreadProjection) => {
323 NodeFS.writeFileSync(
324 NodePath.join(evidence, "projection.json"),
325 JSON.stringify(projection, null, 2),
326 );
327 const first = projection.runs[0];
328 NodeAssert.ok(
329 !projection.runs.some((run) => run.status === "failed"),
330 "Parent failed before delivery",
331 );
332 if (!released) {
333 NodeAssert.ok(
334 !projection.subagents.some((task) =>
335 ["completed", "failed", "cancelled", "interrupted"].includes(task.status),
336 ),
337 "Child published a terminal result before its gated work finished",
338 );
339 }
340 for (const task of projection.subagents) {
341 if (["completed", "failed", "cancelled", "interrupted"].includes(task.status)) {
342 NodeAssert.ok(
343 task.result?.includes(secret),
344 "Child published an incorrect or failed result",
345 );
346 }
347 }
348 if (values.scenario !== "active") parentFinished = first?.status === "completed";
349 release();
350 if (
351 values.scenario === "active" &&
352 projection.subagents.some(
353 (task) =>
354 task.completionDelivery?.state === "delivered" ||
355 task.completionDelivery?.state === "acknowledged",
356 )
357 ) {
358 parentResponse?.end("Completion notification delivered. Read task_status now.");
359 }
360 const acknowledged = projection.messages.some(
361 (message) => message.role === "assistant" && message.text.includes(secret),
362 );
363 return (
364 acknowledged &&
365 projection.runs.every((run) =>
366 ["completed", "failed", "cancelled", "interrupted", "rolled_back"].includes(run.status),
367 )
368 );
369 };
370 const result = yield* client["orchestration.subscribeThread"]({ threadId }).pipe(
371 Stream.mapEffect(
372 Effect.fn(function* (item) {
373 NodeFS.appendFileSync(
374 NodePath.join(evidence, "events.ndjson"),
375 JSON.stringify(item) + "\n",
376 );
377 const projection =
378 item.kind === "snapshot"
379 ? item.projection
380 : yield* client["orchestration.getThreadProjection"]({ threadId });
381 const childThreadId = projection.subagents[0]?.childThreadId;
382 if (values.scenario === "nested" && childThreadId && !watchingChild) {
383 watchingChild = true;
384 yield* client["orchestration.subscribeThread"]({ threadId: childThreadId }).pipe(
385 Stream.runForEach(
386 Effect.fn(function* (childItem) {
387 NodeFS.appendFileSync(
388 NodePath.join(evidence, "nested-events.ndjson"),
389 JSON.stringify(childItem) + "\n",
390 );
391 const child =
392 childItem.kind === "snapshot"
393 ? childItem.projection
394 : yield* client["orchestration.getThreadProjection"]({
395 threadId: childThreadId,
396 });
397 nestedChildFinished =
398 child.runs[0]?.status === "completed" && child.subagents.length === 1;
399 release();
400 }),
401 ),
402 Effect.forkScoped,
403 );
404 }
405 return { projection, complete: check(projection) };
406 }),
407 ),
408 Stream.filter((item) => item.complete),
409 Stream.take(1),
410 Stream.runCollect,
411 );
412 const final = result[0]?.projection;
413 NodeAssert.ok(final, "Stream ended without proof of delivery");
414 NodeAssert.ok(released, "Child did not use the controlled gate");
415 NodeAssert.equal(
416 final.subagents.filter((task) => task.origin === "app_owned").length,
417 values.scenario === "native" ? 0 : 1,
418 "Wrong delegation path",
419 );
420 NodeAssert.ok(
421 final.turnItems.some((item) => item.type === "notification"),
422 "Missing typed notification",
423 );
424 NodeAssert.equal(
425 final.turnItems.filter((item) => item.type === "user_message").length,
426 1,
427 "Background delivery appeared as an ordinary user message",
428 );
429 NodeAssert.ok(!final.runs.some((run) => run.status === "failed"), "A parent run failed");
430 NodeAssert.equal(
431 final.turnItems.filter((item) => item.type === "notification").length,
432 1,
433 "One completion produced multiple notifications",
434 );
435 for (const task of final.subagents.filter((task) => task.origin === "app_owned")) {
436 NodeAssert.equal(task.status, "completed", "Child did not complete successfully");
437 NodeAssert.ok(task.result?.includes(secret), "Published task result lost the actual work");
438 NodeAssert.equal(
439 task.completionDelivery?.state,
440 "acknowledged",
441 "Parent did not read task_status",
442 );
443 }
444 if (values.scenario === "nested") {
445 const childThreadId = final.subagents[0]?.childThreadId;
446 NodeAssert.ok(childThreadId, "Missing nested task owner");
447 const child = yield* client["orchestration.getThreadProjection"]({ threadId: childThreadId });
448 NodeFS.writeFileSync(
449 NodePath.join(evidence, "nested-projection.json"),
450 JSON.stringify(child, null, 2),
451 );
452 NodeAssert.equal(child.subagents.length, 1, "Missing grandchild");
453 NodeAssert.ok(child.subagents[0]?.result?.includes(secret), "Nested result was lost");
454 }
455 NodeAssert.ok(
456 final.providerThreads.every((thread) => (thread.pendingBackgroundTasks?.length ?? 0) === 0),
457 "Stale background work remains after completion",
458 );
459 if (values.scenario === "active") {
460 NodeAssert.equal(final.runs.length, 1, "Active delivery unexpectedly started another run");
461 NodeAssert.ok(parentResponse, "Parent never entered the foreground gate");
462 } else {
463 NodeAssert.ok(
464 final.runs.length >= 2,
465 "Expected a new continuation after the initial turn ended",
466 );
467 }
468 return {
469 status: "passed",
470 scenario: values.scenario,
471 threadId,
472 runCount: final.runs.length,
473 acknowledgementIds: final.messages
474 .filter((message) => message.role === "assistant" && message.text.includes(secret))
475 .map((message) => message.id),
476 };
477 }).pipe(Effect.scoped, Effect.provide(layerProtocol), Effect.timeout(timeoutMs));
478 const finished = await Effect.runPromise(program, { signal: abort.signal });
479 verdict = finished;
480 await stopServer(server);
481 ({ server, ready } = startServer());
482 await ready;
483 const restored = await fetch(`${origin}/api/orchestration/threads/${threadId}`, { headers });
484 NodeAssert.ok(restored.ok, `Restart read failed: ${restored.status}`);
485 const snapshot = await restored.json();
486 NodeFS.writeFileSync(
487 NodePath.join(evidence, "after-restart.json"),
488 JSON.stringify(snapshot, null, 2),
489 );
490 const durable = Schema.decodeUnknownSync(
491 Schema.Struct({
492 projection: Schema.Struct({
493 messages: Schema.Array(
494 Schema.Struct({ id: Schema.String, role: Schema.String, text: Schema.String }),
495 ),
496 runs: Schema.Array(Schema.Struct({ id: Schema.String, status: Schema.String })),
497 }),
498 }),
499 )(snapshot);
500 NodeAssert.deepEqual(
501 durable.projection.messages
502 .filter((message) => message.role === "assistant" && message.text.includes(secret))
503 .map((message) => message.id),
504 finished.acknowledgementIds,
505 "Acknowledgement was lost or duplicated on restart",
506 );
507 NodeAssert.ok(
508 durable.projection.runs.every((run) => run.status === "completed"),
509 "Restart left an unfinished run",
510 );
511 NodeAssert.equal(
512 durable.projection.runs.length,
513 finished.runCount,
514 "Restart created an extra continuation",
515 );
516 verdict = { ...verdict, restart: "passed" };
517 console.log(JSON.stringify(verdict));
518} catch (error) {
519 verdict = {
520 status: "failed",
521 scenario: values.scenario,
522 reason: String(error),
523 milestones: {
524 parentFinished,
525 nestedChildFinished,
526 gateRequested: response !== undefined,
527 gateReleased: released,
528 parentGateRequested: parentResponse !== undefined,
529 },
530 };
531 console.error(String(error));
532 process.exitCode = 1;
533} finally {
534 NodeFS.writeFileSync(NodePath.join(evidence, "verdict.json"), JSON.stringify(verdict, null, 2));
535 response?.destroy();
536 parentResponse?.destroy();
537 gate.closeAllConnections();
538 gate.close();
539 await stopServer(server);
540 process.off("SIGINT", onSignal);
541 process.off("SIGTERM", onSignal);
542}
543
544async function stopServer(server: ReturnType<typeof NodeChildProcess.spawn>) {
545 if (server.exitCode === null && server.signalCode === null) {
546 const stopped = new Promise<void>((done) => server.once("exit", () => done()));
547 server.kill("SIGTERM");
548 const deadline = setTimeout(() => server.kill("SIGKILL"), 10_000);
549 await stopped;
550 clearTimeout(deadline);
551 }
552 if (server.pid) {
553 try {
554 process.kill(-server.pid, "SIGKILL");
555 } catch (error) {
556 if (!(error instanceof Error && "code" in error && error.code === "ESRCH")) throw error;
557 }
558 }
559}