1// An in-process environment for app render scenarios: a fetch stub for its HTTP API and a
2// session that answers the shell, thread and config subscriptions from fixtures. Other RPCs
3// stay pending so nothing retries on timers. Everything above the wire is the app's own.
4import {
5 AuthStandardClientScopes,
6 EnvironmentId,
7 EventId,
8 ExecutionEnvironmentDescriptor,
9 MessageId,
10 OrchestrationV2DispatchCommandError,
11 ORCHESTRATION_PROTOCOL_VERSION,
12 ORCHESTRATION_V2_WS_METHODS,
13 ProjectId,
14 ProviderInstanceId,
15 RunId,
16 ServerConfig,
17 ThreadId,
18 TurnItemId,
19 WS_METHODS,
20 WsRpcGroup,
21 type VcsStatusStreamEvent,
22 type ModelSelection,
23 type OrchestrationProjectShell,
24 type OrchestrationV2Command,
25 type OrchestrationV2ProjectedTurnItem,
26 type OrchestrationV2ProviderThread,
27 type OrchestrationV2Run,
28 type OrchestrationV2ShellStreamItem,
29 type OrchestrationV2ThreadProjection,
30 type OrchestrationV2ThreadShell,
31 type OrchestrationV2ThreadStreamItem,
32 type OrchestrationV2TurnItem,
33 type TerminalMetadataStreamEvent,
34} from "@t3tools/contracts";
35import type {
36 ConnectionAttemptError,
37 ConnectionCatalogEntry,
38 EnvironmentConnectionLease,
39 RouteCheck,
40} from "@t3tools/client-runtime/connection";
41import type { WsRpcProtocolClient } from "@t3tools/client-runtime/rpc";
42import { DEFAULT_RESOLVED_KEYBINDINGS } from "@t3tools/shared/keybindings";
43import { APP_VERSION } from "~/branding";
44import * as DateTime from "effect/DateTime";
45import * as Deferred from "effect/Deferred";
46import * as Effect from "effect/Effect";
47import * as Queue from "effect/Queue";
48import type * as Scope from "effect/Scope";
49import * as Schema from "effect/Schema";
50import * as Stream from "effect/Stream";
51import { RpcSchema } from "effect/rpc";
52
53export const ENVIRONMENT_ID = EnvironmentId.make("environment-render");
54export const PROJECT_ID = ProjectId.make("project-render");
55const CODEX = ProviderInstanceId.make("codex");
56const MODEL = { instanceId: CODEX, model: "gpt-5.4" };
57const AT = DateTime.makeUnsafe("2026-03-17T19:12:28.000Z");
58
59const descriptor = {
60 environmentId: ENVIRONMENT_ID,
61 label: "Render scenarios",
62 platform: { os: "linux", arch: "x64" },
63 // A matching version, since a mismatch adds an update banner a typical user never sees.
64 serverVersion: APP_VERSION,
65 orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
66 capabilities: { repositoryIdentity: true },
67};
68const auth = {
69 policy: "loopback-browser",
70 bootstrapMethods: ["one-time-token"],
71 sessionMethods: ["browser-session-cookie"],
72 sessionCookieName: "t3_session",
73};
74
75const readyProvider = (
76 instanceId: string,
77 driver: string,
78 models: ReadonlyArray<{ readonly slug: string; readonly name: string }>,
79) => ({
80 instanceId,
81 driver,
82 enabled: true,
83 installed: true,
84 version: "1.0.0",
85 status: "ready",
86 auth: { status: "authenticated" },
87 checkedAt: "2026-03-17T19:12:28.000Z",
88 models: models.map((model) => ({ ...model, isCustom: false, capabilities: null })),
89});
90export const CODEX_PROVIDER = readyProvider(CODEX, "codex", [
91 { slug: "gpt-5.4", name: "GPT-5.4" },
92 { slug: "gpt-5.5", name: "GPT-5.5" },
93]);
94export const CLAUDE_PROVIDER = {
95 ...readyProvider("claudeAgent", "claudeAgent", [
96 { slug: "claude-opus-4-6", name: "Claude Opus 4.6" },
97 ]),
98 slashCommands: [{ name: "compact" }],
99};
100
101const decodeServerConfig = Schema.decodeUnknownSync(ServerConfig);
102const makeServerConfig = (providers: ReadonlyArray<unknown>): ServerConfig => ({
103 ...decodeServerConfig({
104 environment: descriptor,
105 auth,
106 cwd: "/workspace/render",
107 keybindingsConfigPath: "/workspace/.t3/keybindings.json",
108 keybindings: [],
109 issues: [],
110 providers,
111 availableEditors: [],
112 observability: {
113 logsDirectoryPath: "/workspace/.t3/logs",
114 localTracingEnabled: false,
115 otlpTracesEnabled: false,
116 otlpMetricsEnabled: false,
117 },
118 settings: {},
119 threadSnapshotPagination: true,
120 threadResumeCompletionMarker: true,
121 shellResumeCompletionMarker: true,
122 }),
123 keybindings: DEFAULT_RESOLVED_KEYBINDINGS,
124});
125let baseServerConfig = makeServerConfig([CODEX_PROVIDER]);
126// A server only leaves threads out of the shell when it serves their families.
127const serverConfig = (): ServerConfig =>
128 threads.some((thread) => thread.olderSettled)
129 ? { ...baseServerConfig, shellThreadFamilies: true }
130 : baseServerConfig;
131
132const project: OrchestrationProjectShell = {
133 id: PROJECT_ID,
134 title: "render",
135 workspaceRoot: "/workspace/render",
136 repositoryIdentity: null,
137 defaultModelSelection: null,
138 scripts: [],
139 createdAt: "2026-03-17T19:12:28.000Z",
140 updatedAt: "2026-03-17T19:12:28.000Z",
141};
142
143const base = (threadId: ThreadId, runId: RunId, id: string, ordinal: number) => ({
144 id: TurnItemId.make(id),
145 threadId,
146 runId,
147 nodeId: null,
148 providerThreadId: null,
149 providerTurnId: null,
150 nativeItemRef: null,
151 parentItemId: null,
152 ordinal,
153 status: "completed" as const,
154 title: null,
155 startedAt: AT,
156 completedAt: AT,
157 updatedAt: AT,
158});
159
160function userItem(threadId: ThreadId, runId: RunId, id: string, ordinal: number, text: string) {
161 return {
162 ...base(threadId, runId, id, ordinal),
163 type: "user_message",
164 messageId: MessageId.make(id),
165 text,
166 attachments: [],
167 inputIntent: "turn_start",
168 createdBy: "user",
169 creationSource: "web",
170 } satisfies OrchestrationV2TurnItem;
171}
172
173export function assistantItem(
174 threadId: ThreadId,
175 runId: RunId,
176 id: string,
177 ordinal: number,
178 text: string,
179 streaming = false,
180) {
181 return {
182 ...base(threadId, runId, id, ordinal),
183 type: "assistant_message",
184 messageId: MessageId.make(id),
185 text,
186 streaming,
187 ...(streaming ? { status: "running" as const, completedAt: null } : {}),
188 } satisfies OrchestrationV2TurnItem;
189}
190
191function run(
192 threadId: ThreadId,
193 id: RunId,
194 ordinal: number,
195 userMessageId: string,
196 modelSelection: ModelSelection = MODEL,
197) {
198 return {
199 id,
200 threadId,
201 ordinal,
202 providerInstanceId: modelSelection.instanceId,
203 modelSelection,
204 providerThreadId: null,
205 userMessageId: MessageId.make(userMessageId),
206 rootNodeId: null,
207 activeAttemptId: null,
208 status: "completed",
209 requestedAt: AT,
210 startedAt: AT,
211 completedAt: AT,
212 checkpointId: null,
213 contextHandoffId: null,
214 } satisfies OrchestrationV2Run;
215}
216
217export interface FakeThread {
218 /** Makes this a side chat of that thread. */
219 readonly sideChatOf?: ThreadId;
220 /** Makes this an older settled thread the shell leaves out; only its family read returns it. */
221 readonly olderSettled?: true;
222 readonly id: ThreadId;
223 readonly title: string;
224 /** Defaults to GPT-5.4 on Codex. */
225 readonly modelSelection?: ModelSelection;
226 /** The first is the thread's active provider thread. */
227 readonly providerThreads?: ReadonlyArray<OrchestrationV2ProviderThread>;
228 readonly runs: ReadonlyArray<OrchestrationV2Run>;
229 readonly items: ReadonlyArray<OrchestrationV2TurnItem>;
230}
231
232/** A thread of `turns` user prompts, each answered in markdown with a list and inline code. */
233export function threadWithHistory(
234 name: string,
235 turns: number,
236 modelSelection: ModelSelection = MODEL,
237): FakeThread {
238 const id = ThreadId.make(name);
239 const runs: OrchestrationV2Run[] = [];
240 const items: OrchestrationV2TurnItem[] = [];
241 for (let turn = 0; turn < turns; turn++) {
242 const runId = RunId.make(`${name}-run-${turn}`);
243 runs.push(run(id, runId, turn + 1, `${name}-user-${turn}`, modelSelection));
244 items.push(
245 userItem(id, runId, `${name}-user-${turn}`, turn * 2, `${name}: question ${turn}`),
246 assistantItem(
247 id,
248 runId,
249 `${name}-assistant-${turn}`,
250 turn * 2 + 1,
251 `Answer ${turn} for ${name}.\n\n- checked \`Sidebar.tsx\`\n- ran the tests\n\n` +
252 "The width is now `clamp(next, MIN_WIDTH, MAX_WIDTH)`, so the layout no longer jumps.",
253 ),
254 );
255 }
256 return { id, title: `Thread ${name}`, modelSelection, runs, items };
257}
258
259function shellOf(thread: FakeThread): OrchestrationV2ThreadShell {
260 const latest = thread.runs.at(-1);
261 const modelSelection = thread.modelSelection ?? MODEL;
262 return {
263 id: thread.id,
264 projectId: PROJECT_ID,
265 title: thread.title,
266 providerInstanceId: modelSelection.instanceId,
267 modelSelection,
268 runtimeMode: "full-access",
269 interactionMode: "default",
270 branch: "main",
271 worktreePath: null,
272 activeProviderThreadId: thread.providerThreads?.[0]?.id ?? null,
273 lineage: {
274 rootThreadId: thread.sideChatOf ?? thread.id,
275 parentThreadId: thread.sideChatOf ?? null,
276 relationshipToParent: null,
277 ...(thread.sideChatOf ? { sideChat: true } : {}),
278 },
279 forkedFrom: null,
280 createdBy: "user",
281 creationSource: "web",
282 latestRunId: latest?.id ?? null,
283 activeRunId: null,
284 // Idle reads as waiting on the user, which shows a per-second duration in the sidebar.
285 status: latest ? "completed" : "idle",
286 pendingRuntimeRequest: null,
287 latestVisibleMessage: null,
288 latestUserMessageAt: AT,
289 hasActionableProposedPlan: false,
290 itemCount: thread.items.length,
291 visibleItemCount: thread.items.length,
292 createdAt: AT,
293 updatedAt: AT,
294 archivedAt: null,
295 settledOverride: thread.olderSettled ? "settled" : null,
296 settledAt: thread.olderSettled ? AT : null,
297 lastVisitedAt: AT,
298 deletedAt: null,
299 };
300}
301
302function projectionOf(thread: FakeThread): OrchestrationV2ThreadProjection {
303 const shell = shellOf(thread);
304 const visibleTurnItems: OrchestrationV2ProjectedTurnItem[] = thread.items.map(
305 (item, position) => ({
306 position,
307 visibility: "local",
308 sourceThreadId: thread.id,
309 sourceItemId: item.id,
310 item,
311 }),
312 );
313 return {
314 thread: {
315 id: shell.id,
316 projectId: shell.projectId,
317 title: shell.title,
318 providerInstanceId: shell.providerInstanceId,
319 modelSelection: shell.modelSelection,
320 runtimeMode: shell.runtimeMode,
321 interactionMode: shell.interactionMode,
322 branch: shell.branch,
323 worktreePath: shell.worktreePath,
324 activeProviderThreadId: shell.activeProviderThreadId,
325 lineage: shell.lineage,
326 forkedFrom: null,
327 createdBy: "user",
328 creationSource: "web",
329 createdAt: AT,
330 updatedAt: AT,
331 archivedAt: null,
332 settledOverride: shell.settledOverride,
333 settledAt: shell.settledAt,
334 lastVisitedAt: AT,
335 deletedAt: null,
336 },
337 runs: [...thread.runs],
338 attempts: [],
339 nodes: [],
340 subagents: [],
341 providerSessions: [],
342 providerThreads: [...(thread.providerThreads ?? [])],
343 providerTurns: [],
344 runtimeRequests: [],
345 messages: [],
346 plans: [],
347 turnItems: [...thread.items],
348 checkpointScopes: [],
349 checkpoints: [],
350 contextHandoffs: [],
351 contextTransfers: [],
352 visibleTurnItems,
353 updatedAt: AT,
354 };
355}
356
357const json = (body: unknown, status = 200) =>
358 new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } });
359
360// Keyed by thread id: a live thread subscription's queue, for pushing events to it.
361const threadSubscribers = new Map<string, Queue.Queue<OrchestrationV2ThreadStreamItem>>();
362let terminalMetadataSubscriber: Queue.Queue<TerminalMetadataStreamEvent> | null = null;
363let threads: ReadonlyArray<FakeThread> = [];
364let dispatched: OrchestrationV2Command[] = [];
365let acceptCommands = false;
366let holdCommand: (command: OrchestrationV2Command) => boolean = () => false;
367let pendingDispatches: Array<{
368 readonly command: OrchestrationV2Command;
369 readonly reply: Deferred.Deferred<never, OrchestrationV2DispatchCommandError>;
370}> = [];
371let sequence = 1;
372let vcsStatusSubscriber: Queue.Queue<VcsStatusStreamEvent> | null = null;
373
374const subscribe = <A>(first: ReadonlyArray<A>, register?: (queue: Queue.Queue<A>) => () => void) =>
375 Stream.unwrap(
376 Effect.gen(function* () {
377 const queue = yield* Queue.unbounded<A>();
378 yield* Queue.offerAll(queue, first);
379 if (register) {
380 const unregister = register(queue);
381 yield* Effect.addFinalizer(() => Effect.sync(unregister));
382 }
383 return Stream.fromQueue(queue);
384 }),
385 );
386
387const handlers: Record<string, (input: never) => unknown> = {
388 // Recorded, and accepted only when the scenario asks, since a pending command holds every
389 // later command on its thread. Accepted commands' events never arrive.
390 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
391 Effect.suspend(() => {
392 dispatched.push(command);
393 if (acceptCommands && !holdCommand(command))
394 return Effect.succeed({ sequence: sequence + 1 });
395 const reply = Deferred.makeUnsafe<never, OrchestrationV2DispatchCommandError>();
396 pendingDispatches.push({ command, reply });
397 return Deferred.await(reply);
398 }),
399 [WS_METHODS.subscribeTerminalMetadata]: () =>
400 subscribe<TerminalMetadataStreamEvent>([{ type: "snapshot", terminals: [] }], (queue) => {
401 terminalMetadataSubscriber = queue;
402 return () => {
403 if (terminalMetadataSubscriber === queue) terminalMetadataSubscriber = null;
404 };
405 }),
406 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () =>
407 subscribe<OrchestrationV2ShellStreamItem>([
408 {
409 kind: "snapshot",
410 snapshot: {
411 schemaVersion: 1,
412 snapshotSequence: sequence,
413 projects: [project],
414 threads: threads.filter((thread) => !thread.olderSettled).map(shellOf),
415 archivedThreads: [],
416 },
417 },
418 { kind: "synchronized" },
419 ]),
420 [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input: { readonly threadId: ThreadId }) => {
421 const thread = threads.find((candidate) => candidate.id === input.threadId);
422 if (!thread) return Stream.never;
423 return subscribe<OrchestrationV2ThreadStreamItem>(
424 [
425 { kind: "snapshot", snapshotSequence: sequence, projection: projectionOf(thread) },
426 { kind: "synchronized" },
427 ],
428 (queue) => {
429 threadSubscribers.set(thread.id, queue);
430
431 // A replaced subscription closes after its successor registers.
432 return () => {
433 if (threadSubscribers.get(thread.id) === queue) threadSubscribers.delete(thread.id);
434 };
435 },
436 );
437 },
438 [ORCHESTRATION_V2_WS_METHODS.getThreadFamily]: (input: { readonly threadId: ThreadId }) => {
439 const rootOf = (thread: FakeThread) => thread.sideChatOf ?? thread.id;
440 const requested = threads.find((candidate) => candidate.id === input.threadId);
441 return Effect.succeed({
442 snapshotSequence: sequence,
443 threads: requested
444 ? threads.filter((thread) => rootOf(thread) === rootOf(requested)).map(shellOf)
445 : [],
446 });
447 },
448 [WS_METHODS.subscribeVcsStatus]: () =>
449 subscribe<VcsStatusStreamEvent>([], (queue) => {
450 vcsStatusSubscriber = queue;
451 return () => {
452 if (vcsStatusSubscriber === queue) vcsStatusSubscriber = null;
453 };
454 }),
455 [WS_METHODS.subscribeServerConfig]: () =>
456 subscribe([{ version: 1 as const, type: "snapshot" as const, config: serverConfig() }]),
457};
458
459// Unanswered calls stay pending, like a server that has not replied yet.
460const client = new Proxy(handlers, {
461 get: (target, method: string) =>
462 target[method] ??
463 (() => {
464 const rpc = WsRpcGroup.requests.get(method);
465 return rpc && RpcSchema.isStreamSchema(rpc.successSchema) ? Stream.never : Effect.never;
466 }),
467});
468
469const session: EnvironmentConnectionLease["session"] = {
470 client: client as unknown as WsRpcProtocolClient,
471 initialConfig: Effect.sync(serverConfig),
472 subscribeServerConfig: () =>
473 subscribe([{ version: 1 as const, type: "snapshot" as const, config: serverConfig() }]),
474 ready: Effect.void,
475 probe: Effect.void,
476 inboundFrames: Effect.succeed(0),
477 closed: Effect.never,
478};
479
480/** Stands in for client-runtime's ConnectionDriver: every connection opens the session above. */
481export const fakeDriver = {
482 connect: (
483 entry: ConnectionCatalogEntry,
484 ): Effect.Effect<EnvironmentConnectionLease, ConnectionAttemptError, Scope.Scope> =>
485 Effect.succeed({
486 prepared: {
487 environmentId: ENVIRONMENT_ID,
488 label: descriptor.label,
489 httpBaseUrl: window.location.origin,
490 socketUrl: window.location.origin.replace(/^http/, "ws"),
491 httpAuthorization: null,
492 target: entry.target,
493 },
494 session,
495 }),
496 preflight: () => Effect.succeed(true),
497 checkHost: () => Effect.succeed<RouteCheck>("unchecked"),
498};
499export type FakeDriver = typeof fakeDriver;
500
501const realFetch = globalThis.fetch;
502const encodedDescriptor = Schema.encodeSync(ExecutionEnvironmentDescriptor)(
503 Schema.decodeUnknownSync(ExecutionEnvironmentDescriptor)(descriptor),
504);
505
506/** Serves the environment's HTTP API and these threads until disposed. */
507export function installFakeEnvironment(
508 fixtures: ReadonlyArray<FakeThread>,
509 options?: {
510 readonly acceptCommands?: boolean;
511 /** Commands left unanswered even when accepting, until `rejectPendingCommand`. */
512 readonly holdCommand?: (command: OrchestrationV2Command) => boolean;
513 readonly providers?: ReadonlyArray<unknown>;
514 },
515): Disposable {
516 threads = fixtures;
517 baseServerConfig = makeServerConfig(options?.providers ?? [CODEX_PROVIDER]);
518 dispatched = [];
519 acceptCommands = options?.acceptCommands === true;
520 holdCommand = options?.holdCommand ?? (() => false);
521 globalThis.fetch = async (input, init) => {
522 const url = new URL(input instanceof Request ? input.url : String(input), window.location.href);
523 if (url.origin !== window.location.origin) return realFetch(input, init);
524 if (url.pathname === "/.well-known/t3/environment") return json(encodedDescriptor);
525 if (url.pathname === "/api/auth/session") {
526 return json({
527 authenticated: true,
528 auth,
529 scopes: AuthStandardClientScopes,
530 sessionMethod: "browser-session-cookie",
531 expiresAt: "2099-01-01T00:00:00.000Z",
532 });
533 }
534 // Snapshot endpoints fall back to the socket, which serves the same fixtures.
535 if (/^\/(api|oauth|\.well-known)\//.test(url.pathname)) return json({}, 404);
536 return realFetch(input, init);
537 };
538 return {
539 [Symbol.dispose]: () => {
540 globalThis.fetch = realFetch;
541 threads = [];
542 baseServerConfig = makeServerConfig([CODEX_PROVIDER]);
543 dispatched = [];
544 acceptCommands = false;
545 holdCommand = () => false;
546 pendingDispatches = [];
547 },
548 };
549}
550
551/** Answers the oldest unanswered command with the server's rejection. */
552export async function rejectPendingCommand(message: string): Promise<void> {
553 const pending = pendingDispatches.shift();
554 if (!pending) throw new Error("No pending command");
555 await Effect.runPromise(
556 Deferred.fail(
557 pending.reply,
558 new OrchestrationV2DispatchCommandError({
559 commandId: pending.command.commandId,
560 commandType: pending.command.type,
561 message,
562 }),
563 ),
564 );
565}
566
567/** Commands the app dispatched to the environment, oldest first. */
568export function dispatchedCommands(): ReadonlyArray<OrchestrationV2Command> {
569 return dispatched;
570}
571
572type ThreadEvent = Extract<OrchestrationV2ThreadStreamItem, { readonly kind: "event" }>["event"];
573
574export async function emitTerminalMetadata(event: TerminalMetadataStreamEvent): Promise<void> {
575 if (terminalMetadataSubscriber === null) throw new Error("No terminal metadata subscription");
576 await Effect.runPromise(Queue.offer(terminalMetadataSubscriber, event));
577}
578
579/** Whether the app holds a live subscription to the thread, so events reach it. */
580export function isThreadSubscribed(threadId: ThreadId): boolean {
581 return threadSubscribers.has(threadId);
582}
583
584/** Delivers one domain event, such as a run or turn-item update, to the thread's subscription. */
585export async function emitThreadEvent(
586 threadId: ThreadId,
587 event: Pick<ThreadEvent, "type" | "payload">,
588): Promise<void> {
589 const queue = threadSubscribers.get(threadId);
590 if (!queue) throw new Error(`No live subscription for ${threadId}`);
591 sequence += 1;
592 await Effect.runPromise(
593 Queue.offer(queue, {
594 kind: "event",
595 sequence,
596 event: { ...event, id: EventId.make(`event-${sequence}`), threadId, occurredAt: AT },
597 } as OrchestrationV2ThreadStreamItem),
598 );
599}
600
601export function runningRun(threadId: ThreadId, id: string, ordinal: number, userMessageId: string) {
602 return {
603 ...run(threadId, RunId.make(id), ordinal, userMessageId),
604 status: "running" as const,
605 completedAt: null,
606 };
607}
608
609export { userItem };
610
611export async function emitVcsStatus(event: VcsStatusStreamEvent): Promise<void> {
612 if (vcsStatusSubscriber === null) throw new Error("No VCS status subscription");
613 await Effect.runPromise(Queue.offer(vcsStatusSubscriber, event));
614}