clover's git

the-knife

1	import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
2	import { assert, describe, it } from "@effect/vitest";
3	import * as NodeServices from "@effect/platform-node/NodeServices";
4	import {
5	CheckpointId,
6	EnvironmentId,
7	NodeId,
8	ProviderInstanceId,
9	ProviderSessionId,
10	ProviderThreadId,
11	ProviderTurnId,
12	RunAttemptId,
13	RunId,
14	ThreadId,
15	type ChatAttachment,
16	type ModelSelection,
17	type OrchestrationV2AppThread,
18	type OrchestrationV2ProviderThread,
19	type OrchestrationV2ProviderTurn,
20	} from "@t3tools/contracts";
21	import * as Cause from "effect/Cause";
22	import * as DateTime from "effect/DateTime";
23	import * as Duration from "effect/Duration";
24	import * as Effect from "effect/Effect";
25	import * as Fiber from "effect/Fiber";
26	import * as Layer from "effect/Layer";
27	import * as PlatformError from "effect/PlatformError";
28	import * as Queue from "effect/Queue";
29	import * as Schema from "effect/Schema";
30	import * as Sink from "effect/Sink";
31	import * as Stream from "effect/Stream";
32	import * as TestClock from "effect/testing/TestClock";
33	import * as ChildProcess from "effect/process/ChildProcess";
34	import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
35
36	import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
37	import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
38	import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
39	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
40	import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
41	import { handoffBudget } from "@t3tools/provider-core/server/handoffBudget";
42	import * as HostProcess from "@t3tools/shared/HostProcess";
43	import { toolOutputImages, compactDynamicToolOutput } from "@t3tools/shared/toolOutput";
44	import {
45	makePiAdapterV2,
46	PiAdapterV2Driver,
47	PI_PROVIDER,
48	type PiAdapterV2Options,
49	} from "./adapter.ts";
50	import { makePiRpcConnection, type PiRpcRecord } from "./rpc.ts";
51	import { loadMcpBridge } from "./mcpBridge.testkit.ts";
52	import { turnItemOutputText } from "../../../client-runtime/src/work-log/itemDetail.ts";
53
54	const layerTest = Layer.mergeAll(
55	NodeServices.layer,
56	IdAllocator.layer,
57	McpProviderSessions.layer,
58	ScriptProgress.layerUnavailable,
59	TestProviderHost.layer().pipe(Layer.provide(NodeServices.layer)),
60	);
61
62	const decodeJsonLine = Schema.decodeSync(Schema.fromJsonString(Schema.Unknown));
63	const encodeJsonLine = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
64
65	const PI_INSTANCE_ID = ProviderInstanceId.make("pi");
66	const THREAD_ID = ThreadId.make("thread-pi-test");
67	const SESSION_ID = ProviderSessionId.make("provider-session-pi-test");
68	const FAKE_SESSION_FILE = "/fake/.pi/agent/sessions/--workspace--/0001_abc.jsonl";
69	/** Deliberately outside the valid pid range so a group-kill can never land. */
70	const FAKE_PID = 999_999_999;
71
72	const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({
73	runtimeMode: "full-access",
74	interactionMode: "default",
75	cwd: null,
76	});
77
78	const modelSelection = (model: string): ModelSelection => ({
79	instanceId: PI_INSTANCE_ID,
80	model,
81	});
82
83	interface FakePi {
84	readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"];
85	readonly emit: (record: PiRpcRecord, chunkBytes?: number) => Effect.Effect<void>;
86	readonly takeRequest: (type: string) => Effect.Effect<PiRpcRecord>;
87	/** Data returned by the next `get_entries` acks, consumed in order. */
88	readonly queueEntries: (data: unknown) => void;
89	/** Data returned by the next active-branch `get_messages` acks. */
90	readonly queueMessages: (data: unknown) => void;
91	/** Make the next `switch_session` ack report an extension veto. */
92	readonly vetoNextSwitch: () => void;
93	/** Fields overriding the recorded idle state in the next `get_state` acks, in order. */
94	readonly queueState: (data: Record<string, unknown>) => void;
95	/** Change Pi's streaming state before its agent_start event reaches T3. */
96	readonly setStreaming: (streaming: boolean) => void;
97	/** Hold the next `get_state` response until the test resolves it. */
98	readonly deferNextState: () => void;
99	/** Resolve the held `get_state` request. */
100	readonly resolveDeferredState: (data: unknown) => Effect.Effect<void>;
101	/** Reject the next `get_state` request. */
102	readonly failNextState: () => void;
103	readonly deferNextLifecycle: (type: "switch_session" | "new_session" | "fork") => void;
104	/** Hold a model-select extension hook until its UI request is answered. */
105	readonly deferNextModelSelection: () => void;
106	readonly queueModels: (models: ReadonlyArray<unknown>) => void;
107	readonly vetoNextNewSession: () => void;
108	/** Every request received by the fake process. */
109	readonly allRequests: () => ReadonlyArray<PiRpcRecord>;
110	/** Data returned by the next `get_session_stats` acks, consumed in order. */
111	readonly queueStats: (data: unknown) => void;
112	/** Close the fake process stdout stream. */
113	readonly closeStdout: Effect.Effect<void>;
114	readonly lastSpawn: () => {
115	readonly args: ReadonlyArray<string>;
116	readonly env: NodeJS.ProcessEnv;
117	};
118	}
119
120	/**
121	* Pi 1.0.0's idle `get_state` reply, taken from the `simple` replay fixture
122	* (fixtures/simple/pi_transcript.ndjson) minus the model object. Pi omits
123	* `model` when none is selected and `sessionName` until one is set.
124	*/
125	const recordedIdleState = (sessionFile: string) => ({
126	thinkingLevel: "high",
127	isStreaming: false,
128	isCompacting: false,
129	steeringMode: "one-at-a-time",
130	followUpMode: "one-at-a-time",
131	sessionFile,
132	sessionId: "00000000-0000-4000-8000-000000000002",
133	autoCompactionEnabled: true,
134	messageCount: 0,
135	pendingMessageCount: 0,
136	});
137
138	/**
139	* In-process fake `pi --mode rpc` for races and failures a live Pi cannot
140	* produce on demand: captures every stdin record, auto-acks requests, and lets
141	* tests push protocol events to stdout. Behaviour a real Pi can show belongs
142	* in a replay fixture instead (see PiAdapterV2.testkit.ts).
143	*/
144	const makeFakePi: Effect.Effect<FakePi> = Effect.gen(function* () {
145	const stdout = yield* Queue.unbounded<Uint8Array, Cause.Done>();
146	const requests = yield* Queue.unbounded<PiRpcRecord>();
147	const entriesQueue: Array<unknown> = [];
148	const messagesQueue: Array<unknown> = [];
149	const stateQueue: Array<Record<string, unknown>> = [];
150	const statsQueue: Array<unknown> = [];
151	const allRequests: Array<PiRpcRecord> = [];
152	let deferState = false;
153	let deferredStateRequest: PiRpcRecord | undefined;
154	let failState = false;
155	let vetoSwitch = false;
156	let vetoNewSession = false;
157	let deferredLifecycle: string | undefined;
158	let sessionFile = FAKE_SESSION_FILE;
159	let sessionGeneration = 0;
160	let models: ReadonlyArray<unknown> = [];
161	let isStreaming: boolean | null = null;
162	let stdinBuffer = "";
163
164	const emit = (record: PiRpcRecord, chunkBytes?: number) =>
165	Effect.gen(function* () {
166	const bytes = new TextEncoder().encode(`${encodeJsonLine(record)}\n`);
167	const size = chunkBytes ?? bytes.byteLength;
168	for (let offset = 0; offset < bytes.byteLength; offset += size) {
169	yield* Queue.offer(stdout, bytes.subarray(offset, offset + size));
170	}
171	});
172
173	const respondTo = (record: PiRpcRecord): PiRpcRecord | null => {
174	if (typeof record["id"] !== "string" && !(record["type"] === "prompt" && isStreaming !== null))
175	return null;
176	const base = {
177	type: "response",
178	id: record["id"],
179	command: String(record["type"]),
180	success: true,
181	};
182	switch (record["type"]) {
183	case "get_state":
184	if (failState) {
185	failState = false;
186	return { ...base, success: false, error: "state unavailable" };
187	}
188	// Queued data overrides fields of the recorded idle state, so a test
189	// that only cares about the session file still gets a real shape.
190	return {
191	...base,
192	data: {
193	...recordedIdleState(sessionFile),
194	isStreaming: isStreaming ?? false,
195	...stateQueue.shift(),
196	},
197	};
198	case "prompt":
199	if (isStreaming && record["streamingBehavior"] === undefined) {
200	return {
201	...base,
202	success: false,
203	error:
204	"Agent is already processing. Specify streamingBehavior ('steer' or 'followUp') to queue the message.",
205	};
206	}
207	return base;
208	case "get_available_models":
209	return { ...base, data: { models } };
210	case "new_session": {
211	const cancelled = vetoNewSession;
212	vetoNewSession = false;
213	if (!cancelled) sessionFile = `/fake/new-${++sessionGeneration}.jsonl`;
214	return { ...base, data: { cancelled } };
215	}
216	case "switch_session": {
217	const cancelled = vetoSwitch;
218	vetoSwitch = false;
219	return { ...base, data: { cancelled } };
220	}
221	case "get_entries":
222	return { ...base, data: entriesQueue.shift() ?? { entries: [], leafId: null } };
223	case "get_messages":
224	return { ...base, data: messagesQueue.shift() ?? { messages: [] } };
225	case "get_session_stats":
226	return { ...base, data: statsQueue.shift() ?? {} };
227	case "fork":
228	return { ...base, data: { text: "Hello pi", cancelled: false } };
229	default:
230	return base;
231	}
232	};
233
234	const handleStdinChunk = (chunk: Uint8Array) =>
235	Effect.gen(function* () {
236	stdinBuffer += new TextDecoder().decode(chunk);
237	while (true) {
238	const newline = stdinBuffer.indexOf("\n");
239	if (newline === -1) return;
240	const line = stdinBuffer.slice(0, newline);
241	stdinBuffer = stdinBuffer.slice(newline + 1);
242	if (line.length === 0) continue;
243	const record = decodeJsonLine(line) as PiRpcRecord;
244	allRequests.push(record);
245	yield* Queue.offer(requests, record);
246	if (record["type"] === "get_state" && deferState) {
247	deferState = false;
248	deferredStateRequest = record;
249	continue;
250	}
251	if (record["type"] === deferredLifecycle) {
252	deferredLifecycle = undefined;
253	continue;
254	}
255	const response = respondTo(record);
256	if (response !== null) yield* emit(response);
257	}
258	});
259
260	let lastSpawn: { readonly args: ReadonlyArray<string>; readonly env: NodeJS.ProcessEnv } = {
261	args: [],
262	env: {},
263	};
264	const spawner = ChildProcessSpawner.make((command) =>
265	Effect.sync(() => {
266	if (ChildProcess.isStandardCommand(command)) {
267	lastSpawn = {
268	args: command.args,
269	env: command.options.env ?? {},
270	};
271	}
272	return ChildProcessSpawner.makeHandle({
273	pid: ChildProcessSpawner.ProcessId(FAKE_PID),
274	exitCode: Effect.never,
275	isRunning: Effect.succeed(true),
276	kill: () => Effect.void,
277	unref: Effect.succeed(Effect.void),
278	stdin: Sink.forEach(handleStdinChunk),
279	stdout: Stream.fromQueue(stdout),
280	stderr: Stream.empty,
281	all: Stream.empty,
282	getInputFd: () => Sink.drain,
283	getOutputFd: () => Stream.empty,
284	});
285	}),
286	);
287
288	const takeRequest = (type: string): Effect.Effect<PiRpcRecord> =>
289	Effect.gen(function* () {
290	while (true) {
291	const record = yield* Queue.take(requests);
292	if (record["type"] === type) return record;
293	}
294	});
295
296	return {
297	spawner,
298	emit,
299	takeRequest,
300	queueEntries: (data) => entriesQueue.push(data),
301	queueMessages: (data) => messagesQueue.push(data),
302	deferNextState: () => {
303	deferState = true;
304	},
305	resolveDeferredState: (data) =>
306	Effect.gen(function* () {
307	const record = deferredStateRequest;
308	assert.isDefined(record);
309	deferredStateRequest = undefined;
310	yield* emit({
311	type: "response",
312	id: record!["id"],
313	command: "get_state",
314	success: true,
315	data,
316	});
317	}),
318	failNextState: () => {
319	failState = true;
320	},
321	deferNextLifecycle: (type) => {
322	deferredLifecycle = type;
323	},
324	deferNextModelSelection: () => {
325	deferredLifecycle = "set_model";
326	},
327	queueModels: (value) => {
328	models = value;
329	},
330	vetoNextNewSession: () => {
331	vetoNewSession = true;
332	},
333	allRequests: () => allRequests,
334	vetoNextSwitch: () => {
335	vetoSwitch = true;
336	},
337	queueState: (data) => stateQueue.push(data),
338	setStreaming: (streaming) => {
339	isStreaming = streaming;
340	},
341	queueStats: (data) => statsQueue.push(data),
342	closeStdout: Queue.end(stdout),
343	lastSpawn: () => lastSpawn,
344	} satisfies FakePi;
345	});
346
347	const makeAdapter = Effect.fnUntraced(function* (
348	fake: FakePi,
349	launchArgs = "",
350	forkFake?: FakePi,
351	continuationRequests?: PiAdapterV2Options["continuationRequests"],
352	) {
353	const spawner =
354	forkFake === undefined
355	? fake.spawner
356	: ChildProcessSpawner.make((command) =>
357	ChildProcess.isStandardCommand(command) && command.args.includes("--fork")
358	? forkFake.spawner.spawn(command)
359	: fake.spawner.spawn(command),
360	);
361	// Continuation cases go through the driver, which wires the offer from the
362	// environment the way production does.
363	if (continuationRequests !== undefined) {
364	return yield* PiAdapterV2Driver.create({
365	instanceId: PI_INSTANCE_ID,
366	displayName: undefined,
367	enabled: true,
368	environment: [],
369	config: { enabled: true, binaryPath: "pi", launchArgs, customModels: [] },
370	}).pipe(
371	Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
372	Effect.provideService(HostProcess.Environment, {}),
373	Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, {
374	...continuationRequests,
375	take: Effect.never,
376	}),
377	);
378	}
379	return yield* makePiAdapterV2({
380	instanceId: PI_INSTANCE_ID,
381	settings: { enabled: true, binaryPath: "pi", launchArgs, customModels: [] },
382	environment: {},
383	}).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
384	});
385
386	const openRuntime = Effect.fnUntraced(function* (
387	fake: FakePi,
388	model = "default",
389	threadId = THREAD_ID,
390	providerSessionId = SESSION_ID,
391	forkFake?: FakePi,
392	continuationRequests?: PiAdapterV2Options["continuationRequests"],
393	) {
394	const adapter = yield* makeAdapter(fake, "", forkFake, continuationRequests);
395	const runtime = yield* adapter.openSession({
396	threadId,
397	providerSessionId,
398	modelSelection: modelSelection(model),
399	runtimePolicy,
400	});
401	const emitted = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
402	yield* runtime.events.pipe(
403	Stream.runForEach((event) => Queue.offer(emitted, event)),
404	Effect.forkScoped,
405	);
406	const takeEvent = (predicate: (event: ProviderAdapter.ProviderAdapterV2Event) => boolean) =>
407	Effect.gen(function* () {
408	while (true) {
409	const event = yield* Queue.take(emitted);
410	if (predicate(event)) return event;
411	}
412	});
413	return { runtime, takeEvent };
414	});
415
416	const makeAppThread = Effect.fnUntraced(function* (model: string, threadId = THREAD_ID) {
417	const now = yield* DateTime.now;
418	return {
419	createdBy: "user",
420	creationSource: "web",
421	id: threadId,
422	projectId: "project:fixture:pi" as OrchestrationV2AppThread["projectId"],
423	title: "Pi test thread",
424	providerInstanceId: PI_INSTANCE_ID,
425	modelSelection: modelSelection(model),
426	runtimeMode: "full-access",
427	interactionMode: "default",
428	branch: null,
429	worktreePath: null,
430	activeProviderThreadId: null,
431	lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
432	forkedFrom: null,
433	createdAt: now,
434	updatedAt: now,
435	archivedAt: null,
436	settledOverride: null,
437	settledAt: null,
438	lastVisitedAt: null,
439	deletedAt: null,
440	} satisfies OrchestrationV2AppThread;
441	});
442
443	const startTurn = Effect.fnUntraced(function* (
444	runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime,
445	providerThread: OrchestrationV2ProviderThread,
446	model = "default",
447	attachments: ReadonlyArray<ChatAttachment> = [],
448	text = "Hello pi",
449	selection?: ModelSelection,
450	runOrdinal = 1,
451	threadId = THREAD_ID,
452	continuation = false,
453	) {
454	const appThread = yield* makeAppThread(model, threadId);
455	const runId = RunId.make(`run:${threadId}:${runOrdinal}`);
456	yield* runtime.startTurn({
457	appThread,
458	threadId,
459	runId,
460	runOrdinal,
461	providerTurnOrdinal: runOrdinal,
462	attemptId: RunAttemptId.make(`run-attempt:${runId}:1`),
463	rootNodeId: NodeId.make(`node:${runId}:root`),
464	providerThread,
465	message: {
466	messageId: `message:${threadId}:${runOrdinal}` as never,
467	text,
468	attachments,
469	createdBy: continuation ? "agent" : "user",
470	creationSource: continuation ? "provider" : "web",
471	},
472	modelSelection: selection ?? modelSelection(model),
473	runtimePolicy,
474	});
475	});
476
477	const expectModelFailure = (errorMessage: string) =>
478	Effect.gen(function* () {
479	const fake = yield* makeFakePi;
480	const { runtime, takeEvent } = yield* openRuntime(fake);
481	const providerThread = yield* runtime.ensureThread({
482	threadId: THREAD_ID,
483	modelSelection: modelSelection("default"),
484	runtimePolicy,
485	});
486	yield* startTurn(runtime, providerThread);
487	yield* fake.takeRequest("prompt");
488	yield* fake.emit({ type: "agent_start" });
489	yield* fake.emit({
490	type: "message_end",
491	message: {
492	role: "assistant",
493	content: [],
494	stopReason: "error",
495	errorMessage,
496	},
497	});
498	yield* fake.emit({ type: "agent_settled" });
499
500	const sessionError = yield* takeEvent(
501	(event) =>
502	event.type === "provider_session.updated" && event.providerSession.status === "error",
503	);
504	assert.isTrue(
505	sessionError.type === "provider_session.updated" &&
506	sessionError.providerSession.lastError === errorMessage,
507	);
508	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
509	assert.isTrue(
510	terminal.type === "turn.terminal" &&
511	terminal.status === "failed" &&
512	terminal.failure.message === errorMessage,
513	);
514	}).pipe(Effect.scoped, Effect.provide(layerTest));
515
516	describe("PiAdapterV2", () => {
517	it.effect.each([false, true])(
518	"completes a resumed user turn before native wake events arrive, Pi streaming=%s",
519	(streaming) =>
520	Effect.gen(function* () {
521	const fake = yield* makeFakePi;
522	const { runtime, takeEvent } = yield* openRuntime(fake);
523	const providerThread = yield* runtime.ensureThread({
524	threadId: THREAD_ID,
525	modelSelection: modelSelection("default"),
526	runtimePolicy,
527	});
528	fake.setStreaming(streaming);
529	yield* runtime.resumeThread({ providerThread });
530	yield* startTurn(runtime, providerThread);
531	yield* fake.takeRequest("prompt");
532	yield* fake.emit({ type: "agent_start" });
533	yield* fake.emit({
534	type: "message_end",
535	message: {
536	role: "assistant",
537	content: [{ type: "text", text: "Resumed prompt completed" }],
538	stopReason: "stop",
539	},
540	});
541	fake.queueState({ isStreaming: false });
542	yield* fake.emit({ type: "agent_settled" });
543	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
544	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
545	}).pipe(Effect.scoped, Effect.provide(layerTest)),
546	);
547
548	it.effect.each([false, true])(
549	"serializes rollback fork hooks with native wake ownership, wake=%s",
550	(nativeWake) =>
551	Effect.gen(function* () {
552	const fake = yield* makeFakePi;
553	const offers =
554	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
555	const { runtime, takeEvent } = yield* openRuntime(
556	fake,
557	"default",
558	THREAD_ID,
559	SESSION_ID,
560	undefined,
561	{
562	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
563	},
564	);
565	const providerThread = yield* runtime.ensureThread({
566	threadId: THREAD_ID,
567	modelSelection: modelSelection("default"),
568	runtimePolicy,
569	});
570	fake.deferNextLifecycle("fork");
571	const rollingBack = yield* runtime
572	.rollbackThread({
573	providerThread,
574	target: {
575	type: "thread_start",
576	checkpointId: CheckpointId.make("checkpoint-start"),
577	appRunOrdinal: 0,
578	},
579	providerThreadTurns: [
580	{
581	id: ProviderTurnId.make("previous-turn"),
582	providerThreadId: providerThread.id,
583	nodeId: NodeId.make("previous-node"),
584	runAttemptId: null,
585	nativeTurnRef: {
586	driver: PI_PROVIDER,
587	nativeId: "user-before-wake",
588	strength: "strong",
589	},
590	ordinal: 1,
591	status: "completed",
592	startedAt: providerThread.createdAt,
593	completedAt: providerThread.createdAt,
594	},
595	],
596	})
597	.pipe(Effect.result, Effect.forkScoped);
598	const fork = yield* fake.takeRequest("fork");
599	const overlappingTurn = yield* startTurn(runtime, providerThread).pipe(Effect.result);
600	assert.equal(overlappingTurn._tag, "Failure");
601	const overlappingRegistration = yield* runtime
602	.ensureThread({
603	threadId: THREAD_ID,
604	modelSelection: modelSelection("default"),
605	runtimePolicy,
606	})
607	.pipe(Effect.result);
608	assert.equal(overlappingRegistration._tag, "Failure");
609	yield* fake.emit({
610	type: "extension_ui_request",
611	id: "fork-hook-dialog",
612	method: "confirm",
613	title: "Restore this branch?",
614	});
615	const forkDialog = yield* takeEvent((event) => event.type === "runtime_request.updated");
616	assert.isTrue(forkDialog.type === "runtime_request.updated");
617	if (forkDialog.type !== "runtime_request.updated") return;
618	assert.isNull(forkDialog.runtimeRequest.providerTurnId);
619	yield* runtime.respondToRuntimeRequest({
620	requestId: forkDialog.runtimeRequest.id,
621	decision: "accept",
622	});
623	yield* fake.takeRequest("extension_ui_response");
624	if (nativeWake) {
625	yield* fake.emit({ type: "agent_start" });
626	yield* fake.emit({ type: "message_start", message: { role: "assistant" } });
627	yield* fake.emit({
628	type: "message_update",
629	assistantMessageEvent: {
630	type: "text_end",
631	contentIndex: 0,
632	content: "Discarded branch wake output",
633	},
634	});
635	yield* fake.emit({ type: "agent_settled" });
636	const retired = yield* takeEvent(
637	(event) =>
638	event.type === "provider_session.updated" &&
639	event.providerSession.status === "stopped",
640	).pipe(Effect.as(true), Effect.raceFirst(Queue.take(offers).pipe(Effect.as(false))));
641	assert.isTrue(retired, "Pi offered native work while rollback was changing its binding");
642	const result = yield* Fiber.join(rollingBack);
643	assert.equal(result._tag, "Failure");
644	assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
645	const staleTurn = yield* startTurn(
646	runtime,
647	providerThread,
648	"default",
649	[],
650	"Synthetic wake",
651	undefined,
652	1,
653	THREAD_ID,
654	true,
655	).pipe(Effect.result);
656	assert.equal(staleTurn._tag, "Failure");
657	assert.isFalse(fake.allRequests().some((request) => request["type"] === "prompt"));
658	return;
659	}
660	fake.queueState({ sessionFile: "/fake/rolled-back-session.jsonl" });
661	yield* fake.emit({
662	type: "response",
663	id: fork["id"],
664	command: "fork",
665	success: true,
666	data: { cancelled: false },
667	});
668	const result = yield* Fiber.join(rollingBack);
669	assert.equal(result._tag, "Success");
670	if (result._tag !== "Success") return;
671	assert.equal(
672	result.success.providerThread.nativeThreadRef?.nativeId,
673	"/fake/rolled-back-session.jsonl",
674	);
675	yield* startTurn(runtime, result.success.providerThread);
676	const prompt = yield* fake.takeRequest("prompt");
677	assert.equal(prompt["message"], "Hello pi");
678	yield* fake.emit({ type: "response", command: "prompt", success: true });
679	yield* fake.emit({ type: "agent_start" });
680	yield* fake.emit({ type: "agent_settled" });
681	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
682	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
683	}).pipe(Effect.scoped, Effect.provide(layerTest)),
684	);
685
686	it.effect.each([false, true])(
687	"retains native wake ownership after a rejected joining prompt, compaction aborted=%s",
688	(aborted) =>
689	Effect.gen(function* () {
690	const fake = yield* makeFakePi;
691	const offers =
692	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
693	const { runtime, takeEvent } = yield* openRuntime(
694	fake,
695	"default",
696	THREAD_ID,
697	SESSION_ID,
698	undefined,
699	{
700	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
701	},
702	);
703	const providerThread = yield* runtime.ensureThread({
704	threadId: THREAD_ID,
705	modelSelection: modelSelection("default"),
706	runtimePolicy,
707	});
708	yield* fake.takeRequest("get_state");
709	yield* fake.emit({ type: "agent_start" });
710	yield* Queue.take(offers);
711	yield* fake.emit({ type: "agent_settled" });
712	yield* fake.emit({ type: "compaction_start", reason: "manual" });
713	yield* startTurn(runtime, providerThread);
714	yield* fake.takeRequest("prompt");
715	const rejection = "Cannot submit a prompt while compaction is in progress.";
716	yield* fake.emit({ type: "response", command: "prompt", success: false, error: rejection });
717	yield* fake.emit({
718	type: "extension_ui_request",
719	id: "native-compaction-dialog",
720	method: "confirm",
721	title: "Continue native work?",
722	});
723	const dialog = yield* takeEvent((event) => event.type === "runtime_request.updated");
724	assert.isTrue(dialog.type === "runtime_request.updated");
725	if (dialog.type !== "runtime_request.updated") return;
726	assert.isNotNull(dialog.runtimeRequest.providerTurnId);
727	yield* runtime.respondToRuntimeRequest({
728	requestId: dialog.runtimeRequest.id,
729	decision: "accept",
730	});
731	yield* fake.takeRequest("extension_ui_response");
732	if (aborted) {
733	yield* fake.emit({ type: "compaction_end", result: null, aborted: true });
734	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
735	assert.isTrue(
736	terminal.type === "turn.terminal" &&
737	terminal.status === "failed" &&
738	terminal.failure.message === rejection,
739	);
740	return;
741	}
742	fake.queueState({ isStreaming: true });
743	yield* fake.emit({
744	type: "compaction_end",
745	result: { summary: "Native compaction finished" },
746	willRetry: true,
747	});
748	yield* fake.emit({ type: "agent_start" });
749	yield* fake.emit({ type: "message_start", message: { role: "assistant" } });
750	yield* fake.emit({
751	type: "message_update",
752	assistantMessageEvent: {
753	type: "text_end",
754	contentIndex: 0,
755	content: "Native work completed",
756	},
757	});
758	const message = yield* takeEvent((event) => event.type === "message.updated");
759	assert.isTrue(message.type === "message.updated" && message.message.runId !== null);
760	yield* fake.takeRequest("get_state");
761	yield* fake.emit({ type: "agent_settled" });
762	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
763	assert.isTrue(
764	terminal.type === "turn.terminal" &&
765	terminal.status === "failed" &&
766	terminal.failure.message === rejection,
767	);
768	}).pipe(Effect.scoped, Effect.provide(layerTest)),
769	);
770
771	it.effect("rejects rollback while native wake output is awaiting ownership", () =>
772	Effect.gen(function* () {
773	const fake = yield* makeFakePi;
774	const offers =
775	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
776	const { runtime, takeEvent } = yield* openRuntime(
777	fake,
778	"default",
779	THREAD_ID,
780	SESSION_ID,
781	undefined,
782	{
783	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
784	},
785	);
786	const providerThread = yield* runtime.ensureThread({
787	threadId: THREAD_ID,
788	modelSelection: modelSelection("default"),
789	runtimePolicy,
790	});
791	yield* fake.emit({ type: "agent_start" });
792	yield* Queue.take(offers);
793	yield* fake.emit({ type: "message_start", message: { role: "assistant" } });
794	yield* fake.emit({
795	type: "message_update",
796	assistantMessageEvent: {
797	type: "text_end",
798	contentIndex: 0,
799	content: "Current branch wake output",
800	},
801	});
802	yield* fake.emit({ type: "agent_settled" });
803	yield* takeEvent(
804	(event) =>
805	event.type === "provider_session.updated" && event.providerSession.status === "ready",
806	);
807	const rolledBack = yield* runtime
808	.rollbackThread({
809	providerThread,
810	target: {
811	type: "thread_start",
812	checkpointId: CheckpointId.make("checkpoint-start"),
813	appRunOrdinal: 0,
814	},
815	providerThreadTurns: [
816	{
817	id: ProviderTurnId.make("previous-turn"),
818	providerThreadId: providerThread.id,
819	nodeId: NodeId.make("previous-node"),
820	runAttemptId: null,
821	nativeTurnRef: {
822	driver: PI_PROVIDER,
823	nativeId: "user-before-wake",
824	strength: "strong",
825	},
826	ordinal: 1,
827	status: "completed",
828	startedAt: providerThread.createdAt,
829	completedAt: providerThread.createdAt,
830	},
831	],
832	})
833	.pipe(Effect.result);
834	assert.equal(
835	rolledBack._tag,
836	"Failure",
837	"Rollback changed the native session while its wake was still buffered",
838	);
839	assert.isFalse(fake.allRequests().some((request) => request["type"] === "fork"));
840	assert.isTrue(yield* runtime.hasPendingBackgroundWork!);
841	yield* startTurn(
842	runtime,
843	providerThread,
844	"default",
845	[],
846	"Synthetic wake",
847	undefined,
848	1,
849	THREAD_ID,
850	true,
851	);
852	const message = yield* takeEvent((event) => event.type === "message.updated");
853	assert.equal(
854	message.type === "message.updated" && message.message.text,
855	"Current branch wake output",
856	);
857	yield* takeEvent((event) => event.type === "turn.terminal");
858	}).pipe(Effect.scoped, Effect.provide(layerTest)),
859	);
860
861	it.effect("answers model-selection dialogs before a user turn claims a native wake", () =>
862	Effect.gen(function* () {
863	const fake = yield* makeFakePi;
864	const offers =
865	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
866	const { runtime, takeEvent } = yield* openRuntime(
867	fake,
868	"default",
869	THREAD_ID,
870	SESSION_ID,
871	undefined,
872	{ offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid) },
873	);
874	const providerThread = yield* runtime.ensureThread({
875	threadId: THREAD_ID,
876	modelSelection: modelSelection("default"),
877	runtimePolicy,
878	});
879	yield* fake.emit({ type: "agent_start" });
880	yield* Queue.take(offers);
881	const requests: Array<
882	Extract<ProviderAdapter.ProviderAdapterV2Event, { type: "runtime_request.updated" }>
883	> = [];
884	fake.deferNextModelSelection();
885	const starting = yield* startTurn(
886	runtime,
887	providerThread,
888	"default",
889	[],
890	"Continue the review",
891	modelSelection("xai/grok-4.6"),
892	).pipe(Effect.forkScoped);
893	const modelRequest = yield* fake.takeRequest("set_model");
894	yield* fake.emit({
895	type: "extension_ui_request",
896	id: "model-selection-dialog",
897	method: "confirm",
898	title: "Use this model?",
899	});
900	yield* fake.emit({ type: "agent_settled" });
901	yield* takeEvent((event) => {
902	if (event.type === "runtime_request.updated") requests.push(event);
903	return (
904	event.type === "provider_session.updated" && event.providerSession.status === "ready"
905	);
906	});
907	const dialog = requests[0]?.runtimeRequest;
908	assert.isDefined(dialog, "Pi buffered the dialog needed to finish model selection");
909	assert.equal(requests[0]?.threadId, THREAD_ID);
910	assert.isNull(dialog!.providerTurnId);
911	yield* runtime.respondToRuntimeRequest({ requestId: dialog!.id, decision: "accept" });
912	const response = yield* fake.takeRequest("extension_ui_response");
913	assert.equal(response["id"], "model-selection-dialog");
914	assert.isTrue(response["confirmed"]);
915	yield* fake.emit({
916	type: "response",
917	id: modelRequest["id"],
918	command: "set_model",
919	success: true,
920	data: { provider: "xai", id: "grok-4.6" },
921	});
922	yield* Fiber.join(starting);
923	const prompt = yield* fake.takeRequest("prompt");
924	assert.equal(prompt["message"], "Continue the review");
925	assert.equal(prompt["streamingBehavior"], "steer");
926	const resolved = yield* takeEvent(
927	(event) =>
928	event.type === "runtime_request.updated" && event.runtimeRequest.id === dialog!.id,
929	);
930	assert.isTrue(
931	resolved.type === "runtime_request.updated" &&
932	resolved.runtimeRequest.status === "resolved",
933	);
934	}).pipe(Effect.scoped, Effect.provide(layerTest)),
935	);
936
937	it.effect("adopts a completed native wake without sending a synthetic prompt", () =>
938	Effect.gen(function* () {
939	const fake = yield* makeFakePi;
940	const offers =
941	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
942	const { runtime, takeEvent } = yield* openRuntime(
943	fake,
944	"default",
945	THREAD_ID,
946	SESSION_ID,
947	undefined,
948	{
949	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
950	},
951	);
952	const providerThread = yield* runtime.ensureThread({
953	threadId: THREAD_ID,
954	modelSelection: modelSelection("default"),
955	runtimePolicy,
956	});
957	yield* fake.emit({ type: "agent_start" });
958	const offer = yield* Queue.take(offers).pipe(
959	Effect.raceFirst(
960	takeEvent(
961	(event) =>
962	event.type === "provider_session.updated" && event.providerSession.status === "error",
963	).pipe(
964	Effect.andThen(Effect.sync(() => assert.fail("Pi killed a routable native wake"))),
965	),
966	),
967	);
968	assert.equal(offer.threadId, THREAD_ID);
969	assert.equal(offer.providerThreadId, providerThread.id);
970	assert.equal(offer.delivery, "adapter_buffered");
971	assert.isTrue(yield* runtime.hasPendingBackgroundWork!);
972	let dispatchCommitted = false;
973	yield* offer.dispatchIfCurrent!(
974	Effect.sync(() => {
975	dispatchCommitted = true;
976	}),
977	);
978	assert.isTrue(dispatchCommitted);
979	assert.isTrue(yield* runtime.hasPendingBackgroundWork!);
980	yield* fake.emit({ type: "message_start", message: { role: "assistant" } });
981	yield* fake.emit({
982	type: "message_update",
983	assistantMessageEvent: {
984	type: "text_delta",
985	contentIndex: 0,
986	delta: "Background review finished",
987	},
988	});
989	yield* fake.emit({ type: "message_end", message: { role: "assistant" } });
990	yield* fake.emit({ type: "agent_settled" });
991	yield* takeEvent(
992	(event) =>
993	event.type === "provider_session.updated" && event.providerSession.status === "ready",
994	);
995	assert.isTrue(yield* runtime.hasPendingBackgroundWork!);
996	yield* startTurn(
997	runtime,
998	providerThread,
999	"default",
1000	[],
1001	"Synthetic wake",
1002	undefined,
1003	1,
1004	THREAD_ID,
1005	true,
1006	);
1007	const message = yield* takeEvent((event) => event.type === "message.updated");
1008	assert.equal(
1009	message.type === "message.updated" && message.message.text,
1010	"Background review finished",
1011	);
1012	yield* takeEvent((event) => event.type === "turn.terminal");
1013	assert.equal(fake.allRequests().filter((request) => request["type"] === "prompt").length, 0);
1014	assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
1015	let dispatched = false;
1016	yield* offer.dispatchIfCurrent!(
1017	Effect.sync(() => {
1018	dispatched = true;
1019	}),
1020	);
1021	assert.isFalse(dispatched);
1022	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1023	);
1024
1025	it.effect.each([false, true])(
1026	"a user prompt takes the wake without premature settlement, native wake settled=%s",
1027	(settled) =>
1028	Effect.gen(function* () {
1029	const fake = yield* makeFakePi;
1030	const offers =
1031	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
1032	const { runtime, takeEvent } = yield* openRuntime(
1033	fake,
1034	"default",
1035	THREAD_ID,
1036	SESSION_ID,
1037	undefined,
1038	{
1039	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
1040	},
1041	);
1042	const providerThread = yield* runtime.ensureThread({
1043	threadId: THREAD_ID,
1044	modelSelection: modelSelection("default"),
1045	runtimePolicy,
1046	});
1047	yield* fake.emit({ type: "agent_start" });
1048	const offer = yield* Queue.take(offers);
1049	if (settled) {
1050	yield* fake.emit({ type: "agent_settled" });
1051	yield* takeEvent(
1052	(event) =>
1053	event.type === "provider_session.updated" && event.providerSession.status === "ready",
1054	);
1055	}
1056	yield* startTurn(runtime, providerThread);
1057	const prompt = yield* fake.takeRequest("prompt");
1058	assert.equal(prompt["message"], "Hello pi");
1059	assert.equal(prompt["streamingBehavior"], "steer");
1060	yield* fake.emit({ type: "response", command: "prompt", success: true });
1061	yield* fake.emit({ type: "agent_start" });
1062	yield* fake.emit({
1063	type: "extension_ui_request",
1064	id: "owned-dialog",
1065	method: "confirm",
1066	title: "Continue?",
1067	});
1068	const dialog = yield* takeEvent((event) => event.type === "runtime_request.updated");
1069	assert.isTrue(dialog.type === "runtime_request.updated");
1070	if (dialog.type !== "runtime_request.updated") return;
1071	assert.isNotNull(dialog.runtimeRequest.providerTurnId);
1072	yield* offer.clearIfCurrent!();
1073	yield* runtime.respondToRuntimeRequest({
1074	requestId: dialog.runtimeRequest.id,
1075	decision: "accept",
1076	});
1077	const response = yield* fake.takeRequest("extension_ui_response");
1078	assert.isTrue(response["confirmed"]);
1079	yield* fake.emit({ type: "message_start", message: { role: "assistant" } });
1080	yield* fake.emit({
1081	type: "message_update",
1082	assistantMessageEvent: {
1083	type: "text_end",
1084	contentIndex: 0,
1085	content: "User prompt completed",
1086	},
1087	});
1088	yield* fake.emit({ type: "agent_settled" });
1089	const message = yield* takeEvent(
1090	(event) =>
1091	event.type === "message.updated" && event.message.text === "User prompt completed",
1092	);
1093	assert.isTrue(message.type === "message.updated" && message.message.runId !== null);
1094	yield* takeEvent((event) => event.type === "turn.terminal");
1095	let dispatched = false;
1096	yield* offer.dispatchIfCurrent!(
1097	Effect.sync(() => {
1098	dispatched = true;
1099	}),
1100	);
1101	assert.isFalse(dispatched);
1102	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1103	);
1104
1105	it.effect.each([false, true])(
1106	"defers a wake's late settlement until the joined user prompt is acknowledged, aborted=%s",
1107	(aborted) =>
1108	Effect.gen(function* () {
1109	const fake = yield* makeFakePi;
1110	const offers =
1111	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
1112	const { runtime, takeEvent } = yield* openRuntime(
1113	fake,
1114	"default",
1115	THREAD_ID,
1116	SESSION_ID,
1117	undefined,
1118	{
1119	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
1120	},
1121	);
1122	const providerThread = yield* runtime.ensureThread({
1123	threadId: THREAD_ID,
1124	modelSelection: modelSelection("default"),
1125	runtimePolicy,
1126	});
1127	yield* fake.emit({ type: "agent_start" });
1128	yield* Queue.take(offers);
1129	yield* startTurn(runtime, providerThread);
1130	yield* fake.takeRequest("prompt");
1131	yield* fake.emit({ type: "agent_settled", aborted });
1132	yield* fake.emit({
1133	type: "extension_ui_request",
1134	method: "notify",
1135	message: "Ordering barrier",
1136	});
1137	yield* takeEvent(
1138	(event) =>
1139	event.type === "turn_item.updated" &&
1140	event.turnItem.type === "dynamic_tool" &&
1141	event.turnItem.toolName === "notify",
1142	);
1143	assert.equal(runtime.providerSession.status, "running");
1144	fake.queueState({ isStreaming: true });
1145	yield* fake.emit({ type: "response", command: "prompt", success: true });
1146	yield* fake.takeRequest("get_state");
1147	yield* fake.emit({ type: "agent_start" });
1148	yield* fake.emit({ type: "agent_settled" });
1149	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
1150	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
1151	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1152	);
1153
1154	it.effect.each([
1155	{ aborted: false, resumes: false, expected: "completed" },
1156	{ aborted: true, resumes: false, expected: "interrupted" },
1157	{ aborted: true, resumes: true, expected: "completed" },
1158	])(
1159	"settles a completed wake, aborted=$aborted, resumes=$resumes",
1160	({ aborted, resumes, expected }) =>
1161	Effect.gen(function* () {
1162	const fake = yield* makeFakePi;
1163	const offers =
1164	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
1165	const { runtime, takeEvent } = yield* openRuntime(
1166	fake,
1167	"default",
1168	THREAD_ID,
1169	SESSION_ID,
1170	undefined,
1171	{
1172	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
1173	},
1174	);
1175	const providerThread = yield* runtime.ensureThread({
1176	threadId: THREAD_ID,
1177	modelSelection: modelSelection("default"),
1178	runtimePolicy,
1179	});
1180	yield* fake.emit({ type: "agent_start" });
1181	yield* Queue.take(offers);
1182	yield* fake.emit({ type: "agent_settled", aborted });
1183	yield* takeEvent(
1184	(event) =>
1185	event.type === "provider_session.updated" && event.providerSession.status === "ready",
1186	);
1187	yield* startTurn(runtime, providerThread, "default", [], "/hello");
1188	yield* fake.takeRequest("prompt");
1189	if (resumes) yield* fake.emit({ type: "agent_start" });
1190	yield* fake.emit({ type: "response", command: "prompt", success: true });
1191	if (resumes) yield* fake.emit({ type: "agent_settled", aborted: false });
1192	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
1193	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === expected);
1194	yield* startTurn(
1195	runtime,
1196	providerThread,
1197	"default",
1198	[],
1199	"Stale continuation",
1200	undefined,
1201	2,
1202	THREAD_ID,
1203	true,
1204	);
1205	yield* takeEvent((event) => event.type === "turn.terminal");
1206	assert.equal(
1207	fake.allRequests().filter((request) => request["type"] === "prompt").length,
1208	1,
1209	);
1210	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1211	);
1212
1213	it.effect("keeps a pending wake on its thread and lets Stop retire its process", () =>
1214	Effect.gen(function* () {
1215	const fake = yield* makeFakePi;
1216	const offers =
1217	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
1218	const { runtime, takeEvent } = yield* openRuntime(
1219	fake,
1220	"default",
1221	THREAD_ID,
1222	SESSION_ID,
1223	undefined,
1224	{
1225	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
1226	},
1227	);
1228	const providerThread = yield* runtime.ensureThread({
1229	threadId: THREAD_ID,
1230	modelSelection: modelSelection("default"),
1231	runtimePolicy,
1232	});
1233	yield* fake.emit({ type: "agent_start" });
1234	const offer = yield* Queue.take(offers);
1235	assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(providerThread));
1236	const otherThread = {
1237	...providerThread,
1238	appThreadId: ThreadId.make("other-thread"),
1239	id: ProviderThreadId.make("other-provider-thread"),
1240	};
1241	assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(otherThread));
1242	const failure = yield* runtime
1243	.resumeThread({ providerThread: otherThread })
1244	.pipe(Effect.flip);
1245	assert.equal(failure._tag, "ProviderAdapterResumeThreadError");
1246	yield* runtime.interruptTurn({
1247	providerThread,
1248	providerTurnId: ProviderTurnId.make("settled-turn"),
1249	requestRuntimeRestart: true,
1250	});
1251	yield* takeEvent(
1252	(event) =>
1253	event.type === "provider_session.updated" && event.providerSession.status === "stopped",
1254	);
1255	assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
1256	let dispatched = false;
1257	yield* offer.dispatchIfCurrent!(
1258	Effect.sync(() => {
1259	dispatched = true;
1260	}),
1261	);
1262	assert.isFalse(dispatched);
1263	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1264	);
1265
1266	it.effect.each(["archive", "stop", "transport death"] as const)(
1267	"retiring a wake by %s cancels visible dialogs and invalidates its offer",
1268	(retirement) =>
1269	Effect.gen(function* () {
1270	const fake = yield* makeFakePi;
1271	const offers =
1272	yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
1273	const { runtime, takeEvent } = yield* openRuntime(
1274	fake,
1275	"default",
1276	THREAD_ID,
1277	SESSION_ID,
1278	undefined,
1279	{
1280	offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
1281	},
1282	);
1283	const providerThread = yield* runtime.ensureThread({
1284	threadId: THREAD_ID,
1285	modelSelection: modelSelection("default"),
1286	runtimePolicy,
1287	});
1288	yield* fake.emit({ type: "agent_start" });
1289	const offer = yield* Queue.take(offers);
1290	yield* fake.emit({
1291	type: "extension_ui_request",
1292	id: "wake-dialog",
1293	method: "confirm",
1294	title: "Continue background work?",
1295	});
1296	const dialog = yield* takeEvent((event) => event.type === "runtime_request.updated");
1297	assert.isTrue(dialog.type === "runtime_request.updated");
1298	if (dialog.type !== "runtime_request.updated") return;
1299	assert.isNull(dialog.runtimeRequest.providerTurnId);
1300	if (retirement === "archive") yield* offer.clearIfCurrent!();
1301	else if (retirement === "stop") {
1302	yield* runtime.interruptTurn({
1303	providerThread,
1304	providerTurnId: ProviderTurnId.make("settled-turn"),
1305	requestRuntimeRestart: true,
1306	});
1307	} else yield* fake.closeStdout;
1308	let requestCancelled = false;
1309	let nodeCancelled = false;
1310	let itemCancelled = false;
1311	yield* takeEvent((event) => {
1312	if (
1313	event.type === "runtime_request.updated" &&
1314	event.runtimeRequest.id === dialog.runtimeRequest.id
1315	)
1316	requestCancelled = event.runtimeRequest.status === "cancelled";
1317	if (event.type === "node.updated" && event.node.id === dialog.runtimeRequest.nodeId)
1318	nodeCancelled = event.node.status === "cancelled";
1319	if (
1320	event.type === "turn_item.updated" &&
1321	event.turnItem.nodeId === dialog.runtimeRequest.nodeId
1322	)
1323	itemCancelled = event.turnItem.status === "cancelled";
1324	return (
1325	event.type === "provider_session.updated" &&
1326	event.providerSession.status ===
1327	(retirement === "transport death" ? "error" : "stopped")
1328	);
1329	});
1330	assert.isTrue(requestCancelled, "Retired wake left a pending runtime request");
1331	assert.isTrue(nodeCancelled);
1332	assert.isTrue(itemCancelled);
1333	assert.isFalse(yield* runtime.hasPendingBackgroundWork!);
1334	let dispatched = false;
1335	yield* offer.dispatchIfCurrent!(
1336	Effect.sync(() => {
1337	dispatched = true;
1338	}),
1339	);
1340	assert.isFalse(dispatched);
1341	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1342	);
1343
1344	it.effect("stops provider-initiated work that has no T3 turn owner", () =>
1345	Effect.gen(function* () {
1346	const fake = yield* makeFakePi;
1347	const { runtime, takeEvent } = yield* openRuntime(fake);
1348	yield* runtime.ensureThread({
1349	threadId: THREAD_ID,
1350	modelSelection: modelSelection("default"),
1351	runtimePolicy,
1352	});
1353
1354	yield* fake.emit({ type: "agent_start" });
1355
1356	const sessionError = yield* takeEvent(
1357	(event) =>
1358	event.type === "provider_session.updated" && event.providerSession.status === "error",
1359	);
1360	assert.isTrue(
1361	sessionError.type === "provider_session.updated" &&
1362	sessionError.providerSession.lastError?.includes("invisible tool execution") === true,
1363	);
1364	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1365	);
1366
1367	it.effect("injects the T3 MCP extension and bearer when a session exists", () =>
1368	Effect.gen(function* () {
1369	yield* (yield* McpProviderSessions.McpProviderSessions).set({
1370	environmentId: EnvironmentId.make("environment-pi-mcp"),
1371	threadId: THREAD_ID,
1372	providerSessionId: "mcp-session-pi",
1373	providerInstanceId: PI_INSTANCE_ID,
1374	endpoint: "http://127.0.0.1:43123/mcp",
1375	authorizationHeader: "Bearer secret-pi-token",
1376	browserToolsAvailable: true,
1377	});
1378	const fake = yield* makeFakePi;
1379	yield* openRuntime(fake);
1380	const spawn = fake.lastSpawn();
1381	assert.isTrue(spawn.args.includes("--extension"));
1382	const extensions = spawn.args.flatMap((arg, index) =>
1383	arg === "--extension" ? [spawn.args[index + 1]] : [],
1384	);
1385	assert.isFalse(spawn.args.includes("--no-extensions"));
1386	assert.isTrue(extensions.some((path) => path?.endsWith("pi-t3-mcp-extension.ts")));
1387	assert.equal(spawn.env.T3_MCP_URL, "http://127.0.0.1:43123/mcp");
1388	assert.equal(spawn.env.T3_MCP_BEARER_TOKEN, "secret-pi-token");
1389	assert.equal(spawn.env.T3_PI_RUNTIME_MODE, "full-access");
1390	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1391	);
1392
1393	it.effect("rejects a resume while a turn is active", () =>
1394	Effect.gen(function* () {
1395	const fake = yield* makeFakePi;
1396	const { runtime } = yield* openRuntime(fake);
1397	const providerThread = yield* runtime.ensureThread({
1398	threadId: THREAD_ID,
1399	modelSelection: modelSelection("default"),
1400	runtimePolicy,
1401	});
1402	yield* startTurn(runtime, providerThread);
1403	yield* fake.takeRequest("prompt");
1404	const error = yield* runtime.resumeThread({ providerThread }).pipe(Effect.flip);
1405	assert.equal(error._tag, "ProviderAdapterResumeThreadError");
1406	assert.match(String(error.cause), /while a turn is active/);
1407	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1408	);
1409
1410	it.effect("waits for a slow Pi resume without starting a replacement", () =>
1411	Effect.gen(function* () {
1412	const fake = yield* makeFakePi;
1413	const { runtime } = yield* openRuntime(fake);
1414	const providerThread = yield* runtime.ensureThread({
1415	threadId: THREAD_ID,
1416	modelSelection: modelSelection("default"),
1417	runtimePolicy,
1418	});
1419	fake.deferNextLifecycle("switch_session");
1420	const resumed = yield* runtime.resumeThread({ providerThread }).pipe(Effect.forkChild);
1421	const request = yield* fake.takeRequest("switch_session");
1422	yield* TestClock.adjust(Duration.millis(16_820));
1423	yield* fake.emit({
1424	type: "response",
1425	id: request.id,
1426	command: "switch_session",
1427	success: true,
1428	data: { cancelled: false },
1429	});
1430	assert.equal((yield* Fiber.join(resumed)).nativeThreadRef?.nativeId, FAKE_SESSION_FILE);
1431	assert.isFalse(fake.allRequests().some((request) => request.type === "new_session"));
1432	yield* startTurn(runtime, providerThread, "default");
1433	yield* fake.takeRequest("prompt");
1434	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1435	);
1436
1437	it.effect("creates a distinct native session after a failed resume", () =>
1438	Effect.gen(function* () {
1439	const fake = yield* makeFakePi;
1440	const { runtime } = yield* openRuntime(fake);
1441	const providerThread = yield* runtime.ensureThread({
1442	threadId: THREAD_ID,
1443	modelSelection: modelSelection("default"),
1444	runtimePolicy,
1445	});
1446	fake.vetoNextSwitch();
1447	yield* runtime.resumeThread({ providerThread }).pipe(Effect.flip);
1448	const replacement = yield* runtime.ensureThread({
1449	threadId: THREAD_ID,
1450	modelSelection: modelSelection("default"),
1451	runtimePolicy,
1452	existingProviderThread: { ...providerThread, nativeThreadRef: null },
1453	});
1454	assert.equal(replacement.id, providerThread.id);
1455	assert.notEqual(
1456	replacement.nativeThreadRef?.nativeId,
1457	providerThread.nativeThreadRef?.nativeId,
1458	);
1459	assert.equal(
1460	fake.allRequests().filter((request) => request.type === "new_session").length,
1461	1,
1462	);
1463	yield* startTurn(runtime, replacement, "default");
1464	yield* fake.takeRequest("prompt");
1465	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1466	);
1467
1468	it.effect.each(["veto", "same identity"] as const)(
1469	"rejects a replacement with %s",
1470	(invalidReplacement) =>
1471	Effect.gen(function* () {
1472	const fake = yield* makeFakePi;
1473	const { runtime } = yield* openRuntime(fake);
1474	const providerThread = yield* runtime.ensureThread({
1475	threadId: THREAD_ID,
1476	modelSelection: modelSelection("default"),
1477	runtimePolicy,
1478	});
1479	if (invalidReplacement === "veto") fake.vetoNextNewSession();
1480	else fake.queueState({ sessionFile: FAKE_SESSION_FILE });
1481	const error = yield* runtime
1482	.ensureThread({
1483	threadId: THREAD_ID,
1484	modelSelection: modelSelection("default"),
1485	runtimePolicy,
1486	existingProviderThread: { ...providerThread, nativeThreadRef: null },
1487	})
1488	.pipe(Effect.flip);
1489	assert.equal(error._tag, "ProviderAdapterEnsureThreadError");
1490	assert.match(
1491	String(error.cause),
1492	invalidReplacement === "veto" ? /cancelled/ : /distinct session/,
1493	);
1494	yield* startTurn(runtime, providerThread, "default").pipe(Effect.flip);
1495	assert.isFalse(fake.allRequests().some((request) => request.type === "prompt"));
1496	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1497	);
1498
1499	it.effect("retires a timed-out lifecycle process before a late switch can race replacement", () =>
1500	Effect.gen(function* () {
1501	const fake = yield* makeFakePi;
1502	const { runtime, takeEvent } = yield* openRuntime(fake);
1503	const providerThread = yield* runtime.ensureThread({
1504	threadId: THREAD_ID,
1505	modelSelection: modelSelection("default"),
1506	runtimePolicy,
1507	});
1508	fake.deferNextLifecycle("switch_session");
1509	const resumed = yield* runtime
1510	.resumeThread({ providerThread })
1511	.pipe(Effect.flip, Effect.forkChild);
1512	const request = yield* fake.takeRequest("switch_session");
1513	yield* TestClock.adjust(Duration.seconds(60));
1514	const error = yield* Fiber.join(resumed);
1515	assert.match(String(error.cause), /timed out after 60000ms/);
1516	yield* takeEvent(
1517	(event) =>
1518	event.type === "provider_session.updated" && event.providerSession.status === "error",
1519	);
1520	yield* fake.emit({
1521	type: "response",
1522	id: request.id,
1523	command: "switch_session",
1524	success: true,
1525	data: { cancelled: false },
1526	});
1527	yield* runtime
1528	.ensureThread({
1529	threadId: THREAD_ID,
1530	modelSelection: modelSelection("default"),
1531	runtimePolicy,
1532	existingProviderThread: { ...providerThread, nativeThreadRef: null },
1533	})
1534	.pipe(Effect.flip);
1535	assert.isFalse(
1536	fake
1537	.allRequests()
1538	.some((request) => request.type === "new_session" || request.type === "prompt"),
1539	);
1540	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1541	);
1542
1543	it.effect(
1544	"uses the lifecycle deadline for fresh sessions and drops replaced native metadata",
1545	() =>
1546	Effect.gen(function* () {
1547	const fake = yield* makeFakePi;
1548	const { runtime } = yield* openRuntime(fake);
1549	const providerThread = yield* runtime.ensureThread({
1550	threadId: THREAD_ID,
1551	modelSelection: modelSelection("default"),
1552	runtimePolicy,
1553	});
1554	fake.deferNextLifecycle("new_session");
1555	fake.queueState({ sessionFile: "/fake/fresh-after-delay.jsonl" });
1556	const replacing = yield* runtime
1557	.ensureThread({
1558	threadId: THREAD_ID,
1559	modelSelection: modelSelection("default"),
1560	runtimePolicy,
1561	existingProviderThread: {
1562	...providerThread,
1563	nativeThreadRef: null,
1564	nativeConversationHeadRef: {
1565	driver: PI_PROVIDER,
1566	nativeId: "old-leaf",
1567	strength: "strong",
1568	},
1569	contextUsage: { usedTokens: 314_551, maxTokens: 1_000_000 },
1570	},
1571	})
1572	.pipe(Effect.forkChild);
1573	const request = yield* fake.takeRequest("new_session");
1574	yield* TestClock.adjust(Duration.millis(16_820));
1575	yield* fake.emit({
1576	type: "response",
1577	id: request.id,
1578	command: "new_session",
1579	success: true,
1580	data: { cancelled: false },
1581	});
1582	const replacement = yield* Fiber.join(replacing);
1583	assert.equal(replacement.nativeThreadRef?.nativeId, "/fake/fresh-after-delay.jsonl");
1584	assert.isNull(replacement.contextUsage);
1585	assert.isNull(replacement.nativeConversationHeadRef);
1586	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1587	);
1588
1589	it.effect("replaces a native session when the first resume's state refresh fails", () =>
1590	Effect.gen(function* () {
1591	const original = yield* makeFakePi;
1592	const originalRuntime = yield* openRuntime(original);
1593	const providerThread = yield* originalRuntime.runtime.ensureThread({
1594	threadId: THREAD_ID,
1595	modelSelection: modelSelection("default"),
1596	runtimePolicy,
1597	});
1598	const fake = yield* makeFakePi;
1599	const { runtime } = yield* openRuntime(fake);
1600	fake.failNextState();
1601	yield* runtime.resumeThread({ providerThread }).pipe(Effect.flip);
1602	const replacement = yield* runtime.ensureThread({
1603	threadId: THREAD_ID,
1604	modelSelection: modelSelection("default"),
1605	runtimePolicy,
1606	existingProviderThread: { ...providerThread, nativeThreadRef: null },
1607	});
1608	assert.notEqual(
1609	replacement.nativeThreadRef?.nativeId,
1610	providerThread.nativeThreadRef?.nativeId,
1611	);
1612	yield* startTurn(runtime, replacement, "default");
1613	yield* fake.takeRequest("prompt");
1614	assert.isTrue(fake.allRequests().some((request) => request.type === "new_session"));
1615	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1616	);
1617
1618	it.effect("retires an interrupted switch before accepting further requests", () =>
1619	Effect.gen(function* () {
1620	const fake = yield* makeFakePi;
1621	const { runtime, takeEvent } = yield* openRuntime(fake);
1622	const providerThread = yield* runtime.ensureThread({
1623	threadId: THREAD_ID,
1624	modelSelection: modelSelection("default"),
1625	runtimePolicy,
1626	});
1627	fake.deferNextLifecycle("switch_session");
1628	const resumed = yield* runtime.resumeThread({ providerThread }).pipe(Effect.forkChild);
1629	yield* fake.takeRequest("switch_session");
1630	yield* Fiber.interrupt(resumed);
1631	yield* takeEvent(
1632	(event) =>
1633	event.type === "provider_session.updated" && event.providerSession.status === "error",
1634	);
1635	yield* runtime
1636	.ensureThread({
1637	threadId: THREAD_ID,
1638	modelSelection: modelSelection("default"),
1639	runtimePolicy,
1640	existingProviderThread: { ...providerThread, nativeThreadRef: null },
1641	})
1642	.pipe(Effect.flip);
1643	assert.isFalse(fake.allRequests().some((request) => request.type === "new_session"));
1644	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1645	);
1646
1647	it.effect("budgets legacy native history with Pi's selected model capacity", () =>
1648	Effect.gen(function* () {
1649	const fake = yield* makeFakePi;
1650	fake.queueState({
1651	sessionFile: FAKE_SESSION_FILE,
1652	model: { provider: "anthropic", id: "large", contextWindow: 1_000_000 },
1653	});
1654	fake.queueModels([{ provider: "anthropic", id: "small", contextWindow: 32_000 }]);
1655	const { runtime } = yield* openRuntime(fake);
1656	const providerThread = yield* runtime.ensureThread({
1657	threadId: THREAD_ID,
1658	modelSelection: modelSelection("default"),
1659	runtimePolicy,
1660	});
1661	const budget = (model: string) =>
1662	handoffBudget({
1663	tokenCap: 16_000,
1664	userText: "$handoff",
1665	attachments: [],
1666	providerThread,
1667	nativeContextEstimate: 307_543,
1668	modelContextWindow: runtime.getModelContextWindow?.(modelSelection(model)),
1669	});
1670	assert.equal(budget("default"), 16_000);
1671	assert.equal(budget("anthropic/large"), 16_000);
1672	assert.equal(runtime.getModelContextWindow?.(modelSelection("anthropic/small")), 32_000);
1673	assert.equal(budget("anthropic/small"), 0);
1674	assert.isUndefined(runtime.getModelContextWindow?.(modelSelection("anthropic/unknown")));
1675	assert.isUndefined(
1676	runtime.getModelContextWindow?.({
1677	instanceId: ProviderInstanceId.make("other-pi"),
1678	model: "anthropic/large",
1679	}),
1680	);
1681	// New native sessions have their own default, even within one process.
1682	fake.queueState({
1683	sessionFile: "/fake/replacement.jsonl",
1684	model: { provider: "anthropic", id: "small", contextWindow: 32_000 },
1685	});
1686	yield* runtime.ensureThread({
1687	threadId: THREAD_ID,
1688	modelSelection: modelSelection("default"),
1689	runtimePolicy,
1690	existingProviderThread: { ...providerThread, nativeThreadRef: null },
1691	});
1692	assert.equal(runtime.getModelContextWindow?.(modelSelection("default")), 32_000);
1693	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1694	);
1695
1696	it.effect("adopts the run's provider thread identity instead of minting a second row", () =>
1697	Effect.gen(function* () {
1698	const fake = yield* makeFakePi;
1699	const { runtime, takeEvent } = yield* openRuntime(fake);
1700	const now = yield* DateTime.now;
1701	// The placeholder row the orchestrator creates for a first run: no
1702	// native identity yet. The adapter must bind the pi session to this
1703	// row instead of registering a second session-file-keyed row, or the
1704	// projection ends up with two live rows per app thread.
1705	const placeholder: OrchestrationV2ProviderThread = {
1706	id: ProviderThreadId.make("thread:provider:pi:native-thread:pending:run:thread-pi-test:1"),
1707	driver: PI_PROVIDER,
1708	providerInstanceId: PI_INSTANCE_ID,
1709	providerSessionId: SESSION_ID,
1710	appThreadId: THREAD_ID,
1711	ownerNodeId: null,
1712	nativeThreadRef: null,
1713	nativeConversationHeadRef: null,
1714	status: "not_loaded",
1715	firstRunOrdinal: 1,
1716	lastRunOrdinal: 1,
1717	handoffIds: [],
1718	forkedFrom: null,
1719	createdAt: now,
1720	updatedAt: now,
1721	};
1722	const providerThread = yield* runtime.ensureThread({
1723	threadId: THREAD_ID,
1724	modelSelection: modelSelection("default"),
1725	runtimePolicy,
1726	existingProviderThread: placeholder,
1727	});
1728	assert.equal(providerThread.id, placeholder.id);
1729	assert.equal(providerThread.nativeThreadRef?.nativeId, FAKE_SESSION_FILE);
1730	assert.isFalse(fake.allRequests().some((request) => request.type === "new_session"));
1731	const updated = yield* takeEvent((event) => event.type === "provider_thread.updated");
1732	assert.isTrue(
1733	updated.type === "provider_thread.updated" && updated.providerThread.id === placeholder.id,
1734	);
1735	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1736	);
1737
1738	it.effect("resets applied thinking when returning to Pi default", () =>
1739	Effect.gen(function* () {
1740	const fake = yield* makeFakePi;
1741	fake.queueState({
1742	model: { provider: "xai", id: "grok-4.6" },
1743	thinkingLevel: "medium",
1744	sessionFile: FAKE_SESSION_FILE,
1745	sessionId: "abc",
1746	});
1747	const { runtime, takeEvent } = yield* openRuntime(fake);
1748	const providerThread = yield* runtime.ensureThread({
1749	threadId: THREAD_ID,
1750	modelSelection: modelSelection("default"),
1751	runtimePolicy,
1752	});
1753
1754	// An explicit effort on a concrete model.
1755	yield* startTurn(runtime, providerThread, "default", [], "Hello pi", {
1756	instanceId: PI_INSTANCE_ID,
1757	model: "xai/grok-4.6",
1758	options: [{ id: "thinking", value: "high" }],
1759	});
1760	const modelRequest = yield* fake.takeRequest("set_model");
1761	assert.equal(modelRequest["provider"], "xai");
1762	assert.equal(modelRequest["modelId"], "grok-4.6");
1763	const levelRequest = yield* fake.takeRequest("set_thinking_level");
1764	assert.equal(levelRequest["level"], "high");
1765	yield* fake.emit({ type: "agent_start" });
1766	yield* fake.emit({ type: "agent_end", messages: [], willRetry: false });
1767	yield* fake.emit({ type: "agent_settled" });
1768	yield* takeEvent((event) => event.type === "turn.terminal");
1769
1770	// Back to Pi default with no explicit thinking choice of its own.
1771	yield* startTurn(
1772	runtime,
1773	providerThread,
1774	"default",
1775	[],
1776	"Hello pi",
1777	{
1778	instanceId: PI_INSTANCE_ID,
1779	model: "default",
1780	},
1781	2,
1782	);
1783	const replayModel = yield* fake.takeRequest("set_model");
1784	assert.equal(replayModel["provider"], "xai");
1785	assert.equal(replayModel["modelId"], "grok-4.6");
1786	const resetLevel = yield* fake.takeRequest("set_thinking_level");
1787	assert.equal(resetLevel["level"], "medium");
1788	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1789	);
1790
1791	it.effect("preserves selected skill references for the Pi extension to expand", () =>
1792	Effect.gen(function* () {
1793	const fake = yield* makeFakePi;
1794	const { runtime } = yield* openRuntime(fake);
1795	const providerThread = yield* runtime.ensureThread({
1796	threadId: THREAD_ID,
1797	modelSelection: modelSelection("default"),
1798	runtimePolicy,
1799	});
1800
1801	yield* startTurn(
1802	runtime,
1803	providerThread,
1804	"default",
1805	[],
1806	"Review this change please $repo-review",
1807	);
1808	const prompt = yield* fake.takeRequest("prompt");
1809	assert.equal(prompt["message"], "Review this change please $repo-review");
1810	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1811	);
1812
1813	it.effect.each([
1814	{ historical: false, label: "the latest turn" },
1815	{ historical: true, label: "a historical turn" },
1816	])("natively forks $label into an independent session", ({ historical }) =>
1817	Effect.gen(function* () {
1818	const fake = yield* makeFakePi;
1819	const forkFake = yield* makeFakePi;
1820	const forkFile = "/fake/forked.jsonl";
1821	const { runtime, takeEvent } = yield* openRuntime(
1822	fake,
1823	"default",
1824	THREAD_ID,
1825	SESSION_ID,
1826	forkFake,
1827	);
1828	const source = yield* runtime.ensureThread({
1829	threadId: THREAD_ID,
1830	modelSelection: modelSelection("default"),
1831	runtimePolicy,
1832	});
1833	const turn = (ordinal: number): OrchestrationV2ProviderTurn => ({
1834	id: ProviderTurnId.make(`turn-${ordinal}`),
1835	providerThreadId: source.id,
1836	nodeId: NodeId.make(`node-${ordinal}`),
1837	runAttemptId: null,
1838	nativeTurnRef: { driver: PI_PROVIDER, nativeId: `u${ordinal}`, strength: "strong" },
1839	ordinal,
1840	status: "completed",
1841	startedAt: null,
1842	completedAt: null,
1843	});
1844	forkFake.queueState({ sessionFile: forkFile });
1845	fake.queueState({ sessionFile: forkFile });
1846	const target = ThreadId.make("fork-target");
1847	const forked = yield* runtime.forkThread({
1848	sourceProviderThread: source,
1849	sourceProviderTurns: historical ? [turn(1), turn(2)] : [turn(1)],
1850	providerTurnId: turn(1).id,
1851	targetThreadId: target,
1852	});
1853	assert.equal(forked.appThreadId, target);
1854	assert.equal(forked.nativeThreadRef?.nativeId, forkFile);
1855	assert.notEqual(forked.id, source.id);
1856	assert.equal(source.nativeThreadRef?.nativeId, FAKE_SESSION_FILE);
1857	const args = forkFake.lastSpawn().args;
1858	assert.equal(args[args.indexOf("--fork") + 1], FAKE_SESSION_FILE);
1859	assert.include(args, "--no-extensions");
1860	assert.include(args, "--no-tools");
1861	assert.notInclude(args, "--no-session");
1862	assert.deepEqual(
1863	forkFake
1864	.allRequests()
1865	.filter((request) => request.type === "fork")
1866	.map((request) => request.entryId),
1867	historical ? ["u2"] : [],
1868	);
1869	assert.isFalse(
1870	fake.allRequests().some((request) => request.type === "fork" || request.type === "clone"),
1871	);
1872	// ProviderTurnStartService adopts the fork into its pending row.
1873	const adopted = { ...forked, id: ProviderThreadId.make("pending-fork-row") };
1874	yield* startTurn(runtime, adopted, "default", [], "Continue", undefined, 1, target);
1875	yield* fake.emit({ type: "agent_start" });
1876	yield* fake.emit({ type: "agent_settled" });
1877	const updated = yield* takeEvent(
1878	(event) =>
1879	event.type === "provider_thread.updated" && event.providerThread.appThreadId === target,
1880	);
1881	assert.isTrue(
1882	updated.type === "provider_thread.updated" && updated.providerThread.id === adopted.id,
1883	);
1884	yield* takeEvent((event) => event.type === "turn.terminal");
1885	yield* runtime.resumeThread({ providerThread: adopted });
1886	assert.equal(
1887	fake.allRequests().findLast((request) => request.type === "switch_session")?.sessionPath,
1888	forkFile,
1889	);
1890	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1891	);
1892
1893	it.effect("presents explicitly namespaced MCP extension tools", () =>
1894	Effect.gen(function* () {
1895	const fake = yield* makeFakePi;
1896	const { runtime, takeEvent } = yield* openRuntime(fake);
1897	const providerThread = yield* runtime.ensureThread({
1898	threadId: THREAD_ID,
1899	modelSelection: modelSelection("default"),
1900	runtimePolicy,
1901	});
1902	yield* startTurn(runtime, providerThread);
1903	yield* fake.takeRequest("prompt");
1904	yield* fake.emit({ type: "agent_start" });
1905	for (const type of ["tool_execution_start", "tool_execution_end"]) {
1906	yield* fake.emit({
1907	type,
1908	toolCallId: "weather-call",
1909	toolName: "mcp__weather__get_weather",
1910	args: { city: "Berlin" },
1911	result: { content: [{ type: "text", text: "Sunny" }] },
1912	isError: false,
1913	});
1914	const event = yield* takeEvent(
1915	(event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
1916	);
1917	if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
1918	return yield* Effect.die("Expected an MCP tool item");
1919	assert.equal(event.turnItem.title, "get weather");
1920	assert.equal(
1921	event.turnItem.status,
1922	type === "tool_execution_start" ? "running" : "completed",
1923	);
1924	assert.deepEqual(event.turnItem.toolSource, {
1925	key: "mcp:weather",
1926	name: "weather",
1927	kind: "integration",
1928	});
1929	assert.deepEqual(event.turnItem.input, { city: "Berlin" });
1930	}
1931	}).pipe(Effect.scoped, Effect.provide(layerTest)),
1932	);
1933
1934	it.effect(
1935	"keeps native and MCP images available to the shared asset reader with typed results",
1936	() =>
1937	Effect.gen(function* () {
1938	const fake = yield* makeFakePi;
1939	const { runtime, takeEvent } = yield* openRuntime(fake);
1940	const providerThread = yield* runtime.ensureThread({
1941	threadId: THREAD_ID,
1942	modelSelection: modelSelection("default"),
1943	runtimePolicy,
1944	});
1945	yield* startTurn(runtime, providerThread);
1946	yield* fake.takeRequest("prompt");
1947	yield* fake.emit({ type: "agent_start" });
1948	const image = {
1949	type: "image",
1950	mimeType: "image/png",
1951	data: Buffer.alloc(64 * 1024, 0x61).toString("base64"),
1952	};
1953	const note = "Read image file [image/png]";
1954	const content = [{ type: "text", text: note }, image];
1955	const structured = { count: 2, ready: true, paths: ["one.png", "two.png"] };
1956	const scriptResult = { content, structuredContent: { threadId: "child-thread" } };
1957	const cases = [
1958	{
1959	toolName: "read",
1960	result: { content, structuredContent: { ...image, note } },
1961	expected: { content },
1962	},
1963	{
1964	toolName: "image_generate",
1965	result: { content, structuredContent: structured },
1966	expected: { content, structuredContent: structured },
1967	},
1968	{
1969	toolName: "mcp__t3-code__preview_snapshot",
1970	result: {
1971	content,
1972	structuredContent: scriptResult,
1973	details: { server: "t3-code", tool: "preview_snapshot" },
1974	},
1975	expected: { content, structuredContent: { threadId: "child-thread" } },
1976	},
1977	{
1978	toolName: "structured_tool",
1979	result: { content: [], structuredContent: [false, 2, null, { ready: true }] },
1980	expected: { content: [], structuredContent: [false, 2, null, { ready: true }] },
1981	},
1982	{
1983	toolName: "mcp__custom__extension",
1984	result: { content, structuredContent: { content: ["domain content"], ready: true } },
1985	expected: { content, structuredContent: { content: ["domain content"], ready: true } },
1986	},
1987	{
1988	toolName: "mcp__t3-code__preview_snapshot",
1989	result: {
1990	content: [
1991	{ type: "text", text: "first" },
1992	{ type: "text", text: "second" },
1993	],
1994	structuredContent: {
1995	content: [
1996	{ type: "text", text: "first" },
1997	{ type: "text", text: "second" },
1998	],
1999	},
2000	details: { server: "t3-code", tool: "preview_snapshot" },
2001	},
2002	expected: {
2003	content: [
2004	{ type: "text", text: "first" },
2005	{ type: "text", text: "second" },
2006	],
2007	},
2008	},
2009	];
2010	for (const [index, test] of cases.entries()) {
2011	for (const phase of ["update", "end"] as const) {
2012	yield* fake.emit({
2013	type: `tool_execution_${phase}`,
2014	toolCallId: `image-${index}`,
2015	toolName: test.toolName,
2016	...(phase === "end" ? { result: test.result } : { partialResult: test.result }),
2017	isError: false,
2018	});
2019	const event = yield* takeEvent(
2020	(event) =>
2021	event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2022	);
2023	if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2024	return yield* Effect.die("Expected an image tool item");
2025	assert.equal(event.turnItem.status, phase === "end" ? "completed" : "running");
2026	assert.deepEqual(event.turnItem.output, test.expected);
2027	assert.deepEqual(
2028	toolOutputImages(event.turnItem.output),
2029	test.expected.content.some((block) => block.type === "image")
2030	? [{ mimeType: "image/png", data: image.data }]
2031	: [],
2032	);
2033	if (index === 0) assert.equal(turnItemOutputText(event.turnItem), note);
2034	if (index === 5) assert.equal(turnItemOutputText(event.turnItem), "first\nsecond");
2035	if (index === 2) {
2036	assert.deepEqual(compactDynamicToolOutput(event.turnItem.output), {
2037	threadId: "child-thread",
2038	});
2039	assert.equal(JSON.stringify(event.turnItem.output).split(image.data).length - 1, 1);
2040	}
2041	}
2042	}
2043	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2044	);
2045
2046	it.effect(
2047	"keeps large tool text readable, drops identical read mirrors, and bounds structured values",
2048	() =>
2049	Effect.gen(function* () {
2050	const fake = yield* makeFakePi;
2051	const { runtime, takeEvent } = yield* openRuntime(fake);
2052	const providerThread = yield* runtime.ensureThread({
2053	threadId: THREAD_ID,
2054	modelSelection: modelSelection("default"),
2055	runtimePolicy,
2056	});
2057	yield* startTurn(runtime, providerThread);
2058	yield* fake.takeRequest("prompt");
2059	yield* fake.emit({ type: "agent_start" });
2060	const text = "hello\nworld\n" + "🙂".repeat(40_000);
2061	const content = [{ type: "text", text }];
2062	const cases = [
2063	{ content, structuredContent: text },
2064	{ content, structuredContent: { oversized: "x".repeat(128 * 1024) } },
2065	{ content, structuredContent: Array.from({ length: 10_000 }, () => 1) },
2066	{
2067	content,
2068	structuredContent: Array.from({ length: 40 }).reduce<unknown>(
2069	(nested) => ({ nested }),
2070	null,
2071	),
2072	},
2073	];
2074	for (const [index, result] of cases.entries()) {
2075	yield* fake.emit({
2076	type: "tool_execution_end",
2077	toolCallId: `structured-${index}`,
2078	toolName: "read",
2079	result,
2080	isError: true,
2081	});
2082	const event = yield* takeEvent(
2083	(event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2084	);
2085	if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2086	return yield* Effect.die("Expected a tool item");
2087	assert.equal(event.turnItem.status, "failed");
2088	assert.notProperty(event.turnItem.output, "structuredContent");
2089	const displayed = turnItemOutputText(event.turnItem);
2090	if (index === 0) {
2091	assert.equal(event.turnItem.output, text);
2092	assert.equal(displayed, text);
2093	} else {
2094	assert.equal(
2095	displayed,
2096	text + "\nStructured output omitted because it exceeds the stored result limit.",
2097	);
2098	assert.deepEqual(Object.keys(event.turnItem.output as object), ["content"]);
2099	}
2100	}
2101	// Distinct typed metadata remains stored even when readable text exists.
2102	yield* fake.emit({
2103	type: "tool_execution_end",
2104	toolCallId: "distinct-metadata",
2105	toolName: "read",
2106	result: { content, structuredContent: { path: "read.ts", ready: true } },
2107	});
2108	const event = yield* takeEvent(
2109	(event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2110	);
2111	if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2112	return yield* Effect.die("Expected a tool item");
2113	assert.deepEqual(event.turnItem.output, {
2114	content,
2115	structuredContent: { path: "read.ts", ready: true },
2116	});
2117	assert.equal(turnItemOutputText(event.turnItem), text);
2118	const note = "Read image file [image/png]";
2119	const image = { type: "image", data: "AAAA", mimeType: "image/png" };
2120	const imageContent = [{ type: "text", text: note }, image];
2121	// An extension can attach distinct typed metadata to a read result.
2122	for (const structuredContent of [
2123	{ ...image, note, width: 1200 },
2124	{ ...image, note: "A distinct note" },
2125	{ ...image, note, data: "AQID" },
2126	]) {
2127	yield* fake.emit({
2128	type: "tool_execution_end",
2129	toolCallId: "distinct-image-metadata",
2130	toolName: "read",
2131	result: { content: imageContent, structuredContent },
2132	});
2133	const event = yield* takeEvent(
2134	(event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2135	);
2136	if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2137	return yield* Effect.die("Expected a read image tool item");
2138	assert.deepEqual(event.turnItem.output, { content: imageContent, structuredContent });
2139	assert.equal(turnItemOutputText(event.turnItem), note);
2140	}
2141	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2142	);
2143
2144	it.effect(
2145	"delivers realistic MCP screenshots through the bridge, JSONL transport, and adapter",
2146	() =>
2147	Effect.gen(function* () {
2148	const fake = yield* makeFakePi;
2149	const { runtime, takeEvent } = yield* openRuntime(fake);
2150	const providerThread = yield* runtime.ensureThread({
2151	threadId: THREAD_ID,
2152	modelSelection: modelSelection("default"),
2153	runtimePolicy,
2154	});
2155	yield* startTurn(runtime, providerThread);
2156	yield* fake.takeRequest("prompt");
2157	yield* fake.emit({ type: "agent_start" });
2158	for (const size of [3.25 * 1024 * 1024, 10 * 1024 * 1024]) {
2159	const pixels = Buffer.alloc(size, 0x61);
2160	const data = pixels.toString("base64");
2161	const content = [
2162	{ type: "text", text: "Screenshot captured" },
2163	{ type: "image", data, mimeType: "image/png" },
2164	];
2165	const metadata = { screenshot: { width: 1200, height: 800 }, ready: true };
2166	const bridge = yield* Effect.promise(() =>
2167	loadMcpBridge({ modern: true, result: { content, structuredContent: metadata } }),
2168	);
2169	const tool = bridge.tools.find((tool) => tool.name === "mcp__t3-code__preview_snapshot")!;
2170	const result = yield* Effect.promise(() => tool.execute("screenshot", {}));
2171	assert.deepEqual(result.structuredContent, { content, structuredContent: metadata });
2172	yield* fake.emit(
2173	{
2174	type: "tool_execution_end",
2175	toolCallId: "screenshot",
2176	toolName: tool.name,
2177	result,
2178	isError: false,
2179	},
2180	64 * 1024,
2181	);
2182	// A following small completion proves a dropped screenshot immediately,
2183	// without waiting for a timeout when framing loses the large event.
2184	yield* fake.emit({
2185	type: "tool_execution_end",
2186	toolCallId: "after-screenshot",
2187	toolName: "marker",
2188	result: { content: [{ type: "text", text: "after" }] },
2189	});
2190	const event = yield* takeEvent(
2191	(event) => event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool",
2192	);
2193	if (event.type !== "turn_item.updated" || event.turnItem.type !== "dynamic_tool")
2194	return yield* Effect.die("Expected a screenshot tool item");
2195	assert.equal(event.turnItem.toolName, tool.name);
2196	assert.equal(event.turnItem.status, "completed");
2197	assert.deepEqual(event.turnItem.output, { content, structuredContent: metadata });
2198	assert.equal(turnItemOutputText(event.turnItem), "Screenshot captured");
2199	const image = toolOutputImages(event.turnItem.output)[0];
2200	assert.equal(image?.mimeType, "image/png");
2201	assert.deepEqual(Buffer.from(image!.data!, "base64"), pixels);
2202	yield* takeEvent(
2203	(event) =>
2204	event.type === "turn_item.updated" &&
2205	event.turnItem.type === "dynamic_tool" &&
2206	event.turnItem.toolName === "marker",
2207	);
2208	}
2209	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2210	);
2211
2212	it.effect("observes official subagent results without inventing child threads", () =>
2213	Effect.gen(function* () {
2214	const fake = yield* makeFakePi;
2215	const { runtime, takeEvent } = yield* openRuntime(fake);
2216	const providerThread = yield* runtime.ensureThread({
2217	threadId: THREAD_ID,
2218	modelSelection: modelSelection("default"),
2219	runtimePolicy,
2220	});
2221	yield* startTurn(runtime, providerThread);
2222	yield* fake.takeRequest("prompt");
2223	yield* fake.emit({ type: "agent_start" });
2224	yield* fake.emit({
2225	type: "tool_execution_update",
2226	toolCallId: "call_sub",
2227	toolName: "subagent",
2228	partialResult: {
2229	content: [{ type: "text", text: "(running...)" }],
2230	details: {
2231	mode: "single",
2232	results: [
2233	{
2234	agent: "scout",
2235	task: "map the repo",
2236	exitCode: 0,
2237	stderr: "",
2238	sessionFile: "/ignored/custom-extension-session.jsonl",
2239	messages: [
2240	{ role: "assistant", content: [{ type: "text", text: "scanning files" }] },
2241	],
2242	},
2243	],
2244	},
2245	},
2246	});
2247	const running = yield* takeEvent(
2248	(event) => event.type === "subagent.updated" && event.subagent.status === "running",
2249	);
2250	assert.isTrue(
2251	running.type === "subagent.updated" &&
2252	running.subagent.title === "scout" &&
2253	running.subagent.prompt === "map the repo" &&
2254	running.subagent.progress === "scanning files" &&
2255	running.subagent.childThreadId === null,
2256	);
2257
2258	yield* fake.emit({
2259	type: "tool_execution_end",
2260	toolCallId: "call_sub",
2261	toolName: "subagent",
2262	isError: false,
2263	result: {
2264	content: [{ type: "text", text: "done" }],
2265	details: {
2266	mode: "single",
2267	results: [
2268	{
2269	agent: "scout",
2270	task: "map the repo",
2271	exitCode: 0,
2272	stopReason: "stop",
2273	stderr: "",
2274	messages: [
2275	{ role: "assistant", content: [{ type: "text", text: "repo has one file" }] },
2276	],
2277	},
2278	],
2279	},
2280	},
2281	});
2282	const doneCard = yield* takeEvent(
2283	(event) => event.type === "subagent.updated" && event.subagent.status === "completed",
2284	);
2285	assert.isTrue(
2286	doneCard.type === "subagent.updated" &&
2287	doneCard.subagent.result === "repo has one file" &&
2288	doneCard.subagent.childThreadId === null,
2289	);
2290	const subagentItem = yield* takeEvent(
2291	(event) =>
2292	event.type === "turn_item.updated" &&
2293	event.turnItem.type === "subagent" &&
2294	event.turnItem.status === "completed",
2295	);
2296	assert.isTrue(
2297	subagentItem.type === "turn_item.updated" &&
2298	subagentItem.turnItem.type === "subagent" &&
2299	subagentItem.turnItem.childThreadId === null,
2300	);
2301	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2302	);
2303
2304	it.effect("persists edit patches and write content on file change items", () =>
2305	Effect.gen(function* () {
2306	const fake = yield* makeFakePi;
2307	const { runtime, takeEvent } = yield* openRuntime(fake);
2308	const providerThread = yield* runtime.ensureThread({
2309	threadId: THREAD_ID,
2310	modelSelection: modelSelection("default"),
2311	runtimePolicy,
2312	});
2313	yield* startTurn(runtime, providerThread);
2314	yield* fake.takeRequest("prompt");
2315	yield* fake.emit({ type: "agent_start" });
2316	const patch = "--- a.ts\n+++ a.ts\n@@ -1 +1 @@\n-old\n+new\n";
2317	yield* fake.emit({
2318	type: "tool_execution_start",
2319	toolCallId: "call_edit",
2320	toolName: "edit",
2321	args: { path: "a.ts", edits: [{ oldText: "old", newText: "new" }] },
2322	});
2323	yield* fake.emit({
2324	type: "tool_execution_end",
2325	toolCallId: "call_edit",
2326	toolName: "edit",
2327	isError: false,
2328	result: {
2329	content: [{ type: "text", text: "Successfully replaced 1 block(s) in a.ts." }],
2330	details: { diff: "-1 old\n+1 new", patch, firstChangedLine: 1 },
2331	},
2332	});
2333	const edit = yield* takeEvent(
2334	(event) =>
2335	event.type === "turn_item.updated" &&
2336	event.turnItem.type === "file_change" &&
2337	event.turnItem.status === "completed",
2338	);
2339	assert.isTrue(
2340	edit.type === "turn_item.updated" &&
2341	edit.turnItem.type === "file_change" &&
2342	edit.turnItem.fileName === "a.ts" &&
2343	edit.turnItem.diffStr === patch,
2344	);
2345
2346	yield* fake.emit({
2347	type: "tool_execution_start",
2348	toolCallId: "call_write",
2349	toolName: "write",
2350	args: { path: "b.ts", content: "export {};\n" },
2351	});
2352	yield* fake.emit({
2353	type: "tool_execution_end",
2354	toolCallId: "call_write",
2355	toolName: "write",
2356	isError: false,
2357	result: { content: [{ type: "text", text: "Successfully wrote to b.ts" }] },
2358	});
2359	const write = yield* takeEvent(
2360	(event) =>
2361	event.type === "turn_item.updated" &&
2362	event.turnItem.type === "file_change" &&
2363	event.turnItem.status === "completed",
2364	);
2365	assert.isTrue(
2366	write.type === "turn_item.updated" &&
2367	write.turnItem.type === "file_change" &&
2368	write.turnItem.fileName === "b.ts" &&
2369	write.turnItem.newStr === "export {};\n",
2370	);
2371
2372	// A failed edit has no patch, so it keeps the error to show when expanded.
2373	yield* fake.emit({
2374	type: "tool_execution_start",
2375	toolCallId: "call_edit_failed",
2376	toolName: "edit",
2377	args: { path: "c.ts", edits: [{ oldText: "missing", newText: "new" }] },
2378	});
2379	yield* fake.emit({
2380	type: "tool_execution_end",
2381	toolCallId: "call_edit_failed",
2382	toolName: "edit",
2383	isError: true,
2384	result: { content: [{ type: "text", text: "Could not find the text in c.ts." }] },
2385	});
2386	const failedEdit = yield* takeEvent(
2387	(event) =>
2388	event.type === "turn_item.updated" &&
2389	event.turnItem.type === "file_change" &&
2390	event.turnItem.status === "failed",
2391	);
2392	assert.isTrue(
2393	failedEdit.type === "turn_item.updated" &&
2394	failedEdit.turnItem.type === "file_change" &&
2395	failedEdit.turnItem.diffStr === "Could not find the text in c.ts.",
2396	);
2397	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2398	);
2399
2400	it.effect.each([false, true])("follows an extension rewind with summarize=%s", (summarize) =>
2401	Effect.gen(function* () {
2402	const fake = yield* makeFakePi;
2403	const { runtime, takeEvent } = yield* openRuntime(fake);
2404	const root = { type: "model_change", id: "root", parentId: null };
2405	const user = (id: string, parentId: string) => ({
2406	type: "message",
2407	id,
2408	parentId,
2409	message: { role: "user", content: [{ type: "text", text: id }] },
2410	});
2411	const reply = (id: string, parentId: string) => ({
2412	type: "message",
2413	id,
2414	parentId,
2415	message: { role: "assistant", content: [] },
2416	});
2417	fake.queueEntries({ entries: [root], leafId: "root" });
2418	const providerThread = yield* runtime.ensureThread({
2419	threadId: THREAD_ID,
2420	modelSelection: modelSelection("default"),
2421	runtimePolicy,
2422	});
2423	// Runs one turn to its terminal. `listings` answer the turn's
2424	// get_entries requests in order.
2425	const settle = Effect.fnUntraced(function* (
2426	text: string,
2427	runOrdinal: number,
2428	...listings: ReadonlyArray<unknown>
2429	) {
2430	yield* startTurn(runtime, providerThread, "default", [], text, undefined, runOrdinal);
2431	yield* fake.takeRequest("prompt");
2432	for (const listing of listings) fake.queueEntries(listing);
2433	// Pi acks every prompt; a command-only prompt then settles from
2434	// an idle probe instead of agent events.
2435	yield* fake.emit({ type: "response", command: "prompt", success: true });
2436	if (!text.startsWith("/")) {
2437	yield* fake.emit({ type: "agent_start" });
2438	yield* fake.emit({ type: "agent_settled" });
2439	}
2440	const turn = yield* takeEvent(
2441	(event) =>
2442	event.type === "provider_turn.updated" && event.providerTurn.status === "completed",
2443	);
2444	const thread = yield* takeEvent(
2445	(event) =>
2446	event.type === "provider_thread.updated" && event.providerThread.status === "idle",
2447	);
2448	yield* takeEvent((event) => event.type === "turn.terminal");
2449	return {
2450	turnRef:
2451	turn.type === "provider_turn.updated" ? turn.providerTurn.nativeTurnRef : undefined,
2452	head:
2453	thread.type === "provider_thread.updated"
2454	? thread.providerThread.nativeConversationHeadRef?.nativeId
2455	: undefined,
2456	retained:
2457	thread.type === "provider_thread.updated" ? thread.retainedNativeTurnIds : undefined,
2458	};
2459	});
2460
2461	yield* settle("prompt A", 1, {
2462	entries: [user("uA", "root"), reply("aA", "uA")],
2463	leafId: "aA",
2464	});
2465	const promptB = yield* settle("prompt B", 2, {
2466	entries: [user("uB", "aA"), reply("aB", "uB")],
2467	leafId: "aB",
2468	});
2469	assert.equal(promptB.turnRef?.nativeId, "uB");
2470	assert.isUndefined(promptB.retained);
2471
2472	// The extension moves the leaf back to A's reply. With a summary,
2473	// pi appends a branch_summary child of that reply as the new leaf.
2474	const rewoundLeaf = summarize ? "summary" : "aA";
2475	const summary = summarize ? [{ type: "branch_summary", id: "summary", parentId: "aA" }] : [];
2476	const rewind = yield* settle(
2477	"/rewind",
2478	3,
2479	{ entries: summary, leafId: rewoundLeaf },
2480	{
2481	entries: [
2482	root,
2483	user("uA", "root"),
2484	reply("aA", "uA"),
2485	user("uB", "aA"),
2486	reply("aB", "uB"),
2487	...summary,
2488	],
2489	leafId: rewoundLeaf,
2490	},
2491	);
2492	assert.deepEqual(rewind.retained, ["uA"]);
2493	assert.equal(rewind.head, rewoundLeaf);
2494	if (summarize) {
2495	// The summary is this turn's only entry, and fork cannot re-root
2496	// before a non-user entry, so rollback may not pass it.
2497	assert.equal(rewind.turnRef?.strength, "weak");
2498	} else {
2499	// Nothing of this turn is on the branch; rollback can pass it.
2500	assert.isNull(rewind.turnRef);
2501	}
2502
2503	// get_entries is append-ordered: after a rewind to an older leaf,
2504	// the abandoned branch still comes first in the next window.
2505	const promptC = yield* settle("prompt C", 4, {
2506	entries: [
2507	...(summarize ? [] : [user("uB", "aA"), reply("aB", "uB")]),
2508	user("uC", rewoundLeaf),
2509	reply("aC", "uC"),
2510	],
2511	leafId: "aC",
2512	});
2513	assert.deepEqual(promptC.turnRef, {
2514	driver: PI_PROVIDER,
2515	nativeId: "uC",
2516	strength: "strong",
2517	});
2518	assert.isUndefined(promptC.retained);
2519	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2520	);
2521
2522	// Corrupted session data can loop its parentId chain. Finalizing the turn
2523	// must still finish, and must not report a branch to roll runs back to.
2524	it.effect.each(["turn window", "full tree"] as const)(
2525	"finishes a turn whose %s has a parentId cycle",
2526	(cycleIn) =>
2527	Effect.gen(function* () {
2528	const fake = yield* makeFakePi;
2529	const { runtime, takeEvent } = yield* openRuntime(fake);
2530	const root = { type: "model_change", id: "root", parentId: null };
2531	const cycle = [
2532	{ type: "message", id: "x", parentId: "y", message: { role: "user", content: [] } },
2533	{ type: "message", id: "y", parentId: "x", message: { role: "assistant", content: [] } },
2534	];
2535	fake.queueEntries({ entries: [root], leafId: "root" });
2536	const providerThread = yield* runtime.ensureThread({
2537	threadId: THREAD_ID,
2538	modelSelection: modelSelection("default"),
2539	runtimePolicy,
2540	});
2541	yield* startTurn(runtime, providerThread, "default", [], "prompt", undefined, 1);
2542	yield* fake.takeRequest("prompt");
2543	if (cycleIn === "turn window") {
2544	fake.queueEntries({ entries: cycle, leafId: "x" });
2545	} else {
2546	// The leaf is outside the `since` window, so this reads as a rewind
2547	// and the full tree is walked for the surviving branch.
2548	fake.queueEntries({ entries: [], leafId: "x" });
2549	fake.queueEntries({ entries: [root, ...cycle], leafId: "x" });
2550	}
2551	yield* fake.emit({ type: "response", command: "prompt", success: true });
2552	yield* fake.emit({ type: "agent_start" });
2553	yield* fake.emit({ type: "agent_settled" });
2554	yield* takeEvent(
2555	(event) =>
2556	event.type === "provider_turn.updated" && event.providerTurn.status === "completed",
2557	);
2558	const thread = yield* takeEvent(
2559	(event) =>
2560	event.type === "provider_thread.updated" && event.providerThread.status === "idle",
2561	);
2562	assert.isUndefined(
2563	thread.type === "provider_thread.updated" ? thread.retainedNativeTurnIds : null,
2564	);
2565	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2566	);
2567
2568	it.effect.each(["missing parent", "invalid parent", "duplicate id"] as const)(
2569	"does not reconcile malformed session entries: %s",
2570	(malformed) =>
2571	Effect.gen(function* () {
2572	const fake = yield* makeFakePi;
2573	const { runtime, takeEvent } = yield* openRuntime(fake);
2574	const root = { type: "model_change", id: "root", parentId: null };
2575	fake.queueEntries({ entries: [root], leafId: "root" });
2576	const providerThread = yield* runtime.ensureThread({
2577	threadId: THREAD_ID,
2578	modelSelection: modelSelection("default"),
2579	runtimePolicy,
2580	});
2581	yield* startTurn(runtime, providerThread);
2582	yield* fake.takeRequest("prompt");
2583	const entry = {
2584	type: "message",
2585	id: "x",
2586	message: { role: "user", content: [] },
2587	...(malformed === "missing parent"
2588	? {}
2589	: { parentId: malformed === "invalid parent" ? 42 : null }),
2590	};
2591	fake.queueEntries({ entries: [], leafId: "x" });
2592	fake.queueEntries({
2593	entries: [
2594	root,
2595	entry,
2596	...(malformed === "duplicate id" ? [{ ...entry, parentId: "root" }] : []),
2597	],
2598	leafId: "x",
2599	});
2600	yield* fake.emit({ type: "response", command: "prompt", success: true });
2601	yield* fake.emit({ type: "agent_start" });
2602	yield* fake.emit({ type: "agent_settled" });
2603	yield* takeEvent(
2604	(event) =>
2605	event.type === "provider_turn.updated" && event.providerTurn.status === "completed",
2606	);
2607	const thread = yield* takeEvent(
2608	(event) =>
2609	event.type === "provider_thread.updated" && event.providerThread.status === "idle",
2610	);
2611	assert.isUndefined(
2612	thread.type === "provider_thread.updated" ? thread.retainedNativeTurnIds : null,
2613	);
2614	yield* takeEvent((event) => event.type === "turn.terminal");
2615	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2616	);
2617
2618	it.effect("does not reconcile a replacement session tree as a native rewind", () =>
2619	Effect.gen(function* () {
2620	const fake = yield* makeFakePi;
2621	fake.queueEntries({
2622	entries: [{ type: "model_change", id: "old-leaf", parentId: null }],
2623	leafId: "old-leaf",
2624	});
2625	const { runtime, takeEvent } = yield* openRuntime(fake);
2626	const providerThread = yield* runtime.ensureThread({
2627	threadId: THREAD_ID,
2628	modelSelection: modelSelection("default"),
2629	runtimePolicy,
2630	});
2631	const finish = Effect.fnUntraced(function* () {
2632	yield* startTurn(runtime, providerThread);
2633	yield* fake.takeRequest("prompt");
2634	yield* fake.emit({ type: "response", command: "prompt", success: true });
2635	yield* fake.emit({ type: "agent_start" });
2636	yield* fake.emit({ type: "agent_settled" });
2637	const update = yield* takeEvent(
2638	(event) =>
2639	event.type === "provider_thread.updated" && event.providerThread.status === "idle",
2640	);
2641	yield* takeEvent((event) => event.type === "turn.terminal");
2642	assert.isUndefined(
2643	update.type === "provider_thread.updated" ? update.retainedNativeTurnIds : null,
2644	);
2645	});
2646	// A failed cursor read is followed by a full listing from a new session.
2647	fake.queueEntries({ entries: [] });
2648	yield* finish();
2649	fake.queueEntries({
2650	entries: [
2651	{
2652	type: "message",
2653	id: "new-user",
2654	parentId: null,
2655	message: { role: "user", content: [] },
2656	},
2657	],
2658	leafId: "new-user",
2659	});
2660	yield* finish();
2661	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2662	);
2663
2664	it.effect("rolls back past turns that left nothing in the session tree", () =>
2665	Effect.gen(function* () {
2666	const fake = yield* makeFakePi;
2667	const { runtime } = yield* openRuntime(fake);
2668	const providerThread = yield* runtime.ensureThread({
2669	threadId: THREAD_ID,
2670	modelSelection: modelSelection("default"),
2671	runtimePolicy,
2672	});
2673	const turn = (ordinal: number, entryId: string | null): OrchestrationV2ProviderTurn => ({
2674	id: ProviderTurnId.make(`turn-${ordinal}`),
2675	providerThreadId: providerThread.id,
2676	nodeId: NodeId.make(`node-${ordinal}`),
2677	runAttemptId: null,
2678	nativeTurnRef:
2679	entryId === null ? null : { driver: PI_PROVIDER, nativeId: entryId, strength: "strong" },
2680	ordinal,
2681	status: "completed",
2682	startedAt: null,
2683	completedAt: null,
2684	});
2685	const rollBackToFirst = (turns: ReadonlyArray<OrchestrationV2ProviderTurn>) =>
2686	runtime.rollbackThread({
2687	providerThread,
2688	target: {
2689	type: "provider_turn",
2690	checkpointId: CheckpointId.make("checkpoint-1"),
2691	appRunOrdinal: 1,
2692	providerTurn: turns[0]!,
2693	},
2694	providerThreadTurns: turns,
2695	});
2696	const forkedEntries = () =>
2697	fake
2698	.allRequests()
2699	.filter((request) => request.type === "fork")
2700	.map((request) => request.entryId);
2701
2702	// Only a turn that left nothing follows the target: already there.
2703	yield* rollBackToFirst([turn(1, "u1"), turn(2, null)]);
2704	assert.deepEqual(forkedEntries(), []);
2705	// Pi re-roots before the first later turn that left entries.
2706	yield* rollBackToFirst([turn(1, "u1"), turn(2, null), turn(3, "u3")]);
2707	assert.deepEqual(forkedEntries(), ["u3"]);
2708	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2709	);
2710
2711	it.effect("settles a command-only prompt from its deferred ack and idle probe", () =>
2712	Effect.gen(function* () {
2713	const fake = yield* makeFakePi;
2714	const { runtime, takeEvent } = yield* openRuntime(fake);
2715	const providerThread = yield* runtime.ensureThread({
2716	threadId: THREAD_ID,
2717	modelSelection: modelSelection("default"),
2718	runtimePolicy,
2719	});
2720	yield* startTurn(runtime, providerThread, "default", [], "/command-only");
2721	yield* fake.takeRequest("prompt");
2722	// A pure extension command: dialog + notify, then the deferred ack —
2723	// pi emits no agent_start/agent_settled at all.
2724	yield* fake.emit({
2725	type: "extension_ui_request",
2726	id: "ui-cmd",
2727	method: "notify",
2728	message: "done",
2729	notifyType: "info",
2730	});
2731	yield* fake.emit({ type: "response", command: "prompt", success: true });
2732	// The adapter probes get_state (auto-acked idle by the fake), then
2733	// settles the turn as completed.
2734	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
2735	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
2736	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2737	);
2738
2739	it.effect("leaves /compacted as an ordinary prompt", () =>
2740	Effect.gen(function* () {
2741	const fake = yield* makeFakePi;
2742	const { runtime } = yield* openRuntime(fake);
2743	const providerThread = yield* runtime.ensureThread({
2744	threadId: THREAD_ID,
2745	modelSelection: modelSelection("default"),
2746	runtimePolicy,
2747	});
2748	yield* startTurn(runtime, providerThread, "default", [], "/compacted please");
2749	const prompt = yield* fake.takeRequest("prompt");
2750	assert.equal(prompt["message"], "/compacted please");
2751	assert.isFalse(fake.allRequests().some((request) => request["type"] === "compact"));
2752	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2753	);
2754
2755	it.effect("fails a compact that never started", () =>
2756	Effect.gen(function* () {
2757	const fake = yield* makeFakePi;
2758	const { runtime, takeEvent } = yield* openRuntime(fake);
2759	const providerThread = yield* runtime.ensureThread({
2760	threadId: THREAD_ID,
2761	modelSelection: modelSelection("default"),
2762	runtimePolicy,
2763	});
2764	yield* startTurn(runtime, providerThread, "default", [], "/compact");
2765	yield* fake.takeRequest("compact");
2766	yield* fake.emit({
2767	type: "response",
2768	command: "compact",
2769	success: false,
2770	error: "Nothing to compact (session too small)",
2771	});
2772	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
2773	assert.isTrue(
2774	terminal.type === "turn.terminal" &&
2775	terminal.status === "failed" &&
2776	terminal.failure.message === "Nothing to compact (session too small)",
2777	);
2778	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2779	);
2780
2781	it.effect("restarts Pi when Stop interrupts a user compact", () =>
2782	Effect.gen(function* () {
2783	const fake = yield* makeFakePi;
2784	const { runtime, takeEvent } = yield* openRuntime(fake);
2785	const providerThread = yield* runtime.ensureThread({
2786	threadId: THREAD_ID,
2787	modelSelection: modelSelection("default"),
2788	runtimePolicy,
2789	});
2790	yield* startTurn(runtime, providerThread, "default", [], "/compact");
2791	yield* fake.takeRequest("compact");
2792	const running = yield* takeEvent(
2793	(event) =>
2794	event.type === "provider_turn.updated" && event.providerTurn.status === "running",
2795	);
2796	const providerTurnId =
2797	running.type === "provider_turn.updated" ? running.providerTurn.id : undefined;
2798	assert.isDefined(providerTurnId);
2799	yield* fake.emit({ type: "compaction_start", reason: "manual" });
2800	yield* takeEvent(
2801	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
2802	);
2803	yield* runtime.interruptTurn({ providerThread, providerTurnId: providerTurnId! });
2804	assert.isFalse(fake.allRequests().some((request) => request["type"] === "abort"));
2805	yield* fake.closeStdout;
2806	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
2807	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "interrupted");
2808	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2809	);
2810
2811	it.effect("steers /compact as RPC compact instead of a prompt", () =>
2812	Effect.gen(function* () {
2813	const fake = yield* makeFakePi;
2814	const { runtime, takeEvent } = yield* openRuntime(fake);
2815	const providerThread = yield* runtime.ensureThread({
2816	threadId: THREAD_ID,
2817	modelSelection: modelSelection("default"),
2818	runtimePolicy,
2819	});
2820	yield* startTurn(runtime, providerThread);
2821	yield* fake.takeRequest("prompt");
2822	const running = yield* takeEvent(
2823	(event) =>
2824	event.type === "provider_turn.updated" && event.providerTurn.status === "running",
2825	);
2826	const providerTurnId =
2827	running.type === "provider_turn.updated" ? running.providerTurn.id : undefined;
2828	yield* fake.emit({ type: "agent_start" });
2829	yield* runtime.steerTurn({
2830	threadId: THREAD_ID,
2831	runId: RunId.make("run:thread-pi-test:1"),
2832	providerThread,
2833	providerTurnId: providerTurnId!,
2834	message: {
2835	messageId: "message:thread-pi-test:steer-compact" as never,
2836	text: "/compact keep the tests",
2837	attachments: [],
2838	createdBy: "user",
2839	creationSource: "web",
2840	},
2841	});
2842	const compact = yield* fake.takeRequest("compact");
2843	assert.equal(compact["customInstructions"], "keep the tests");
2844	assert.isUndefined(compact["streamingBehavior"]);
2845	yield* fake.emit({ type: "compaction_start", reason: "manual" });
2846	yield* takeEvent(
2847	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
2848	);
2849	fake.queueState({ isStreaming: false, isCompacting: false, pendingMessageCount: 0 });
2850	yield* fake.emit({
2851	type: "compaction_end",
2852	reason: "manual",
2853	result: { summary: "smaller", tokensBefore: 10_000, estimatedTokensAfter: 2_000 },
2854	aborted: false,
2855	willRetry: false,
2856	});
2857	yield* takeEvent(
2858	(event) =>
2859	event.type === "turn_item.updated" &&
2860	event.turnItem.type === "compaction" &&
2861	event.turnItem.status === "completed",
2862	);
2863	yield* fake.emit({ type: "response", command: "compact", success: true });
2864	yield* fake.emit({ type: "agent_settled" });
2865	yield* fake.takeRequest("get_state");
2866	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
2867	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
2868	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2869	);
2870
2871	it.effect("persists current xAI capacity text for the thread error banner", () =>
2872	expectModelFailure("The model is currently at capacity due to high demand."),
2873	);
2874
2875	it.effect("persists extension-normalized xAI capacity text for the thread error banner", () =>
2876	expectModelFailure(
2877	"Provider overloaded: The model is currently at capacity due to high demand.",
2878	),
2879	);
2880
2881	it.effect.each([
2882	{ toolName: "bash", exitCode: 0, legacy: false },
2883	{ toolName: "powershell", exitCode: 2, legacy: false },
2884	{ toolName: "bash", exitCode: 1, legacy: true },
2885	{ toolName: "powershell", exitCode: 0, legacy: true },
2886	])(
2887	"normalizes $toolName command results, exit=$exitCode legacy=$legacy",
2888	({ toolName, exitCode, legacy }) =>
2889	Effect.gen(function* () {
2890	const fake = yield* makeFakePi;
2891	const { runtime, takeEvent } = yield* openRuntime(fake);
2892	const providerThread = yield* runtime.ensureThread({
2893	threadId: THREAD_ID,
2894	modelSelection: modelSelection("default"),
2895	runtimePolicy,
2896	});
2897	yield* startTurn(runtime, providerThread);
2898	yield* fake.takeRequest("prompt");
2899	yield* fake.emit({
2900	type: "tool_execution_start",
2901	toolCallId: "shell-call",
2902	toolName,
2903	args: { command: "echo hello" },
2904	});
2905	const running = yield* takeEvent(
2906	(event) =>
2907	event.type === "turn_item.updated" && event.turnItem.type === "command_execution",
2908	);
2909	assert.isTrue(
2910	running.type === "turn_item.updated" &&
2911	running.turnItem.type === "command_execution" &&
2912	running.turnItem.status === "running" &&
2913	running.turnItem.input === "echo hello",
2914	);
2915	yield* fake.emit({
2916	type: "tool_execution_end",
2917	toolCallId: "shell-call",
2918	toolName,
2919	isError: exitCode !== 0,
2920	result: {
2921	content: [{ type: "text", text: "hello" }],
2922	details: { exitCode: legacy ? exitCode : 99 },
2923	...(legacy ? {} : { structuredContent: { exit_code: exitCode } }),
2924	},
2925	});
2926	const completed = yield* takeEvent(
2927	(event) =>
2928	event.type === "turn_item.updated" && event.turnItem.type === "command_execution",
2929	);
2930	assert.isTrue(
2931	completed.type === "turn_item.updated" &&
2932	completed.turnItem.type === "command_execution" &&
2933	completed.turnItem.title === toolName &&
2934	completed.turnItem.input === "echo hello" &&
2935	completed.turnItem.output === "hello" &&
2936	completed.turnItem.exitCode === exitCode &&
2937	completed.turnItem.status === (exitCode === 0 ? "completed" : "failed"),
2938	);
2939	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2940	);
2941
2942	it.effect("emits session-start dialogs before a turn exists", () =>
2943	Effect.gen(function* () {
2944	const fake = yield* makeFakePi;
2945	const { runtime, takeEvent } = yield* openRuntime(fake);
2946	yield* runtime.ensureThread({
2947	threadId: THREAD_ID,
2948	modelSelection: modelSelection("default"),
2949	runtimePolicy,
2950	});
2951	// Project-trust style prompt before any turn exists.
2952	yield* fake.emit({
2953	type: "extension_ui_request",
2954	id: "ui-trust",
2955	method: "confirm",
2956	title: "Run project extensions?",
2957	message: "This project has .pi/extensions.",
2958	});
2959	const pending = yield* takeEvent(
2960	(event) =>
2961	event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending",
2962	);
2963	const requestId =
2964	pending.type === "runtime_request.updated" ? pending.runtimeRequest.id : undefined;
2965	yield* runtime.respondToRuntimeRequest({ requestId: requestId!, decision: "accept" });
2966	const uiResponse = yield* fake.takeRequest("extension_ui_response");
2967	assert.equal(uiResponse["id"], "ui-trust");
2968	assert.equal(uiResponse["confirmed"], true);
2969	}).pipe(Effect.scoped, Effect.provide(layerTest)),
2970	);
2971
2972	it.effect("expires native dialogs and rejects stale answers without caching approval", () =>
2973	Effect.gen(function* () {
2974	const fake = yield* makeFakePi;
2975	const { runtime, takeEvent } = yield* openRuntime(fake);
2976	yield* runtime.ensureThread({
2977	threadId: THREAD_ID,
2978	modelSelection: modelSelection("default"),
2979	runtimePolicy,
2980	});
2981	const dialog = {
2982	type: "extension_ui_request",
2983	method: "confirm",
2984	title: "Allow extension?",
2985	message: "Approve this operation",
2986	};
2987	yield* fake.emit({ ...dialog, id: "timed-dialog", timeout: 1000 });
2988	const pending = yield* takeEvent(
2989	(event) => event.type === "turn_item.updated" && event.turnItem.type === "approval_request",
2990	);
2991	assert.isTrue(
2992	pending.type === "turn_item.updated" && pending.turnItem.type === "approval_request",
2993	);
2994	if (pending.type !== "turn_item.updated" || pending.turnItem.type !== "approval_request")
2995	return;
2996	const requestId = pending.turnItem.requestId;
2997	yield* TestClock.adjust(Duration.seconds(1));
2998	yield* takeEvent(
2999	(event) =>
3000	event.type === "runtime_request.updated" &&
3001	event.runtimeRequest.id === requestId &&
3002	event.runtimeRequest.status === "cancelled",
3003	);
3004	assert.equal((yield* fake.takeRequest("extension_ui_response"))["cancelled"], true);
3005	const late = yield* Effect.result(
3006	runtime.respondToRuntimeRequest({
3007	requestId: pending.turnItem.requestId,
3008	decision: "acceptForSession",
3009	}),
3010	);
3011	assert.equal(late._tag, "Failure");
3012	yield* fake.emit({ ...dialog, id: "fresh-dialog", timeout: 1000 });
3013	const fresh = yield* takeEvent(
3014	(event) =>
3015	event.type === "runtime_request.updated" &&
3016	event.runtimeRequest.status === "pending" &&
3017	event.runtimeRequest.nativeRequestRef?.nativeId === "fresh-dialog",
3018	);
3019	if (fresh.type !== "runtime_request.updated") return;
3020	yield* runtime.respondToRuntimeRequest({
3021	requestId: fresh.runtimeRequest.id,
3022	decision: "accept",
3023	});
3024	assert.equal((yield* fake.takeRequest("extension_ui_response"))["id"], "fresh-dialog");
3025	// The old expiry must not cancel an already answered request, and zero disables expiry.
3026	yield* fake.emit({ ...dialog, id: "zero-timeout", timeout: 0 });
3027	const zero = yield* takeEvent(
3028	(event) =>
3029	event.type === "runtime_request.updated" &&
3030	event.runtimeRequest.status === "pending" &&
3031	event.runtimeRequest.nativeRequestRef?.nativeId === "zero-timeout",
3032	);
3033	yield* TestClock.adjust(Duration.seconds(60));
3034	if (zero.type !== "runtime_request.updated") return;
3035	yield* runtime.respondToRuntimeRequest({
3036	requestId: zero.runtimeRequest.id,
3037	decision: "decline",
3038	});
3039	const response = yield* fake.takeRequest("extension_ui_response");
3040	assert.equal(response["id"], "zero-timeout");
3041	assert.equal(response["confirmed"], false);
3042	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3043	);
3044
3045	it.effect("cancels startup dialogs when Pi exits before a turn exists", () =>
3046	Effect.gen(function* () {
3047	const fake = yield* makeFakePi;
3048	const { runtime, takeEvent } = yield* openRuntime(fake);
3049	yield* runtime.ensureThread({
3050	threadId: THREAD_ID,
3051	modelSelection: modelSelection("default"),
3052	runtimePolicy,
3053	});
3054	yield* fake.emit({
3055	type: "extension_ui_request",
3056	id: "startup",
3057	method: "confirm",
3058	title: "Trust project?",
3059	});
3060	const pending = yield* takeEvent(
3061	(event) =>
3062	event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending",
3063	);
3064	if (pending.type !== "runtime_request.updated") return;
3065	yield* fake.closeStdout;
3066	const cancelled = yield* takeEvent(
3067	(event) =>
3068	event.type === "runtime_request.updated" &&
3069	event.runtimeRequest.id === pending.runtimeRequest.id &&
3070	event.runtimeRequest.status === "cancelled",
3071	);
3072	assert.isTrue(cancelled.type === "runtime_request.updated");
3073	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3074	);
3075
3076	it.effect("remembers session approvals only for identical confirmation content", () =>
3077	Effect.gen(function* () {
3078	const fake = yield* makeFakePi;
3079	const { runtime, takeEvent } = yield* openRuntime(fake);
3080	yield* runtime.ensureThread({
3081	threadId: THREAD_ID,
3082	modelSelection: modelSelection("default"),
3083	runtimePolicy,
3084	});
3085	// Project-trust style prompt before any turn exists.
3086	yield* fake.emit({
3087	type: "extension_ui_request",
3088	id: "ui-trust",
3089	method: "confirm",
3090	title: "Run project extensions?",
3091	message: "This project has .pi/extensions.",
3092	});
3093	const pending = yield* takeEvent(
3094	(event) =>
3095	event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending",
3096	);
3097	const requestId =
3098	pending.type === "runtime_request.updated" ? pending.runtimeRequest.id : undefined;
3099	yield* runtime.respondToRuntimeRequest({
3100	requestId: requestId!,
3101	decision: "acceptForSession",
3102	});
3103	const uiResponse = yield* fake.takeRequest("extension_ui_response");
3104	assert.equal(uiResponse["id"], "ui-trust");
3105	assert.equal(uiResponse["confirmed"], true);
3106	yield* fake.emit({
3107	type: "extension_ui_request",
3108	id: "ui-trust-again",
3109	method: "confirm",
3110	title: "Run project extensions?",
3111	message: "This project has .pi/extensions.",
3112	});
3113	assert.equal((yield* fake.takeRequest("extension_ui_response"))["id"], "ui-trust-again");
3114	yield* fake.emit({
3115	type: "extension_ui_request",
3116	id: "ui-other",
3117	method: "confirm",
3118	title: "Run project extensions?",
3119	message: "A different project.",
3120	});
3121	const other = yield* takeEvent(
3122	(event) =>
3123	event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending",
3124	);
3125	assert.isTrue(
3126	other.type === "runtime_request.updated" &&
3127	other.runtimeRequest.nativeRequestRef?.nativeId === "ui-other",
3128	);
3129	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3130	);
3131
3132	it.effect("offers an explicit empty value for extension input dialogs", () =>
3133	Effect.gen(function* () {
3134	const fake = yield* makeFakePi;
3135	const { runtime, takeEvent } = yield* openRuntime(fake);
3136	yield* runtime.ensureThread({
3137	threadId: THREAD_ID,
3138	modelSelection: modelSelection("default"),
3139	runtimePolicy,
3140	});
3141	yield* fake.emit({
3142	type: "extension_ui_request",
3143	id: "ui-input",
3144	method: "input",
3145	title: "Optional value",
3146	});
3147	const event = yield* takeEvent(
3148	(event) =>
3149	event.type === "turn_item.updated" && event.turnItem.type === "user_input_request",
3150	);
3151	assert.isTrue(
3152	event.type === "turn_item.updated" && event.turnItem.type === "user_input_request",
3153	);
3154	if (event.type !== "turn_item.updated" || event.turnItem.type !== "user_input_request")
3155	return;
3156	assert.equal(event.turnItem.questions[0]?.options[0]?.value, "");
3157	yield* runtime.respondToRuntimeRequest({
3158	requestId: event.turnItem.requestId,
3159	answers: { "ui-input": "" },
3160	});
3161	const response = yield* fake.takeRequest("extension_ui_response");
3162	assert.equal(response["value"], "");
3163	assert.isUndefined(response["cancelled"]);
3164	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3165	);
3166
3167	it.effect("keeps the complete editor prefill editable and returns exact edited text", () =>
3168	Effect.gen(function* () {
3169	const fake = yield* makeFakePi;
3170	const { runtime, takeEvent } = yield* openRuntime(fake);
3171	yield* runtime.ensureThread({
3172	threadId: THREAD_ID,
3173	modelSelection: modelSelection("default"),
3174	runtimePolicy,
3175	});
3176	const prefill = ` fix(server): retain full editor content\n\n${"detail\n".repeat(500)} `;
3177	yield* fake.emit({
3178	type: "extension_ui_request",
3179	id: "ui-editor",
3180	method: "editor",
3181	title: "Commit message",
3182	prefill,
3183	});
3184	const event = yield* takeEvent(
3185	(event) =>
3186	event.type === "turn_item.updated" && event.turnItem.type === "user_input_request",
3187	);
3188	if (event.type !== "turn_item.updated" || event.turnItem.type !== "user_input_request")
3189	return;
3190	assert.equal(event.turnItem.questions[0]?.initialAnswer, prefill);
3191	assert.deepEqual(event.turnItem.questions[0]?.options, []);
3192	assert.equal(event.turnItem.questions[0]?.question, "Commit message");
3193	const edited = " fix(server): edited commit message\n\n";
3194	yield* runtime.respondToRuntimeRequest({
3195	requestId: event.turnItem.requestId,
3196	answers: { "ui-editor": edited },
3197	});
3198	const response = yield* fake.takeRequest("extension_ui_response");
3199	assert.equal(response["value"], edited);
3200	assert.isUndefined(response["cancelled"]);
3201	yield* fake.emit({
3202	type: "extension_ui_request",
3203	id: "ui-editor-cleared",
3204	method: "editor",
3205	title: "Commit message",
3206	prefill,
3207	});
3208	const cleared = yield* takeEvent(
3209	(event) =>
3210	event.type === "turn_item.updated" &&
3211	event.turnItem.type === "user_input_request" &&
3212	event.turnItem.questions.some((question) => question.id === "ui-editor-cleared"),
3213	);
3214	if (cleared.type !== "turn_item.updated" || cleared.turnItem.type !== "user_input_request")
3215	return;
3216	yield* runtime.respondToRuntimeRequest({
3217	requestId: cleared.turnItem.requestId,
3218	answers: { "ui-editor-cleared": "" },
3219	});
3220	const emptyResponse = yield* fake.takeRequest("extension_ui_response");
3221	assert.equal(emptyResponse["value"], "");
3222	assert.isUndefined(emptyResponse["cancelled"]);
3223	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3224	);
3225
3226	it.effect("raises bridge edit confirmations as file-change approvals", () =>
3227	Effect.gen(function* () {
3228	const fake = yield* makeFakePi;
3229	const { runtime, takeEvent } = yield* openRuntime(fake);
3230	const providerThread = yield* runtime.ensureThread({
3231	threadId: THREAD_ID,
3232	modelSelection: modelSelection("default"),
3233	runtimePolicy,
3234	});
3235	yield* startTurn(runtime, providerThread);
3236	yield* fake.takeRequest("prompt");
3237	for (const [id, title, requestKind] of [
3238	["ui-edit", "Allow edit?", "file-change"],
3239	["ui-bash", "Allow bash?", "command"],
3240	["ui-ext", "Deploy to staging?", "command"],
3241	] as const) {
3242	yield* fake.emit({ type: "extension_ui_request", id, method: "confirm", title });
3243	const item = yield* takeEvent(
3244	(event) =>
3245	event.type === "turn_item.updated" && event.turnItem.type === "approval_request",
3246	);
3247	assert.isTrue(
3248	item.type === "turn_item.updated" &&
3249	item.turnItem.type === "approval_request" &&
3250	item.turnItem.requestKind === requestKind,
3251	`${title} should be ${requestKind}`,
3252	);
3253	}
3254	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3255	);
3256
3257	it.effect("reads a thread snapshot from pi's active branch", () =>
3258	Effect.gen(function* () {
3259	const fake = yield* makeFakePi;
3260	const { runtime } = yield* openRuntime(fake);
3261	const providerThread = yield* runtime.ensureThread({
3262	threadId: THREAD_ID,
3263	modelSelection: modelSelection("default"),
3264	runtimePolicy,
3265	});
3266	fake.queueMessages({
3267	messages: [
3268	{
3269	role: "user",
3270	content: "hello pi",
3271	timestamp: 1700000000000,
3272	},
3273	{
3274	role: "assistant",
3275	content: [{ type: "text", text: "hello back" }],
3276	timestamp: 1700000001000,
3277	},
3278	{ role: "toolResult", content: [] },
3279	],
3280	});
3281	const snapshot = yield* runtime.readThreadSnapshot({ providerThread });
3282	assert.equal(snapshot.messages.length, 2);
3283	assert.equal(snapshot.messages[0]!.role, "user");
3284	assert.equal(snapshot.messages[0]!.text, "hello pi");
3285	assert.equal(snapshot.messages[1]!.role, "assistant");
3286	assert.equal(snapshot.messages[1]!.text, "hello back");
3287	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3288	);
3289
3290	it.effect("keeps snapshot message identities distinct across native sessions", () =>
3291	Effect.gen(function* () {
3292	const fake = yield* makeFakePi;
3293	const { runtime } = yield* openRuntime(fake);
3294	const first = yield* runtime.ensureThread({
3295	threadId: THREAD_ID,
3296	modelSelection: modelSelection("default"),
3297	runtimePolicy,
3298	});
3299	const messages = {
3300	messages: [{ role: "user", content: "same text", timestamp: 1700000000000 }],
3301	};
3302	fake.queueMessages(messages);
3303	const a = yield* runtime.readThreadSnapshot({ providerThread: first });
3304	fake.queueState({ sessionFile: "/fake/another-session.jsonl" });
3305	const second = yield* runtime.ensureThread({
3306	threadId: ThreadId.make("second-thread"),
3307	modelSelection: modelSelection("default"),
3308	runtimePolicy,
3309	existingProviderThread: {
3310	...first,
3311	nativeThreadRef: {
3312	driver: PI_PROVIDER,
3313	nativeId: "/fake/another-session.jsonl",
3314	strength: "strong",
3315	},
3316	},
3317	});
3318	fake.queueMessages(messages);
3319	const b = yield* runtime.readThreadSnapshot({ providerThread: second });
3320	assert.notEqual(a.messages[0]!.id, b.messages[0]!.id);
3321	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3322	);
3323
3324	it.effect("rejects a nonpersistent session UUID instead of treating it as a resumable path", () =>
3325	Effect.gen(function* () {
3326	const fake = yield* makeFakePi;
3327	const { runtime } = yield* openRuntime(fake);
3328	// A --no-session Pi keeps its session in memory and reports no file.
3329	fake.queueState({ sessionFile: undefined });
3330	const result = yield* runtime
3331	.ensureThread({
3332	threadId: THREAD_ID,
3333	modelSelection: modelSelection("default"),
3334	runtimePolicy,
3335	})
3336	.pipe(Effect.result);
3337	assert.equal(result._tag, "Failure");
3338	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3339	);
3340
3341	it.effect("keeps a settled turn's late prompt rejection off the next turn", () =>
3342	Effect.gen(function* () {
3343	const fake = yield* makeFakePi;
3344	const { runtime, takeEvent } = yield* openRuntime(fake);
3345	const providerThread = yield* runtime.ensureThread({
3346	threadId: THREAD_ID,
3347	modelSelection: modelSelection("default"),
3348	runtimePolicy,
3349	});
3350	// An extension command can hold its prompt ack open past settlement.
3351	yield* startTurn(runtime, providerThread, "default", [], "/my-command");
3352	yield* fake.takeRequest("prompt");
3353	yield* fake.emit({ type: "agent_start" });
3354	yield* fake.emit({ type: "agent_settled" });
3355	const firstTerminal = yield* takeEvent((event) => event.type === "turn.terminal");
3356	assert.isTrue(firstTerminal.type === "turn.terminal" && firstTerminal.status === "completed");
3357
3358	yield* startTurn(runtime, providerThread, "default", [], "Second turn", undefined, 2);
3359	yield* fake.takeRequest("prompt");
3360	// The rejection answers the first turn's prompt. It must not consume or
3361	// fail the second turn's prompt acknowledgement.
3362	yield* fake.emit({
3363	type: "response",
3364	command: "prompt",
3365	success: false,
3366	error: "late command rejection",
3367	});
3368	yield* fake.emit({ type: "agent_start" });
3369	yield* fake.emit({ type: "agent_settled" });
3370	const secondTerminal = yield* takeEvent((event) => event.type === "turn.terminal");
3371	assert.isTrue(
3372	secondTerminal.type === "turn.terminal" && secondTerminal.status === "completed",
3373	);
3374	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3375	);
3376
3377	it.effect("shows compaction progress and completes the same activity row", () =>
3378	Effect.gen(function* () {
3379	const fake = yield* makeFakePi;
3380	const { runtime, takeEvent } = yield* openRuntime(fake);
3381	const providerThread = yield* runtime.ensureThread({
3382	threadId: THREAD_ID,
3383	modelSelection: modelSelection("default"),
3384	runtimePolicy,
3385	});
3386	yield* startTurn(runtime, providerThread);
3387	yield* fake.takeRequest("prompt");
3388	yield* fake.emit({ type: "agent_start" });
3389	yield* fake.emit({ type: "compaction_start", reason: "threshold" });
3390
3391	const runningNode = yield* takeEvent(
3392	(event) => event.type === "node.updated" && event.node.kind === "system",
3393	);
3394	const runningItem = yield* takeEvent(
3395	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3396	);
3397	assert.isTrue(
3398	runningNode.type === "node.updated" &&
3399	runningNode.node.status === "running" &&
3400	runningItem.type === "turn_item.updated" &&
3401	runningItem.turnItem.type === "compaction" &&
3402	runningItem.turnItem.status === "running" &&
3403	runningItem.turnItem.title === "Compacting context...",
3404	);
3405
3406	yield* fake.emit({
3407	type: "compaction_end",
3408	reason: "threshold",
3409	result: { summary: "smaller", tokensBefore: 200_000, estimatedTokensAfter: 3_400 },
3410	aborted: false,
3411	willRetry: false,
3412	});
3413	const completedNode = yield* takeEvent(
3414	(event) => event.type === "node.updated" && event.node.kind === "system",
3415	);
3416	const completedItem = yield* takeEvent(
3417	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3418	);
3419	assert.isTrue(
3420	runningNode.type === "node.updated" &&
3421	completedNode.type === "node.updated" &&
3422	runningItem.type === "turn_item.updated" &&
3423	runningItem.turnItem.type === "compaction" &&
3424	completedItem.type === "turn_item.updated" &&
3425	completedItem.turnItem.type === "compaction" &&
3426	completedNode.node.id === runningNode.node.id &&
3427	completedNode.node.status === "completed" &&
3428	completedItem.turnItem.id === runningItem.turnItem.id &&
3429	completedItem.turnItem.ordinal === runningItem.turnItem.ordinal &&
3430	completedItem.turnItem.startedAt === runningItem.turnItem.startedAt &&
3431	completedItem.turnItem.status === "completed" &&
3432	completedItem.turnItem.title === "Context compacted" &&
3433	completedItem.turnItem.beforeTokenCount === 200_000 &&
3434	completedItem.turnItem.afterTokenCount === 3_400,
3435	);
3436
3437	yield* fake.emit({ type: "agent_settled" });
3438	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3439	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
3440	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3441	);
3442
3443	it.effect("uses distinct compaction IDs for first turns in separate threads", () =>
3444	Effect.gen(function* () {
3445	const firstFake = yield* makeFakePi;
3446	const { runtime: firstRuntime, takeEvent: takeFirstEvent } = yield* openRuntime(firstFake);
3447	const firstProviderThread = yield* firstRuntime.ensureThread({
3448	threadId: THREAD_ID,
3449	modelSelection: modelSelection("default"),
3450	runtimePolicy,
3451	});
3452	yield* startTurn(firstRuntime, firstProviderThread);
3453	yield* firstFake.takeRequest("prompt");
3454	yield* firstFake.emit({ type: "agent_start" });
3455	yield* firstFake.emit({ type: "compaction_start", reason: "threshold" });
3456	const first = yield* takeFirstEvent(
3457	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3458	);
3459
3460	const secondThreadId = ThreadId.make("thread-pi-test-second");
3461	const secondFake = yield* makeFakePi;
3462	const { runtime: secondRuntime, takeEvent: takeSecondEvent } = yield* openRuntime(
3463	secondFake,
3464	"default",
3465	secondThreadId,
3466	ProviderSessionId.make("provider-session-pi-test-second"),
3467	);
3468	const secondProviderThread = yield* secondRuntime.ensureThread({
3469	threadId: secondThreadId,
3470	modelSelection: modelSelection("default"),
3471	runtimePolicy,
3472	});
3473	yield* startTurn(
3474	secondRuntime,
3475	secondProviderThread,
3476	"default",
3477	[],
3478	"Hello from another thread",
3479	undefined,
3480	1,
3481	secondThreadId,
3482	);
3483	yield* secondFake.takeRequest("prompt");
3484	yield* secondFake.emit({ type: "agent_start" });
3485	yield* secondFake.emit({ type: "compaction_start", reason: "threshold" });
3486	const second = yield* takeSecondEvent(
3487	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3488	);
3489
3490	assert.isTrue(
3491	first.type === "turn_item.updated" &&
3492	first.turnItem.type === "compaction" &&
3493	second.type === "turn_item.updated" &&
3494	second.turnItem.type === "compaction" &&
3495	first.turnItem.ordinal === second.turnItem.ordinal &&
3496	first.turnItem.id !== second.turnItem.id,
3497	);
3498	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3499	);
3500
3501	it.effect("shows aborted compactions as stopped", () =>
3502	Effect.gen(function* () {
3503	const fake = yield* makeFakePi;
3504	const { runtime, takeEvent } = yield* openRuntime(fake);
3505	const providerThread = yield* runtime.ensureThread({
3506	threadId: THREAD_ID,
3507	modelSelection: modelSelection("default"),
3508	runtimePolicy,
3509	});
3510	yield* startTurn(runtime, providerThread);
3511	yield* fake.takeRequest("prompt");
3512	yield* fake.emit({ type: "agent_start" });
3513	yield* fake.emit({ type: "compaction_start", reason: "manual" });
3514	const running = yield* takeEvent(
3515	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3516	);
3517	yield* fake.emit({
3518	type: "compaction_end",
3519	reason: "manual",
3520	aborted: true,
3521	willRetry: false,
3522	});
3523	const stopped = yield* takeEvent(
3524	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3525	);
3526	assert.isTrue(
3527	running.type === "turn_item.updated" &&
3528	running.turnItem.type === "compaction" &&
3529	stopped.type === "turn_item.updated" &&
3530	stopped.turnItem.type === "compaction" &&
3531	stopped.turnItem.id === running.turnItem.id &&
3532	stopped.turnItem.status === "cancelled" &&
3533	stopped.turnItem.title === "Context compaction stopped",
3534	);
3535	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3536	);
3537
3538	it.effect("keeps the turn open and updates one retry row through final failure", () =>
3539	Effect.gen(function* () {
3540	const fake = yield* makeFakePi;
3541	const { runtime, takeEvent } = yield* openRuntime(fake);
3542	const providerThread = yield* runtime.ensureThread({
3543	threadId: THREAD_ID,
3544	modelSelection: modelSelection("default"),
3545	runtimePolicy,
3546	});
3547	yield* startTurn(runtime, providerThread);
3548	yield* fake.takeRequest("prompt");
3549
3550	yield* fake.emit({ type: "agent_start" });
3551	yield* fake.emit({ type: "agent_end", messages: [], willRetry: true });
3552	yield* fake.emit({
3553	type: "auto_retry_start",
3554	attempt: 1,
3555	maxAttempts: 3,
3556	delayMs: 3_000,
3557	errorMessage: "529 overloaded",
3558	});
3559	const firstRetry = yield* takeEvent(
3560	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3561	);
3562	assert.isTrue(
3563	firstRetry.type === "turn_item.updated" &&
3564	firstRetry.turnItem.type === "error" &&
3565	firstRetry.turnItem.status === "running" &&
3566	firstRetry.turnItem.title === "Provider retry" &&
3567	firstRetry.turnItem.failure.retryable === true &&
3568	firstRetry.turnItem.retry?.attempt === 1 &&
3569	firstRetry.turnItem.retry.maxAttempts === 3 &&
3570	firstRetry.turnItem.retry.retryDelayMs === 3_000,
3571	);
3572
3573	yield* fake.emit({
3574	type: "auto_retry_start",
3575	attempt: 3,
3576	maxAttempts: 3,
3577	delayMs: 12_000,
3578	errorMessage: "529 still overloaded",
3579	});
3580	const lastRetry = yield* takeEvent(
3581	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3582	);
3583	assert.isTrue(
3584	firstRetry.type === "turn_item.updated" &&
3585	firstRetry.turnItem.type === "error" &&
3586	lastRetry.type === "turn_item.updated" &&
3587	lastRetry.turnItem.type === "error" &&
3588	lastRetry.turnItem.id === firstRetry.turnItem.id &&
3589	lastRetry.turnItem.startedAt === firstRetry.turnItem.startedAt &&
3590	lastRetry.turnItem.retry?.attempt === 3,
3591	);
3592
3593	yield* fake.emit({
3594	type: "auto_retry_end",
3595	success: false,
3596	attempt: 3,
3597	finalError: "529 overloaded",
3598	});
3599	const failedRetry = yield* takeEvent(
3600	(event) =>
3601	event.type === "turn_item.updated" &&
3602	event.turnItem.type === "error" &&
3603	event.turnItem.status === "failed",
3604	);
3605	assert.isTrue(
3606	firstRetry.type === "turn_item.updated" &&
3607	firstRetry.turnItem.type === "error" &&
3608	failedRetry.type === "turn_item.updated" &&
3609	failedRetry.turnItem.type === "error" &&
3610	failedRetry.turnItem.id === firstRetry.turnItem.id &&
3611	failedRetry.turnItem.title === "Provider error" &&
3612	failedRetry.turnItem.failure.retryable === false &&
3613	failedRetry.turnItem.retry?.attempt === 3 &&
3614	failedRetry.turnItem.retry.maxAttempts === 3,
3615	);
3616	yield* fake.emit({ type: "agent_settled" });
3617
3618	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3619	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "failed");
3620	assert.isTrue(
3621	firstRetry.type === "turn_item.updated" &&
3622	firstRetry.turnItem.type === "error" &&
3623	terminal.type === "turn.terminal" &&
3624	terminal.status === "failed" &&
3625	terminal.failure.message.includes("overloaded") &&
3626	terminal.retry?.attempt === 3 &&
3627	terminal.retry.maxAttempts === 3 &&
3628	terminal.retryStartedAt === firstRetry.turnItem.startedAt,
3629	);
3630	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3631	);
3632
3633	it.effect("preserves exhausted retry failure through non-retrying compaction", () =>
3634	Effect.gen(function* () {
3635	const fake = yield* makeFakePi;
3636	const { runtime, takeEvent } = yield* openRuntime(fake);
3637	const providerThread = yield* runtime.ensureThread({
3638	threadId: THREAD_ID,
3639	modelSelection: modelSelection("default"),
3640	runtimePolicy,
3641	});
3642	yield* startTurn(runtime, providerThread);
3643	yield* fake.takeRequest("prompt");
3644	yield* fake.emit({ type: "agent_start" });
3645	yield* fake.emit({
3646	type: "auto_retry_start",
3647	attempt: 5,
3648	maxAttempts: 5,
3649	delayMs: 48_000,
3650	errorMessage: "socket timed out",
3651	});
3652	yield* takeEvent(
3653	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3654	);
3655	yield* fake.emit({
3656	type: "auto_retry_end",
3657	success: false,
3658	attempt: 5,
3659	finalError: "socket timed out",
3660	});
3661	yield* takeEvent(
3662	(event) =>
3663	event.type === "turn_item.updated" &&
3664	event.turnItem.type === "error" &&
3665	event.turnItem.status === "failed",
3666	);
3667	yield* fake.emit({ type: "compaction_start", reason: "threshold" });
3668	yield* takeEvent(
3669	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3670	);
3671	yield* fake.emit({
3672	type: "compaction_end",
3673	reason: "threshold",
3674	result: { summary: "smaller", tokensBefore: 200_000, estimatedTokensAfter: 3_400 },
3675	aborted: false,
3676	willRetry: false,
3677	});
3678	yield* takeEvent(
3679	(event) =>
3680	event.type === "turn_item.updated" &&
3681	event.turnItem.type === "compaction" &&
3682	event.turnItem.status === "completed",
3683	);
3684	yield* fake.emit({ type: "agent_settled" });
3685
3686	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3687	assert.isTrue(
3688	terminal.type === "turn.terminal" &&
3689	terminal.status === "failed" &&
3690	terminal.failure.message === "socket timed out" &&
3691	terminal.retry?.attempt === 5 &&
3692	terminal.retry.maxAttempts === 5,
3693	);
3694	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3695	);
3696
3697	it.effect("marks retry progress recovered when Pi succeeds", () =>
3698	Effect.gen(function* () {
3699	const fake = yield* makeFakePi;
3700	const { runtime, takeEvent } = yield* openRuntime(fake);
3701	const providerThread = yield* runtime.ensureThread({
3702	threadId: THREAD_ID,
3703	modelSelection: modelSelection("default"),
3704	runtimePolicy,
3705	});
3706	yield* startTurn(runtime, providerThread);
3707	yield* fake.takeRequest("prompt");
3708	yield* fake.emit({ type: "agent_start" });
3709	yield* fake.emit({
3710	type: "message_end",
3711	message: {
3712	role: "assistant",
3713	content: [],
3714	stopReason: "error",
3715	errorMessage: "temporary network failure",
3716	},
3717	});
3718	yield* fake.emit({
3719	type: "auto_retry_start",
3720	attempt: 1,
3721	maxAttempts: 5,
3722	delayMs: 3_000,
3723	errorMessage: "temporary network failure",
3724	});
3725	const running = yield* takeEvent(
3726	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3727	);
3728	yield* fake.emit({ type: "auto_retry_end", success: true, attempt: 1 });
3729	const recovered = yield* takeEvent(
3730	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3731	);
3732	assert.isTrue(
3733	running.type === "turn_item.updated" &&
3734	running.turnItem.type === "error" &&
3735	recovered.type === "turn_item.updated" &&
3736	recovered.turnItem.type === "error" &&
3737	recovered.turnItem.id === running.turnItem.id &&
3738	recovered.turnItem.status === "completed" &&
3739	recovered.turnItem.title === "Provider recovered",
3740	);
3741
3742	yield* fake.emit({ type: "agent_settled" });
3743	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3744	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
3745	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3746	);
3747
3748	it.effect.each([undefined, false, true])(
3749	"preserves the native settlement outcome, aborted=%s",
3750	(aborted) =>
3751	Effect.gen(function* () {
3752	const fake = yield* makeFakePi;
3753	const { runtime, takeEvent } = yield* openRuntime(fake);
3754	const providerThread = yield* runtime.ensureThread({
3755	threadId: THREAD_ID,
3756	modelSelection: modelSelection("default"),
3757	runtimePolicy,
3758	});
3759	yield* startTurn(runtime, providerThread);
3760	yield* fake.takeRequest("prompt");
3761	yield* fake.emit({ type: "agent_start" });
3762	yield* fake.emit({
3763	type: "agent_settled",
3764	...(aborted === undefined ? {} : { aborted }),
3765	});
3766	const completed = yield* takeEvent(
3767	(event) =>
3768	event.type === "provider_turn.updated" && event.providerTurn.status !== "running",
3769	);
3770	const status = aborted === true ? "interrupted" : "completed";
3771	assert.isTrue(
3772	completed.type === "provider_turn.updated" && completed.providerTurn.status === status,
3773	);
3774	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3775	assert.isTrue(
3776	terminal.type === "turn.terminal" &&
3777	terminal.status === status &&
3778	terminal.failure === null,
3779	);
3780	assert.equal(runtime.providerSession.status, "ready");
3781	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3782	);
3783
3784	it.effect("stops retry progress when Pi aborts the native run", () =>
3785	Effect.gen(function* () {
3786	const fake = yield* makeFakePi;
3787	const { runtime, takeEvent } = yield* openRuntime(fake);
3788	const providerThread = yield* runtime.ensureThread({
3789	threadId: THREAD_ID,
3790	modelSelection: modelSelection("default"),
3791	runtimePolicy,
3792	});
3793	yield* startTurn(runtime, providerThread);
3794	yield* fake.takeRequest("prompt");
3795	yield* fake.emit({ type: "agent_start" });
3796	yield* fake.emit({
3797	type: "auto_retry_start",
3798	attempt: 2,
3799	maxAttempts: 5,
3800	delayMs: 6_000,
3801	errorMessage: "temporary network failure",
3802	});
3803	const retrying = yield* takeEvent(
3804	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3805	);
3806	yield* fake.emit({ type: "agent_settled", aborted: true });
3807	const stopped = yield* takeEvent(
3808	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3809	);
3810	assert.isTrue(
3811	retrying.type === "turn_item.updated" &&
3812	stopped.type === "turn_item.updated" &&
3813	stopped.turnItem.id === retrying.turnItem.id &&
3814	stopped.turnItem.status === "interrupted",
3815	);
3816	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3817	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "interrupted");
3818	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3819	);
3820
3821	it.effect("stops active retry progress when the turn is interrupted", () =>
3822	Effect.gen(function* () {
3823	const fake = yield* makeFakePi;
3824	const { runtime, takeEvent } = yield* openRuntime(fake);
3825	const providerThread = yield* runtime.ensureThread({
3826	threadId: THREAD_ID,
3827	modelSelection: modelSelection("default"),
3828	runtimePolicy,
3829	});
3830	yield* startTurn(runtime, providerThread);
3831	yield* fake.takeRequest("prompt");
3832	yield* fake.emit({ type: "agent_start" });
3833	const runningTurn = yield* takeEvent(
3834	(event) =>
3835	event.type === "provider_turn.updated" && event.providerTurn.status === "running",
3836	);
3837	const providerTurnId =
3838	runningTurn.type === "provider_turn.updated" ? runningTurn.providerTurn.id : undefined;
3839	assert.isDefined(providerTurnId);
3840
3841	yield* fake.emit({
3842	type: "auto_retry_start",
3843	attempt: 2,
3844	maxAttempts: 5,
3845	delayMs: 6_000,
3846	errorMessage: "temporary network failure",
3847	});
3848	const retrying = yield* takeEvent(
3849	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3850	);
3851	yield* runtime.interruptTurn({ providerThread, providerTurnId: providerTurnId! });
3852	yield* fake.takeRequest("abort");
3853	yield* fake.emit({ type: "agent_settled" });
3854
3855	const stopped = yield* takeEvent(
3856	(event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3857	);
3858	assert.isTrue(
3859	retrying.type === "turn_item.updated" &&
3860	retrying.turnItem.type === "error" &&
3861	stopped.type === "turn_item.updated" &&
3862	stopped.turnItem.type === "error" &&
3863	stopped.turnItem.id === retrying.turnItem.id &&
3864	stopped.turnItem.status === "interrupted" &&
3865	stopped.turnItem.title === "Provider retry stopped",
3866	);
3867	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3868	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "interrupted");
3869	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3870	);
3871
3872	it.effect("keeps extension-started compaction and recovery in the settled turn", () =>
3873	Effect.gen(function* () {
3874	const fake = yield* makeFakePi;
3875	const { runtime, takeEvent } = yield* openRuntime(fake);
3876	const providerThread = yield* runtime.ensureThread({
3877	threadId: THREAD_ID,
3878	modelSelection: modelSelection("default"),
3879	runtimePolicy,
3880	});
3881	yield* startTurn(runtime, providerThread);
3882	yield* fake.takeRequest("prompt");
3883	yield* fake.emit({ type: "agent_start" });
3884
3885	// Extension ctx.compact() waits for this first settlement, then starts
3886	// compaction in a detached continuation.
3887	fake.queueState({ isStreaming: false, isCompacting: true, pendingMessageCount: 0 });
3888	yield* fake.emit({ type: "agent_settled" });
3889	yield* fake.takeRequest("get_state");
3890	yield* fake.emit({ type: "compaction_start", reason: "manual" });
3891	yield* takeEvent(
3892	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3893	);
3894
3895	fake.queueState({ isStreaming: true, isCompacting: false, pendingMessageCount: 0 });
3896	yield* fake.emit({
3897	type: "compaction_end",
3898	reason: "manual",
3899	result: { summary: "smaller", tokensBefore: 10_000, estimatedTokensAfter: 2_000 },
3900	aborted: false,
3901	willRetry: false,
3902	});
3903	yield* fake.emit({ type: "agent_start" });
3904	yield* takeEvent(
3905	(event) =>
3906	event.type === "turn_item.updated" &&
3907	event.turnItem.type === "compaction" &&
3908	event.turnItem.status === "completed",
3909	);
3910	yield* fake.takeRequest("get_state");
3911
3912	fake.queueState({ isStreaming: false, isCompacting: false, pendingMessageCount: 0 });
3913	yield* fake.emit({ type: "agent_settled" });
3914	yield* fake.takeRequest("get_state");
3915	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3916	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
3917	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3918	);
3919
3920	it.effect("keeps working after a settle probe fails before detached compaction", () =>
3921	Effect.gen(function* () {
3922	const fake = yield* makeFakePi;
3923	const { runtime, takeEvent } = yield* openRuntime(fake);
3924	const providerThread = yield* runtime.ensureThread({
3925	threadId: THREAD_ID,
3926	modelSelection: modelSelection("default"),
3927	runtimePolicy,
3928	});
3929	yield* startTurn(runtime, providerThread);
3930	yield* fake.takeRequest("prompt");
3931	yield* fake.emit({ type: "agent_start" });
3932
3933	fake.failNextState();
3934	yield* fake.emit({ type: "agent_settled" });
3935	yield* fake.takeRequest("get_state");
3936	yield* fake.emit({ type: "compaction_start", reason: "manual" });
3937	yield* takeEvent(
3938	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3939	);
3940
3941	fake.queueState({ isStreaming: false, isCompacting: false, pendingMessageCount: 0 });
3942	yield* fake.emit({
3943	type: "compaction_end",
3944	reason: "manual",
3945	result: { summary: "smaller", tokensBefore: 10_000, estimatedTokensAfter: 2_000 },
3946	aborted: false,
3947	willRetry: false,
3948	});
3949	yield* takeEvent(
3950	(event) =>
3951	event.type === "turn_item.updated" &&
3952	event.turnItem.type === "compaction" &&
3953	event.turnItem.status === "completed",
3954	);
3955	yield* fake.takeRequest("get_state");
3956	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3957	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
3958	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3959	);
3960
3961	it.effect("restarts Pi when Stop interrupts detached compaction", () =>
3962	Effect.gen(function* () {
3963	const fake = yield* makeFakePi;
3964	const { runtime, takeEvent } = yield* openRuntime(fake);
3965	const providerThread = yield* runtime.ensureThread({
3966	threadId: THREAD_ID,
3967	modelSelection: modelSelection("default"),
3968	runtimePolicy,
3969	});
3970	yield* startTurn(runtime, providerThread);
3971	yield* fake.takeRequest("prompt");
3972	const running = yield* takeEvent(
3973	(event) =>
3974	event.type === "provider_turn.updated" && event.providerTurn.status === "running",
3975	);
3976	assert.equal(running.type, "provider_turn.updated");
3977	const providerTurnId =
3978	running.type === "provider_turn.updated" ? running.providerTurn.id : undefined;
3979	assert.isDefined(providerTurnId);
3980	yield* fake.emit({ type: "agent_start" });
3981
3982	fake.queueState({ isStreaming: false, isCompacting: true, pendingMessageCount: 0 });
3983	yield* fake.emit({ type: "agent_settled" });
3984	yield* fake.takeRequest("get_state");
3985	yield* fake.emit({ type: "compaction_start", reason: "manual" });
3986	yield* takeEvent(
3987	(event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction",
3988	);
3989
3990	yield* runtime.interruptTurn({ providerThread, providerTurnId: providerTurnId! });
3991	assert.isFalse(fake.allRequests().some((request) => request["type"] === "abort"));
3992	yield* fake.closeStdout;
3993	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3994	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "interrupted");
3995	}).pipe(Effect.scoped, Effect.provide(layerTest)),
3996	);
3997
3998	it.effect("ignores an idle snapshot made stale by a steer", () =>
3999	Effect.gen(function* () {
4000	const fake = yield* makeFakePi;
4001	const { runtime, takeEvent } = yield* openRuntime(fake);
4002	const providerThread = yield* runtime.ensureThread({
4003	threadId: THREAD_ID,
4004	modelSelection: modelSelection("default"),
4005	runtimePolicy,
4006	});
4007	yield* startTurn(runtime, providerThread);
4008	yield* fake.takeRequest("prompt");
4009	const running = yield* takeEvent(
4010	(event) =>
4011	event.type === "provider_turn.updated" && event.providerTurn.status === "running",
4012	);
4013	assert.equal(running.type, "provider_turn.updated");
4014	const providerTurnId =
4015	running.type === "provider_turn.updated" ? running.providerTurn.id : undefined;
4016	assert.isDefined(providerTurnId);
4017	yield* fake.emit({ type: "agent_start" });
4018
4019	fake.deferNextState();
4020	yield* fake.emit({ type: "agent_settled" });
4021	yield* fake.takeRequest("get_state");
4022	yield* runtime.steerTurn({
4023	threadId: THREAD_ID,
4024	runId: RunId.make("run:thread-pi-test:1"),
4025	providerThread,
4026	providerTurnId: providerTurnId!,
4027	message: {
4028	messageId: "message:thread-pi-test:late-steer" as never,
4029	text: "Continue after settlement",
4030	attachments: [],
4031	createdBy: "user",
4032	creationSource: "web",
4033	},
4034	});
4035	yield* fake.takeRequest("prompt");
4036	yield* fake.resolveDeferredState({
4037	isStreaming: false,
4038	isCompacting: false,
4039	pendingMessageCount: 0,
4040	});
4041
4042	yield* fake.emit({ type: "agent_start" });
4043	yield* fake.emit({ type: "message_start", message: { role: "assistant" } });
4044	yield* fake.emit({
4045	type: "message_update",
4046	assistantMessageEvent: { type: "text_end", contentIndex: 0, content: "Recovered" },
4047	});
4048	yield* fake.emit({
4049	type: "message_end",
4050	message: {
4051	role: "assistant",
4052	content: [{ type: "text", text: "Recovered" }],
4053	stopReason: "stop",
4054	},
4055	});
4056	const assistantItem = yield* takeEvent(
4057	(event) =>
4058	event.type === "turn_item.updated" &&
4059	event.turnItem.type === "assistant_message" &&
4060	event.turnItem.streaming === false,
4061	);
4062	assert.isTrue(
4063	assistantItem.type === "turn_item.updated" &&
4064	assistantItem.turnItem.type === "assistant_message" &&
4065	assistantItem.turnItem.text === "Recovered",
4066	);
4067
4068	fake.queueState({ isStreaming: false, isCompacting: false, pendingMessageCount: 0 });
4069	yield* fake.emit({ type: "agent_settled" });
4070	yield* fake.takeRequest("get_state");
4071	const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
4072	assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
4073	}).pipe(Effect.scoped, Effect.provide(layerTest)),
4074	);
4075	});
4076
4077	describe("PiRpc framing", () => {
4078	it.effect("reassembles records across chunk boundaries and strips CR", () =>
4079	Effect.gen(function* () {
4080	const stdout = yield* Queue.unbounded<Uint8Array>();
4081	const spawner = ChildProcessSpawner.make(() =>
4082	Effect.succeed(
4083	ChildProcessSpawner.makeHandle({
4084	pid: ChildProcessSpawner.ProcessId(FAKE_PID),
4085	exitCode: Effect.never,
4086	isRunning: Effect.succeed(true),
4087	kill: () => Effect.void,
4088	unref: Effect.succeed(Effect.void),
4089	stdin: Sink.drain,
4090	stdout: Stream.fromQueue(stdout),
4091	stderr: Stream.empty,
4092	all: Stream.empty,
4093	getInputFd: () => Sink.drain,
4094	getOutputFd: () => Stream.empty,
4095	}),
4096	),
4097	);
4098	const connection = yield* makePiRpcConnection({
4099	command: "pi",
4100	args: ["--mode", "rpc"],
4101	cwd: undefined,
4102	env: {},
4103	}).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
4104
4105	const push = (text: string) =>
4106	Queue.offer(stdout, new TextEncoder().encode(text)).pipe(Effect.asVoid);
4107	yield* push('{"type":"agent_');
4108	yield* push('start"}\r\n{"type":"agent_settled"}\nnot json\n{"type":"queue_update"}\n');
4109
4110	yield* push('{"type":"must_not_emit","text":"');
4111	const chunk = new TextEncoder().encode("x".repeat(8 * 1024 * 1024));
4112	for (let index = 0; index < 32; index++) yield* Queue.offer(stdout, chunk);
4113	yield* push('"}\n{"type":"after_oversized"}\n');
4114
4115	const first = yield* Queue.take(connection.events);
4116	assert.equal(first["type"], "agent_start");
4117	const second = yield* Queue.take(connection.events);
4118	assert.equal(second["type"], "agent_settled");
4119	const third = yield* Queue.take(connection.events);
4120	assert.equal(third["type"], "queue_update");
4121	assert.equal((yield* Queue.take(connection.events))["type"], "after_oversized");
4122	}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
4123	);
4124	});
4125
4126	// This fails before a provider transcript exists, so a replay fixture is not
4127	// an honest fit. The boundary is the stdio transport seeing stdout end.
4128	describe("PiRpc early process exit", () => {
4129	const makeHandle = (options: {
4130	readonly exitCode: Effect.Effect<ChildProcessSpawner.ExitCode>;
4131	readonly stderr: Stream.Stream<Uint8Array>;
4132	}) =>
4133	ChildProcessSpawner.makeHandle({
4134	pid: ChildProcessSpawner.ProcessId(FAKE_PID),
4135	exitCode: options.exitCode,
4136	isRunning: Effect.succeed(true),
4137	kill: () => Effect.void,
4138	unref: Effect.succeed(Effect.void),
4139	stdin: Sink.drain,
4140	stdout: Stream.empty,
4141	stderr: options.stderr,
4142	all: Stream.empty,
4143	getInputFd: () => Sink.drain,
4144	getOutputFd: () => Stream.empty,
4145	});
4146
4147	it.effect("reports a nonzero exit code instead of an unexplained stdout close", () =>
4148	Effect.gen(function* () {
4149	const secret = "API_KEY=super-secret\n";
4150	const spawner = ChildProcessSpawner.make(() =>
4151	Effect.succeed(
4152	makeHandle({
4153	exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)),
4154	stderr: Stream.fromIterable([new TextEncoder().encode(secret)]),
4155	}),
4156	),
4157	);
4158	const connection = yield* makePiRpcConnection({
4159	command: "pi",
4160	args: ["--mode", "rpc"],
4161	cwd: undefined,
4162	env: {},
4163	}).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
4164
4165	const error = yield* Queue.take(connection.events).pipe(Effect.flip);
4166	assert.equal(error._tag, "PiRpcError");
4167	assert.equal(error.operation, "read");
4168	assert.equal(error.detail, "pi process exited with code 1");
4169	assert.isFalse((error.detail ?? "").includes("API_KEY"));
4170	assert.isFalse((error.detail ?? "").includes("super-secret"));
4171	assert.isFalse(error.message.includes("API_KEY"));
4172	assert.isFalse(error.message.includes("super-secret"));
4173	}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
4174	);
4175
4176	it.effect("keeps the unexplained stdout-close message when the process has not exited", () =>
4177	Effect.gen(function* () {
4178	const spawner = ChildProcessSpawner.make(() =>
4179	Effect.succeed(
4180	makeHandle({
4181	exitCode: Effect.never,
4182	stderr: Stream.empty,
4183	}),
4184	),
4185	);
4186	const connection = yield* makePiRpcConnection({
4187	command: "pi",
4188	args: ["--mode", "rpc"],
4189	cwd: undefined,
4190	env: {},
4191	}).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
4192
4193	const fiber = yield* Effect.forkChild(Queue.take(connection.events).pipe(Effect.flip));
4194	yield* TestClock.adjust(Duration.millis(300));
4195	const error = yield* Fiber.join(fiber);
4196	assert.equal(error._tag, "PiRpcError");
4197	assert.equal(error.operation, "read");
4198	assert.equal(error.detail, "pi process closed stdout");
4199	}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
4200	);
4201
4202	it.effect("keeps the exit-code diagnosis when stdin breaks while exit is still pending", () =>
4203	Effect.gen(function* () {
4204	const spawner = ChildProcessSpawner.make(() =>
4205	Effect.succeed(
4206	ChildProcessSpawner.makeHandle({
4207	pid: ChildProcessSpawner.ProcessId(FAKE_PID),
4208	exitCode: Effect.sleep(Duration.millis(50)).pipe(
4209	Effect.andThen(Effect.succeed(ChildProcessSpawner.ExitCode(1))),
4210	),
4211	isRunning: Effect.succeed(true),
4212	kill: () => Effect.void,
4213	unref: Effect.succeed(Effect.void),
4214	stdin: Sink.fail(
4215	PlatformError.systemError({
4216	_tag: "Unknown",
4217	module: "ChildProcess",
4218	method: "stdin",
4219	description: "broken pipe",
4220	}),
4221	),
4222	stdout: Stream.empty,
4223	stderr: Stream.empty,
4224	all: Stream.empty,
4225	getInputFd: () => Sink.drain,
4226	getOutputFd: () => Stream.empty,
4227	}),
4228	),
4229	);
4230	const connection = yield* makePiRpcConnection({
4231	command: "pi",
4232	args: ["--mode", "rpc"],
4233	cwd: undefined,
4234	env: {},
4235	}).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
4236
4237	const fiber = yield* Effect.forkChild(Queue.take(connection.events).pipe(Effect.flip));
4238	yield* TestClock.adjust(Duration.millis(300));
4239	const error = yield* Fiber.join(fiber);
4240	assert.equal(error._tag, "PiRpcError");
4241	assert.equal(error.operation, "read");
4242	assert.equal(error.detail, "pi process exited with code 1");
4243	}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
4244	);
4245	});