clover's git

the-knife

1	import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
2	/**
3	* PiAdapterV2 — orchestrator-v2 adapter for the Pi coding agent
4	* (https://pi.dev), driving `pi --mode rpc` over stdio JSONL via `PiRpc.ts`.
5	*
6	* Design intent: honor the user's Pi customizations. The process is spawned
7	* with no `--no-*` flags, so the user's extensions, skills, prompt templates,
8	* AGENTS.md / SYSTEM.md context, settings.json, custom models, and auth all
9	* load exactly as they do in the `pi` TUI. Sessions are stored by Pi itself
10	* (default `~/.pi/agent/sessions/`), and the session file path is the durable
11	* `nativeThreadRef`, so a thread started in T3 can be resumed from the TUI
12	* and vice versa.
13	*
14	* Turn lifecycle: `agent_settled` is the only terminal signal. `agent_end`
15	* merely closes one low-level run — compaction retries, auto-retries, and
16	* queued continuations may still follow it, so the turn stays open until Pi
17	* reports the session settled. An extension can start detached compaction as
18	* that signal unwinds, so the adapter confirms Pi is idle before terminalizing.
19	*
20	* Extension UI: Pi extensions raise dialogs through `extension_ui_request`.
21	* Dialog methods become v2 runtime requests (`confirm` → approval_request,
22	* `select`/`input`/`editor` → user_input_request); answers travel back as
23	* `extension_ui_response`. `notify` becomes a completed activity item.
24	* Terminal-only decoration such as status, widget, title, and editor-text
25	* updates has no matching T3 surface and is ignored.
26	*/
27	import * as HostProcess from "@t3tools/shared/HostProcess";
28	import { AgentScope } from "@t3tools/shared/AgentScope";
29	import { getModelSelectionStringOptionValue } from "@t3tools/shared/model";
30	import { readToolOutputImage } from "@t3tools/shared/toolOutput";
31	import * as Predicate from "effect/Predicate";
32	import {
33	defaultInstanceIdForDriver,
34	ProviderDriverKind,
35	type ChatAttachment,
36	type ModelSelection,
37	type OrchestrationV2ExecutionNode,
38	type OrchestrationV2ProviderCapabilities,
39	type OrchestrationV2ProviderFailure,
40	type OrchestrationV2ProviderRef,
41	type OrchestrationV2ProviderRetry,
42	type OrchestrationV2ProviderSession,
43	type OrchestrationV2ProviderThread,
44	type OrchestrationV2ProviderTurn,
45	type OrchestrationV2RuntimeRequest,
46	type OrchestrationV2TurnItem,
47	type OrchestrationV2UserInputQuestion,
48	type ProviderApprovalDecision,
49	type ProviderInstanceId,
50	type OrchestrationV2ProviderTurnTokenUsage,
51	} from "@t3tools/contracts";
52	import { PiSettings } from "../settings.ts";
53	import * as Cause from "effect/Cause";
54	import * as DateTime from "effect/DateTime";
55	import * as Deferred from "effect/Deferred";
56	import * as Option from "effect/Option";
57	import * as Duration from "effect/Duration";
58	import * as Effect from "effect/Effect";
59	import * as FileSystem from "effect/FileSystem";
60	import * as Fiber from "effect/Fiber";
61	import * as Layer from "effect/Layer";
62	import * as Queue from "effect/Queue";
63	import * as Schema from "effect/Schema";
64	import * as Semaphore from "effect/Semaphore";
65	import * as Stream from "effect/Stream";
66	import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
67
68	import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost";
69	import { mcpToolPresentation } from "@t3tools/provider-core/server/mcpToolPresentation";
70	import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
71	import { parsePiCompactCommand, type PiCompactCommand } from "./commands.ts";
72	import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment";
73	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
74	import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
75	import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
76	import {
77	ProviderAdapterDriverCreateError,
78	type ProviderAdapterDriver,
79	type ProviderAdapterDriverCreateInput,
80	} from "@t3tools/provider-core/server/adapterDriver";
81	import {
82	makeProviderFailure,
83	makeProviderRetryTurnItem,
84	} from "@t3tools/provider-core/server/failure";
85	import { turnScopedSelectionTransition } from "@t3tools/provider-core/server/selectionTransition";
86	import {
87	makePiRpcConnection,
88	parsePiModelSlug,
89	piRecordField as recordField,
90	piRecordNumber as recordNumber,
91	piRecordString as recordString,
92	type PiRpcConnection,
93	type PiRpcRecord,
94	} from "./rpc.ts";
95	import {
96	buildPiRpcLaunch,
97	materializePiT3McpExtension,
98	resolvePiLaunchArgs,
99	} from "./mcpInjection.ts";
100	import { PI_FILE_CHANGE_TOOLS } from "./mcpExtensionSource.ts";
101
102	export const PI_PROVIDER = ProviderDriverKind.make("pi");
103	const PI_DRIVER_KIND = PI_PROVIDER;
104	const PI_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(PI_DRIVER_KIND);
105	const DEFAULT_PI_SETTINGS = Schema.decodeSync(PiSettings)({});
106
107	/**
108	* Sentinel model slug meaning "do not call set_model": Pi resolves the model
109	* from the user's own settings.json (`defaultProvider`/`defaultModel`).
110	*/
111	const PI_INHERIT_MODEL_SLUG = "default";
112
113	const STREAM_FLUSH_MS = 50;
114	const PI_REQUEST_TIMEOUT_MS = 15_000;
115	// Session lifecycle hooks reload extensions, MCP servers and language servers.
116	const PI_SESSION_TIMEOUT_MS = 60_000;
117	const PI_UNSOLICITED_ACTIVITY_ERROR =
118	"Pi started agent work outside an active T3 turn. The session was stopped to prevent invisible tool execution.";
119	const SETTLE_PROBE_MAX_ATTEMPTS = 3;
120	const SETTLE_PROBE_RETRY_DELAY = Duration.millis(100);
121
122	const PiProviderCapabilitiesV2 = {
123	runtimePolicy: { enforcement: "client-boundary" },
124	sessions: {
125	supportsMultipleProviderThreadsPerSession: false,
126	supportsModelSwitchInSession: true,
127	supportsProviderSwitchingViaHandoff: true,
128	// Mode changes restart this process so the injected permission hook gets
129	// one immutable policy for its whole lifetime.
130	supportsRuntimeModeSwitchInSession: false,
131	pendingRequestsSurviveRestart: false,
132	},
133	threads: {
134	canCreateEmptyThread: true,
135	canReadThreadSnapshot: true,
136	canRollbackThread: true,
137	canForkThread: true,
138	canForkFromTurn: true,
139	canForkFromSubagentThread: false,
140	exposesNativeThreadId: true,
141	},
142	turns: {
143	exposesNativeTurnId: false,
144	emitsTurnStarted: true,
145	emitsTurnCompleted: true,
146	supportsInterrupt: true,
147	supportsActiveSteering: true,
148	supportsSteeringByInterruptRestart: false,
149	supportsQueuedMessages: true,
150	terminalStatusQuality: "strong",
151	},
152	streaming: {
153	streamsAssistantText: true,
154	streamsReasoning: true,
155	streamsToolOutput: true,
156	streamsPlanText: false,
157	emitsMessageCompleted: true,
158	},
159	tools: {
160	exposesToolItemIds: true,
161	emitsToolStarted: true,
162	emitsToolCompleted: true,
163	emitsToolOutput: true,
164	supportsMcpTools: true,
165	supportsDynamicToolCallbacks: false,
166	},
167	approvals: {
168	// Pi exposes a blocking tool_call extension hook. The T3 bridge uses it
169	// for supervised and auto-accept modes and forwards its confirmations
170	// through the same extension UI protocol as user-installed extensions.
171	supportsCommandApproval: true,
172	supportsFileReadApproval: false,
173	supportsFileChangeApproval: true,
174	supportsApplyPatchApproval: false,
175	approvalsHaveNativeRequestIds: true,
176	approvalCallbacksAreLiveOnly: true,
177	approvalsCanOriginateFromSubagents: false,
178	},
179	planning: {
180	emitsPlanUpdated: false,
181	emitsTodoList: false,
182	emitsProposedPlan: false,
183	supportsStructuredQuestions: true,
184	planDeltasHaveItemIds: false,
185	},
186	subagents: {
187	// T3 delegation uses the shared MCP `delegate_task` path. Installed Pi
188	// subagent extensions are observed best-effort, but their official tool
189	// runs children with --no-session and exposes no resumable child id.
190	supportsSubagents: true,
191	exposesSubagentThreadIds: false,
192	emitsSubagentLifecycle: true,
193	canWaitForSubagents: false,
194	canCloseSubagents: false,
195	canForkSubagentThread: false,
196	},
197	context: {
198	acceptsSystemContext: false,
199	acceptsDeveloperContext: false,
200	acceptsSyntheticUserContext: true,
201	canGenerateSummaries: false,
202	canConsumeHandoffSummaries: true,
203	// T3 delivers both full and delta handoffs through Pi's normal user-message
204	// input, so neither strategy depends on a Pi-specific context hook.
205	supportsDeltaHandoff: true,
206	supportsFullThreadHandoff: true,
207	maxRecommendedHandoffChars: null,
208	},
209	checkpointing: {
210	appCanCheckpointFilesystem: true,
211	supportsNestedCheckpointScopes: false,
212	providerCanRollbackConversation: true,
213	// CommandPolicy.ensureRollback requires the snapshot whenever provider
214	// rollback is enabled; rollbackThread returns the updated provider thread.
215	providerRollbackReturnsSnapshot: true,
216	providerCanReadConversationSnapshot: true,
217	},
218	identity: {
219	nativeThreadIds: "strong",
220	nativeTurnIds: "weak",
221	nativeItemIds: "strong",
222	nativeRequestIds: "strong",
223	},
224	} satisfies OrchestrationV2ProviderCapabilities;
225
226	export interface PiAdapterV2Options {
227	readonly instanceId: ProviderInstanceId;
228	readonly settings: PiSettings;
229	readonly environment: NodeJS.ProcessEnv;
230	/**
231	* Where to offer a wake turn for provider-initiated work. Without one, such
232	* work has no owner and the adapter stops it.
233	*/
234	readonly continuationRequests?: {
235	readonly offer: (
236	request: ProviderContinuationRequests.ProviderContinuationRequest,
237	) => Effect.Effect<void>;
238	};
239	}
240
241	/** Concatenate the `text` fields of a Pi content-block array. */
242	function contentText(content: unknown): string {
243	if (!Array.isArray(content)) {
244	return typeof content === "string" ? content : "";
245	}
246	return content
247	.map((block) => {
248	if (recordField(block, "type") === "text") return recordString(block, "text") ?? "";
249	return "";
250	})
251	.join("");
252	}
253
254	const MAX_STRUCTURED_TOOL_OUTPUT_BYTES = 64 * 1024;
255
256	/** Preserve complete structured JSON only when its size and traversal remain bounded. */
257	function isBoundedStructuredToolValue(value: unknown): boolean {
258	let bytes = MAX_STRUCTURED_TOOL_OUTPUT_BYTES;
259	let nodes = 4_096;
260	const visit = (entry: unknown, depth: number): boolean => {
261	if (depth > 32 || --nodes < 0 || bytes < 0) return false;
262	if (Array.isArray(entry)) {
263	bytes -= 2 + entry.length;
264	return entry.every((item) => visit(item, depth + 1));
265	}
266	if (Predicate.isObject(entry)) {
267	bytes -= 2;
268	for (const key in entry) {
269	if (!Object.hasOwn(entry, key)) continue;
270	if (key.length > bytes) return false;
271	bytes -= Buffer.byteLength(JSON.stringify(key), "utf8") + 2;
272	if (!visit(entry[key], depth + 1)) return false;
273	}
274	return bytes >= 0;
275	}
276	if (typeof entry === "string" && entry.length > bytes) return false;
277	bytes -= Buffer.byteLength(JSON.stringify(entry) ?? "null", "utf8");
278	return bytes >= 0;
279	};
280	return visit(value, 0);
281	}
282
283	/** Keep model content and distinct typed values; server ingestion owns image byte policy. */
284	function piToolOutput(result: unknown, toolName: string): unknown {
285	const rawContent = recordField(result, "content");
286	const content = Array.isArray(rawContent)
287	? rawContent
288	: typeof rawContent === "string"
289	? [{ type: "text", text: rawContent }]
290	: [];
291	let structured = recordField(result, "structuredContent");
292	// Pi MCP tools return the full CallToolResult to scripts. The model content
293	// is already captured, so store only its distinct structured payload.
294	const details = recordField(result, "details");
295	const isMcpResult =
296	toolName.startsWith("mcp__") &&
297	recordString(details, "server") !== undefined &&
298	recordString(details, "tool") !== undefined &&
299	Array.isArray(recordField(structured, "content"));
300	if (isMcpResult) {
301	structured = recordField(structured, "structuredContent");
302	}
303	if (typeof structured === "string" && structured === contentText(content)) {
304	structured = undefined;
305	}
306	// Native read repeats its image and first text note as structured output.
307	// Drop only that exact mirror; an extension's extra metadata stays distinct.
308	if (toolName === "read" && Predicate.isObject(structured)) {
309	const image = readToolOutputImage(structured);
310	const note =
311	recordString(
312	content.find((block) => recordField(block, "type") === "text"),
313	"text",
314	) ?? "";
315	if (
316	image !== null &&
317	Object.keys(structured).every((key) => ["type", "data", "mimeType", "note"].includes(key)) &&
318	recordString(structured, "note") === note &&
319	content.some((block) => {
320	const match = readToolOutputImage(block);
321	return match !== null && match.mimeType === image.mimeType && match.data === image.data;
322	})
323	) {
324	structured = undefined;
325	}
326	}
327	if (structured !== undefined && !isBoundedStructuredToolValue(structured)) {
328	return {
329	content: [
330	...content,
331	{
332	type: "text",
333	text: "Structured output omitted because it exceeds the stored result limit.",
334	},
335	],
336	};
337	}
338	if (
339	!isMcpResult &&
340	structured === undefined &&
341	content.every((block) => recordField(block, "type") === "text")
342	) {
343	const text = contentText(content);
344	return text.length > 0 ? text : undefined;
345	}
346	return {
347	content,
348	...(structured === undefined ? {} : { structuredContent: structured }),
349	};
350	}
351
352	function providerRef(
353	nativeId: string,
354	strength: "strong" | "weak" = "strong",
355	): OrchestrationV2ProviderRef {
356	return { driver: PI_PROVIDER, nativeId, strength };
357	}
358
359	const PI_THINKING_LEVELS = new Set(["off", "minimal", "low", "medium", "high", "xhigh", "max"]);
360
361	// ── per-session state ─────────────────────────────────────────
362
363	interface PiStreamItemState {
364	readonly nativeItemId: string;
365	readonly kind: "assistant_message" | "reasoning";
366	text: string;
367	completed: boolean;
368	flushScheduled: boolean;
369	readonly startedAt: DateTime.Utc;
370	}
371
372	type PiCompactionStatus = "running" | "completed" | "failed" | "cancelled";
373
374	interface PiCompactionState {
375	readonly nativeItemId: string;
376	readonly startedAt: DateTime.Utc;
377	}
378
379	interface PiProviderRetryState {
380	readonly retry: OrchestrationV2ProviderRetry;
381	readonly failure: OrchestrationV2ProviderFailure;
382	readonly startedAt: DateTime.Utc;
383	readonly itemOrdinal: number;
384	}
385
386	function compactionTitle(status: PiCompactionStatus): string {
387	switch (status) {
388	case "running":
389	return "Compacting context...";
390	case "completed":
391	return "Context compacted";
392	case "failed":
393	return "Context compaction failed";
394	case "cancelled":
395	return "Context compaction stopped";
396	}
397	}
398
399	interface ActivePiTurn {
400	readonly turnInput: ProviderAdapter.ProviderAdapterV2TurnInput;
401	readonly providerTurn: OrchestrationV2ProviderTurn;
402	readonly startedAt: DateTime.Utc;
403	readonly itemOrdinals: Map<string, number>;
404	nextItemOrdinal: number;
405	/** Increments on assistant `message_start` so content indexes stay unique. */
406	messageOrdinal: number;
407	readonly streamItems: Map<string, PiStreamItemState>;
408	readonly toolArgs: Map<string, unknown>;
409	/**
410	* First-seen time per `toolCallId`. Later update/end events reuse it so a
411	* tool keeps one start timestamp and reports a real duration.
412	*/
413	readonly toolStartedAt: Map<string, DateTime.Utc>;
414	interrupted: boolean;
415	/** Cancellation of the latest native run; new work can recover before settlement. */
416	nativeAborted: boolean;
417	/**
418	* Whether any agent run activity was observed. Command-only prompts (pure
419	* extension slash commands) never start an agent run and never emit
420	* `agent_settled`; their deferred prompt ack plus an idle probe settles
421	* the turn instead.
422	*/
423	sawAgentActivity: boolean;
424	/** A user prompt joined provider-native work whose settlement may already be queued. */
425	adoptedWake: boolean;
426	/** Only slash-command prompts can complete without starting an agent run. */
427	readonly promptMayBeCommandOnly: boolean;
428	/** Pi reports context as unknown immediately after compaction; keep its estimate for the meter. */
429	latestCompactionAfterTokens: number | null;
430	/** Last streamed usage total already emitted on the running turn. */
431	lastLiveUsedTokens: number | null;
432	/** Invalidates idle snapshots when new work starts after a settle probe. */
433	settleProbeGeneration: number;
434	/** An extension may start compaction immediately after Pi emits agent_settled. */
435	settleWhenIdle: boolean;
436	sawCompaction: boolean;
437	/** RPC compact is in flight; Pi abort does not cancel it. */
438	manualCompactInFlight: boolean;
439	activeCompaction: PiCompactionState | null;
440	activeProviderRetry: PiProviderRetryState | null;
441	failure: ReturnType<typeof makeProviderFailure> | null;
442	/** Native retry can recover its work without accepting a rejected joining prompt. */
443	rejectedPromptFailure: ReturnType<typeof makeProviderFailure> | null;
444	/** Session-tree refs read just before Stop terminates Pi, when no read is possible later. */
445	stopTreeRefs?: PiTurnTreeRefs | null;
446	}
447
448	interface PiTurnTreeRefs {
449	/** Omitted keeps the synthetic ref; null means the turn left nothing on the active branch. */
450	readonly nativeTurnRef: OrchestrationV2ProviderRef | null | undefined;
451	readonly leafId: string | null;
452	/** Set after a native rewind: every user entry still on the active branch. */
453	readonly retainedNativeTurnIds: ReadonlyArray<string> | undefined;
454	}
455
456	interface PendingPiPrompt {
457	readonly nativeRequestId: string;
458	readonly method: "select" | "confirm" | "input" | "editor";
459	readonly questionId: string;
460	readonly approvalKey: string;
461	readonly expiresAt: number | null;
462	expirationFiber: Fiber.Fiber<void> | null;
463	runtimeRequest: OrchestrationV2RuntimeRequest;
464	readonly node: OrchestrationV2ExecutionNode;
465	readonly turnItem: OrchestrationV2TurnItem;
466	}
467
468	/**
469	* The T3 bridge confirms tool calls as `Allow <tool>?`. Edits surface as
470	* file-change approvals so clients render them like other providers' edits;
471	* every other confirmation, including ones from user extensions, is a command.
472	*/
473	function piApprovalRequestKind(title: string): "command" | "file-change" {
474	const toolName = /^Allow (\S+)\?$/.exec(title)?.[1];
475	return toolName !== undefined &&
476	(PI_FILE_CHANGE_TOOLS as ReadonlyArray<string>).includes(toolName)
477	? "file-change"
478	: "command";
479	}
480
481	interface PiThreadState {
482	providerThread: OrchestrationV2ProviderThread;
483	activeTurn: ActivePiTurn | null;
484	}
485
486	interface PiWake {
487	readonly state: PiThreadState;
488	readonly events: Array<PiRpcRecord>;
489	}
490
491	// ── adapter ───────────────────────────────────────────────────
492
493	export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
494	options: PiAdapterV2Options,
495	) {
496	const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
497	const agentScope = yield* AgentScope;
498	const fileSystem = yield* FileSystem.FileSystem;
499	const idAllocator = yield* IdAllocator.IdAllocatorV2;
500	const host = yield* ProviderHost.ProviderHost;
501	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
502	const scriptProgress = yield* ScriptProgress.ScriptProgress;
503	const { continuationRequests } = options;
504
505	const protocolError = (detail: string, payload?: unknown) =>
506	new ProviderAdapter.ProviderAdapterProtocolError({
507	driver: PI_PROVIDER,
508	detail,
509	...(payload === undefined ? {} : { payload }),
510	});
511
512	return ProviderAdapter.ProviderAdapterV2.of({
513	instanceId: options.instanceId,
514	driver: PI_PROVIDER,
515	getCapabilities: () => Effect.succeed(PiProviderCapabilitiesV2),
516	planSelectionTransition: () => Effect.succeed(turnScopedSelectionTransition()),
517	openSession: Effect.fn("PiAdapterV2.openSession")(function* (
518	input: ProviderAdapter.ProviderAdapterV2OpenSessionInput,
519	) {
520	const scope = yield* Effect.scope;
521	const cwd = input.runtimePolicy.cwd ?? host.paths.cwd;
522	const mcpSession = yield* mcpSessions.read(input.threadId);
523	const provideCacheFs = <A, E>(effect: Effect.Effect<A, E, FileSystem.FileSystem>) =>
524	effect.pipe(
525	Effect.provideService(FileSystem.FileSystem, fileSystem),
526	Effect.mapError(
527	(cause) =>
528	new ProviderAdapter.ProviderAdapterOpenSessionError({
529	driver: PI_PROVIDER,
530	providerSessionId: input.providerSessionId,
531	cause,
532	}),
533	),
534	);
535	// The extension owns both the optional MCP bridge and Pi's permission
536	// hook. Materialize it even when this session has no MCP credential so
537	// Supervised never silently degrades to unrestricted tool execution.
538	const extensionPath = yield* provideCacheFs(
539	materializePiT3McpExtension(host.paths.providerStatusCacheDir),
540	);
541	const resolvedLaunchArgs = resolvePiLaunchArgs(options.settings.launchArgs);
542	if (!resolvedLaunchArgs.ok) {
543	return yield* protocolError(resolvedLaunchArgs.message);
544	}
545	const launch = buildPiRpcLaunch({
546	launchArgs: resolvedLaunchArgs.args,
547	environment: ScriptProgress.withScriptProgressEnvironment(
548	options.environment,
549	yield* scriptProgress.environmentForThread(input.threadId),
550	),
551	mcpSession,
552	extensionPath,
553	runtimeMode: input.runtimePolicy.runtimeMode,
554	});
555	const scopedLaunch = yield* agentScope.wrap({
556	command: options.settings.binaryPath || "pi",
557	args: launch.args,
558	name: "pi",
559	threadId: input.threadId,
560	env: launch.env,
561	});
562	const connection: PiRpcConnection = yield* makePiRpcConnection({
563	command: scopedLaunch.command,
564	args: scopedLaunch.args,
565	cwd,
566	env: launch.env,
567	}).pipe(
568	Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
569	Effect.mapError(
570	(cause) =>
571	new ProviderAdapter.ProviderAdapterOpenSessionError({
572	driver: PI_PROVIDER,
573	providerSessionId: input.providerSessionId,
574	cause,
575	}),
576	),
577	);
578	const now = yield* DateTime.now;
579	let sessionEntity: OrchestrationV2ProviderSession = {
580	id: input.providerSessionId,
581	driver: PI_PROVIDER,
582	providerInstanceId: options.instanceId,
583	status: "ready",
584	cwd,
585	model: input.modelSelection.model,
586	capabilities: PiProviderCapabilitiesV2,
587	createdAt: now,
588	updatedAt: now,
589	lastError: null,
590	};
591	const events = yield* Queue.unbounded<
592	ProviderAdapter.ProviderAdapterV2Event,
593	ProviderAdapter.ProviderAdapterV2Error | Cause.Done
594	>();
595	const pendingPrompts = new Map<string, PendingPiPrompt>();
596	const sessionApprovals = new Set<string>();
597	// Answering a dialog and terminalizing a turn both publish lifecycle
598	// events. Pi can settle immediately after `extension_ui_response`, so
599	// serialize the two paths to stop `turn.terminal` from overtaking the
600	// dialog's own resolution updates.
601	const sessionEventPermit = yield* Semaphore.make(1);
602	let threadState: PiThreadState | null = null;
603	let registrationAttempted = false;
604	let lastNativeThreadId: string | null = null;
605	// User Stop intentionally tears down this RPC process after aborting.
606	// Keep that intent beyond turn finalization so the later stdout close is
607	// not mistaken for an unexpected transport failure.
608	let stopRequested = false;
609	// Native work is held until an orchestrator-owned continuation or user
610	// turn attaches. The guard remains for sessions with no routable owner.
611	let unsolicitedActivityDetected = false;
612	let pendingWake: PiWake | null = null;
613	let rollbackBarrier: Deferred.Deferred<
614	void,
615	ProviderAdapter.ProviderAdapterProtocolError
616	> | null = null;
617	let closed = false;
618	yield* Effect.addFinalizer(() =>
619	Effect.gen(function* () {
620	closed = true;
621	pendingWake = null;
622	if (rollbackBarrier !== null)
623	yield* Deferred.fail(
624	rollbackBarrier,
625	protocolError("Pi session closed during rollback"),
626	);
627	}),
628	);
629	let appliedModel: string | null = null;
630	let appliedThinking: string | null = null;
631	/** Last thread title synced into pi's session name (`/resume` listing). */
632	let appliedSessionName: string | null = null;
633	/** Extension failures raised during startup are attached to the next turn. */
634	const outOfTurnExtensionErrors: Array<PiRpcRecord> = [];
635	/**
636	* Leaf entry id of the pi session tree as of the last turn boundary.
637	* Turn-start user entries are located relative to it, giving each
638	* provider turn a durable native ref for session-tree rollback.
639	*/
640	let lastKnownLeaf: string | null = null;
641	/**
642	* Set when a `get_entries` capture failed. Pi may have advanced past
643	* `lastKnownLeaf` since, so the cursor no longer bounds a single turn
644	* and the next capture re-syncs it instead of trusting it.
645	*/
646	let leafCursorStale = false;
647	// Pi's own configured defaults, captured from the first `get_state` so
648	// that selecting the displayed "Pi default" again can restore them. Pi
649	// has no "unset" commands, so the baselines have to be replayed
650	// explicitly.
651	let baselineModel: { provider: string; modelId: string } | null = null;
652	let baselineThinking: string | null = null;
653	/** Context window of the model Pi currently runs, from get_state and set_model. */
654	let contextWindow: number | null = null;
655	const modelContextWindows = new Map<string, number>();
656	let modelsDiscovered = false;
657	// Prompt responses carry no id. Keep their session-wide send order and
658	// owner so a late ack from a settled turn cannot affect the next turn.
659	const pendingPromptResponses: Array<{
660	readonly providerTurnId: OrchestrationV2ProviderTurn["id"];
661	readonly kind: "turn_start" | "steer";
662	}> = [];
663	const pendingCompactResponses: Array<{
664	readonly providerTurnId: OrchestrationV2ProviderTurn["id"];
665	readonly kind: "turn_start" | "steer";
666	}> = [];
667
668	const compactRpcRecord = (command: PiCompactCommand): PiRpcRecord =>
669	command.customInstructions === undefined
670	? { type: "compact" }
671	: { type: "compact", customInstructions: command.customInstructions };
672
673	const emit = (event: ProviderAdapter.ProviderAdapterV2Event) =>
674	Queue.offer(events, event).pipe(Effect.asVoid);
675
676	const updateProviderSession = (
677	status: OrchestrationV2ProviderSession["status"],
678	lastError: string | null = sessionEntity.lastError,
679	) =>
680	Effect.gen(function* () {
681	const updatedAt = yield* DateTime.now;
682	sessionEntity = { ...sessionEntity, status, lastError, updatedAt };
683	yield* emit({
684	type: "provider_session.updated",
685	driver: PI_PROVIDER,
686	providerSession: sessionEntity,
687	});
688	});
689
690	const updateProviderThread = (
691	state: PiThreadState,
692	patch: Partial<OrchestrationV2ProviderThread>,
693	retainedNativeTurnIds?: ReadonlyArray<string>,
694	) =>
695	Effect.gen(function* () {
696	const updatedAt = yield* DateTime.now;
697	state.providerThread = { ...state.providerThread, ...patch, updatedAt };
698	yield* emit({
699	type: "provider_thread.updated",
700	driver: PI_PROVIDER,
701	providerThread: state.providerThread,
702	...(retainedNativeTurnIds === undefined ? {} : { retainedNativeTurnIds }),
703	});
704	});
705
706	const itemOrdinal = (turn: ActivePiTurn, nativeItemId: string): number => {
707	const existing = turn.itemOrdinals.get(nativeItemId);
708	if (existing !== undefined) return existing;
709	const ordinal = turn.nextItemOrdinal++;
710	turn.itemOrdinals.set(nativeItemId, ordinal);
711	return ordinal;
712	};
713
714	const request = (record: PiRpcRecord, timeoutMs = PI_REQUEST_TIMEOUT_MS) =>
715	connection.request(record, timeoutMs);
716
717	const nonNegativeInteger = (input: unknown, key: string): number | undefined => {
718	const value = recordNumber(input, key);
719	return value === undefined ? undefined : Math.max(0, Math.trunc(value));
720	};
721
722	const rememberModelContextWindow = (model: unknown): number | null => {
723	const provider = recordString(model, "provider");
724	const id = recordString(model, "id");
725	const capacity = nonNegativeInteger(model, "contextWindow");
726	if (provider !== undefined && id !== undefined && capacity !== undefined && capacity > 0) {
727	modelContextWindows.set(`${provider}/${id}`, capacity);
728	return capacity;
729	}
730	return null;
731	};
732
733	const lifecycleRequest = (record: PiRpcRecord) =>
734	request(record, PI_SESSION_TIMEOUT_MS).pipe(
735	// A local timeout does not cancel Pi's lifecycle hook. Retire the
736	// process before fallback can race its eventual switch/new-session.
737	Effect.tapError((error) =>
738	Effect.logWarning("Pi session lifecycle request failed", {
739	providerSessionId: input.providerSessionId,
740	operation: record["type"],
741	errorTag: error._tag,
742	}),
743	),
744	Effect.catchTags({
745	PiRpcTimeoutError: (error) =>
746	connection.terminate.pipe(Effect.andThen(Effect.fail(error))),
747	}),
748	Effect.onInterrupt(() => connection.terminate),
749	);
750
751	const tokenUsageFromStats = (
752	stats: unknown,
753	fallbackUsedTokens: number | null,
754	updatedAt: DateTime.Utc,
755	): OrchestrationV2ProviderTurnTokenUsage | undefined => {
756	const contextUsage = recordField(stats, "contextUsage");
757	const maxTokens = nonNegativeInteger(contextUsage, "contextWindow");
758	const usedTokens =
759	nonNegativeInteger(contextUsage, "tokens") ?? fallbackUsedTokens ?? undefined;
760	if (usedTokens === undefined || maxTokens === undefined || maxTokens === 0)
761	return undefined;
762
763	const totals = recordField(stats, "tokens");
764	const inputTokens = nonNegativeInteger(totals, "input");
765	const cachedInputTokens = nonNegativeInteger(totals, "cacheRead");
766	const outputTokens = nonNegativeInteger(totals, "output");
767	return {
768	usedTokens,
769	maxTokens,
770	...(inputTokens === undefined ? {} : { inputTokens }),
771	...(cachedInputTokens === undefined ? {} : { cachedInputTokens }),
772	...(outputTokens === undefined ? {} : { outputTokens }),
773	updatedAt: DateTime.formatIso(updatedAt),
774	};
775	};
776
777	/**
778	* Pi only reports context usage through `get_session_stats`, so the
779	* settled turn carries it on the base's per-turn `tokenUsage` (#8144).
780	* Usage is secondary telemetry: the request is bounded and a provider
781	* version without stats simply leaves the turn without a report, which
782	* keeps the meter on the last turn that had one.
783	*/
784	const readTokenUsage = (fallbackUsedTokens: number | null, updatedAt: DateTime.Utc) =>
785	request({ type: "get_session_stats" }, 2_000).pipe(
786	Effect.map((stats) => tokenUsageFromStats(stats, fallbackUsedTokens, updatedAt)),
787	Effect.orElseSucceed(() => undefined),
788	);
789
790	/**
791	* Pi attaches the current message's cumulative usage to every streaming
792	* update (0.84.2+). Emit it on the running turn only when the total
793	* changes, so the meter moves live without a burst of no-op updates.
794	*/
795	const reportLiveUsage = (turn: ActivePiTurn, usage: unknown) =>
796	Effect.gen(function* () {
797	const usedTokens = nonNegativeInteger(usage, "totalTokens");
798	if (
799	usedTokens === undefined ||
800	usedTokens === 0 ||
801	usedTokens === turn.lastLiveUsedTokens ||
802	contextWindow === null ||
803	contextWindow === 0
804	) {
805	return;
806	}
807	turn.lastLiveUsedTokens = usedTokens;
808	const inputTokens = nonNegativeInteger(usage, "input");
809	const cachedInputTokens = nonNegativeInteger(usage, "cacheRead");
810	const outputTokens = nonNegativeInteger(usage, "output");
811	const updatedAt = yield* DateTime.now;
812	yield* emit({
813	type: "provider_turn.updated",
814	driver: PI_PROVIDER,
815	threadId: turn.turnInput.threadId,
816	providerTurn: {
817	...turn.providerTurn,
818	tokenUsage: {
819	usedTokens,
820	maxTokens: contextWindow,
821	...(inputTokens === undefined ? {} : { inputTokens }),
822	...(cachedInputTokens === undefined ? {} : { cachedInputTokens }),
823	...(outputTokens === undefined ? {} : { outputTokens }),
824	updatedAt: DateTime.formatIso(updatedAt),
825	},
826	},
827	});
828	});
829
830	const baseItemFields = (
831	turn: ActivePiTurn,
832	nativeItemId: string,
833	startedAt: DateTime.Utc,
834	updatedAt: DateTime.Utc,
835	) => ({
836	id: idAllocator.derive.turnItemFromProviderItem({
837	driver: PI_PROVIDER,
838	nativeItemId,
839	}),
840	threadId: turn.turnInput.threadId,
841	runId: turn.turnInput.runId,
842	nodeId: idAllocator.derive.nodeFromProviderItem({
843	driver: PI_PROVIDER,
844	nativeItemId,
845	}),
846	providerThreadId: turn.turnInput.providerThread.id,
847	providerTurnId: turn.providerTurn.id,
848	nativeItemRef: providerRef(nativeItemId),
849	parentItemId: null,
850	ordinal: itemOrdinal(turn, nativeItemId),
851	startedAt,
852	updatedAt,
853	});
854
855	const emitItemNode = (
856	turn: ActivePiTurn,
857	nativeItemId: string,
858	kind: OrchestrationV2ExecutionNode["kind"],
859	status: OrchestrationV2ExecutionNode["status"],
860	startedAt: DateTime.Utc,
861	completedAt: DateTime.Utc | null,
862	) =>
863	emit({
864	type: "node.updated",
865	driver: PI_PROVIDER,
866	node: {
867	id: idAllocator.derive.nodeFromProviderItem({
868	driver: PI_PROVIDER,
869	nativeItemId,
870	}),
871	threadId: turn.turnInput.threadId,
872	runId: turn.turnInput.runId,
873	parentNodeId: turn.turnInput.rootNodeId,
874	rootNodeId: turn.turnInput.rootNodeId,
875	kind,
876	status,
877	countsForRun: false,
878	providerThreadId: turn.turnInput.providerThread.id,
879	providerTurnId: turn.providerTurn.id,
880	nativeItemRef: providerRef(nativeItemId),
881	runtimeRequestId: null,
882	checkpointScopeId: null,
883	startedAt,
884	completedAt,
885	},
886	});
887
888	const emitProviderRetry = Effect.fnUntraced(function* (
889	turn: ActivePiTurn,
890	providerRetry: PiProviderRetryState,
891	status: "running" | "completed" | "failed" | "interrupted" | "cancelled",
892	updatedAt: DateTime.Utc,
893	) {
894	yield* emit({
895	type: "turn_item.updated",
896	driver: PI_PROVIDER,
897	turnItem: makeProviderRetryTurnItem({
898	driver: PI_PROVIDER,
899	threadId: turn.turnInput.threadId,
900	runId: turn.turnInput.runId,
901	nodeId: turn.turnInput.rootNodeId,
902	providerThreadId: turn.turnInput.providerThread.id,
903	providerTurnId: turn.providerTurn.id,
904	itemOrdinal: providerRetry.itemOrdinal,
905	failure: providerRetry.failure,
906	retry: providerRetry.retry,
907	status,
908	startedAt: providerRetry.startedAt,
909	updatedAt,
910	}),
911	});
912	});
913
914	const compactionNativeItemId = (turn: ActivePiTurn): string =>
915	`compaction:${turn.providerTurn.id}:${turn.nextItemOrdinal}`;
916
917	const emitCompaction = Effect.fnUntraced(function* (
918	turn: ActivePiTurn,
919	compaction: PiCompactionState,
920	status: PiCompactionStatus,
921	details: {
922	readonly summary?: string;
923	readonly beforeTokenCount?: number;
924	readonly afterTokenCount?: number;
925	} = {},
926	) {
927	const emittedAt = yield* DateTime.now;
928	const completedAt = status === "running" ? null : emittedAt;
929	yield* emitItemNode(
930	turn,
931	compaction.nativeItemId,
932	"system",
933	status,
934	compaction.startedAt,
935	completedAt,
936	);
937	yield* emit({
938	type: "turn_item.updated",
939	driver: PI_PROVIDER,
940	turnItem: {
941	...baseItemFields(turn, compaction.nativeItemId, compaction.startedAt, emittedAt),
942	status,
943	title: compactionTitle(status),
944	completedAt,
945	type: "compaction",
946	driver: PI_PROVIDER,
947	...details,
948	},
949	});
950	});
951
952	// ── streaming text / reasoning ────────────────────────
953
954	const emitStreamItem = (turn: ActivePiTurn, item: PiStreamItemState, streaming: boolean) =>
955	Effect.gen(function* () {
956	const emittedAt = yield* DateTime.now;
957	const base = baseItemFields(turn, item.nativeItemId, item.startedAt, emittedAt);
958	yield* emitItemNode(
959	turn,
960	item.nativeItemId,
961	item.kind,
962	streaming ? "running" : "completed",
963	item.startedAt,
964	streaming ? null : emittedAt,
965	);
966	if (item.kind === "assistant_message") {
967	const messageId = idAllocator.derive.messageFromProviderItem({
968	driver: PI_PROVIDER,
969	nativeItemId: item.nativeItemId,
970	});
971	yield* emit({
972	type: "turn_item.updated",
973	driver: PI_PROVIDER,
974	turnItem: {
975	...base,
976	status: streaming ? "running" : "completed",
977	title: null,
978	completedAt: streaming ? null : emittedAt,
979	type: "assistant_message",
980	messageId,
981	text: item.text,
982	streaming,
983	},
984	});
985	yield* emit({
986	type: "message.updated",
987	driver: PI_PROVIDER,
988	message: {
989	id: messageId,
990	threadId: turn.turnInput.threadId,
991	runId: turn.turnInput.runId,
992	nodeId: idAllocator.derive.nodeFromProviderItem({
993	driver: PI_PROVIDER,
994	nativeItemId: item.nativeItemId,
995	}),
996	role: "assistant",
997	text: item.text,
998	attachments: [],
999	streaming,
1000	createdBy: "agent",
1001	creationSource: "provider",
1002	createdAt: item.startedAt,
1003	updatedAt: emittedAt,
1004	},
1005	});
1006	return;
1007	}
1008	yield* emit({
1009	type: "turn_item.updated",
1010	driver: PI_PROVIDER,
1011	turnItem: {
1012	...base,
1013	status: streaming ? "running" : "completed",
1014	title: null,
1015	completedAt: streaming ? null : emittedAt,
1016	type: "reasoning",
1017	text: item.text,
1018	streaming,
1019	},
1020	});
1021	});
1022
1023	const scheduleStreamFlush = (turn: ActivePiTurn, item: PiStreamItemState) =>
1024	Effect.gen(function* () {
1025	if (item.flushScheduled || item.completed) return;
1026	item.flushScheduled = true;
1027	yield* Effect.sleep(Duration.millis(STREAM_FLUSH_MS)).pipe(
1028	Effect.andThen(
1029	Effect.suspend(() => {
1030	item.flushScheduled = false;
1031	return item.completed ? Effect.void : emitStreamItem(turn, item, true);
1032	}),
1033	),
1034	Effect.forkIn(scope),
1035	);
1036	});
1037
1038	const streamItemFor = Effect.fnUntraced(function* (
1039	turn: ActivePiTurn,
1040	kind: PiStreamItemState["kind"],
1041	contentIndex: number,
1042	) {
1043	const nativeItemId = `${turn.providerTurn.id}:m${turn.messageOrdinal}:c${contentIndex}`;
1044	const existing = turn.streamItems.get(nativeItemId);
1045	if (existing !== undefined) return existing;
1046	const startedAt = yield* DateTime.now;
1047	const item: PiStreamItemState = {
1048	nativeItemId,
1049	kind,
1050	text: "",
1051	completed: false,
1052	flushScheduled: false,
1053	startedAt,
1054	};
1055	turn.streamItems.set(nativeItemId, item);
1056	// Ordinal reserved on first delta so items appear in stream order.
1057	itemOrdinal(turn, nativeItemId);
1058	return item;
1059	});
1060
1061	const completeStreamItem = (turn: ActivePiTurn, item: PiStreamItemState, text?: string) =>
1062	Effect.suspend(() => {
1063	if (item.completed) return Effect.void;
1064	item.completed = true;
1065	if (text !== undefined && text.length > 0) item.text = text;
1066	return item.text.length === 0 ? Effect.void : emitStreamItem(turn, item, false);
1067	});
1068
1069	const completeOpenStreamItems = (turn: ActivePiTurn) =>
1070	Effect.forEach(
1071	Array.from(turn.streamItems.values()).filter((item) => !item.completed),
1072	(item) => completeStreamItem(turn, item),
1073	{ discard: true },
1074	);
1075
1076	// ── tools ─────────────────────────────────────────────
1077
1078	const emitToolItem = Effect.fnUntraced(function* (
1079	turn: ActivePiTurn,
1080	event: PiRpcRecord,
1081	phase: "start" | "update" | "end",
1082	) {
1083	const toolCallId = recordString(event, "toolCallId");
1084	const toolName = recordString(event, "toolName") ?? "tool";
1085	if (toolCallId === undefined) return;
1086	if (phase === "start") {
1087	turn.toolArgs.set(toolCallId, event["args"]);
1088	}
1089	const args = event["args"] ?? turn.toolArgs.get(toolCallId);
1090	const emittedAt = yield* DateTime.now;
1091	const startedAt = turn.toolStartedAt.get(toolCallId) ?? emittedAt;
1092	turn.toolStartedAt.set(toolCallId, startedAt);
1093	const completed = phase === "end";
1094	const isError = event["isError"] === true;
1095	const resultRecord = completed ? event["result"] : event["partialResult"];
1096	const outputText = contentText(recordField(resultRecord, "content"));
1097	// A Stop aborts in-flight tools, and pi reports those as error ends.
1098	// Present them as interrupted (matching the run) rather than failed.
1099	const status = completed
1100	? isError
1101	? turn.interrupted
1102	? "interrupted"
1103	: "failed"
1104	: "completed"
1105	: "running";
1106	const base = baseItemFields(turn, toolCallId, startedAt, emittedAt);
1107	yield* emitItemNode(
1108	turn,
1109	toolCallId,
1110	"tool_call",
1111	status,
1112	startedAt,
1113	completed ? emittedAt : null,
1114	);
1115	const shared = {
1116	...base,
1117	status,
1118	completedAt: completed ? emittedAt : null,
1119	} as const;
1120	if (toolName === "bash" || toolName === "powershell") {
1121	const exitCode =
1122	recordNumber(recordField(resultRecord, "structuredContent"), "exit_code") ??
1123	recordNumber(recordField(resultRecord, "details"), "exitCode");
1124	yield* emit({
1125	type: "turn_item.updated",
1126	driver: PI_PROVIDER,
1127	turnItem: {
1128	...shared,
1129	title: toolName,
1130	type: "command_execution",
1131	input: recordString(args, "command") ?? "",
1132	...(outputText.length > 0 ? { output: outputText } : {}),
1133	...(exitCode === undefined ? {} : { exitCode }),
1134	},
1135	});
1136	return;
1137	}
1138	if (toolName === "edit" || toolName === "write") {
1139	const fileName = recordString(args, "path") ?? recordString(args, "file_path");
1140	if (fileName !== undefined) {
1141	// edit reports a unified patch in its result details; write only
1142	// carries the new content in its args. A failed call keeps its error.
1143	const diffStr =
1144	recordString(recordField(resultRecord, "details"), "patch") ??
1145	(isError && outputText.trim().length > 0 ? outputText : undefined);
1146	const newStr = toolName === "write" ? recordString(args, "content") : undefined;
1147	yield* emit({
1148	type: "turn_item.updated",
1149	driver: PI_PROVIDER,
1150	turnItem: {
1151	...shared,
1152	title: toolName,
1153	type: "file_change",
1154	fileName,
1155	...(diffStr === undefined ? {} : { diffStr }),
1156	...(newStr === undefined ? {} : { newStr }),
1157	},
1158	});
1159	return;
1160	}
1161	}
1162	const output = piToolOutput(resultRecord, toolName);
1163	yield* emit({
1164	type: "turn_item.updated",
1165	driver: PI_PROVIDER,
1166	turnItem: {
1167	...shared,
1168	title: toolName,
1169	type: "dynamic_tool",
1170	...mcpToolPresentation({ toolName }),
1171	toolName,
1172	input: args ?? {},
1173	...(output === undefined ? {} : { output }),
1174	},
1175	});
1176	if (toolName === "subagent") {
1177	yield* emitSubagentTasks(turn, toolCallId, resultRecord, completed);
1178	}
1179	});
1180
1181	/**
1182	* Observe the result shape from Pi's official example subagent extension.
1183	* The extension runs children with --no-session, so these entries are
1184	* visible in T3's shared subagent UI without inventing a child thread.
1185	* Unknown or changed result shapes stay ordinary dynamic tool output.
1186	*/
1187	const emitSubagentTasks = Effect.fnUntraced(function* (
1188	turn: ActivePiTurn,
1189	toolCallId: string,
1190	resultRecord: unknown,
1191	completed: boolean,
1192	) {
1193	const results = recordField(recordField(resultRecord, "details"), "results");
1194	if (!Array.isArray(results)) return;
1195	const emittedAt = yield* DateTime.now;
1196	const parentNodeId = idAllocator.derive.nodeFromProviderItem({
1197	driver: PI_PROVIDER,
1198	nativeItemId: toolCallId,
1199	});
1200	for (const [index, result] of results.entries()) {
1201	const agent = recordString(result, "agent");
1202	const task = recordString(result, "task");
1203	if (agent === undefined || task === undefined) continue;
1204	const nativeTaskId = `${toolCallId}:subagent:${recordNumber(result, "step") ?? index}`;
1205	const subagentId = idAllocator.derive.nodeFromProviderItem({
1206	driver: PI_PROVIDER,
1207	nativeItemId: nativeTaskId,
1208	});
1209	const startedAt = turn.toolStartedAt.get(nativeTaskId) ?? emittedAt;
1210	turn.toolStartedAt.set(nativeTaskId, startedAt);
1211	const finished = completed || recordField(result, "finished") === true;
1212	const stopReason = recordString(result, "stopReason");
1213	const interrupted = finished && stopReason === "aborted";
1214	const failed =
1215	finished &&
1216	!interrupted &&
1217	((recordNumber(result, "exitCode") ?? 0) !== 0 || stopReason === "error");
1218	const status = interrupted
1219	? "interrupted"
1220	: failed
1221	? "failed"
1222	: finished
1223	? "completed"
1224	: "running";
1225	const outputText = piSubagentOutput(result);
1226	const progress =
1227	!finished && outputText.length > 0 ? { progress: outputText.slice(0, 200) } : {};
1228	const resultText = finished && outputText.length > 0 ? outputText.slice(0, 10_000) : null;
1229	yield* emit({
1230	type: "subagent.updated",
1231	driver: PI_PROVIDER,
1232	subagent: {
1233	id: subagentId,
1234	threadId: turn.turnInput.threadId,
1235	runId: turn.turnInput.runId,
1236	parentNodeId,
1237	origin: "provider_native",
1238	createdBy: "agent",
1239	driver: PI_PROVIDER,
1240	providerInstanceId: options.instanceId,
1241	providerThreadId: turn.turnInput.providerThread.id,
1242	childThreadId: null,
1243	nativeTaskRef: providerRef(nativeTaskId),
1244	prompt: task,
1245	title: agent,
1246	model: recordString(result, "model") ?? null,
1247	status,
1248	...progress,
1249	result: resultText,
1250	startedAt,
1251	completedAt: finished ? emittedAt : null,
1252	updatedAt: emittedAt,
1253	},
1254	});
1255	yield* emit({
1256	type: "turn_item.updated",
1257	driver: PI_PROVIDER,
1258	turnItem: {
1259	...baseItemFields(turn, nativeTaskId, startedAt, emittedAt),
1260	status,
1261	title: agent,
1262	completedAt: finished ? emittedAt : null,
1263	type: "subagent",
1264	subagentId,
1265	origin: "provider_native",
1266	driver: PI_PROVIDER,
1267	providerInstanceId: options.instanceId,
1268	childThreadId: null,
1269	prompt: task,
1270	...progress,
1271	result: resultText,
1272	},
1273	});
1274	}
1275	});
1276
1277	// ── extension UI prompts ──────────────────────────────
1278
1279	const cancelPrompt = (pending: PendingPiPrompt, resolvedAt: DateTime.Utc) =>
1280	Effect.gen(function* () {
1281	if (pending.expirationFiber !== null) yield* Fiber.interrupt(pending.expirationFiber);
1282	yield* connection
1283	.send({
1284	type: "extension_ui_response",
1285	id: pending.nativeRequestId,
1286	cancelled: true,
1287	})
1288	.pipe(Effect.ignore);
1289	pending.runtimeRequest = {
1290	...pending.runtimeRequest,
1291	status: "cancelled",
1292	resolvedAt,
1293	};
1294	yield* emit({
1295	type: "runtime_request.updated",
1296	driver: PI_PROVIDER,
1297	threadId: pending.node.threadId,
1298	runtimeRequest: pending.runtimeRequest,
1299	});
1300	yield* emit({
1301	type: "node.updated",
1302	driver: PI_PROVIDER,
1303	node: { ...pending.node, status: "cancelled", completedAt: resolvedAt },
1304	});
1305	yield* emit({
1306	type: "turn_item.updated",
1307	driver: PI_PROVIDER,
1308	turnItem: {
1309	...pending.turnItem,
1310	status: "cancelled",
1311	completedAt: resolvedAt,
1312	updatedAt: resolvedAt,
1313	},
1314	});
1315	});
1316
1317	const cancelPendingPrompts = (resolvedAt: DateTime.Utc) =>
1318	Effect.gen(function* () {
1319	const pending = Array.from(pendingPrompts.values());
1320	pendingPrompts.clear();
1321	yield* Effect.forEach(pending, (prompt) => cancelPrompt(prompt, resolvedAt), {
1322	discard: true,
1323	});
1324	});
1325
1326	const handleExtensionUiRequest = Effect.fnUntraced(function* (event: PiRpcRecord) {
1327	const method = recordString(event, "method");
1328	const nativeRequestId = recordString(event, "id");
1329	if (method === undefined) return;
1330	if (method === "notify") {
1331	const state = threadState;
1332	const turn = state?.activeTurn ?? null;
1333	const message = recordString(event, "message") ?? "";
1334	if (turn === null || message.length === 0) return;
1335	const emittedAt = yield* DateTime.now;
1336	const nativeItemId = `notify:${turn.nextItemOrdinal}`;
1337	yield* emitItemNode(turn, nativeItemId, "system", "completed", emittedAt, emittedAt);
1338	yield* emit({
1339	type: "turn_item.updated",
1340	driver: PI_PROVIDER,
1341	turnItem: {
1342	...baseItemFields(turn, nativeItemId, emittedAt, emittedAt),
1343	status: "completed",
1344	completedAt: emittedAt,
1345	title: "notify",
1346	type: "dynamic_tool",
1347	toolName: "notify",
1348	input: {
1349	message,
1350	notifyType: recordString(event, "notifyType") ?? "info",
1351	},
1352	},
1353	});
1354	return;
1355	}
1356	if (
1357	method !== "select" &&
1358	method !== "confirm" &&
1359	method !== "input" &&
1360	method !== "editor"
1361	) {
1362	// Terminal decoration has no matching T3 surface.
1363	yield* Effect.logDebug("Ignoring pi extension UI update.", { method });
1364	return;
1365	}
1366	if (nativeRequestId === undefined) return;
1367	const approvalTitle = recordString(event, "title") ?? "";
1368	const approvalKey = `${approvalTitle.length}:${approvalTitle}${recordString(event, "message") ?? ""}`;
1369	if (method === "confirm" && sessionApprovals.has(approvalKey)) {
1370	yield* connection.send({
1371	type: "extension_ui_response",
1372	id: nativeRequestId,
1373	confirmed: true,
1374	});
1375	return;
1376	}
1377	const state = threadState;
1378	const turn = state?.activeTurn ?? null;
1379	const createdAt = yield* DateTime.now;
1380	const timeout = recordNumber(event, "timeout");
1381	// Pi treats zero as no timeout and negative durations as immediate expiry.
1382	const expiresAt =
1383	timeout === undefined || timeout === 0
1384	? null
1385	: DateTime.toEpochMillis(createdAt) + Math.max(0, timeout);
1386	const requestId = yield* idAllocator.allocate.runtimeRequest({
1387	driver: PI_PROVIDER,
1388	...(turn === null ? {} : { providerTurnId: turn.providerTurn.id }),
1389	nativeRequestId,
1390	});
1391	const nodeId = idAllocator.derive.approvalNode({ requestId });
1392	const title = recordString(event, "title") ?? method;
1393	const threadId =
1394	turn?.turnInput.threadId ?? state?.providerThread.appThreadId ?? input.threadId;
1395	const providerThreadId = state?.providerThread.id ?? null;
1396	const providerTurnId = turn?.providerTurn.id ?? null;
1397	const runtimeRequest: OrchestrationV2RuntimeRequest = {
1398	id: requestId,
1399	nodeId,
1400	providerTurnId,
1401	nativeRequestRef: providerRef(nativeRequestId),
1402	kind: method === "confirm" ? "command" : "user_input",
1403	status: "pending",
1404	responseCapability: { type: "live", providerSessionId: input.providerSessionId },
1405	createdAt,
1406	resolvedAt: null,
1407	};
1408	const node: OrchestrationV2ExecutionNode = {
1409	id: nodeId,
1410	threadId,
1411	runId: turn?.turnInput.runId ?? null,
1412	parentNodeId: turn?.turnInput.rootNodeId ?? null,
1413	rootNodeId: turn?.turnInput.rootNodeId ?? nodeId,
1414	kind: method === "confirm" ? "approval_request" : "user_input_request",
1415	status: "waiting",
1416	countsForRun: false,
1417	providerThreadId,
1418	providerTurnId,
1419	nativeItemRef: providerRef(nativeRequestId),
1420	runtimeRequestId: requestId,
1421	checkpointScopeId: null,
1422	startedAt: createdAt,
1423	completedAt: null,
1424	};
1425	const itemBase = {
1426	id: idAllocator.derive.approvalTurnItem({ requestId }),
1427	threadId,
1428	runId: turn?.turnInput.runId ?? null,
1429	nodeId,
1430	providerThreadId,
1431	providerTurnId,
1432	nativeItemRef: providerRef(nativeRequestId),
1433	parentItemId: null,
1434	// Runless startup/session-switch requests are normalized into the
1435	// thread-level ordinal range by TurnItemPositionStore.
1436	ordinal: turn === null ? 0 : itemOrdinal(turn, nativeRequestId),
1437	status: "waiting" as const,
1438	title,
1439	startedAt: createdAt,
1440	completedAt: null,
1441	updatedAt: createdAt,
1442	};
1443	const turnItem: OrchestrationV2TurnItem =
1444	method === "confirm"
1445	? {
1446	...itemBase,
1447	type: "approval_request",
1448	requestId,
1449	requestKind: piApprovalRequestKind(title),
1450	prompt: recordString(event, "message") ?? title,
1451	}
1452	: {
1453	...itemBase,
1454	type: "user_input_request",
1455	requestId,
1456	questions: [piQuestion(nativeRequestId, method, title, event)],
1457	};
1458	const pending: PendingPiPrompt = {
1459	nativeRequestId,
1460	method,
1461	questionId: nativeRequestId,
1462	approvalKey,
1463	expiresAt,
1464	expirationFiber: null,
1465	runtimeRequest,
1466	node,
1467	turnItem,
1468	};
1469	pendingPrompts.set(String(requestId), pending);
1470	yield* emit({
1471	type: "runtime_request.updated",
1472	driver: PI_PROVIDER,
1473	threadId,
1474	runtimeRequest,
1475	});
1476	yield* emit({ type: "node.updated", driver: PI_PROVIDER, node });
1477	yield* emit({ type: "turn_item.updated", driver: PI_PROVIDER, turnItem });
1478	if (expiresAt !== null) {
1479	const now = yield* DateTime.now;
1480	pending.expirationFiber = yield* Effect.sleep(
1481	Duration.millis(Math.max(0, expiresAt - DateTime.toEpochMillis(now))),
1482	).pipe(
1483	Effect.andThen(
1484	Queue.offer(connection.events, {
1485	type: "t3.ui_request_expired",
1486	requestId: String(requestId),
1487	pending,
1488	}),
1489	),
1490	Effect.asVoid,
1491	Effect.forkIn(scope),
1492	);
1493	}
1494	});
1495
1496	const emitExtensionError = Effect.fnUntraced(function* (event: PiRpcRecord) {
1497	const state = threadState;
1498	const turn = state?.activeTurn ?? null;
1499	if (turn === null) {
1500	outOfTurnExtensionErrors.push(event);
1501	return;
1502	}
1503	const emittedAt = yield* DateTime.now;
1504	const nativeItemId = `extension-error:${turn.nextItemOrdinal}`;
1505	const extensionName = piExtensionDisplayName(recordString(event, "extensionPath"));
1506	const extensionEvent = recordString(event, "event");
1507	const detail = recordString(event, "error")?.trim();
1508	const message = [
1509	`${extensionName} failed${extensionEvent === undefined ? "" : ` during ${extensionEvent}`}.`,
1510	detail === undefined || detail.length === 0 ? undefined : detail.slice(0, 2_000),
1511	]
1512	.filter((part): part is string => part !== undefined)
1513	.join("\n\n");
1514	const failure = makeProviderFailure({
1515	message,
1516	class: "provider_error",
1517	retryable: false,
1518	});
1519	yield* emitItemNode(turn, nativeItemId, "system", "failed", emittedAt, emittedAt);
1520	yield* emit({
1521	type: "turn_item.updated",
1522	driver: PI_PROVIDER,
1523	turnItem: {
1524	...baseItemFields(turn, nativeItemId, emittedAt, emittedAt),
1525	status: "failed",
1526	title: extensionName,
1527	completedAt: emittedAt,
1528	type: "error",
1529	failure,
1530	},
1531	});
1532	});
1533
1534	// ── turn lifecycle ────────────────────────────────────
1535
1536	/** One `get_entries` listing, or null when the request failed or named no leaf. */
1537	const listSessionEntries = (since: string | null, timeoutMs: number) =>
1538	request({ type: "get_entries", ...(since === null ? {} : { since }) }, timeoutMs).pipe(
1539	Effect.map((data) => {
1540	const leafId = recordField(data, "leafId");
1541	const entries = piSessionEntries(recordField(data, "entries"));
1542	return entries !== null && (typeof leafId === "string" || leafId === null)
1543	? { entries, leafId }
1544	: null;
1545	}),
1546	Effect.orElseSucceed(() => null),
1547	);
1548
1549	/**
1550	* Locate this turn's first user entry and the new leaf in pi's session
1551	* tree. The user-entry id becomes the provider turn's native ref (the
1552	* point `fork` rolls back to); the leaf becomes the conversation head.
1553	* `get_entries` is append-ordered and keeps abandoned branches, so both
1554	* come from walking the leaf's `parentId` chain, never from list order.
1555	* A leaf that no longer descends from the previous one means something
1556	* outside T3 rewound the session, such as an extension calling
1557	* `navigateTree`. The user entries left on the active branch are then
1558	* returned so orchestration can roll back runs that fell off it.
1559	* Pure bookkeeping: failures degrade to the synthetic refs.
1560	*/
1561	const captureTurnTreeRefs = Effect.fnUntraced(function* (
1562	timeoutMs = PI_REQUEST_TIMEOUT_MS,
1563	): Effect.fn.Return<PiTurnTreeRefs | null> {
1564	const cursorWasStale = leafCursorStale;
1565	const previousLeaf = lastKnownLeaf;
1566	const fullListing = cursorWasStale || previousLeaf === null;
1567	const listing = yield* listSessionEntries(fullListing ? null : previousLeaf, timeoutMs);
1568	const turnWalk =
1569	listing === null ? null : walkPiBranch(listing.entries, listing.leafId, previousLeaf);
1570	const rewound = turnWalk !== null && previousLeaf !== null && !turnWalk.reachedStop;
1571	// A `since` window ends at the previous leaf; the rewound branch
1572	// continues past it, so its ancestry needs the whole tree.
1573	const tree = rewound && !fullListing ? yield* listSessionEntries(null, timeoutMs) : listing;
1574	if (listing === null || turnWalk === null || tree === null) {
1575	// The read failed or its ancestry loops. Pi may have advanced past
1576	// `lastKnownLeaf` meanwhile, so the cursor can no longer be trusted
1577	// to bound a single turn.
1578	leafCursorStale = true;
1579	return null;
1580	}
1581	lastKnownLeaf = listing.leafId;
1582	leafCursorStale = false;
1583	const turnStartEntryId = turnWalk.userEntryIds.at(-1);
1584	// Only a branch walked to its root lists every survivor; a partial or
1585	// cyclic walk would roll back runs that are still in the conversation.
1586	// A replaced session has a different tree. Only an old leaf still in
1587	// this tree proves an in-session rewind rather than a session switch.
1588	const branch =
1589	rewound && previousLeaf !== null && tree.entries.has(previousLeaf)
1590	? walkPiBranch(tree.entries, listing.leafId, null)
1591	: undefined;
1592	return {
1593	// Without a trustworthy cursor the walk spans more than one turn, so
1594	// its earliest user entry belongs to an earlier turn. Keep the
1595	// synthetic ref rather than pointing rollback too far back; the next
1596	// turn gets an accurate ref again. A turn that left nothing on the
1597	// active branch (a command, or a rewind) gets no ref: a rollback
1598	// past it needs no fork of its own.
1599	nativeTurnRef: cursorWasStale
1600	? undefined
1601	: turnStartEntryId !== undefined
1602	? providerRef(turnStartEntryId)
1603	: turnWalk.entryCount === 0
1604	? null
1605	: undefined,
1606	leafId: listing.leafId,
1607	retainedNativeTurnIds: branch?.reachedStop === true ? branch.userEntryIds : undefined,
1608	};
1609	});
1610
1611	const finalizeTurn = Effect.fnUntraced(function* (state: PiThreadState, readUsage = true) {
1612	const turn = state.activeTurn;
1613	if (turn === null) return;
1614	state.activeTurn = null;
1615	const interrupted = turn.interrupted || turn.nativeAborted;
1616	const completedAt = yield* DateTime.now;
1617	yield* completeOpenStreamItems(turn);
1618	if (turn.activeCompaction !== null) {
1619	const status = interrupted ? "cancelled" : turn.failure === null ? "completed" : "failed";
1620	yield* emitCompaction(turn, turn.activeCompaction, status);
1621	turn.activeCompaction = null;
1622	}
1623	if (turn.activeProviderRetry !== null) {
1624	if (interrupted) {
1625	yield* emitProviderRetry(turn, turn.activeProviderRetry, "interrupted", completedAt);
1626	turn.activeProviderRetry = null;
1627	} else if (turn.failure === null) {
1628	yield* emitProviderRetry(turn, turn.activeProviderRetry, "completed", completedAt);
1629	turn.activeProviderRetry = null;
1630	}
1631	}
1632	yield* cancelPendingPrompts(completedAt);
1633	const treeRefs =
1634	turn.stopTreeRefs !== undefined ? turn.stopTreeRefs : yield* captureTurnTreeRefs();
1635	const tokenUsage = readUsage
1636	? yield* readTokenUsage(turn.latestCompactionAfterTokens, completedAt)
1637	: undefined;
1638	const failure = interrupted ? null : (turn.rejectedPromptFailure ?? turn.failure);
1639	yield* emit({
1640	type: "provider_turn.updated",
1641	driver: PI_PROVIDER,
1642	threadId: turn.turnInput.threadId,
1643	providerTurn: {
1644	...turn.providerTurn,
1645	...(treeRefs === null || treeRefs.nativeTurnRef === undefined
1646	? {}
1647	: { nativeTurnRef: treeRefs.nativeTurnRef }),
1648	status: interrupted ? "interrupted" : failure !== null ? "failed" : "completed",
1649	completedAt,
1650	...(tokenUsage === undefined ? {} : { tokenUsage }),
1651	},
1652	});
1653	yield* updateProviderThread(
1654	state,
1655	{
1656	status: "idle",
1657	...(treeRefs === null
1658	? {}
1659	: {
1660	nativeConversationHeadRef:
1661	treeRefs.leafId === null ? null : providerRef(treeRefs.leafId),
1662	}),
1663	},
1664	treeRefs?.retainedNativeTurnIds,
1665	);
1666	yield* updateProviderSession(
1667	failure !== null ? "error" : "ready",
1668	failure?.message ?? null,
1669	);
1670	if (failure !== null) {
1671	const failureItemId = `terminal-failure:${turn.providerTurn.id}`;
1672	if (turn.activeProviderRetry !== null) {
1673	yield* emitProviderRetry(
1674	turn,
1675	{ ...turn.activeProviderRetry, failure },
1676	"failed",
1677	completedAt,
1678	);
1679	} else {
1680	yield* emit({
1681	type: "turn_item.updated",
1682	driver: PI_PROVIDER,
1683	turnItem: {
1684	...baseItemFields(turn, failureItemId, completedAt, completedAt),
1685	status: "failed",
1686	title: null,
1687	completedAt,
1688	type: "error",
1689	failure,
1690	},
1691	});
1692	}
1693	yield* emit({
1694	type: "turn.terminal",
1695	driver: PI_PROVIDER,
1696	providerThreadId: state.providerThread.id,
1697	providerTurnId: turn.providerTurn.id,
1698	runOrdinal: turn.turnInput.runOrdinal,
1699	failureItemOrdinal: itemOrdinal(turn, failureItemId),
1700	status: "failed",
1701	failure,
1702	...(turn.activeProviderRetry === null
1703	? {}
1704	: {
1705	retry: turn.activeProviderRetry.retry,
1706	retryStartedAt: turn.activeProviderRetry.startedAt,
1707	}),
1708	threadDisposition: "reusable",
1709	});
1710	} else {
1711	yield* emit({
1712	type: "turn.terminal",
1713	driver: PI_PROVIDER,
1714	providerThreadId: state.providerThread.id,
1715	providerTurnId: turn.providerTurn.id,
1716	runOrdinal: turn.turnInput.runOrdinal,
1717	status: interrupted ? "interrupted" : "completed",
1718	failure: null,
1719	threadDisposition: "reusable",
1720	});
1721	}
1722	});
1723
1724	// ── event pump ────────────────────────────────────────
1725
1726	const scheduleSettleProbe = (
1727	turn: ActivePiTurn,
1728	settleAfterAgentActivity = false,
1729	attempt = 1,
1730	) => {
1731	const providerTurnId = turn.providerTurn.id;
1732	const settleProbeGeneration = turn.settleProbeGeneration;
1733	return request({ type: "get_state" }, 2_000).pipe(
1734	Effect.matchEffect({
1735	onSuccess: (data) =>
1736	Queue.offer(connection.events, {
1737	type: "t3.settle_probe",
1738	providerTurnId,
1739	settleAfterAgentActivity,
1740	settleProbeGeneration,
1741	attempt,
1742	data,
1743	}),
1744	// A failed probe still has to reach the pump. Dropping it would
1745	// leave a command-only turn active forever, because Pi never emits
1746	// agent events for one.
1747	onFailure: () =>
1748	Queue.offer(connection.events, {
1749	type: "t3.settle_probe",
1750	providerTurnId,
1751	settleAfterAgentActivity,
1752	settleProbeGeneration,
1753	attempt,
1754	probeFailed: true,
1755	}),
1756	}),
1757	Effect.ignore,
1758	Effect.forkIn(scope),
1759	);
1760	};
1761
1762	const stopPendingWake = Effect.fnUntraced(function* (wake: PiWake) {
1763	if (pendingWake !== wake || threadState !== wake.state || closed) return;
1764	pendingWake = null;
1765	stopRequested = true;
1766	yield* cancelPendingPrompts(yield* DateTime.now);
1767	yield* connection.terminate;
1768	});
1769
1770	const handleSessionEvent = Effect.fnUntraced(function* (event: PiRpcRecord) {
1771	// Model-selection hooks can await a dialog before startTurn can claim
1772	// a wake. Keep interactive requests session-scoped until a turn owns
1773	// them; buffer native output and notifications for that turn.
1774	const interactiveUiRequest =
1775	event["type"] === "extension_ui_request" &&
1776	["select", "confirm", "input", "editor"].includes(String(event["method"]));
1777	if (
1778	threadState?.activeTurn === null &&
1779	pendingWake !== null &&
1780	!interactiveUiRequest &&
1781	!String(event["type"]).startsWith("t3.") &&
1782	event["type"] !== "response"
1783	) {
1784	pendingWake.events.push(event);
1785	if (event["type"] === "agent_start") yield* updateProviderSession("running", null);
1786	if (event["type"] === "agent_settled") yield* updateProviderSession("ready", null);
1787	return;
1788	}
1789	const state = threadState;
1790	const turn = state?.activeTurn ?? null;
1791	switch (event["type"]) {
1792	case "agent_start": {
1793	if (
1794	turn === null &&
1795	state?.providerThread.appThreadId != null &&
1796	continuationRequests !== undefined &&
1797	!closed &&
1798	!stopRequested
1799	) {
1800	const wake: PiWake = { state, events: [event] };
1801	pendingWake = wake;
1802	if (rollbackBarrier !== null) {
1803	yield* stopPendingWake(wake);
1804	return;
1805	}
1806	yield* updateProviderSession("running", null);
1807	yield* continuationRequests
1808	.offer({
1809	threadId: state.providerThread.appThreadId,
1810	providerThreadId: state.providerThread.id,
1811	driver: PI_PROVIDER,
1812	detail: null,
1813	delivery: "adapter_buffered",
1814	dispatchIfCurrent: (dispatch) =>
1815	sessionEventPermit.withPermits(1)(
1816	Effect.suspend(() =>
1817	pendingWake === wake && threadState === state && !closed
1818	? dispatch.pipe(
1819	Effect.map(Option.some),
1820	Effect.tapError(() => stopPendingWake(wake)),
1821	)
1822	: Effect.succeed(Option.none()),
1823	),
1824	),
1825	clearIfCurrent: () => sessionEventPermit.withPermits(1)(stopPendingWake(wake)),
1826	})
1827	.pipe(Effect.tapError(() => stopPendingWake(wake)));
1828	return;
1829	}
1830	if (turn === null) {
1831	unsolicitedActivityDetected = true;
1832	yield* updateProviderSession("error", PI_UNSOLICITED_ACTIVITY_ERROR);
1833	yield* connection.terminate;
1834	return;
1835	}
1836	turn.sawAgentActivity = true;
1837	turn.nativeAborted = false;
1838	turn.settleProbeGeneration += 1;
1839	return;
1840	}
1841	case "message_start": {
1842	if (turn !== null && recordString(event["message"], "role") === "assistant") {
1843	turn.sawAgentActivity = true;
1844	turn.messageOrdinal += 1;
1845	}
1846	return;
1847	}
1848	case "message_update": {
1849	if (turn === null) return;
1850	turn.sawAgentActivity = true;
1851	yield* reportLiveUsage(turn, event["usage"]);
1852	const delta = event["assistantMessageEvent"];
1853	const deltaType = recordString(delta, "type");
1854	const contentIndex = recordNumber(delta, "contentIndex") ?? 0;
1855	if (deltaType === "text_delta" || deltaType === "thinking_delta") {
1856	const item = yield* streamItemFor(
1857	turn,
1858	deltaType === "text_delta" ? "assistant_message" : "reasoning",
1859	contentIndex,
1860	);
1861	item.text += recordString(delta, "delta") ?? "";
1862	yield* scheduleStreamFlush(turn, item);
1863	return;
1864	}
1865	if (deltaType === "text_end" || deltaType === "thinking_end") {
1866	const item = yield* streamItemFor(
1867	turn,
1868	deltaType === "text_end" ? "assistant_message" : "reasoning",
1869	contentIndex,
1870	);
1871	yield* completeStreamItem(
1872	turn,
1873	item,
1874	recordString(delta, "content") ?? recordString(delta, "thinking"),
1875	);
1876	return;
1877	}
1878	return;
1879	}
1880	case "message_end": {
1881	if (turn === null) return;
1882	const message = event["message"];
1883	if (recordString(message, "role") !== "assistant") return;
1884	yield* completeOpenStreamItems(turn);
1885	if (recordString(message, "stopReason") === "error" && turn.failure === null) {
1886	turn.failure = makeProviderFailure({
1887	message: recordString(message, "errorMessage") ?? "Pi reported a model error.",
1888	class: "provider_error",
1889	});
1890	}
1891	return;
1892	}
1893	case "tool_execution_start":
1894	if (turn !== null) {
1895	turn.sawAgentActivity = true;
1896	yield* emitToolItem(turn, event, "start");
1897	}
1898	return;
1899	case "tool_execution_update":
1900	if (turn !== null) yield* emitToolItem(turn, event, "update");
1901	return;
1902	case "tool_execution_end":
1903	if (turn !== null) yield* emitToolItem(turn, event, "end");
1904	return;
1905	case "compaction_start": {
1906	if (turn === null) return;
1907	turn.settleProbeGeneration += 1;
1908	turn.sawCompaction = true;
1909	if (turn.activeCompaction !== null) {
1910	yield* emitCompaction(turn, turn.activeCompaction, "cancelled");
1911	}
1912	const startedAt = yield* DateTime.now;
1913	const compaction = {
1914	nativeItemId: compactionNativeItemId(turn),
1915	startedAt,
1916	} satisfies PiCompactionState;
1917	turn.activeCompaction = compaction;
1918	yield* emitCompaction(turn, compaction, "running");
1919	return;
1920	}
1921	case "compaction_end": {
1922	if (turn === null) return;
1923	const observedAt = yield* DateTime.now;
1924	const compaction = turn.activeCompaction ?? {
1925	nativeItemId: compactionNativeItemId(turn),
1926	startedAt: observedAt,
1927	};
1928	turn.activeCompaction = null;
1929	const result = event["result"];
1930	if (result === null || result === undefined) {
1931	if (event["aborted"] === true) {
1932	yield* emitCompaction(turn, compaction, "cancelled");
1933	if (turn.settleWhenIdle || !turn.sawAgentActivity) {
1934	yield* scheduleSettleProbe(turn, turn.settleWhenIdle);
1935	}
1936	return;
1937	}
1938	const errorMessage =
1939	recordString(event, "errorMessage") ?? "Pi context compaction failed.";
1940	yield* emitCompaction(turn, compaction, "failed", {
1941	summary: errorMessage.slice(0, 1_000),
1942	});
1943	if (turn.settleWhenIdle || !turn.sawAgentActivity) {
1944	yield* scheduleSettleProbe(turn, turn.settleWhenIdle);
1945	}
1946	return;
1947	}
1948	// An overflow can surface as a model error (`message_end` with
1949	// stopReason error) before Pi compacts and retries the turn. Clear
1950	// that failure only when Pi confirms that compaction will retry;
1951	// a successful non-retrying compaction must not erase an exhausted
1952	// provider retry.
1953	if (event["willRetry"] === true) turn.failure = null;
1954	turn.latestCompactionAfterTokens =
1955	nonNegativeInteger(result, "estimatedTokensAfter") ?? null;
1956	const summary = recordString(result, "summary");
1957	const beforeTokenCount = nonNegativeInteger(result, "tokensBefore");
1958	const afterTokenCount = nonNegativeInteger(result, "estimatedTokensAfter");
1959	yield* emitCompaction(turn, compaction, "completed", {
1960	...(summary === undefined ? {} : { summary }),
1961	...(beforeTokenCount === undefined ? {} : { beforeTokenCount }),
1962	...(afterTokenCount === undefined ? {} : { afterTokenCount }),
1963	});
1964	if (turn.settleWhenIdle || !turn.sawAgentActivity) {
1965	yield* scheduleSettleProbe(turn, turn.settleWhenIdle);
1966	}
1967	return;
1968	}
1969	case "auto_retry_start": {
1970	if (turn === null) return;
1971	const emittedAt = yield* DateTime.now;
1972	const attempt = Math.max(1, Math.trunc(recordNumber(event, "attempt") ?? 1));
1973	const maxAttempts = Math.max(
1974	attempt,
1975	Math.trunc(recordNumber(event, "maxAttempts") ?? attempt),
1976	);
1977	const retryDelayMs = Math.max(0, Math.trunc(recordNumber(event, "delayMs") ?? 0));
1978	const failure = makeProviderFailure({
1979	message: recordString(event, "errorMessage") ?? "Pi provider request failed.",
1980	class: "provider_error",
1981	retryable: true,
1982	});
1983	const current = turn.activeProviderRetry;
1984	const providerRetry = {
1985	retry: { attempt, maxAttempts, retryDelayMs },
1986	failure,
1987	startedAt: current?.startedAt ?? emittedAt,
1988	itemOrdinal:
1989	current?.itemOrdinal ??
1990	itemOrdinal(turn, `terminal-failure:${turn.providerTurn.id}`),
1991	} satisfies PiProviderRetryState;
1992	turn.activeProviderRetry = providerRetry;
1993	yield* emitProviderRetry(turn, providerRetry, "running", emittedAt);
1994	return;
1995	}
1996	case "auto_retry_end": {
1997	if (turn === null) return;
1998	const emittedAt = yield* DateTime.now;
1999	if (event["success"] === true) {
2000	// The retry recovered. Pi emits the erroring `message_end`
2001	// before retrying, so leaving that failure in place would make
2002	// `agent_settled` terminalize a successful turn as failed.
2003	if (turn.activeProviderRetry !== null) {
2004	const attempt = Math.max(
2005	1,
2006	Math.trunc(
2007	recordNumber(event, "attempt") ?? turn.activeProviderRetry.retry.attempt,
2008	),
2009	);
2010	const recoveredRetry = {
2011	...turn.activeProviderRetry,
2012	retry: { ...turn.activeProviderRetry.retry, attempt },
2013	};
2014	yield* emitProviderRetry(turn, recoveredRetry, "completed", emittedAt);
2015	turn.activeProviderRetry = null;
2016	}
2017	turn.failure = null;
2018	return;
2019	}
2020	const failure = makeProviderFailure({
2021	message: recordString(event, "finalError") ?? "Pi auto-retry failed.",
2022	class: "provider_error",
2023	retryable: false,
2024	});
2025	const attempt = Math.max(1, Math.trunc(recordNumber(event, "attempt") ?? 1));
2026	const current = turn.activeProviderRetry;
2027	const providerRetry = {
2028	retry: {
2029	attempt,
2030	maxAttempts: current?.retry.maxAttempts ?? attempt,
2031	retryDelayMs: current?.retry.retryDelayMs ?? null,
2032	},
2033	failure,
2034	startedAt: current?.startedAt ?? emittedAt,
2035	itemOrdinal:
2036	current?.itemOrdinal ??
2037	itemOrdinal(turn, `terminal-failure:${turn.providerTurn.id}`),
2038	} satisfies PiProviderRetryState;
2039	turn.activeProviderRetry = providerRetry;
2040	turn.failure = failure;
2041	yield* emitProviderRetry(turn, providerRetry, "failed", emittedAt);
2042	return;
2043	}
2044	case "extension_ui_request":
2045	yield* handleExtensionUiRequest(event);
2046	return;
2047	case "extension_error": {
2048	yield* emitExtensionError(event);
2049	return;
2050	}
2051	case "agent_settled": {
2052	if (turn?.interrupted === true) {
2053	if (state !== null) yield* finalizeTurn(state);
2054	return;
2055	}
2056	if (turn !== null) {
2057	turn.nativeAborted = event["aborted"] === true;
2058	turn.settleWhenIdle = true;
2059	turn.settleProbeGeneration += 1;
2060	// A wake can settle while the newly joined user prompt is still
2061	// expanding. Its acknowledgement opens the idle-probe barrier.
2062	if (
2063	turn.adoptedWake &&
2064	pendingPromptResponses.some(
2065	(pending) =>
2066	pending.providerTurnId === turn.providerTurn.id &&
2067	pending.kind === "turn_start",
2068	)
2069	)
2070	return;
2071	yield* scheduleSettleProbe(turn, true);
2072	}
2073	return;
2074	}
2075	case "response": {
2076	// Correlated responses never reach the pump; an id-less response
2077	// is the deferred ack of a fire-and-forget prompt/steer/compact.
2078	const command = recordString(event, "command");
2079	if (command === "compact") {
2080	const pendingCompact = pendingCompactResponses.shift();
2081	const compactTurn =
2082	pendingCompact?.providerTurnId === turn?.providerTurn.id ? turn : null;
2083	if (compactTurn !== null) compactTurn.manualCompactInFlight = false;
2084	if (event["success"] === true) {
2085	if (
2086	pendingCompact?.kind === "turn_start" &&
2087	compactTurn !== null &&
2088	compactTurn.promptMayBeCommandOnly &&
2089	!compactTurn.sawAgentActivity
2090	) {
2091	yield* scheduleSettleProbe(compactTurn);
2092	}
2093	return;
2094	}
2095	if (event["success"] !== false) return;
2096	if (compactTurn === null) return;
2097	if (compactTurn.activeCompaction !== null) return;
2098	if (pendingCompact?.kind === "steer") {
2099	yield* Effect.logWarning("Pi rejected a compact steer.", {
2100	errorLength: recordString(event, "error")?.length,
2101	});
2102	return;
2103	}
2104	if (!compactTurn.sawCompaction) {
2105	compactTurn.failure = makeProviderFailure({
2106	message: recordString(event, "error") ?? "Pi compact failed.",
2107	class: "provider_error",
2108	});
2109	if (state !== null) yield* finalizeTurn(state);
2110	return;
2111	}
2112	if (!compactTurn.sawAgentActivity) {
2113	yield* scheduleSettleProbe(compactTurn);
2114	}
2115	return;
2116	}
2117	const pendingPrompt = command === "prompt" ? pendingPromptResponses.shift() : undefined;
2118	const responseTurn =
2119	pendingPrompt?.providerTurnId === turn?.providerTurn.id ? turn : null;
2120	if (event["success"] === true) {
2121	// Deferred success ack. Command-only prompts (pure extension
2122	// slash commands) never start an agent run and never emit
2123	// `agent_settled`, so probe for idleness. The probe result is
2124	// re-queued behind any events Pi emitted before answering
2125	// get_state, which keeps the check stream-ordered.
2126	if (
2127	pendingPrompt?.kind === "turn_start" &&
2128	responseTurn !== null &&
2129	((responseTurn.promptMayBeCommandOnly && !responseTurn.sawAgentActivity) ||
2130	(responseTurn.adoptedWake &&
2131	(responseTurn.settleWhenIdle || responseTurn.promptMayBeCommandOnly)))
2132	) {
2133	yield* scheduleSettleProbe(responseTurn, responseTurn.adoptedWake);
2134	}
2135	return;
2136	}
2137	if (event["success"] !== false) return;
2138	if (command === "steer" || pendingPrompt?.kind === "steer") {
2139	// A rejected steer only means that one message was refused. The
2140	// turn it was aimed at is still running on Pi, so terminalizing
2141	// here would report a failure while output keeps streaming.
2142	yield* Effect.logWarning("Pi rejected a steer message.", {
2143	errorLength: recordString(event, "error")?.length,
2144	});
2145	return;
2146	}
2147	const failedTurn =
2148	command === "prompt" && pendingPrompt?.kind === "turn_start"
2149	? responseTurn
2150	: command === "parse"
2151	? turn
2152	: null;
2153	if (failedTurn !== null) {
2154	const failure = makeProviderFailure({
2155	message: recordString(event, "error") ?? "Pi rejected the prompt.",
2156	class: "provider_error",
2157	});
2158	if (failedTurn.adoptedWake && command === "prompt") {
2159	failedTurn.rejectedPromptFailure = failure;
2160	failedTurn.settleWhenIdle = true;
2161	yield* scheduleSettleProbe(failedTurn, true);
2162	} else {
2163	failedTurn.failure = failure;
2164	if (state !== null) yield* finalizeTurn(state);
2165	}
2166	}
2167	return;
2168	}
2169	case "t3.ui_request_expired": {
2170	const requestId = recordString(event, "requestId");
2171	if (requestId === undefined) return;
2172	const pending = pendingPrompts.get(requestId);
2173	if (pending === undefined || pending !== event["pending"]) return;
2174	pendingPrompts.delete(requestId);
2175	yield* cancelPrompt(pending, yield* DateTime.now);
2176	return;
2177	}
2178	case "t3.flush_extension_errors": {
2179	// Startup extension failures are informational and do not block
2180	// Pi, so attach them to the next real turn instead of creating a
2181	// standalone failed run.
2182	for (const extensionError of outOfTurnExtensionErrors.splice(0)) {
2183	yield* emitExtensionError(extensionError);
2184	}
2185	return;
2186	}
2187	case "t3.rollback_barrier": {
2188	if (rollbackBarrier !== null && event["barrier"] === rollbackBarrier)
2189	yield* Deferred.succeed(rollbackBarrier, undefined);
2190	return;
2191	}
2192	case "t3.settle_probe": {
2193	// New work increments the generation before the pump can consume
2194	// a stale idle snapshot, so only a current snapshot may settle.
2195	const data = event["data"];
2196	const probeFailed = event["probeFailed"] === true;
2197	const settleAfterAgentActivity = event["settleAfterAgentActivity"] === true;
2198	const attempt = Math.max(1, Math.trunc(recordNumber(event, "attempt") ?? 1));
2199	if (
2200	turn === null ||
2201	turn.providerTurn.id !== event["providerTurnId"] ||
2202	turn.settleProbeGeneration !== event["settleProbeGeneration"] ||
2203	(!settleAfterAgentActivity && turn.sawAgentActivity) ||
2204	turn.activeCompaction !== null
2205	) {
2206	return;
2207	}
2208	if (probeFailed) {
2209	if (!settleAfterAgentActivity) {
2210	if (state !== null) yield* finalizeTurn(state);
2211	return;
2212	}
2213	if (attempt < SETTLE_PROBE_MAX_ATTEMPTS) {
2214	yield* Effect.sleep(SETTLE_PROBE_RETRY_DELAY).pipe(
2215	Effect.andThen(scheduleSettleProbe(turn, true, attempt + 1)),
2216	Effect.forkIn(scope),
2217	);
2218	return;
2219	}
2220	stopRequested = true;
2221	yield* connection.terminate;
2222	return;
2223	}
2224	if (
2225	recordField(data, "isStreaming") !== true &&
2226	recordField(data, "isCompacting") !== true &&
2227	(recordNumber(data, "pendingMessageCount") ?? 0) === 0
2228	) {
2229	turn.settleWhenIdle = false;
2230	if (state !== null) yield* finalizeTurn(state);
2231	}
2232	return;
2233	}
2234	default:
2235	return;
2236	}
2237	});
2238
2239	yield* Effect.gen(function* () {
2240	while (true) {
2241	const event = yield* Queue.take(connection.events);
2242	yield* sessionEventPermit.withPermits(1)(handleSessionEvent(event));
2243	}
2244	}).pipe(
2245	Effect.catchCause((cause) =>
2246	sessionEventPermit.withPermits(1)(
2247	Effect.gen(function* () {
2248	// Transport death finalizes any live turn. Stop-with-restart
2249	// closes the provider stream cleanly; only an unexpected death
2250	// is surfaced as an event-stream failure.
2251	const state = threadState;
2252	if (rollbackBarrier !== null)
2253	yield* Deferred.fail(
2254	rollbackBarrier,
2255	protocolError("Pi transport closed during rollback"),
2256	);
2257	pendingWake = null;
2258	const interrupted = state?.activeTurn?.interrupted === true;
2259	if (state?.activeTurn != null) {
2260	state.activeTurn.failure = interrupted
2261	? null
2262	: makeProviderFailure({
2263	cause,
2264	message: "Pi process exited unexpectedly.",
2265	class: "transport_error",
2266	});
2267	yield* finalizeTurn(state, false);
2268	} else {
2269	yield* cancelPendingPrompts(yield* DateTime.now);
2270	}
2271	if (unsolicitedActivityDetected) {
2272	yield* updateProviderSession("error", PI_UNSOLICITED_ACTIVITY_ERROR);
2273	yield* Queue.end(events);
2274	} else if (stopRequested) {
2275	yield* updateProviderSession("stopped", null);
2276	yield* Queue.end(events);
2277	} else {
2278	yield* updateProviderSession(
2279	"error",
2280	interrupted ? "Pi process was stopped." : "Pi process exited unexpectedly.",
2281	);
2282	yield* Queue.fail(
2283	events,
2284	new ProviderAdapter.ProviderAdapterEventStreamError({
2285	driver: PI_PROVIDER,
2286	providerSessionId: input.providerSessionId,
2287	cause,
2288	}),
2289	);
2290	}
2291	}),
2292	),
2293	),
2294	Effect.forkIn(scope),
2295	);
2296
2297	// ── session runtime ───────────────────────────────────
2298
2299	const registerThread = Effect.fnUntraced(function* (
2300	threadInput: ProviderAdapter.ProviderAdapterV2EnsureThreadInput,
2301	publish = true,
2302	) {
2303	if (rollbackBarrier !== null)
2304	return yield* protocolError("Cannot register a Pi thread during rollback");
2305	if (threadState !== null && threadState.activeTurn !== null) {
2306	return yield* protocolError("Cannot register a Pi thread while a turn is active");
2307	}
2308	if (pendingWake !== null) {
2309	const state = threadState;
2310	if (
2311	state !== null &&
2312	state === pendingWake.state &&
2313	state.providerThread.appThreadId === threadInput.threadId &&
2314	(threadInput.existingProviderThread === undefined ||
2315	threadInput.existingProviderThread.nativeThreadRef?.nativeId ===
2316	state.providerThread.nativeThreadRef?.nativeId)
2317	)
2318	return state.providerThread;
2319	return yield* protocolError(
2320	"Cannot switch Pi threads while native work is awaiting a turn",
2321	);
2322	}
2323	const existing = threadInput.existingProviderThread;
2324	const resumeId = existing?.nativeThreadRef?.nativeId;
2325	const needsNewSession = resumeId == null && registrationAttempted;
2326	registrationAttempted = true;
2327	if (resumeId != null || needsNewSession) {
2328	lastNativeThreadId = resumeId ?? lastNativeThreadId;
2329	// Even a failed lifecycle operation can change Pi's native session.
2330	// Never leave the old app binding or model defaults usable afterward.
2331	threadState = null;
2332	appliedModel = null;
2333	appliedThinking = null;
2334	appliedSessionName = null;
2335	baselineModel = null;
2336	baselineThinking = null;
2337	contextWindow = null;
2338	const result = yield* lifecycleRequest(
2339	resumeId != null
2340	? { type: "switch_session", sessionPath: resumeId }
2341	: { type: "new_session" },
2342	);
2343	if (recordField(result, "cancelled") === true) {
2344	return yield* protocolError("A Pi extension cancelled the session switch");
2345	}
2346	}
2347	const stateData = yield* request({ type: "get_state" });
2348	if (!modelsDiscovered) {
2349	const modelsData = yield* request({ type: "get_available_models" }).pipe(
2350	Effect.orElseSucceed(() => undefined),
2351	);
2352	const models = recordField(modelsData, "models");
2353	if (Array.isArray(models)) {
2354	for (const model of models) rememberModelContextWindow(model);
2355	modelsDiscovered = true;
2356	}
2357	}
2358	contextWindow = rememberModelContextWindow(recordField(stateData, "model"));
2359	// Each baseline is captured independently, and only while nothing has
2360	// been applied yet, so a `get_state` that arrives after our own
2361	// selection cannot record that selection as Pi's default.
2362	if (baselineModel === null && appliedModel === null) {
2363	const stateModel = recordField(stateData, "model");
2364	const provider = recordString(stateModel, "provider");
2365	const modelId = recordString(stateModel, "id");
2366	if (provider !== undefined && modelId !== undefined) {
2367	baselineModel = { provider, modelId };
2368	}
2369	}
2370	if (baselineThinking === null && appliedThinking === null) {
2371	baselineThinking = recordString(stateData, "thinkingLevel") ?? null;
2372	}
2373	// switch_session accepts a file path, not Pi's display session UUID.
2374	const nativeId = recordString(stateData, "sessionFile");
2375	if (nativeId === undefined) {
2376	return yield* protocolError("get_state returned no persisted sessionFile", stateData);
2377	}
2378	if (needsNewSession && nativeId === lastNativeThreadId) {
2379	return yield* protocolError("Pi did not create a distinct session file");
2380	}
2381	lastNativeThreadId = nativeId;
2382	const createdAt = yield* DateTime.now;
2383	const providerThread: OrchestrationV2ProviderThread =
2384	existing !== undefined
2385	? {
2386	...existing,
2387	providerSessionId: input.providerSessionId,
2388	nativeThreadRef: providerRef(nativeId),
2389	...(needsNewSession ? { nativeConversationHeadRef: null, contextUsage: null } : {}),
2390	status: "idle",
2391	updatedAt: createdAt,
2392	}
2393	: {
2394	id: idAllocator.derive.providerThread({
2395	driver: PI_PROVIDER,
2396	nativeThreadId: nativeId,
2397	}),
2398	driver: PI_PROVIDER,
2399	providerInstanceId: options.instanceId,
2400	providerSessionId: input.providerSessionId,
2401	appThreadId: threadInput.threadId,
2402	ownerNodeId: null,
2403	nativeThreadRef: providerRef(nativeId),
2404	nativeConversationHeadRef: null,
2405	status: "idle",
2406	firstRunOrdinal: null,
2407	lastRunOrdinal: null,
2408	handoffIds: [],
2409	forkedFrom: null,
2410	pendingBackgroundTasks: [],
2411	createdAt,
2412	updatedAt: createdAt,
2413	};
2414	threadState = { providerThread, activeTurn: null };
2415	// Baseline the session-tree leaf so the first turn's user entry can
2416	// be located with a `since` cursor instead of a full entry scan.
2417	const baselineEntries = yield* request({ type: "get_entries" }).pipe(
2418	Effect.orElseSucceed(() => undefined),
2419	);
2420	lastKnownLeaf = recordString(baselineEntries, "leafId") ?? null;
2421	// A successful full baseline makes the cursor trustworthy again. Only
2422	// a failed one leaves it stale, so a recovered session does not keep
2423	// skipping turn refs. An empty tree is a success with no leafId.
2424	leafCursorStale = baselineEntries === undefined;
2425	if (publish)
2426	yield* emit({
2427	type: "provider_thread.updated",
2428	driver: PI_PROVIDER,
2429	providerThread,
2430	});
2431	return providerThread;
2432	});
2433
2434	const applySelection = Effect.fnUntraced(function* (modelSelection: ModelSelection) {
2435	const thinking = getModelSelectionStringOptionValue(modelSelection, "thinking");
2436	if (modelSelection.model === PI_INHERIT_MODEL_SLUG) {
2437	// Returning to "Pi default" after an explicit pick has to replay the
2438	// captured baseline, otherwise Pi stays on the last model applied.
2439	if (appliedModel !== null && baselineModel !== null) {
2440	const restoredModel = yield* request({ type: "set_model", ...baselineModel });
2441	contextWindow = rememberModelContextWindow(restoredModel);
2442	appliedModel = null;
2443	const updatedAt = yield* DateTime.now;
2444	sessionEntity = { ...sessionEntity, model: PI_INHERIT_MODEL_SLUG, updatedAt };
2445	yield* emit({
2446	type: "provider_session.updated",
2447	driver: PI_PROVIDER,
2448	providerSession: sessionEntity,
2449	});
2450	}
2451	// The "Pi default" model advertises no thinking choices of its own,
2452	// so an unqualified return also restores Pi's configured level
2453	// instead of silently keeping the effort a previous pick applied.
2454	if (
2455	thinking === undefined &&
2456	appliedThinking !== null &&
2457	baselineThinking !== null &&
2458	appliedThinking !== baselineThinking
2459	) {
2460	yield* request({ type: "set_thinking_level", level: baselineThinking });
2461	appliedThinking = null;
2462	}
2463	} else if (modelSelection.model !== appliedModel) {
2464	const parsed = parsePiModelSlug(modelSelection.model);
2465	if (parsed === null) {
2466	return yield* protocolError(
2467	`Pi model '${modelSelection.model}' must use provider/model format`,
2468	);
2469	}
2470	const selectedModel = yield* request({
2471	type: "set_model",
2472	provider: parsed.provider,
2473	modelId: parsed.modelId,
2474	});
2475	contextWindow = rememberModelContextWindow(selectedModel);
2476	appliedModel = modelSelection.model;
2477	const updatedAt = yield* DateTime.now;
2478	sessionEntity = { ...sessionEntity, model: modelSelection.model, updatedAt };
2479	yield* emit({
2480	type: "provider_session.updated",
2481	driver: PI_PROVIDER,
2482	providerSession: sessionEntity,
2483	});
2484	}
2485	if (
2486	thinking !== undefined &&
2487	thinking !== appliedThinking &&
2488	PI_THINKING_LEVELS.has(thinking)
2489	) {
2490	yield* request({ type: "set_thinking_level", level: thinking });
2491	appliedThinking = thinking;
2492	}
2493	});
2494
2495	const resolvePromptPayload = Effect.fnUntraced(function* (
2496	text: string,
2497	attachments: ReadonlyArray<ChatAttachment>,
2498	) {
2499	const images: Array<{ type: "image"; data: string; mimeType: string }> = [];
2500	const extraLines: Array<string> = [];
2501	for (const attachment of attachments) {
2502	const path = host.resolveAttachmentPath(attachment);
2503	if (path === null) continue;
2504	if (attachment.mimeType.startsWith("image/")) {
2505	const bytes = yield* fileSystem.readFile(path);
2506	images.push({
2507	type: "image",
2508	data: Buffer.from(bytes).toString("base64"),
2509	mimeType: attachment.mimeType,
2510	});
2511	} else {
2512	extraLines.push(`[Attachment saved at ${path}]`);
2513	}
2514	}
2515	const message = extraLines.length === 0 ? text : `${text}\n\n${extraLines.join("\n")}`;
2516	return { message, images };
2517	});
2518
2519	const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = {
2520	instanceId: options.instanceId,
2521	driver: PI_PROVIDER,
2522	providerSessionId: input.providerSessionId,
2523	get providerSession() {
2524	return sessionEntity;
2525	},
2526	events: Stream.fromQueue(events),
2527	hasPendingBackgroundWork: Effect.sync(() => pendingWake !== null),
2528	hasPendingBackgroundWorkForThread: (providerThread) =>
2529	Effect.sync(
2530	() => pendingWake !== null && pendingWake.state.providerThread.id === providerThread.id,
2531	),
2532	getModelContextWindow: (selection) => {
2533	if (selection.instanceId !== options.instanceId) return undefined;
2534	const slug =
2535	selection.model === PI_INHERIT_MODEL_SLUG
2536	? baselineModel === null
2537	? undefined
2538	: `${baselineModel.provider}/${baselineModel.modelId}`
2539	: selection.model;
2540	return slug === undefined ? undefined : modelContextWindows.get(slug);
2541	},
2542	ensureThread: (threadInput) =>
2543	registerThread(threadInput).pipe(
2544	Effect.mapError(
2545	(cause) =>
2546	new ProviderAdapter.ProviderAdapterEnsureThreadError({
2547	driver: PI_PROVIDER,
2548	threadId: threadInput.threadId,
2549	cause,
2550	}),
2551	),
2552	),
2553	resumeThread: (threadInput) =>
2554	registerThread({
2555	threadId:
2556	threadInput.threadId ?? threadInput.providerThread.appThreadId ?? input.threadId,
2557	modelSelection: threadInput.modelSelection ?? input.modelSelection,
2558	runtimePolicy: threadInput.runtimePolicy ?? input.runtimePolicy,
2559	existingProviderThread: threadInput.providerThread,
2560	}).pipe(
2561	Effect.mapError(
2562	(cause) =>
2563	new ProviderAdapter.ProviderAdapterResumeThreadError({
2564	driver: PI_PROVIDER,
2565	providerSessionId: input.providerSessionId,
2566	providerThreadId: threadInput.providerThread.id,
2567	cause,
2568	}),
2569	),
2570	),
2571	// The run executor routes a bare `/compact` here instead of startTurn.
2572	// Pi's start path already turns that text into the RPC compact call.
2573	compactThread: (turnInput) =>
2574	runtime.startTurn({
2575	...turnInput,
2576	message: { ...turnInput.message, text: "/compact" },
2577	}),
2578	startTurn: (turnInput) =>
2579	Effect.gen(function* () {
2580	if (rollbackBarrier !== null)
2581	return yield* protocolError("Cannot start a Pi turn during rollback");
2582	const state = threadState;
2583	if (state === null) {
2584	return yield* protocolError("Pi session has no registered thread");
2585	}
2586	if (state.activeTurn !== null) {
2587	return yield* protocolError(
2588	`Pi provider thread ${turnInput.providerThread.id} already has an active turn`,
2589	);
2590	}
2591	if (
2592	state.providerThread.nativeThreadRef?.nativeId !==
2593	turnInput.providerThread.nativeThreadRef?.nativeId
2594	) {
2595	return yield* protocolError("Pi turn requested for a different native session");
2596	}
2597	// The orchestrator adopts a fork under its already-allocated row.
2598	// Future session updates must retain that authoritative identity.
2599	const continuation =
2600	turnInput.message.createdBy === "agent" &&
2601	turnInput.message.creationSource === "provider";
2602	if (!continuation) yield* applySelection(turnInput.modelSelection);
2603	// Mirror the thread title into pi's session name so the session
2604	// stays identifiable in pi's own /resume listing. Best-effort:
2605	// naming must never block a turn.
2606	if (turnInput.appThread.title !== appliedSessionName) {
2607	yield* request({
2608	type: "set_session_name",
2609	name: turnInput.appThread.title,
2610	}).pipe(
2611	Effect.tap(() =>
2612	Effect.sync(() => (appliedSessionName = turnInput.appThread.title)),
2613	),
2614	Effect.ignore,
2615	);
2616	}
2617	// Resolved before the turn is installed: a failure here (an
2618	// unreadable attachment) must not leave `activeTurn` set, which
2619	// would reject every later turn as already active.
2620	// Orchestration instructions reach pi through the T3 MCP
2621	// extension's before_agent_start system-prompt hook, never by
2622	// wrapping the user text: a wrapped first message would no
2623	// longer start with "/" and slash commands would stop expanding.
2624	const compactCommand = continuation
2625	? null
2626	: parsePiCompactCommand(turnInput.message.text);
2627	const payload =
2628	!continuation && compactCommand === null
2629	? yield* resolvePromptPayload(turnInput.message.text, turnInput.message.attachments)
2630	: null;
2631	const startedAt = yield* DateTime.now;
2632	const syntheticNativeTurnId = `${state.providerThread.id}:attempt:${turnInput.attemptId}`;
2633	const providerTurn: OrchestrationV2ProviderTurn = {
2634	id: idAllocator.derive.providerTurn({
2635	driver: PI_PROVIDER,
2636	nativeTurnId: syntheticNativeTurnId,
2637	}),
2638	providerThreadId: turnInput.providerThread.id,
2639	nodeId: turnInput.rootNodeId,
2640	runAttemptId: turnInput.attemptId,
2641	nativeTurnRef: providerRef(syntheticNativeTurnId, "weak"),
2642	ordinal: turnInput.providerTurnOrdinal,
2643	status: "running",
2644	startedAt,
2645	completedAt: null,
2646	};
2647	const activeTurn: ActivePiTurn = {
2648	turnInput,
2649	providerTurn,
2650	startedAt,
2651	itemOrdinals: new Map(),
2652	nextItemOrdinal: turnInput.providerTurnOrdinal * 100 + 1,
2653	messageOrdinal: 0,
2654	streamItems: new Map(),
2655	toolArgs: new Map(),
2656	toolStartedAt: new Map(),
2657	interrupted: false,
2658	nativeAborted: false,
2659	sawAgentActivity: false,
2660	adoptedWake: false,
2661	promptMayBeCommandOnly:
2662	compactCommand !== null || (payload?.message.trimStart().startsWith("/") ?? false),
2663	latestCompactionAfterTokens: null,
2664	lastLiveUsedTokens: null,
2665	settleProbeGeneration: 0,
2666	settleWhenIdle: false,
2667	sawCompaction: false,
2668	manualCompactInFlight: compactCommand !== null,
2669	activeCompaction: null,
2670	activeProviderRetry: null,
2671	failure: null,
2672	rejectedPromptFailure: null,
2673	};
2674	// Only the install/send/start-event boundary excludes the event
2675	// pump. Earlier correlated requests must leave the pump free so
2676	// project trust, login, and session-switch dialogs can be shown
2677	// and answered instead of deadlocking the caller.
2678	yield* Effect.gen(function* () {
2679	if (
2680	threadState !== state ||
2681	state.activeTurn !== null ||
2682	closed ||
2683	stopRequested ||
2684	rollbackBarrier !== null ||
2685	state.providerThread.nativeThreadRef?.nativeId !==
2686	turnInput.providerThread.nativeThreadRef?.nativeId
2687	) {
2688	return yield* protocolError("Pi turn ownership changed before it could start");
2689	}
2690	const wake = pendingWake?.state === state ? pendingWake : null;
2691	if (wake !== null && wake.state.providerThread.appThreadId !== turnInput.threadId) {
2692	return yield* protocolError("Pi native wake belongs to a different app thread");
2693	}
2694	pendingWake = null;
2695	state.providerThread = turnInput.providerThread;
2696	state.activeTurn = activeTurn;
2697	activeTurn.adoptedWake = wake !== null && !continuation;
2698	if (compactCommand !== null) {
2699	yield* connection.send(compactRpcRecord(compactCommand));
2700	pendingCompactResponses.push({
2701	providerTurnId: providerTurn.id,
2702	kind: "turn_start",
2703	});
2704	} else if (payload !== null) {
2705	// Pi may already be streaming before its wake reaches T3.
2706	// Let Pi atomically queue the prompt or start an idle run.
2707	yield* connection.send({
2708	type: "prompt",
2709	message: payload.message,
2710	streamingBehavior: "steer",
2711	...(payload.images.length === 0 ? {} : { images: payload.images }),
2712	});
2713	pendingPromptResponses.push({
2714	providerTurnId: providerTurn.id,
2715	kind: "turn_start",
2716	});
2717	}
2718	yield* emit({
2719	type: "provider_turn.updated",
2720	driver: PI_PROVIDER,
2721	threadId: turnInput.threadId,
2722	providerTurn,
2723	});
2724	yield* updateProviderThread(state, {
2725	status: "active",
2726	firstRunOrdinal: state.providerThread.firstRunOrdinal ?? turnInput.runOrdinal,
2727	lastRunOrdinal: turnInput.runOrdinal,
2728	});
2729	yield* updateProviderSession("running", null);
2730	if (wake !== null) {
2731	for (const event of wake.events) {
2732	if (!continuation && event["type"] === "agent_settled") {
2733	activeTurn.nativeAborted = event["aborted"] === true;
2734	activeTurn.settleWhenIdle = true;
2735	continue;
2736	}
2737	yield* handleSessionEvent(event);
2738	}
2739	} else if (continuation) {
2740	// A user turn, teardown, or newer binding already took the wake.
2741	yield* finalizeTurn(state);
2742	}
2743	if (outOfTurnExtensionErrors.length > 0) {
2744	yield* Queue.offer(connection.events, { type: "t3.flush_extension_errors" });
2745	}
2746	}).pipe(
2747	sessionEventPermit.withPermits(1),
2748	Effect.tapError(() =>
2749	Effect.sync(() => {
2750	if (state.activeTurn === activeTurn) state.activeTurn = null;
2751	}),
2752	),
2753	);
2754	// Pi acks `prompt` only after slash-command expansion completes,
2755	// and extension commands may block on user dialogs indefinitely.
2756	// Rejections therefore return later as id-less response records
2757	// handled by the event pump.
2758	}).pipe(
2759	Effect.mapError(
2760	(cause) =>
2761	new ProviderAdapter.ProviderAdapterTurnStartError({
2762	driver: PI_PROVIDER,
2763	threadId: turnInput.threadId,
2764	providerThreadId: turnInput.providerThread.id,
2765	runId: turnInput.runId,
2766	cause,
2767	}),
2768	),
2769	),
2770	steerTurn: (steerInput: ProviderAdapter.ProviderAdapterV2SteerInput) =>
2771	Effect.gen(function* () {
2772	const turn = threadState?.activeTurn ?? null;
2773	if (turn === null || turn.providerTurn.id !== steerInput.providerTurnId) {
2774	return yield* protocolError(`Pi turn ${steerInput.providerTurnId} is not active`);
2775	}
2776	const compactCommand = parsePiCompactCommand(steerInput.message.text);
2777	const payload =
2778	compactCommand === null
2779	? yield* resolvePromptPayload(
2780	steerInput.message.text,
2781	steerInput.message.attachments,
2782	)
2783	: null;
2784	// Prompt with streamingBehavior steer is atomic on Pi's side: it
2785	// queues during an active run and starts a new run if settlement
2786	// won the race. A direct `steer` sent after Pi became idle would
2787	// remain queued forever. Send fire-and-forget under the session
2788	// permit so a slash-command dialog cannot block the turn, and so
2789	// settlement cannot overtake the active-turn check.
2790	// /compact is not a prompt: Pi's compact RPC aborts the agent first.
2791	yield* sessionEventPermit.withPermits(1)(
2792	Effect.gen(function* () {
2793	if (threadState?.activeTurn !== turn) {
2794	return yield* protocolError(`Pi turn ${steerInput.providerTurnId} is not active`);
2795	}
2796	if (compactCommand !== null) {
2797	turn.manualCompactInFlight = true;
2798	yield* connection.send(compactRpcRecord(compactCommand));
2799	pendingCompactResponses.push({
2800	providerTurnId: turn.providerTurn.id,
2801	kind: "steer",
2802	});
2803	} else if (payload !== null) {
2804	yield* connection.send({
2805	type: "prompt",
2806	message: payload.message,
2807	streamingBehavior: "steer",
2808	...(payload.images.length === 0 ? {} : { images: payload.images }),
2809	});
2810	pendingPromptResponses.push({
2811	providerTurnId: turn.providerTurn.id,
2812	kind: "steer",
2813	});
2814	}
2815	turn.settleProbeGeneration += 1;
2816	}),
2817	);
2818	}).pipe(
2819	Effect.mapError(
2820	(cause) =>
2821	new ProviderAdapter.ProviderAdapterSteerRunError({
2822	driver: PI_PROVIDER,
2823	providerThreadId: steerInput.providerThread.id,
2824	providerTurnId: steerInput.providerTurnId,
2825	cause,
2826	}),
2827	),
2828	),
2829	interruptTurn: (interruptInput) =>
2830	Effect.gen(function* () {
2831	const turn = threadState?.activeTurn ?? null;
2832	// Stop on a settled turn: Pi runs nothing between prompts, so
2833	// nothing of that turn is left to stop.
2834	if (turn === null && interruptInput.requestRuntimeRestart === true) {
2835	if (pendingWake !== null)
2836	yield* sessionEventPermit.withPermits(1)(stopPendingWake(pendingWake));
2837	return;
2838	}
2839	if (turn === null || turn.providerTurn.id !== interruptInput.providerTurnId) {
2840	return yield* protocolError(`Pi turn ${interruptInput.providerTurnId} is not active`);
2841	}
2842	turn.interrupted = true;
2843	if (
2844	interruptInput.requestRuntimeRestart === true ||
2845	turn.settleWhenIdle ||
2846	turn.activeCompaction !== null ||
2847	turn.manualCompactInFlight
2848	) {
2849	// Pi's generic abort does not cancel manual compaction. Terminate
2850	// so Stop covers user /compact as well as detached recovery compact.
2851	stopRequested = true;
2852	if (interruptInput.requestRuntimeRestart === true && !turn.settleWhenIdle) {
2853	yield* request({ type: "abort" }, 2_000).pipe(Effect.ignore);
2854	}
2855	// Terminating fails every later request, so read the stopped
2856	// turn's session-tree refs first: rolling back past this turn
2857	// forks at its user entry. Holding the event permit also lets a
2858	// finalize that is already reading them finish before the kill.
2859	yield* sessionEventPermit.withPermits(1)(
2860	Effect.gen(function* () {
2861	if (threadState?.activeTurn === turn && turn.stopTreeRefs === undefined) {
2862	turn.stopTreeRefs = yield* captureTurnTreeRefs(2_000);
2863	}
2864	yield* connection.terminate;
2865	}),
2866	);
2867	return;
2868	}
2869	yield* request({ type: "abort" }).pipe(
2870	Effect.tapError(() => Effect.sync(() => (turn.interrupted = false))),
2871	);
2872	}).pipe(
2873	Effect.mapError(
2874	(cause) =>
2875	new ProviderAdapter.ProviderAdapterInterruptError({
2876	driver: PI_PROVIDER,
2877	providerThreadId: interruptInput.providerThread.id,
2878	providerTurnId: interruptInput.providerTurnId,
2879	cause,
2880	}),
2881	),
2882	),
2883	respondToRuntimeRequest: (requestInput) =>
2884	Effect.gen(function* () {
2885	const pending = pendingPrompts.get(String(requestInput.requestId));
2886	if (pending === undefined) {
2887	return yield* protocolError(
2888	`No pending Pi extension request ${requestInput.requestId}`,
2889	);
2890	}
2891	const answeredAt = yield* DateTime.now;
2892	if (
2893	pending.expiresAt !== null &&
2894	DateTime.toEpochMillis(answeredAt) >= pending.expiresAt
2895	) {
2896	pendingPrompts.delete(String(requestInput.requestId));
2897	yield* cancelPrompt(pending, answeredAt);
2898	return yield* protocolError(
2899	`Pi extension request ${requestInput.requestId} has expired`,
2900	);
2901	}
2902	const response = piUiResponse(pending, requestInput.decision, requestInput.answers);
2903	yield* connection.send({
2904	type: "extension_ui_response",
2905	id: pending.nativeRequestId,
2906	...response,
2907	});
2908	// Dropped only once Pi has the answer, so a failed send leaves the
2909	// request retryable and still cancellable during teardown.
2910	pendingPrompts.delete(String(requestInput.requestId));
2911	if (pending.expirationFiber !== null) yield* Fiber.interrupt(pending.expirationFiber);
2912	if (pending.method === "confirm" && requestInput.decision === "acceptForSession") {
2913	sessionApprovals.add(pending.approvalKey);
2914	}
2915	const resolvedAt = yield* DateTime.now;
2916	pending.runtimeRequest = {
2917	...pending.runtimeRequest,
2918	status: "resolved",
2919	resolvedAt,
2920	};
2921	yield* emit({
2922	type: "runtime_request.updated",
2923	driver: PI_PROVIDER,
2924	threadId: pending.node.threadId,
2925	runtimeRequest: pending.runtimeRequest,
2926	});
2927	yield* emit({
2928	type: "node.updated",
2929	driver: PI_PROVIDER,
2930	node: { ...pending.node, status: "completed", completedAt: resolvedAt },
2931	});
2932	yield* emit({
2933	type: "turn_item.updated",
2934	driver: PI_PROVIDER,
2935	turnItem: {
2936	...pending.turnItem,
2937	status: "completed",
2938	completedAt: resolvedAt,
2939	updatedAt: resolvedAt,
2940	},
2941	});
2942	}).pipe(
2943	sessionEventPermit.withPermits(1),
2944	Effect.mapError(
2945	(cause) =>
2946	new ProviderAdapter.ProviderAdapterRuntimeRequestResponseError({
2947	driver: PI_PROVIDER,
2948	requestId: requestInput.requestId,
2949	cause,
2950	}),
2951	),
2952	),
2953	readThreadSnapshot: (snapshotInput) =>
2954	Effect.gen(function* () {
2955	const state = threadState;
2956	const boundNativeId = state?.providerThread.nativeThreadRef?.nativeId;
2957	const wantedNativeId = snapshotInput.providerThread.nativeThreadRef?.nativeId;
2958	if (state === null || wantedNativeId == null || boundNativeId !== wantedNativeId) {
2959	return yield* protocolError(
2960	"Pi snapshot requested for a thread this session does not host",
2961	);
2962	}
2963	// get_messages is Pi's active-branch view. get_entries returns the
2964	// whole session tree, including abandoned branches after /tree or
2965	// fork, which would leak discarded conversation into handoffs.
2966	const messagesData = yield* request({ type: "get_messages" });
2967	const activeMessages = recordField(messagesData, "messages");
2968	const threadId = state.providerThread.appThreadId ?? input.threadId;
2969	const messages = (Array.isArray(activeMessages) ? activeMessages : []).flatMap(
2970	(message, index) => {
2971	const role = recordString(message, "role");
2972	if (role !== "user" && role !== "assistant") return [];
2973	const text = contentText(recordField(message, "content"));
2974	if (text.length === 0) return [];
2975	const timestamp = recordNumber(message, "timestamp");
2976	const at = Option.getOrElse(
2977	DateTime.make(timestamp ?? Number.NaN),
2978	() => state.providerThread.createdAt,
2979	);
2980	return [
2981	{
2982	id: idAllocator.derive.messageFromProviderItem({
2983	driver: PI_PROVIDER,
2984	// RPC messages do not expose session-tree entry ids. The
2985	// active-branch index is stable for the lifetime of this
2986	// snapshot and keeps abandoned branch ids out of it.
2987	nativeItemId: `${wantedNativeId}:snapshot-message:${index}`,
2988	}),
2989	threadId,
2990	runId: null,
2991	nodeId: null,
2992	role: role as "user" | "assistant",
2993	text,
2994	attachments: [],
2995	streaming: false,
2996	createdBy: role === "user" ? ("user" as const) : ("agent" as const),
2997	creationSource: "provider" as const,
2998	createdAt: at,
2999	updatedAt: at,
3000	},
3001	];
3002	},
3003	);
3004	return {
3005	providerThread: state.providerThread,
3006	providerTurns: [],
3007	messages,
3008	runtimeRequests: [],
3009	};
3010	}).pipe(
3011	Effect.mapError(
3012	(cause) =>
3013	new ProviderAdapter.ProviderAdapterReadThreadSnapshotError({
3014	driver: PI_PROVIDER,
3015	providerThreadId: snapshotInput.providerThread.id,
3016	cause,
3017	}),
3018	),
3019	),
3020	rollbackThread: (rollbackInput) =>
3021	Effect.gen(function* () {
3022	const state = threadState;
3023	if (state === null) {
3024	return yield* protocolError("Pi session has no registered thread");
3025	}
3026	if (state.providerThread.id !== rollbackInput.providerThread.id) {
3027	return yield* protocolError(
3028	"Pi rollback requested for a thread this session does not host",
3029	);
3030	}
3031	if (state.activeTurn !== null) {
3032	return yield* protocolError("Cannot roll back while a Pi turn is active");
3033	}
3034	if (pendingWake !== null) {
3035	return yield* protocolError("Cannot roll back while native Pi work awaits a turn");
3036	}
3037	// `fork(entryId)` re-roots the active branch before that user
3038	// message, so the rollback boundary is the first user entry of
3039	// the earliest turn being discarded.
3040	const forkEntryId = piRollbackForkEntry(rollbackInput);
3041	if (forkEntryId === null) {
3042	// Nothing after the target: the conversation is already there.
3043	return piThreadSnapshot(state.providerThread);
3044	}
3045	if (forkEntryId === undefined) {
3046	return yield* protocolError("Pi rollback target has no captured session-tree entry");
3047	}
3048	return yield* Effect.acquireUseRelease(
3049	sessionEventPermit.withPermits(1)(
3050	Effect.gen(function* () {
3051	if (
3052	rollbackBarrier !== null ||
3053	threadState !== state ||
3054	state.activeTurn !== null ||
3055	pendingWake !== null ||
3056	closed ||
3057	stopRequested
3058	)
3059	return yield* protocolError(
3060	"Pi rollback ownership changed before its fork could start",
3061	);
3062	const barrier = yield* Deferred.make<
3063	void,
3064	ProviderAdapter.ProviderAdapterProtocolError
3065	>();
3066	rollbackBarrier = barrier;
3067	return barrier;
3068	}),
3069	),
3070	(barrier) =>
3071	Effect.gen(function* () {
3072	const forkData = yield* lifecycleRequest({ type: "fork", entryId: forkEntryId });
3073	if (recordField(forkData, "cancelled") === true)
3074	return yield* protocolError("A Pi extension cancelled the session fork");
3075	const forkState = yield* request({ type: "get_state" }).pipe(
3076	Effect.onError(() =>
3077	Effect.sync(() => {
3078	threadState = null;
3079	}),
3080	),
3081	);
3082	const forkSessionFile = recordString(forkState, "sessionFile");
3083	if (forkSessionFile === undefined) {
3084	threadState = null;
3085	return yield* protocolError("Pi fork did not return a persisted session file");
3086	}
3087	const entriesData = yield* request({ type: "get_entries" }).pipe(
3088	Effect.orElseSucceed(() => undefined),
3089	);
3090	const leafId = recordString(entriesData, "leafId") ?? null;
3091	// RPC responses bypass the pump. Drain preceding native events
3092	// before deciding whether this fork can replace their binding.
3093	const queued = yield* Queue.offer(connection.events, {
3094	type: "t3.rollback_barrier",
3095	barrier,
3096	});
3097	if (!queued) return yield* protocolError("Pi transport closed during rollback");
3098	yield* Deferred.await(barrier);
3099	return yield* sessionEventPermit.withPermits(1)(
3100	Effect.gen(function* () {
3101	if (pendingWake !== null) {
3102	yield* stopPendingWake(pendingWake);
3103	return yield* protocolError(
3104	"Native Pi work started while rollback was changing the session",
3105	);
3106	}
3107	if (
3108	threadState !== state ||
3109	state.activeTurn !== null ||
3110	closed ||
3111	stopRequested
3112	)
3113	return yield* protocolError(
3114	"Pi rollback ownership changed during its fork",
3115	);
3116	lastNativeThreadId = forkSessionFile;
3117	appliedModel = null;
3118	appliedThinking = null;
3119	appliedSessionName = null;
3120	lastKnownLeaf = leafId;
3121	leafCursorStale = entriesData === undefined;
3122	yield* updateProviderThread(state, {
3123	nativeThreadRef: providerRef(forkSessionFile),
3124	nativeConversationHeadRef: leafId === null ? null : providerRef(leafId),
3125	});
3126	rollbackBarrier = null;
3127	return piThreadSnapshot(state.providerThread);
3128	}),
3129	);
3130	}).pipe(Effect.onInterrupt(() => connection.terminate)),
3131	(barrier) =>
3132	Effect.sync(() => {
3133	if (rollbackBarrier === barrier) rollbackBarrier = null;
3134	}),
3135	);
3136	}).pipe(
3137	Effect.mapError(
3138	(cause) =>
3139	new ProviderAdapter.ProviderAdapterRollbackThreadError({
3140	driver: PI_PROVIDER,
3141	providerThreadId: rollbackInput.providerThread.id,
3142	checkpointId: rollbackInput.target.checkpointId,
3143	cause,
3144	}),
3145	),
3146	),
3147	forkThread: (forkInput) =>
3148	Effect.gen(function* () {
3149	if (threadState?.activeTurn != null) {
3150	return yield* protocolError("Cannot fork while a Pi turn is active");
3151	}
3152	const source = forkInput.sourceProviderThread;
3153	const sourceFile = source.nativeThreadRef?.nativeId;
3154	if (sourceFile == null)
3155	return yield* protocolError("Pi fork source has no session file");
3156	const sourceTurns = (forkInput.sourceProviderTurns ?? []).filter(
3157	(turn) => turn.providerThreadId === source.id,
3158	);
3159	const target = sourceTurns.find((turn) => turn.id === forkInput.providerTurnId);
3160	if (forkInput.providerTurnId !== undefined && target === undefined) {
3161	return yield* protocolError("Pi fork target turn is missing");
3162	}
3163	const beforeEntry =
3164	target === undefined
3165	? null
3166	: piRollbackForkEntry({
3167	target: { type: "provider_turn", providerTurn: target },
3168	providerThreadTurns: sourceTurns,
3169	});
3170	if (beforeEntry === undefined) {
3171	return yield* protocolError("Pi fork boundary has no captured session-tree entry");
3172	}
3173	// CLI --fork uses Pi's own session format and sets the destination
3174	// cwd. RPC switch_session/clone alone would retain the source cwd.
3175	// This short-lived process never prompts or runs user extensions.
3176	const forkLaunch = buildPiRpcLaunch({
3177	launchArgs: resolvedLaunchArgs.args,
3178	environment: options.environment,
3179	mcpSession: undefined,
3180	extensionPath: undefined,
3181	disableExtensions: true,
3182	disableTools: true,
3183	});
3184	const nativeId = yield* Effect.scoped(
3185	Effect.gen(function* () {
3186	const forkConnection = yield* makePiRpcConnection({
3187	command: options.settings.binaryPath || "pi",
3188	args: [...forkLaunch.args, "--fork", sourceFile],
3189	cwd,
3190	env: forkLaunch.env,
3191	});
3192	yield* Stream.fromQueue(forkConnection.events).pipe(
3193	Stream.runDrain,
3194	Effect.ignore,
3195	Effect.forkScoped,
3196	);
3197	if (beforeEntry !== null) {
3198	const result = yield* forkConnection.request({
3199	type: "fork",
3200	entryId: beforeEntry,
3201	});
3202	if (recordField(result, "cancelled") === true) {
3203	return yield* protocolError("A Pi extension cancelled the session fork");
3204	}
3205	}
3206	const state = yield* forkConnection.request({ type: "get_state" });
3207	const file = recordString(state, "sessionFile");
3208	if (file === undefined || file === sourceFile) {
3209	return yield* protocolError("Pi fork did not create a distinct session file");
3210	}
3211	return file;
3212	}),
3213	).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner));
3214	const now = yield* DateTime.now;
3215	return yield* registerThread(
3216	{
3217	threadId: forkInput.targetThreadId,
3218	modelSelection: forkInput.modelSelection ?? input.modelSelection,
3219	runtimePolicy: forkInput.runtimePolicy ?? input.runtimePolicy,
3220	existingProviderThread: {
3221	...source,
3222	id: idAllocator.derive.providerThread({
3223	driver: PI_PROVIDER,
3224	nativeThreadId: nativeId,
3225	}),
3226	appThreadId: forkInput.targetThreadId,
3227	providerSessionId: input.providerSessionId,
3228	providerInstanceId: options.instanceId,
3229	ownerNodeId: forkInput.ownerNodeId ?? null,
3230	nativeThreadRef: providerRef(nativeId),
3231	nativeConversationHeadRef: null,
3232	firstRunOrdinal: null,
3233	lastRunOrdinal: null,
3234	handoffIds: [],
3235	pendingBackgroundTasks: [],
3236	forkedFrom: {
3237	providerThreadId: source.id,
3238	...(forkInput.providerTurnId === undefined
3239	? {}
3240	: { providerTurnId: forkInput.providerTurnId }),
3241	},
3242	createdAt: now,
3243	updatedAt: now,
3244	},
3245	},
3246	false,
3247	);
3248	}).pipe(
3249	Effect.mapError(
3250	(cause) =>
3251	new ProviderAdapter.ProviderAdapterForkThreadError({
3252	driver: PI_PROVIDER,
3253	providerThreadId: forkInput.sourceProviderThread.id,
3254	cause,
3255	}),
3256	),
3257	),
3258	};
3259	return runtime;
3260	}),
3261	});
3262	});
3263
3264	interface PiSessionEntry {
3265	readonly parentId: string | null;
3266	readonly isUserMessage: boolean;
3267	}
3268
3269	/** Index a `get_entries` listing, rejecting ambiguous ancestry. */
3270	function piSessionEntries(entries: unknown): ReadonlyMap<string, PiSessionEntry> | null {
3271	const byId = new Map<string, PiSessionEntry>();
3272	if (!Array.isArray(entries)) return null;
3273	for (const entry of entries) {
3274	const id = recordString(entry, "id");
3275	const parentId = recordField(entry, "parentId");
3276	if (id === undefined || byId.has(id) || (parentId !== null && typeof parentId !== "string"))
3277	return null;
3278	byId.set(id, {
3279	parentId,
3280	isUserMessage:
3281	recordField(entry, "type") === "message" &&
3282	recordString(recordField(entry, "message"), "role") === "user",
3283	});
3284	}
3285	return byId;
3286	}
3287
3288	/**
3289	* Walk pi's active branch from `leafId` towards the root through `entries`,
3290	* stopping at `stopAt`. Returns how many entries it passed, the user entries
3291	* among them (newest first), and whether the walk reached `stopAt`. Leaving a
3292	* partial listing anywhere else means the branch does not descend from `stopAt`.
3293	* Returns null when the `parentId` chain loops, which only corrupted session
3294	* data can produce; callers treat that like a failed listing.
3295	*/
3296	function walkPiBranch(
3297	entries: ReadonlyMap<string, PiSessionEntry>,
3298	leafId: string | null,
3299	stopAt: string | null,
3300	): {
3301	readonly entryCount: number;
3302	readonly userEntryIds: ReadonlyArray<string>;
3303	readonly reachedStop: boolean;
3304	} | null {
3305	const visited = new Set<string>();
3306	const userEntryIds: Array<string> = [];
3307	let current = leafId;
3308	while (current !== null && current !== stopAt) {
3309	const entry = entries.get(current);
3310	if (entry === undefined) break;
3311	if (visited.has(current)) return null;
3312	visited.add(current);
3313	if (entry.isUserMessage) userEntryIds.push(current);
3314	current = entry.parentId;
3315	}
3316	return { entryCount: visited.size, userEntryIds, reachedStop: current === stopAt };
3317	}
3318
3319	/**
3320	* Resolve the pi session-tree entry `fork` should re-root at for a rollback.
3321	* Turns without a ref left nothing on the active branch, so the boundary is
3322	* the first discarded turn that has one. Returns `null` when no such turn
3323	* follows the target (nothing to discard) and `undefined` when the boundary
3324	* turn has no captured entry ref (only turn-boundary refs recorded by
3325	* `captureTurnTreeRefs` are strong).
3326	*/
3327	function piRollbackForkEntry(input: {
3328	readonly target:
3329	| { readonly type: "thread_start" }
3330	| { readonly type: "provider_turn"; readonly providerTurn: OrchestrationV2ProviderTurn };
3331	readonly providerThreadTurns: ReadonlyArray<OrchestrationV2ProviderTurn>;
3332	}): string | null | undefined {
3333	const boundaryOrdinal =
3334	input.target.type === "thread_start" ? 0 : input.target.providerTurn.ordinal;
3335	const ref = input.providerThreadTurns
3336	.filter((turn) => turn.ordinal > boundaryOrdinal && turn.nativeTurnRef !== null)
3337	.sort((a, b) => a.ordinal - b.ordinal)[0]?.nativeTurnRef;
3338	if (ref == null) return null;
3339	if (ref.strength !== "strong" || ref.nativeId === null) return undefined;
3340	return ref.nativeId;
3341	}
3342
3343	/**
3344	* Human-readable output for one subagent-extension task result: the last
3345	* assistant text from its transcript, or the error/stderr when it failed.
3346	*/
3347	function piSubagentOutput(result: unknown): string {
3348	const stopReason = recordString(result, "stopReason");
3349	const failed =
3350	(recordNumber(result, "exitCode") ?? 0) !== 0 ||
3351	stopReason === "error" ||
3352	stopReason === "aborted";
3353	if (failed) {
3354	// Falsy fallback, not `??`: an empty `errorMessage` must not suppress a
3355	// non-empty `stderr`, which is often the only description of the failure.
3356	const failure = recordString(result, "errorMessage") || recordString(result, "stderr");
3357	if (failure !== undefined && failure.length > 0) return failure;
3358	}
3359	const messages = recordField(result, "messages");
3360	if (!Array.isArray(messages)) return "";
3361	for (let index = messages.length - 1; index >= 0; index -= 1) {
3362	const message = messages[index];
3363	if (recordString(message, "role") !== "assistant") continue;
3364	const text = contentText(recordField(message, "content"));
3365	if (text.length > 0) return text;
3366	}
3367	return "";
3368	}
3369
3370	function piExtensionDisplayName(extensionPath: string | undefined): string {
3371	if (extensionPath === undefined) return "Pi extension";
3372	const normalized = extensionPath.replace(/\\/g, "/").replace(/\/+$/, "");
3373	const name = normalized.slice(normalized.lastIndexOf("/") + 1).replace(/\.[^.]+$/, "");
3374	return name.length === 0 ? "Pi extension" : name;
3375	}
3376
3377	function piThreadSnapshot(
3378	providerThread: OrchestrationV2ProviderThread,
3379	): ProviderAdapter.ProviderAdapterV2ThreadSnapshot {
3380	return { providerThread, providerTurns: [], messages: [], runtimeRequests: [] };
3381	}
3382
3383	function piQuestion(
3384	questionId: string,
3385	method: "select" | "input" | "editor",
3386	title: string,
3387	event: PiRpcRecord,
3388	): OrchestrationV2UserInputQuestion {
3389	const options =
3390	method === "select" && Array.isArray(event["options"])
3391	? event["options"]
3392	.filter((option): option is string => typeof option === "string")
3393	.map((option) => ({ label: option || "Empty value", description: option, value: option }))
3394	: method === "editor"
3395	? []
3396	: [
3397	{
3398	label: "Submit empty value",
3399	description: "Send an empty string to the extension.",
3400	value: "",
3401	},
3402	];
3403	const question = recordString(event, "message") ?? recordString(event, "placeholder") ?? title;
3404	return {
3405	id: questionId,
3406	header: title,
3407	question,
3408	...(method === "editor" ? { initialAnswer: recordString(event, "prefill") ?? "" } : {}),
3409	options,
3410	};
3411	}
3412
3413	function piUiResponse(
3414	pending: PendingPiPrompt,
3415	decision: ProviderApprovalDecision | undefined,
3416	answers: Record<string, unknown> | undefined,
3417	): PiRpcRecord {
3418	if (pending.method === "confirm") {
3419	if (decision === "accept" || decision === "acceptForSession") return { confirmed: true };
3420	if (decision === "decline") return { confirmed: false };
3421	return { cancelled: true };
3422	}
3423	const answer = answers?.[pending.questionId];
3424	// An empty string is a valid dialog value per the RPC spec (the extension
3425	// receives ""), distinct from cancelling (the extension receives undefined).
3426	if (typeof answer === "string") return { value: answer };
3427	return { cancelled: true };
3428	}
3429
3430	// ── driver ────────────────────────────────────────────────────
3431
3432	export type PiAdapterV2DriverEnv =
3433	| ChildProcessSpawner.ChildProcessSpawner
3434	| FileSystem.FileSystem
3435	| IdAllocator.IdAllocatorV2
3436	| ScriptProgress.ScriptProgress
3437	| McpProviderSessions.McpProviderSessions
3438	| ProviderHost.ProviderHost;
3439
3440	export const PiAdapterV2Driver: ProviderAdapterDriver<PiSettings, PiAdapterV2DriverEnv> = {
3441	driverKind: PI_DRIVER_KIND,
3442	configSchema: PiSettings,
3443	defaultConfig: (): PiSettings => DEFAULT_PI_SETTINGS,
3444	create: Effect.fn("PiAdapterV2Driver.create")(
3445	function* (input: ProviderAdapterDriverCreateInput<PiSettings>) {
3446	const hostEnvironment = yield* HostProcess.Environment;
3447	const continuationRequests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
3448	return yield* makePiAdapterV2({
3449	instanceId: input.instanceId,
3450	settings: { ...input.config, enabled: input.enabled },
3451	environment: yield* mergeProviderInstanceEnvironment(input.environment, hostEnvironment),
3452	continuationRequests,
3453	});
3454	},
3455	(effect, input) =>
3456	effect.pipe(
3457	Effect.mapError(
3458	(cause) =>
3459	new ProviderAdapterDriverCreateError({
3460	driver: PI_DRIVER_KIND,
3461	instanceId: input.instanceId,
3462	detail: "Failed to create Pi adapter.",
3463	cause,
3464	}),
3465	),
3466	),
3467	),
3468	};
3469
3470	const layer: Layer.Layer<ProviderAdapter.ProviderAdapterV2, never, PiAdapterV2DriverEnv> =
3471	Layer.effect(
3472	ProviderAdapter.ProviderAdapterV2,
3473	Effect.gen(function* () {
3474	const hostEnvironment = yield* HostProcess.Environment;
3475	const continuationRequests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
3476	return yield* makePiAdapterV2({
3477	instanceId: PI_DEFAULT_INSTANCE_ID,
3478	settings: DEFAULT_PI_SETTINGS,
3479	environment: hostEnvironment,
3480	continuationRequests,
3481	});
3482	}),
3483	);