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