clover's git

the-knife

1	import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
2	import * as MergebotQueue from "../mergebot/MergebotQueue.ts";
3	import { SourceControlProviderKind } from "@t3tools/contracts";
4	import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts";
5	import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
6	import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";
7	import * as NodeServices from "@effect/platform-node/NodeServices";
8	import { assert, it } from "@effect/vitest";
9	import { vi } from "vite-plus/test";
10	import {
11	type ApplicationStoredEvent,
12	CheckpointId,
13	CheckpointRef,
14	CommandId,
15	ContextTransferId,
16	EventId,
17	MessageId,
18	NodeId,
19	RuntimeRequestId,
20	TurnItemId,
21	type ModelSelection,
22	type OrchestrationV2Run,
23	ProjectId,
24	type PullRequestDetail,
25	type PullRequestComment,
26	PullRequestOperationError,
27	ProviderDriverKind,
28	ProviderInstanceId,
29	ProviderThreadId,
30	ProviderTurnId,
31	RunId,
32	ThreadId,
33	} from "@t3tools/contracts";
34	import * as Effect from "effect/Effect";
35	import * as DateTime from "effect/DateTime";
36	import * as Deferred from "effect/Deferred";
37	import * as Fiber from "effect/Fiber";
38	import * as Layer from "effect/Layer";
39	import * as Logger from "effect/Logger";
40	import * as Path from "effect/Path";
41	import * as Option from "effect/Option";
42	import * as Queue from "effect/Queue";
43	import * as Stream from "effect/Stream";
44	import * as TestClock from "effect/testing/TestClock";
45	import * as SqlClient from "effect/sql/SqlClient";
46
47	import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
48	import * as GitWorkflow from "../git/GitWorkflowService.ts";
49	import * as ServerConfig from "../config.ts";
50	import * as SqlitePersistence from "../persistence/Sqlite.ts";
51	import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
52	import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
53	import * as ProjectService from "../project/ProjectService.ts";
54	import * as ServerSettings from "../serverSettings.ts";
55	import * as McpSessionRegistryTestkit from "../mcp/McpSessionRegistry.testkit.ts";
56	import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts";
57	import type { ProviderInstance } from "@t3tools/provider-core/server/driver";
58	import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
59	import * as VcsProcess from "../vcs/VcsProcess.ts";
60	import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
61	import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
62	import * as Orchestrator from "./Orchestrator.ts";
63	import { ROLLBACK_FAILED_MESSAGE } from "./CheckpointRollbackService.ts";
64	import * as EffectWorker from "./EffectWorker.ts";
65	import * as EffectOutbox from "./EffectOutbox.ts";
66	import * as EventSink from "./EventSink.ts";
67	import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryService.ts";
68	import * as ProjectionMaintenance from "./ProjectionMaintenance.ts";
69	import * as ProjectionStore from "./ProjectionStore.ts";
70	import * as PullRequestWatchReactor from "./PullRequestWatchReactor.ts";
71	import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
72	import * as PullRequestService from "../pullRequest/PullRequestService.ts";
73	import * as ProjectStore from "./ProjectStore.ts";
74	import * as ProviderSessionManager from "./ProviderSessionManager.ts";
75	import * as RuntimeLayer from "./runtimeLayer.ts";
76	import { shellStreamItemFromThreadShell } from "./ShellStream.ts";
77	import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
78	import * as ThreadManagementService from "./ThreadManagementService.ts";
79	import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
80	import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
81
82	const layerPlatformTest = Layer.merge(
83	NodeServices.layer,
84	Layer.mock(SourceControlProviderRegistry.SourceControlProviderRegistry)({
85	resolveLink: () => Effect.die("unused title link"),
86	}),
87	);
88
89	const layerServerConfig = ServerConfig.layerTest(process.cwd(), {
90	prefix: "t3-orchestration-v2-runtime-layer-",
91	});
92
93	const modelSelection = {
94	instanceId: ProviderInstanceId.make("codex"),
95	model: "gpt-5.4",
96	} satisfies ModelSelection;
97	const alternateInstanceId = ProviderInstanceId.make("codex_alternate");
98
99	const layerVcsDriverRegistryTest = VcsDriverRegistry.layer.pipe(
100	Layer.provide(VcsProcess.layer),
101	Layer.provide(layerServerConfig),
102	Layer.provide(layerPlatformTest),
103	);
104
105	const layerCheckpointStoreTest = CheckpointStore.layer.pipe(
106	Layer.provide(layerVcsDriverRegistryTest),
107	);
108	const layerGitWorkflowTest = Layer.mock(GitWorkflow.GitWorkflowService)({
109	pruneWorktrees: () => Effect.void,
110	createWorktree: () => Effect.succeed({} as never),
111	});
112	const layerProjectServiceTest = Layer.mock(ProjectService.ProjectService)({
113	getById: () => Effect.succeed(Option.none()),
114	});
115
116	const driver = ProviderDriverKind.make("codex");
117	const orchestrationAdapter = {
118	instanceId: modelSelection.instanceId,
119	driver,
120	getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
121	planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
122	openSession: () => Effect.die("sessions are not used by lifecycle tests"),
123	} as ProviderAdapter.ProviderAdapterV2["Service"];
124	const providerInstance = {
125	instanceId: modelSelection.instanceId,
126	driverKind: driver,
127	continuationIdentity: {
128	driverKind: driver,
129	continuationKey: "codex:test",
130	},
131	displayName: "Codex test",
132	enabled: true,
133	// No supportedRuntimeModes: every runtime mode runs as stored.
134	snapshot: { getSnapshot: Effect.succeed({}) } as unknown as ProviderInstance["snapshot"],
135	orchestrationAdapter,
136	textGeneration: {} as ProviderInstance["textGeneration"],
137	} satisfies ProviderInstance;
138	const alternateProviderInstance = {
139	...providerInstance,
140	instanceId: alternateInstanceId,
141	continuationIdentity: {
142	driverKind: driver,
143	continuationKey: "codex:test:alternate",
144	},
145	displayName: "Codex alternate test",
146	orchestrationAdapter: {
147	...orchestrationAdapter,
148	instanceId: alternateInstanceId,
149	},
150	} satisfies ProviderInstance;
151
152	const layerTestProviderInstanceRegistry = Layer.succeed(
153	ProviderInstanceRegistry.ProviderInstanceRegistry,
154	{
155	getInstance: (instanceId) =>
156	Effect.succeed(
157	[providerInstance, alternateProviderInstance].find(
158	(instance) => instanceId === instance.instanceId,
159	),
160	),
161	listInstances: Effect.succeed([providerInstance, alternateProviderInstance]),
162	listUnavailable: Effect.succeed([]),
163	streamChanges: Stream.empty,
164	subscribeChanges: Effect.never,
165	},
166	);
167
168	/** Seed a project row the way a committed `project.created` event folds into it. */
169	/** A watched pull request as the host reports it: open, mergeable, with a failing lint check. */
170	const watchedPullRequestDetail = (input: {
171	readonly projectId: ProjectId;
172	readonly number: number;
173	readonly at: string;
174	}): PullRequestDetail => ({
175	provider: SourceControlProviderKind.make("github"),
176	capabilities: {
177	diff: true,
178	comment: true,
179	actions: [],
180	mergeMethods: [],
181	search: false,
182	review: { inlineComment: false, reply: false, resolve: false, verdicts: [] },
183	reviewers: { request: false, listCandidates: false },
184	},
185	viewerPermissions: {
186	actions: [],
187	comment: true,
188	resolve: true,
189	verdicts: [],
190	requestReviewers: false,
191	},
192	projectId: input.projectId,
193	projectTitle: "Watch wake",
194	workspaceRoot: "/workspace/watch",
195	repository: "pingdotgg/t3code",
196	number: input.number,
197	title: "Watched pull request",
198	body: "",
199	url: `https://github.com/pingdotgg/t3code/pull/${input.number}`,
200	author: { login: "agent-user", name: null, avatarUrl: null },
201	state: "open",
202	isDraft: false,
203	mergeability: "mergeable",
204	additions: 1,
205	deletions: 0,
206	changedFiles: 1,
207	headBranch: "feature",
208	headSha: "abc1234def",
209	baseBranch: "main",
210	createdAt: input.at,
211	updatedAt: input.at,
212	mergedAt: null,
213	closedAt: null,
214	reviewers: [],
215	labels: [],
216	checks: [{ name: "lint", status: "failure", description: null, url: null }],
217	mergeCapabilities: { merge: true, squash: true, rebase: true },
218	viewer: "agent-user",
219	});
220
221	const seedProject = (input: {
222	readonly projectId: ProjectId;
223	readonly title: string;
224	readonly workspaceRoot: string;
225	readonly defaultModelSelection: ModelSelection | null;
226	readonly createdAt: string;
227	}) =>
228	Effect.flatMap(ProjectStore.ProjectStoreV2, (projects) =>
229	projects.apply({
230	sequence: 0,
231	eventId: EventId.make(`seed:${input.projectId}`),
232	aggregateKind: "project",
233	aggregateId: input.projectId,
234	occurredAt: input.createdAt,
235	commandId: null,
236	causationEventId: null,
237	correlationId: null,
238	metadata: {},
239	type: "project.created",
240	payload: {
241	projectId: input.projectId,
242	title: input.title,
243	workspaceRoot: input.workspaceRoot,
244	defaultModelSelection: input.defaultModelSelection,
245	scripts: [],
246	createdAt: input.createdAt,
247	updatedAt: input.createdAt,
248	},
249	}),
250	);
251
252	/** Move a seeded project the way a committed `project.meta-updated` event does. */
253	const moveProject = (projectId: ProjectId, workspaceRoot: string, updatedAt: string) =>
254	Effect.flatMap(ProjectStore.ProjectStoreV2, (projects) =>
255	projects.apply({
256	sequence: 0,
257	eventId: EventId.make(`move:${projectId}:${workspaceRoot}`),
258	aggregateKind: "project",
259	aggregateId: projectId,
260	occurredAt: updatedAt,
261	commandId: null,
262	causationEventId: null,
263	correlationId: null,
264	metadata: {},
265	type: "project.meta-updated",
266	payload: { projectId, workspaceRoot, updatedAt },
267	}),
268	);
269
270	const layerTest = Layer.mergeAll(
271	RuntimeLayer.layer,
272	RuntimeLayer.layerEventSink,
273	ProjectStore.layer,
274	ProjectionStore.layer,
275	EffectOutbox.layer,
276	ThreadCommandExecutor.layer,
277	).pipe(
278	Layer.provide(McpSessionRegistryTestkit.layer),
279	Layer.provide(Layer.merge(McpProviderSessions.layer, ScriptProgress.layerUnavailable)),
280	Layer.provide(SqlitePersistence.layerMemory),
281	Layer.provide(layerCheckpointStoreTest),
282	Layer.provide(layerServerConfig),
283	Layer.provide(ServerSettings.layerTest()),
284	Layer.provide(layerTestProviderInstanceRegistry),
285	Layer.provide(layerGitWorkflowTest),
286	Layer.provide(layerProjectServiceTest),
287	Layer.provide(layerPlatformTest),
288	);
289
290	const layerLegacyImportTest = RuntimeLayer.layer.pipe(
291	Layer.provide(McpSessionRegistryTestkit.layer),
292	Layer.provide(Layer.merge(McpProviderSessions.layer, ScriptProgress.layerUnavailable)),
293	Layer.provideMerge(SqlitePersistence.layerMemory),
294	Layer.provide(layerCheckpointStoreTest),
295	Layer.provide(layerServerConfig),
296	Layer.provide(ServerSettings.layerTest()),
297	Layer.provide(layerTestProviderInstanceRegistry),
298	Layer.provide(layerGitWorkflowTest),
299	Layer.provide(layerProjectServiceTest),
300	Layer.provide(layerPlatformTest),
301	);
302
303	const layerProjectDeletionTest = Layer.mergeAll(
304	RuntimeLayer.layer.pipe(Layer.provide(RuntimeLayer.layerProjectService)),
305	RuntimeLayer.layerProjectService,
306	RuntimeLayer.layerEventSink,
307	ThreadCommandExecutor.layer,
308	).pipe(
309	Layer.provide(
310	Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({
311	peek: () =>
312	Effect.succeed({
313	repositoryIdentity: null,
314	faviconPath: null,
315	repositoryIdentityResolved: false,
316	}),
317	getAvailable: () =>
318	Effect.succeed({
319	repositoryIdentity: null,
320	faviconPath: null,
321	repositoryIdentityResolved: false,
322	}),
323	invalidate: () => Effect.void,
324	}),
325	),
326	Layer.provide(
327	Layer.mock(WorkspacePaths.WorkspacePaths)({
328	normalizeWorkspaceRoot: (workspaceRoot) => Effect.succeed(workspaceRoot),
329	}),
330	),
331	Layer.provide(McpSessionRegistryTestkit.layer),
332	Layer.provide(Layer.merge(McpProviderSessions.layer, ScriptProgress.layerUnavailable)),
333	Layer.provide(SqlitePersistence.layerMemory),
334	Layer.provide(layerCheckpointStoreTest),
335	Layer.provide(layerServerConfig),
336	Layer.provide(ServerSettings.layerTest()),
337	Layer.provide(layerTestProviderInstanceRegistry),
338	Layer.provide(layerGitWorkflowTest),
339	Layer.provide(layerPlatformTest),
340	);
341
342	it.layer(layerProjectDeletionTest)("project deletion during thread commands", (it) => {
343	it.effect("waits for an in-flight thread update before planning deletion", () =>
344	Effect.gen(function* () {
345	const projects = yield* ProjectService.ProjectService;
346	const orchestrator = yield* Orchestrator.OrchestratorV2;
347	const eventSink = yield* EventSink.EventSinkV2;
348	const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor;
349	const projectId = ProjectId.make("runtime-project-delete-concurrent");
350	const threadId = ThreadId.make("runtime-thread-delete-concurrent");
351	const updateCommandId = CommandId.make("runtime-thread-delete-concurrent-update");
352	yield* projects.create({
353	commandId: CommandId.make("runtime-project-delete-concurrent-create"),
354	projectId,
355	title: "Concurrent deletion",
356	workspaceRoot: "/work/concurrent-deletion",
357	});
358	yield* orchestrator.dispatch({
359	type: "thread.create",
360	commandId: CommandId.make("runtime-thread-delete-concurrent-create"),
361	createdBy: "user",
362	creationSource: "web",
363	threadId,
364	projectId,
365	title: "Original title",
366	modelSelection,
367	runtimeMode: "full-access",
368	interactionMode: "default",
369	branch: null,
370	worktreePath: null,
371	});
372
373	const updateReady = yield* Deferred.make<void>();
374	const releaseUpdate = yield* Deferred.make<void>();
375	const deletionQueued = yield* Deferred.make<void>();
376	const commitCommand = eventSink.commitCommand;
377	const withLock = executor.withLock;
378	let threadLockRequests = 0;
379	const commitSpy = vi
380	.spyOn(eventSink, "commitCommand")
381	.mockImplementation((input) =>
382	input.commandId === updateCommandId
383	? Deferred.succeed(updateReady, undefined).pipe(
384	Effect.andThen(Deferred.await(releaseUpdate)),
385	Effect.andThen(commitCommand(input)),
386	)
387	: commitCommand(input),
388	);
389	const observeLock: ThreadCommandExecutor.ThreadCommandExecutor["Service"]["withLock"] = (
390	key,
391	effect,
392	) => {
393	if (key !== threadId || ++threadLockRequests !== 2) return withLock(key, effect);
394	return Deferred.succeed(deletionQueued, undefined).pipe(
395	Effect.andThen(withLock(key, effect)),
396	);
397	};
398	const lockSpy = vi.spyOn(executor, "withLock").mockImplementation(observeLock);
399	yield* Effect.gen(function* () {
400	const updateFiber = yield* orchestrator
401	.dispatch({
402	type: "thread.metadata.update",
403	commandId: updateCommandId,
404	threadId,
405	title: "Updated before deletion",
406	})
407	.pipe(Effect.forkChild({ startImmediately: true }));
408	yield* Effect.raceFirst(
409	Deferred.await(updateReady),
410	Fiber.join(updateFiber).pipe(
411	Effect.andThen(Effect.die("The update completed before reaching its commit barrier.")),
412	),
413	);
414	const deleteFiber = yield* projects
415	.delete({
416	commandId: CommandId.make("runtime-project-delete-concurrent-delete"),
417	projectId,
418	force: true,
419	})
420	.pipe(Effect.forkChild({ startImmediately: true }));
421	yield* Effect.raceFirst(
422	Deferred.await(deletionQueued),
423	Fiber.join(deleteFiber).pipe(
424	Effect.andThen(Effect.die("Project deletion bypassed the in-flight thread command.")),
425	),
426	);
427	yield* Deferred.succeed(releaseUpdate, undefined);
428	yield* Fiber.join(updateFiber);
429	const deletedProject = yield* Fiber.join(deleteFiber);
430	const projection = yield* orchestrator.getThreadProjection(threadId);
431	assert.isNotNull(deletedProject.deletedAt);
432	assert.isNotNull(projection.thread.deletedAt);
433	assert.equal(projection.thread.title, "Updated before deletion");
434	}).pipe(
435	Effect.ensuring(
436	Effect.sync(() => {
437	commitSpy.mockRestore();
438	lockSpy.mockRestore();
439	}),
440	),
441	);
442	}),
443	);
444	});
445
446	const layerSharedApplicationDataPlaneTest = Layer.mergeAll(
447	RuntimeLayer.layer.pipe(Layer.provide(RuntimeLayer.layerProjectService)),
448	RuntimeLayer.layerProjectService,
449	RuntimeLayer.layerEventSink,
450	RuntimeLayer.layerEventInfrastructure,
451	).pipe(
452	Layer.provide(
453	Layer.mock(WorkspacePaths.WorkspacePaths)({
454	normalizeWorkspaceRoot: (workspaceRoot) => Effect.succeed(workspaceRoot),
455	}),
456	),
457	Layer.provide(
458	Layer.succeed(ProjectEnrichmentService.ProjectEnrichmentService, {
459	peek: () =>
460	Effect.succeed({
461	repositoryIdentity: null,
462	faviconPath: null,
463	repositoryIdentityResolved: false,
464	}),
465	request: () => Effect.void,
466	getAvailable: () =>
467	Effect.succeed({
468	repositoryIdentity: null,
469	faviconPath: null,
470	repositoryIdentityResolved: false,
471	}),
472	invalidate: () => Effect.void,
473	subscribeChanges: Effect.never,
474	}),
475	),
476	Layer.provide(McpSessionRegistryTestkit.layer),
477	Layer.provide(Layer.merge(McpProviderSessions.layer, ScriptProgress.layerUnavailable)),
478	Layer.provideMerge(SqlitePersistence.layerMemory),
479	Layer.provide(layerCheckpointStoreTest),
480	Layer.provide(layerServerConfig),
481	Layer.provide(ServerSettings.layerTest()),
482	Layer.provide(layerTestProviderInstanceRegistry),
483	Layer.provide(layerGitWorkflowTest),
484	Layer.provide(layerPlatformTest),
485	);
486
487	it.layer(layerTest)("OrchestrationV2LayerLive", (it) => {
488	it.effect("emits model updates separately from provider switches", () =>
489	Effect.gen(function* () {
490	const orchestrator = yield* Orchestrator.OrchestratorV2;
491	const threadId = ThreadId.make("runtime-layer-model-selection-events");
492
493	yield* orchestrator.dispatch({
494	type: "thread.create",
495	createdBy: "user",
496	creationSource: "web",
497	commandId: CommandId.make("runtime-layer-model-selection-events-create"),
498	threadId,
499	projectId: ProjectId.make("runtime-layer-model-selection-events-project"),
500	title: "Model selection events",
501	modelSelection,
502	runtimeMode: "full-access",
503	interactionMode: "default",
504	branch: null,
505	worktreePath: null,
506	});
507
508	const sameInstance = yield* orchestrator.dispatch({
509	type: "thread.model-selection.set",
510	commandId: CommandId.make("runtime-layer-model-selection-events-update"),
511	threadId,
512	modelSelection: { ...modelSelection, model: "gpt-5.5" },
513	});
514	assert.deepEqual(
515	sameInstance.storedEvents.map((stored) => stored.event.type),
516	["thread.model-selection-updated"],
517	);
518
519	const differentInstance = yield* orchestrator.dispatch({
520	type: "thread.model-selection.set",
521	commandId: CommandId.make("runtime-layer-model-selection-events-switch"),
522	threadId,
523	modelSelection: { instanceId: alternateInstanceId, model: "gpt-5.5" },
524	});
525	assert.deepEqual(
526	differentInstance.storedEvents.map((stored) => stored.event.type),
527	["thread.provider-switched"],
528	);
529	}),
530	);
531
532	/**
533	* A thread with one ready checkpoint and no queued turn start. Every provider
534	* rollback on it fails, so each rollback effect retries until it gives up.
535	*/
536	const seedFailingRollbackThread = (name: string) =>
537	Effect.gen(function* () {
538	const orchestrator = yield* Orchestrator.OrchestratorV2;
539	const eventSink = yield* EventSink.EventSinkV2;
540	const outbox = yield* EffectOutbox.EffectOutboxV2;
541	const threadId = ThreadId.make(name);
542	yield* orchestrator.dispatch({
543	type: "thread.create",
544	createdBy: "user",
545	creationSource: "web",
546	commandId: CommandId.make(`${name}-create`),
547	threadId,
548	projectId: ProjectId.make(`${name}-project`),
549	title: "Rollback failure",
550	modelSelection,
551	runtimeMode: "full-access",
552	interactionMode: "default",
553	branch: null,
554	// Its own path, so other rollback tests keep an isolated worktree.
555	worktreePath: `/tmp/t3-${name}`,
556	});
557	yield* orchestrator.dispatch({
558	type: "message.dispatch",
559	createdBy: "user",
560	creationSource: "web",
561	commandId: CommandId.make(`${name}-message`),
562	threadId,
563	messageId: MessageId.make(`${name}-message`),
564	text: "Create the provider thread and checkpoint scope.",
565	attachments: [],
566	dispatchMode: { type: "start_immediately" },
567	});
568	const scope = (yield* orchestrator.getThreadProjection(threadId)).checkpointScopes[0]!;
569	const now = yield* DateTime.now;
570	const checkpointId = CheckpointId.make(`${name}-checkpoint`);
571	yield* eventSink.write({
572	commandId: CommandId.make(`${name}-seed`),
573	events: [
574	{
575	id: EventId.make(`${name}-checkpoint-event`),
576	type: "checkpoint.captured",
577	threadId,
578	occurredAt: now,
579	payload: {
580	id: checkpointId,
581	threadId,
582	scopeId: scope.id,
583	runId: null,
584	nodeId: scope.nodeId,
585	parentCheckpointId: null,
586	ordinalWithinScope: 0,
587	appRunOrdinal: null,
588	ref: CheckpointRef.make(`refs/t3/${name}`),
589	status: "ready",
590	files: [],
591	capturedAt: now,
592	},
593	},
594	],
595	});
596	// These tests cover rollback only, so drop the first message's start.
597	yield* outbox.cancelUnsettled({
598	threadId,
599	effectTypes: ["provider-turn.start"],
600	reason: "not under test",
601	});
602	return {
603	threadId,
604	rollback: (commandId: CommandId) =>
605	orchestrator.dispatch({
606	type: "checkpoint.rollback",
607	commandId,
608	threadId,
609	checkpointId,
610	scopeId: scope.id,
611	restoreFiles: false,
612	}),
613	};
614	});
615
616	it.effect("projects a rollback that fails every attempt and clears it on the next one", () =>
617	Effect.gen(function* () {
618	const orchestrator = yield* Orchestrator.OrchestratorV2;
619	const outbox = yield* EffectOutbox.EffectOutboxV2;
620	const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
621	const { threadId, rollback } = yield* seedFailingRollbackThread("runtime-rollback-failure");
622
623	const rollbackCommandId = CommandId.make("runtime-rollback-failure-rollback");
624	yield* rollback(rollbackCommandId);
625	// Retries back off on the clock; advance it until the worker gives up.
626	for (let attempt = 0; attempt < 5; attempt++) {
627	yield* worker.drain();
628	yield* TestClock.adjust("30 seconds");
629	}
630
631	const [rollbackEffect] = yield* outbox.listByCommandId(rollbackCommandId);
632	assert.equal(rollbackEffect?.status, "failed");
633	const failed = yield* orchestrator.getThreadProjection(threadId);
634	assert.deepEqual(failed.thread.rollbackFailure, {
635	requestId: rollbackCommandId,
636	message: ROLLBACK_FAILED_MESSAGE,
637	});
638
639	yield* rollback(CommandId.make("runtime-rollback-failure-retry"));
640	const retried = yield* orchestrator.getThreadProjection(threadId);
641	assert.isNull(retried.thread.rollbackFailure);
642	}),
643	);
644
645	it.effect("ignores a late failure from a rollback that a newer one superseded", () =>
646	Effect.gen(function* () {
647	const orchestrator = yield* Orchestrator.OrchestratorV2;
648	const outbox = yield* EffectOutbox.EffectOutboxV2;
649	const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
650	const { threadId, rollback } = yield* seedFailingRollbackThread(
651	"runtime-rollback-superseded",
652	);
653
654	const olderCommandId = CommandId.make("runtime-rollback-superseded-older");
655	yield* rollback(olderCommandId);
656	// Spend every attempt but the last.
657	for (let attempt = 0; attempt < 4; attempt++) {
658	yield* worker.drain();
659	yield* TestClock.adjust("30 seconds");
660	}
661	const newerCommandId = CommandId.make("runtime-rollback-superseded-newer");
662	yield* rollback(newerCommandId);
663	// The older rollback's last attempt fails after the newer one started.
664	yield* worker.drain();
665
666	const [olderEffect] = yield* outbox.listByCommandId(olderCommandId);
667	assert.equal(olderEffect?.status, "failed");
668	const superseded = yield* orchestrator.getThreadProjection(threadId);
669	assert.isNull(superseded.thread.rollbackFailure);
670
671	for (let attempt = 0; attempt < 5; attempt++) {
672	yield* TestClock.adjust("30 seconds");
673	yield* worker.drain();
674	}
675	const [newerEffect] = yield* outbox.listByCommandId(newerCommandId);
676	assert.equal(newerEffect?.status, "failed");
677	const failed = yield* orchestrator.getThreadProjection(threadId);
678	assert.deepEqual(failed.thread.rollbackFailure, {
679	requestId: newerCommandId,
680	message: ROLLBACK_FAILED_MESSAGE,
681	});
682	}),
683	);
684
685	it.effect("rejects non-ready rollback targets before persisting events or effects", () =>
686	Effect.gen(function* () {
687	const orchestrator = yield* Orchestrator.OrchestratorV2;
688	const eventSink = yield* EventSink.EventSinkV2;
689	const outbox = yield* EffectOutbox.EffectOutboxV2;
690	const threadId = ThreadId.make("runtime-rollback-readiness");
691	yield* orchestrator.dispatch({
692	type: "thread.create",
693	createdBy: "user",
694	creationSource: "web",
695	commandId: CommandId.make("runtime-rollback-readiness-create"),
696	threadId,
697	projectId: ProjectId.make("runtime-rollback-readiness-project"),
698	title: "Rollback readiness",
699	modelSelection,
700	runtimeMode: "full-access",
701	interactionMode: "default",
702	branch: null,
703	worktreePath: process.cwd(),
704	});
705	yield* orchestrator.dispatch({
706	type: "message.dispatch",
707	createdBy: "user",
708	creationSource: "web",
709	commandId: CommandId.make("runtime-rollback-readiness-message"),
710	threadId,
711	messageId: MessageId.make("runtime-rollback-readiness-message"),
712	text: "Create the provider thread and checkpoint scope.",
713	attachments: [],
714	dispatchMode: { type: "start_immediately" },
715	});
716	const projection = yield* orchestrator.getThreadProjection(threadId);
717	const scope = projection.checkpointScopes[0]!;
718	const now = yield* DateTime.now;
719
720	for (const status of ["missing", "error", "stale", "ready"] as const) {
721	const checkpointId = CheckpointId.make("runtime-rollback-checkpoint");
722	const commandId = CommandId.make(`runtime-rollback-${status}`);
723	yield* eventSink.write({
724	commandId: CommandId.make(`runtime-rollback-${status}-seed`),
725	events: [
726	{
727	id: EventId.make(`runtime-rollback-${status}-event`),
728	type: "checkpoint.captured",
729	threadId,
730	occurredAt: now,
731	payload: {
732	id: checkpointId,
733	threadId,
734	scopeId: scope.id,
735	runId: null,
736	nodeId: scope.nodeId,
737	parentCheckpointId: null,
738	ordinalWithinScope: 0,
739	appRunOrdinal: null,
740	ref: CheckpointRef.make(`refs/t3/runtime-rollback-${status}`),
741	status,
742	files: [],
743	capturedAt: now,
744	},
745	},
746	],
747	});
748	const previousSequence = yield* orchestrator.getThreadEventSequence(threadId);
749	const rollback = orchestrator.dispatch({
750	type: "checkpoint.rollback",
751	commandId,
752	threadId,
753	checkpointId,
754	scopeId: scope.id,
755	});
756
757	if (status === "ready") {
758	const accepted = yield* rollback;
759	assert.deepEqual(
760	accepted.storedEvents.map((stored) => stored.event.type),
761	["thread.metadata-updated", "checkpoint.rollback-requested"],
762	);
763	assert.deepEqual(
764	(yield* outbox.listByCommandId(commandId)).map((effect) => effect.request.type),
765	["provider-thread.rollback"],
766	);
767	const path = yield* Path.Path.pipe(Effect.provide(NodeServices.layer));
768	yield* orchestrator.dispatch({
769	type: "thread.create",
770	createdBy: "user",
771	creationSource: "web",
772	commandId: CommandId.make("runtime-rollback-ancestor-create"),
773	threadId: ThreadId.make("runtime-rollback-ancestor"),
774	projectId: ProjectId.make("runtime-rollback-readiness-project"),
775	title: "Ancestor workspace owner",
776	modelSelection,
777	runtimeMode: "full-access",
778	interactionMode: "default",
779	branch: null,
780	worktreePath: path.dirname(process.cwd()),
781	});
782	const overlapCommandId = CommandId.make("runtime-rollback-overlap");
783	const overlapSequence = yield* orchestrator.getThreadEventSequence(threadId);
784	const overlap = yield* orchestrator
785	.dispatch({
786	type: "checkpoint.rollback",
787	commandId: overlapCommandId,
788	threadId,
789	checkpointId,
790	scopeId: scope.id,
791	})
792	.pipe(Effect.flip);
793	assert.match(String(overlap.cause), /isolated worktree/);
794	assert.equal(yield* orchestrator.getThreadEventSequence(threadId), overlapSequence);
795	assert.deepEqual(yield* outbox.listByCommandId(overlapCommandId), []);
796	yield* orchestrator.dispatch({
797	type: "thread.metadata.update",
798	commandId: CommandId.make("runtime-rollback-share"),
799	threadId,
800	worktreePath: null,
801	});
802	const sharedCommandId = CommandId.make("runtime-rollback-shared");
803	const sequence = yield* orchestrator.getThreadEventSequence(threadId);
804	const shared = yield* orchestrator
805	.dispatch({
806	type: "checkpoint.rollback",
807	commandId: sharedCommandId,
808	threadId,
809	checkpointId,
810	scopeId: scope.id,
811	})
812	.pipe(Effect.flip);
813	assert.match(String(shared.cause), /isolated worktree/);
814	assert.equal(yield* orchestrator.getThreadEventSequence(threadId), sequence);
815	assert.deepEqual(yield* outbox.listByCommandId(sharedCommandId), []);
816	const conversationOnly = yield* orchestrator.dispatch({
817	type: "checkpoint.rollback",
818	commandId: CommandId.make("runtime-rollback-conversation"),
819	threadId,
820	checkpointId,
821	scopeId: scope.id,
822	restoreFiles: false,
823	});
824	assert.deepEqual(
825	conversationOnly.storedEvents.map((stored) => stored.event.type),
826	["thread.metadata-updated", "checkpoint.rollback-requested"],
827	);
828	} else {
829	const error = yield* rollback.pipe(Effect.flip);
830	assert.instanceOf(error, Orchestrator.OrchestratorDispatchError);
831	assert.equal(
832	error.cause,
833	`Checkpoint ${checkpointId} is ${status} and cannot be restored.`,
834	);
835	assert.equal(yield* orchestrator.getThreadEventSequence(threadId), previousSequence);
836	assert.deepEqual(
837	yield* eventSink.readByCommandId({ commandId }).pipe(Stream.runCollect),
838	[],
839	);
840	assert.deepEqual(yield* outbox.listByCommandId(commandId), []);
841	}
842	}
843	}).pipe(Effect.provide(Layer.fresh(layerTest))),
844	);
845
846	it.effect("resolves delivery intent against the active run and starts after it completes", () =>
847	Effect.gen(function* () {
848	const orchestrator = yield* Orchestrator.OrchestratorV2;
849	const eventSink = yield* EventSink.EventSinkV2;
850	const outbox = yield* EffectOutbox.EffectOutboxV2;
851	const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
852	const threadId = ThreadId.make("runtime-delivery-intent");
853	yield* orchestrator.dispatch({
854	type: "thread.create",
855	createdBy: "user",
856	creationSource: "web",
857	commandId: CommandId.make("runtime-delivery-intent-create"),
858	threadId,
859	projectId: ProjectId.make("runtime-delivery-intent-project"),
860	title: "Delivery intent",
861	modelSelection,
862	runtimeMode: "full-access",
863	interactionMode: "default",
864	branch: null,
865	worktreePath: process.cwd(),
866	});
867	yield* orchestrator.dispatch({
868	type: "message.dispatch",
869	createdBy: "user",
870	creationSource: "web",
871	commandId: CommandId.make("runtime-delivery-intent-first"),
872	threadId,
873	messageId: MessageId.make("runtime-delivery-intent-first"),
874	text: "Start work.",
875	attachments: [],
876	dispatchMode: { type: "start_immediately" },
877	});
878	const initial = yield* orchestrator.getThreadProjection(threadId);
879	const run = initial.runs[0]!;
880	const providerThread = initial.providerThreads[0]!;
881	const now = yield* DateTime.now;
882	const providerSession = {
883	id: providerThread.providerSessionId!,
884	driver,
885	providerInstanceId: modelSelection.instanceId,
886	status: "running" as const,
887	cwd: process.cwd(),
888	model: modelSelection.model,
889	capabilities: CodexProviderCapabilitiesV2,
890	createdAt: now,
891	updatedAt: now,
892	lastError: null,
893	};
894	const providerTurn = {
895	id: ProviderTurnId.make("runtime-delivery-intent-turn"),
896	providerThreadId: providerThread.id,
897	nodeId: run.rootNodeId!,
898	runAttemptId: run.activeAttemptId,
899	nativeTurnRef: null,
900	ordinal: 1,
901	status: "running" as const,
902	startedAt: now,
903	completedAt: null,
904	};
905	yield* eventSink.write({
906	commandId: CommandId.make("runtime-delivery-intent-running"),
907	events: [
908	{
909	id: EventId.make("runtime-delivery-intent-run-event"),
910	type: "run.updated",
911	threadId,
912	runId: run.id,
913	occurredAt: now,
914	payload: { ...run, status: "running", startedAt: now },
915	},
916	{
917	id: EventId.make("runtime-delivery-intent-session-event"),
918	type: "provider-session.attached",
919	threadId,
920	occurredAt: now,
921	payload: providerSession,
922	},
923	{
924	id: EventId.make("runtime-delivery-intent-turn-event"),
925	type: "provider-turn.updated",
926	threadId,
927	runId: run.id,
928	occurredAt: now,
929	payload: providerTurn,
930	},
931	],
932	});
933	const sessionSpy = vi
934	.spyOn(sessions, "get")
935	.mockReturnValue(
936	Effect.succeed(
937	Option.some({ providerSession } as ProviderAdapter.ProviderAdapterV2SessionRuntime),
938	),
939	);
940	yield* Effect.addFinalizer(() => Effect.sync(() => sessionSpy.mockRestore()));
941
942	const steerCommandId = CommandId.make("runtime-delivery-intent-auto");
943	const steerMessageId = MessageId.make("runtime-delivery-intent-auto");
944	yield* orchestrator.dispatch({
945	type: "message.dispatch",
946	createdBy: "user",
947	creationSource: "web",
948	commandId: steerCommandId,
949	threadId,
950	messageId: steerMessageId,
951	text: "Include this in the active work.",
952	attachments: [],
953	dispatchMode: { type: "start_immediately" },
954	deliveryIntent: "auto",
955	});
956	const steered = yield* orchestrator.getThreadProjection(threadId);
957	assert.lengthOf(steered.runs, 1);
958	assert.equal(
959	steered.messages.find((message) => message.id === steerMessageId)?.runId,
960	run.id,
961	);
962	assert.deepEqual(
963	(yield* outbox.listByCommandId(steerCommandId)).map((effect) => effect.request),
964	[
965	{
966	type: "provider-turn.steer",
967	providerSessionId: providerSession.id,
968	providerThreadId: providerThread.id,
969	providerTurnId: providerTurn.id,
970	messageId: steerMessageId,
971	},
972	],
973	);
974
975	yield* eventSink.write({
976	commandId: CommandId.make("runtime-delivery-intent-completed"),
977	events: [
978	{
979	id: EventId.make("runtime-delivery-intent-run-completed"),
980	type: "run.updated",
981	threadId,
982	runId: run.id,
983	occurredAt: now,
984	payload: { ...run, status: "completed", startedAt: now, completedAt: now },
985	},
986	{
987	id: EventId.make("runtime-delivery-intent-turn-completed"),
988	type: "provider-turn.updated",
989	threadId,
990	runId: run.id,
991	occurredAt: now,
992	payload: { ...providerTurn, status: "completed", completedAt: now },
993	},
994	],
995	});
996	const nextCommandId = CommandId.make("runtime-delivery-intent-next");
997	yield* orchestrator.dispatch({
998	type: "message.dispatch",
999	createdBy: "user",
1000	creationSource: "web",
1001	commandId: nextCommandId,
1002	threadId,
1003	messageId: MessageId.make("runtime-delivery-intent-next"),
1004	text: "The previous run finished before this arrived.",
1005	attachments: [],
1006	dispatchMode: { type: "start_immediately" },
1007	deliveryIntent: "restart",
1008	});
1009	const restarted = yield* orchestrator.getThreadProjection(threadId);
1010	assert.deepEqual(
1011	restarted.runs.map((candidate) => candidate.status),
1012	["completed", "starting"],
1013	);
1014	assert.deepEqual(
1015	(yield* outbox.listByCommandId(nextCommandId)).map((effect) => effect.request.type),
1016	["provider-turn.start"],
1017	);
1018	}),
1019	);
1020
1021	it.effect("answers an async question after its provider exits and commits the answer once", () =>
1022	Effect.gen(function* () {
1023	const orchestrator = yield* Orchestrator.OrchestratorV2;
1024	const eventSink = yield* EventSink.EventSinkV2;
1025	const now = yield* DateTime.now;
1026	const threadId = ThreadId.make("runtime-async-question");
1027	const requestId = RuntimeRequestId.make("runtime-async-question-request");
1028	const nodeId = NodeId.make("runtime-async-question-node");
1029	const itemId = TurnItemId.make("runtime-async-question-item");
1030	yield* orchestrator.dispatch({
1031	type: "thread.create",
1032	commandId: CommandId.make("runtime-async-question-create"),
1033	createdBy: "user",
1034	creationSource: "web",
1035	threadId,
1036	projectId: ProjectId.make("runtime-async-question-project"),
1037	title: "Async question",
1038	modelSelection,
1039	runtimeMode: "full-access",
1040	interactionMode: "default",
1041	branch: null,
1042	worktreePath: process.cwd(),
1043	});
1044	yield* eventSink.write({
1045	commandId: CommandId.make("runtime-async-question-seed"),
1046	events: [
1047	{
1048	id: EventId.make("runtime-async-question-node-event"),
1049	type: "node.updated",
1050	threadId,
1051	nodeId,
1052	occurredAt: now,
1053	payload: {
1054	id: nodeId,
1055	threadId,
1056	runId: null,
1057	parentNodeId: null,
1058	rootNodeId: nodeId,
1059	kind: "user_input_request",
1060	status: "waiting",
1061	countsForRun: false,
1062	providerThreadId: null,
1063	providerTurnId: null,
1064	nativeItemRef: null,
1065	runtimeRequestId: requestId,
1066	checkpointScopeId: null,
1067	startedAt: now,
1068	completedAt: null,
1069	},
1070	},
1071	{
1072	id: EventId.make("runtime-async-question-request-event"),
1073	type: "runtime-request.updated",
1074	threadId,
1075	nodeId,
1076	occurredAt: now,
1077	payload: {
1078	id: requestId,
1079	nodeId,
1080	providerTurnId: null,
1081	nativeRequestRef: null,
1082	kind: "user_input",
1083	status: "pending",
1084	responseCapability: { type: "message" },
1085	createdAt: now,
1086	resolvedAt: null,
1087	},
1088	},
1089	{
1090	id: EventId.make("runtime-async-question-item-event"),
1091	type: "turn-item.updated",
1092	threadId,
1093	nodeId,
1094	occurredAt: now,
1095	payload: {
1096	id: itemId,
1097	type: "user_input_request",
1098	threadId,
1099	runId: null,
1100	nodeId,
1101	providerThreadId: null,
1102	providerTurnId: null,
1103	nativeItemRef: null,
1104	parentItemId: null,
1105	ordinal: 0,
1106	status: "waiting",
1107	title: null,
1108	startedAt: now,
1109	completedAt: null,
1110	updatedAt: now,
1111	requestId,
1112	responseMode: "message",
1113	questions: [{ id: "color", header: "Color", question: "Which color?", options: [] }],
1114	},
1115	},
1116	],
1117	});
1118	const invalid = yield* orchestrator
1119	.dispatch({
1120	type: "runtime-request.respond",
1121	commandId: CommandId.make("runtime-async-question-blank"),
1122	threadId,
1123	requestId,
1124	answers: { color: " " },
1125	})
1126	.pipe(Effect.result);
1127	assert.equal(invalid._tag, "Failure");
1128	const unanswered = yield* orchestrator.getThreadProjection(threadId);
1129	assert.equal(unanswered.runtimeRequests[0]?.status, "pending");
1130	assert.deepEqual(unanswered.messages, []);
1131
1132	const command = {
1133	type: "runtime-request.respond" as const,
1134	commandId: CommandId.make("runtime-async-question-answer"),
1135	threadId,
1136	requestId,
1137	answers: { color: " Blue " },
1138	};
1139	const accepted = yield* orchestrator.dispatch(command);
1140	const repeated = yield* orchestrator.dispatch(command);
1141	assert.equal(repeated.sequence, accepted.sequence);
1142	const answered = yield* orchestrator.getThreadProjection(threadId);
1143	assert.equal(answered.runtimeRequests[0]?.status, "resolved");
1144	assert.deepEqual(answered.runtimeRequests[0]?.answers, command.answers);
1145	assert.equal(answered.nodes.find((node) => node.id === nodeId)?.status, "completed");
1146	assert.equal(answered.turnItems.find((item) => item.id === itemId)?.status, "completed");
1147	const answeredItem = answered.turnItems.find((item) => item.id === itemId);
1148	assert.equal(answeredItem?.type, "user_input_request");
1149	if (answeredItem?.type === "user_input_request") {
1150	assert.deepEqual(answeredItem.questionAnswer, {
1151	requestId,
1152	answers: command.answers,
1153	attachmentsByQuestionId: {},
1154	questionTextById: { color: "Which color?" },
1155	});
1156	}
1157	assert.equal(answered.messages.length, 1);
1158	assert.equal(answered.messages[0]?.text, "Which color?\nBlue");
1159	assert.equal(answered.messages[0]?.role, "user");
1160	assert.equal(answered.runs.length, 1);
1161
1162	const duplicate = yield* orchestrator
1163	.dispatch({
1164	...command,
1165	commandId: CommandId.make("runtime-async-question-duplicate"),
1166	})
1167	.pipe(Effect.result);
1168	assert.equal(duplicate._tag, "Failure");
1169	assert.equal((yield* orchestrator.getThreadProjection(threadId)).messages.length, 1);
1170	}),
1171	);
1172
1173	it.effect("dismisses message-capable questions directly and while settling", () =>
1174	Effect.gen(function* () {
1175	const orchestrator = yield* Orchestrator.OrchestratorV2;
1176	const eventSink = yield* EventSink.EventSinkV2;
1177
1178	const seedQuestion = Effect.fn("runtimeLayerTest.seedQuestion")(function* (name: string) {
1179	const threadId = ThreadId.make(`${name}-thread`);
1180	const requestId = RuntimeRequestId.make(`${name}-request`);
1181	const nodeId = NodeId.make(`${name}-node`);
1182	const itemId = TurnItemId.make(`${name}-item`);
1183	const now = yield* DateTime.now;
1184	yield* orchestrator.dispatch({
1185	type: "thread.create",
1186	commandId: CommandId.make(`${name}-create`),
1187	createdBy: "user",
1188	creationSource: "web",
1189	threadId,
1190	projectId: ProjectId.make(`${name}-project`),
1191	title: "Dismissible question",
1192	modelSelection,
1193	runtimeMode: "full-access",
1194	interactionMode: "default",
1195	branch: null,
1196	worktreePath: process.cwd(),
1197	});
1198	yield* eventSink.write({
1199	commandId: CommandId.make(`${name}-seed`),
1200	events: [
1201	{
1202	id: EventId.make(`${name}-node-event`),
1203	type: "node.updated",
1204	threadId,
1205	nodeId,
1206	occurredAt: now,
1207	payload: {
1208	id: nodeId,
1209	threadId,
1210	runId: null,
1211	parentNodeId: null,
1212	rootNodeId: nodeId,
1213	kind: "user_input_request",
1214	status: "waiting",
1215	countsForRun: false,
1216	providerThreadId: null,
1217	providerTurnId: null,
1218	nativeItemRef: null,
1219	runtimeRequestId: requestId,
1220	checkpointScopeId: null,
1221	startedAt: now,
1222	completedAt: null,
1223	},
1224	},
1225	{
1226	id: EventId.make(`${name}-request-event`),
1227	type: "runtime-request.updated",
1228	threadId,
1229	nodeId,
1230	occurredAt: now,
1231	payload: {
1232	id: requestId,
1233	nodeId,
1234	providerTurnId: null,
1235	nativeRequestRef: null,
1236	kind: "user_input",
1237	status: "pending",
1238	responseCapability: { type: "message" },
1239	createdAt: now,
1240	resolvedAt: null,
1241	},
1242	},
1243	{
1244	id: EventId.make(`${name}-item-event`),
1245	type: "turn-item.updated",
1246	threadId,
1247	nodeId,
1248	occurredAt: now,
1249	payload: {
1250	id: itemId,
1251	type: "user_input_request",
1252	threadId,
1253	runId: null,
1254	nodeId,
1255	providerThreadId: null,
1256	providerTurnId: null,
1257	nativeItemRef: null,
1258	parentItemId: null,
1259	ordinal: 0,
1260	status: "waiting",
1261	title: null,
1262	startedAt: now,
1263	completedAt: null,
1264	updatedAt: now,
1265	requestId,
1266	responseMode: "message",
1267	questions: [{ id: "choice", header: "Choice", question: "Continue?", options: [] }],
1268	},
1269	},
1270	],
1271	});
1272	return { threadId, requestId, nodeId, itemId };
1273	});
1274
1275	const dismissed = yield* seedQuestion("runtime-dismiss-question");
1276	yield* orchestrator.dispatch({
1277	type: "thread.user-input.dismiss",
1278	commandId: CommandId.make("runtime-dismiss-question-command"),
1279	threadId: dismissed.threadId,
1280	requestId: dismissed.requestId,
1281	});
1282	const dismissedProjection = yield* orchestrator.getThreadProjection(dismissed.threadId);
1283	assert.equal(dismissedProjection.runtimeRequests[0]?.status, "resolved");
1284	assert.equal(dismissedProjection.runtimeRequests[0]?.decision, "cancel");
1285	assert.equal(
1286	dismissedProjection.nodes.find((node) => node.id === dismissed.nodeId)?.status,
1287	"cancelled",
1288	);
1289	assert.equal(
1290	dismissedProjection.turnItems.find((item) => item.id === dismissed.itemId)?.status,
1291	"cancelled",
1292	);
1293	assert.lengthOf(dismissedProjection.messages, 0);
1294
1295	const settled = yield* seedQuestion("runtime-settle-question");
1296	yield* orchestrator.dispatch({
1297	type: "thread.settle",
1298	commandId: CommandId.make("runtime-settle-question-command"),
1299	threadId: settled.threadId,
1300	});
1301	const settledProjection = yield* orchestrator.getThreadProjection(settled.threadId);
1302	assert.equal(settledProjection.thread.settledOverride, "settled");
1303	assert.equal(settledProjection.runtimeRequests[0]?.status, "resolved");
1304	assert.equal(settledProjection.runtimeRequests[0]?.decision, "cancel");
1305	assert.equal(
1306	settledProjection.nodes.find((node) => node.id === settled.nodeId)?.status,
1307	"cancelled",
1308	);
1309	assert.equal(
1310	settledProjection.turnItems.find((item) => item.id === settled.itemId)?.status,
1311	"cancelled",
1312	);
1313	}),
1314	);
1315
1316	it.effect("merges an explicit provider-finished run while checkpoint capture is pending", () =>
1317	Effect.gen(function* () {
1318	const orchestrator = yield* Orchestrator.OrchestratorV2;
1319	const eventSink = yield* EventSink.EventSinkV2;
1320	const now = yield* DateTime.now;
1321	const projectId = ProjectId.make("runtime-layer-waiting-merge-project");
1322	const targetThreadId = ThreadId.make("runtime-layer-waiting-merge-target");
1323	const sourceThreadId = ThreadId.make("runtime-layer-waiting-merge-source");
1324	const baseRunId = RunId.make("runtime-layer-waiting-merge-base-run");
1325	const sourceRunId = RunId.make("runtime-layer-waiting-merge-source-run");
1326	const sourceProviderThreadId = ProviderThreadId.make(
1327	"runtime-layer-waiting-merge-provider-thread",
1328	);
1329	const forkTransferId = ContextTransferId.make("runtime-layer-waiting-merge-fork-transfer");
1330
1331	yield* orchestrator.dispatch({
1332	type: "thread.create",
1333	createdBy: "user",
1334	creationSource: "web",
1335	commandId: CommandId.make("runtime-layer-waiting-merge-create-target"),
1336	threadId: targetThreadId,
1337	projectId,
1338	title: "Waiting merge target",
1339	modelSelection,
1340	runtimeMode: "full-access",
1341	interactionMode: "default",
1342	branch: null,
1343	worktreePath: null,
1344	});
1345	const target = yield* orchestrator.getThreadProjection(targetThreadId);
1346
1347	yield* eventSink.write({
1348	commandId: CommandId.make("runtime-layer-waiting-merge-seed"),
1349	events: [
1350	{
1351	id: EventId.make("runtime-layer-waiting-merge-source-thread-event"),
1352	type: "thread.created",
1353	threadId: sourceThreadId,
1354	providerInstanceId: modelSelection.instanceId,
1355	occurredAt: now,
1356	payload: {
1357	...target.thread,
1358	id: sourceThreadId,
1359	title: "Waiting merge source",
1360	activeProviderThreadId: null,
1361	lineage: {
1362	parentThreadId: targetThreadId,
1363	relationshipToParent: "fork",
1364	rootThreadId: targetThreadId,
1365	},
1366	forkedFrom: {
1367	type: "run",
1368	threadId: targetThreadId,
1369	runId: baseRunId,
1370	},
1371	createdAt: now,
1372	updatedAt: now,
1373	},
1374	},
1375	{
1376	id: EventId.make("runtime-layer-waiting-merge-fork-transfer-event"),
1377	type: "context-transfer.created",
1378	threadId: sourceThreadId,
1379	providerInstanceId: modelSelection.instanceId,
1380	occurredAt: now,
1381	payload: {
1382	id: forkTransferId,
1383	type: "fork",
1384	sourceThreadId: targetThreadId,
1385	targetThreadId: sourceThreadId,
1386	sourcePoint: { threadId: targetThreadId, runId: baseRunId },
1387	basePoint: null,
1388	sourceProviderInstanceId: modelSelection.instanceId,
1389	targetProviderInstanceId: modelSelection.instanceId,
1390	targetRunId: null,
1391	status: "consumed",
1392	resolution: null,
1393	createdBy: "user",
1394	error: null,
1395	createdAt: now,
1396	updatedAt: now,
1397	consumedAt: now,
1398	},
1399	},
1400	{
1401	id: EventId.make("runtime-layer-waiting-merge-provider-thread-event"),
1402	type: "provider-thread.updated",
1403	threadId: sourceThreadId,
1404	driver,
1405	providerInstanceId: modelSelection.instanceId,
1406	occurredAt: now,
1407	payload: {
1408	id: sourceProviderThreadId,
1409	driver,
1410	providerInstanceId: modelSelection.instanceId,
1411	providerSessionId: null,
1412	appThreadId: sourceThreadId,
1413	ownerNodeId: null,
1414	nativeThreadRef: {
1415	driver,
1416	nativeId: "native-waiting-merge-source",
1417	strength: "strong",
1418	},
1419	nativeConversationHeadRef: null,
1420	status: "idle",
1421	firstRunOrdinal: 1,
1422	lastRunOrdinal: 1,
1423	handoffIds: [],
1424	forkedFrom: null,
1425	createdAt: now,
1426	updatedAt: now,
1427	},
1428	},
1429	{
1430	id: EventId.make("runtime-layer-waiting-merge-source-run-event"),
1431	type: "run.created",
1432	threadId: sourceThreadId,
1433	runId: sourceRunId,
1434	providerInstanceId: modelSelection.instanceId,
1435	occurredAt: now,
1436	payload: {
1437	id: sourceRunId,
1438	threadId: sourceThreadId,
1439	ordinal: 1,
1440	providerInstanceId: modelSelection.instanceId,
1441	modelSelection,
1442	providerThreadId: sourceProviderThreadId,
1443	userMessageId: MessageId.make("runtime-layer-waiting-merge-message"),
1444	rootNodeId: null,
1445	activeAttemptId: null,
1446	status: "waiting",
1447	queuePosition: null,
1448	requestedAt: now,
1449	startedAt: now,
1450	completedAt: null,
1451	checkpointId: null,
1452	contextHandoffId: null,
1453	},
1454	},
1455	],
1456	});
1457
1458	yield* orchestrator.dispatch({
1459	type: "thread.merge_back",
1460	createdBy: "user",
1461	creationSource: "mobile",
1462	commandId: CommandId.make("runtime-layer-waiting-merge"),
1463	sourceThreadId,
1464	targetThreadId,
1465	sourcePoint: { type: "run", runId: sourceRunId },
1466	createdAt: now,
1467	});
1468
1469	const mergedTarget = yield* orchestrator.getThreadProjection(targetThreadId);
1470	const transfer = mergedTarget.contextTransfers.find(
1471	(candidate) => candidate.type === "merge_back",
1472	);
1473	assert.isDefined(transfer);
1474	assert.equal(transfer.status, "pending");
1475	assert.equal(transfer.sourceThreadId, sourceThreadId);
1476	assert.equal(transfer.targetThreadId, targetThreadId);
1477	assert.equal(transfer.sourcePoint.runId, sourceRunId);
1478	assert.isUndefined(transfer.sourcePoint.checkpointId);
1479	assert.equal(transfer.sourcePoint.providerThreadRef?.nativeId, "native-waiting-merge-source");
1480	assert.equal(transfer.basePoint?.runId, baseRunId);
1481	assert.isNull(transfer.error);
1482	}),
1483	);
1484	});
1485
1486	it.layer(layerLegacyImportTest)("OrchestrationV2 legacy import", (it) => {
1487	it.effect("hydrates imported transcripts before commands and propagates hydration failures", () =>
1488	Effect.gen(function* () {
1489	const sql = yield* SqlClient.SqlClient;
1490	const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
1491	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
1492	const orchestrator = yield* Orchestrator.OrchestratorV2;
1493	const threadManagement = yield* ThreadManagementService.ThreadManagementService;
1494	const metadataThreadId = ThreadId.make("runtime-layer-legacy-metadata-thread");
1495	const failureThreadId = ThreadId.make("runtime-layer-legacy-failure-thread");
1496	const projectId = ProjectId.make("runtime-layer-legacy-project");
1497
1498	yield* sql`
1499	INSERT INTO projection_projects (
1500	project_id,
1501	title,
1502	workspace_root,
1503	default_model_selection_json,
1504	scripts_json,
1505	created_at,
1506	updated_at,
1507	deleted_at
1508	) VALUES (
1509	${projectId},
1510	'Legacy project',
1511	'/tmp/runtime-layer-legacy-project',
1512	'{"instanceId":"codex","model":"gpt-5.4"}',
1513	'[]',
1514	'2026-01-01T00:00:00.000Z',
1515	'2026-01-04T00:00:00.000Z',
1516	NULL
1517	)
1518	`;
1519	yield* sql`
1520	INSERT INTO projection_threads (
1521	thread_id,
1522	project_id,
1523	title,
1524	model_selection_json,
1525	runtime_mode,
1526	interaction_mode,
1527	branch,
1528	worktree_path,
1529	latest_turn_id,
1530	created_at,
1531	updated_at,
1532	archived_at,
1533	settled_override,
1534	settled_at,
1535	deleted_at
1536	) VALUES
1537	(
1538	${metadataThreadId},
1539	${projectId},
1540	'Legacy metadata title',
1541	'{"instanceId":"codex","model":"gpt-5.4"}',
1542	'full-access',
1543	'default',
1544	'main',
1545	'/tmp/runtime-layer-legacy-project',
1546	NULL,
1547	'2026-01-01T00:00:00.000Z',
1548	'2026-01-04T00:00:00.000Z',
1549	NULL,
1550	NULL,
1551	NULL,
1552	NULL
1553	),
1554	(
1555	${failureThreadId},
1556	${projectId},
1557	'Legacy failure title',
1558	'{"instanceId":"codex","model":"gpt-5.4"}',
1559	'full-access',
1560	'default',
1561	'main',
1562	'/tmp/runtime-layer-legacy-project',
1563	NULL,
1564	'2026-01-01T00:00:00.000Z',
1565	'2026-01-04T00:00:00.000Z',
1566	NULL,
1567	NULL,
1568	NULL,
1569	NULL
1570	)
1571	`;
1572	yield* sql`
1573	INSERT INTO projection_thread_messages (
1574	message_id,
1575	thread_id,
1576	turn_id,
1577	role,
1578	text,
1579	attachments_json,
1580	is_streaming,
1581	created_at,
1582	updated_at
1583	) VALUES
1584	(
1585	'message:runtime-layer-legacy:1',
1586	${metadataThreadId},
1587	NULL,
1588	'user',
1589	'First imported question',
1590	'[]',
1591	0,
1592	'2026-01-01T01:00:00.000Z',
1593	'2026-01-01T01:00:00.000Z'
1594	),
1595	(
1596	'message:runtime-layer-legacy:2',
1597	${metadataThreadId},
1598	NULL,
1599	'assistant',
1600	'First imported answer',
1601	'[]',
1602	0,
1603	'2026-01-02T01:00:00.000Z',
1604	'2026-01-02T01:00:00.000Z'
1605	),
1606	(
1607	'message:runtime-layer-legacy:3',
1608	${metadataThreadId},
1609	NULL,
1610	'user',
1611	'Latest imported question',
1612	'[]',
1613	0,
1614	'2026-01-03T01:00:00.000Z',
1615	'2026-01-03T01:00:00.000Z'
1616	),
1617	(
1618	'message:runtime-layer-legacy:failure',
1619	${failureThreadId},
1620	NULL,
1621	'user',
1622	'Imported context must load before archive',
1623	'[]',
1624	0,
1625	'2026-01-03T01:00:00.000Z',
1626	'2026-01-03T01:00:00.000Z'
1627	)
1628	`;
1629
1630	yield* importer.reconcileShells;
1631	assert.isTrue((yield* maintenance.verify).valid);
1632
1633	yield* threadManagement.dispatch({
1634	type: "thread.metadata.update",
1635	commandId: CommandId.make("runtime-layer-legacy-metadata-update"),
1636	threadId: metadataThreadId,
1637	title: "Updated after import",
1638	});
1639	const updatedProjection = yield* threadManagement.getThreadProjection(metadataThreadId);
1640	assert.equal(updatedProjection.thread.title, "Updated after import");
1641	assert.deepEqual(
1642	updatedProjection.messages.map((message) => message.text),
1643	["First imported question", "First imported answer", "Latest imported question"],
1644	);
1645
1646	yield* sql`
1647	ALTER TABLE projection_thread_messages
1648	RENAME TO projection_thread_messages_unavailable
1649	`;
1650	const { projectionFailure, hydrationFailure } = yield* Effect.all({
1651	projectionFailure: threadManagement.getThreadProjection(failureThreadId).pipe(Effect.flip),
1652	hydrationFailure: threadManagement
1653	.dispatch({
1654	type: "thread.archive",
1655	commandId: CommandId.make("runtime-layer-legacy-failed-archive"),
1656	threadId: failureThreadId,
1657	})
1658	.pipe(Effect.flip),
1659	}).pipe(
1660	Effect.ensuring(
1661	sql`
1662	ALTER TABLE projection_thread_messages_unavailable
1663	RENAME TO projection_thread_messages
1664	`.pipe(Effect.orDie),
1665	),
1666	);
1667	assert.instanceOf(projectionFailure, Orchestrator.OrchestratorProjectionError);
1668	assert.instanceOf(projectionFailure.cause, LegacyV1ThreadImporter.LegacyV1ThreadImportError);
1669	assert.instanceOf(hydrationFailure, Orchestrator.OrchestratorDispatchError);
1670	assert.instanceOf(hydrationFailure.cause, LegacyV1ThreadImporter.LegacyV1ThreadImportError);
1671
1672	const projectionAfterFailure = yield* orchestrator.getThreadProjection(failureThreadId);
1673	assert.isNull(projectionAfterFailure.thread.archivedAt);
1674
1675	yield* threadManagement.dispatch({
1676	type: "thread.archive",
1677	commandId: CommandId.make("runtime-layer-legacy-retried-archive"),
1678	threadId: failureThreadId,
1679	});
1680	const projectionAfterRetry = yield* threadManagement.getThreadProjection(failureThreadId);
1681	assert.isNotNull(projectionAfterRetry.thread.archivedAt);
1682	}),
1683	);
1684	});
1685
1686	it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
1687	it.effect("applies lifecycle commands idempotently and emits archive/removal shell deltas", () =>
1688	Effect.gen(function* () {
1689	const orchestrator = yield* Orchestrator.OrchestratorV2;
1690	const threadId = ThreadId.make("runtime-layer-lifecycle-thread");
1691	const projectId = ProjectId.make("runtime-layer-lifecycle-project");
1692	const project = {
1693	projectId,
1694	title: "Lifecycle project",
1695	workspaceRoot: "/workspace/project",
1696	defaultModelSelection: null,
1697	createdAt: "2026-09-07T00:00:00.000Z",
1698	} as const;
1699	yield* seedProject(project);
1700	const create = {
1701	type: "thread.create" as const,
1702	createdBy: "user" as const,
1703	creationSource: "web" as const,
1704	commandId: CommandId.make("runtime-layer-lifecycle-create"),
1705	threadId,
1706	projectId,
1707	title: "Lifecycle thread",
1708	modelSelection,
1709	runtimeMode: "full-access" as const,
1710	interactionMode: "default" as const,
1711	branch: null,
1712	worktreePath: null,
1713	};
1714
1715	const firstCreate = yield* orchestrator.dispatch(create);
1716	const retriedCreate = yield* orchestrator.dispatch(create);
1717	assert.equal(retriedCreate.sequence, firstCreate.sequence);
1718	assert.lengthOf(retriedCreate.storedEvents, 1);
1719
1720	yield* orchestrator.dispatch({
1721	type: "thread.metadata.update",
1722	commandId: CommandId.make("runtime-layer-lifecycle-metadata"),
1723	threadId,
1724	title: "Renamed lifecycle thread",
1725	branch: "feature/v2",
1726	worktreePath: "/tmp/t3-v2-worktree",
1727	});
1728	const staleWorkspaceUpdate = yield* orchestrator
1729	.dispatch({
1730	type: "thread.metadata.update",
1731	commandId: CommandId.make("runtime-layer-lifecycle-stale-workspace"),
1732	threadId,
1733	branch: "feature/stale",
1734	worktreePath: "/tmp/stale-worktree",
1735	expectedWorktreePath: null,
1736	})
1737	.pipe(Effect.flip);
1738	assert.instanceOf(staleWorkspaceUpdate, Orchestrator.OrchestratorDispatchError);
1739	const projectionAfterStaleWorkspaceUpdate = yield* orchestrator.getThreadProjection(threadId);
1740	assert.equal(projectionAfterStaleWorkspaceUpdate.thread.branch, "feature/v2");
1741	assert.equal(projectionAfterStaleWorkspaceUpdate.thread.worktreePath, "/tmp/t3-v2-worktree");
1742	const pullRequestSnapshot = yield* orchestrator.getShellSnapshot();
1743	const pullRequest = {
1744	projectId,
1745	repository: "owner/repository",
1746	number: 24,
1747	url: "https://github.com/owner/repository/pull/24",
1748	};
1749	yield* moveProject(projectId, "/workspace/moved", "2026-09-07T00:01:00.000Z");
1750	const staleProjectWorkspace = yield* orchestrator
1751	.dispatch({
1752	type: "thread.pull-request.sync",
1753	commandId: CommandId.make("runtime-layer-lifecycle-pr-sync-stale-project-workspace"),
1754	threadId,
1755	projectId,
1756	snapshotSequence: pullRequestSnapshot.snapshotSequence,
1757	expected: {
1758	workspaceRoot: "/workspace/project",
1759	branch: "feature/v2",
1760	worktreePath: "/tmp/t3-v2-worktree",
1761	linkedPullRequest: null,
1762	branchPullRequest: null,
1763	},
1764	branchPullRequest: pullRequest,
1765	})
1766	.pipe(Effect.flip);
1767	assert.instanceOf(staleProjectWorkspace, Orchestrator.OrchestratorDispatchError);
1768	yield* moveProject(projectId, project.workspaceRoot, "2026-09-07T00:02:00.000Z");
1769	yield* orchestrator.dispatch({
1770	type: "thread.pull-request.sync",
1771	commandId: CommandId.make("runtime-layer-lifecycle-pr-sync"),
1772	threadId,
1773	projectId,
1774	snapshotSequence: pullRequestSnapshot.snapshotSequence,
1775	expected: {
1776	workspaceRoot: "/workspace/project",
1777	branch: "feature/v2",
1778	worktreePath: "/tmp/t3-v2-worktree",
1779	linkedPullRequest: null,
1780	branchPullRequest: null,
1781	},
1782	branchPullRequest: pullRequest,
1783	});
1784	assert.deepEqual(
1785	(yield* orchestrator.getThreadProjection(threadId)).thread.branchPullRequest,
1786	pullRequest,
1787	);
1788	const stalePullRequestSync = yield* orchestrator
1789	.dispatch({
1790	type: "thread.pull-request.sync",
1791	commandId: CommandId.make("runtime-layer-lifecycle-pr-sync-stale"),
1792	threadId,
1793	projectId,
1794	snapshotSequence: pullRequestSnapshot.snapshotSequence,
1795	expected: {
1796	workspaceRoot: "/workspace/project",
1797	branch: "feature/v2",
1798	worktreePath: "/tmp/t3-v2-worktree",
1799	linkedPullRequest: null,
1800	branchPullRequest: null,
1801	},
1802	branchPullRequest: null,
1803	})
1804	.pipe(Effect.flip);
1805	assert.instanceOf(stalePullRequestSync, Orchestrator.OrchestratorDispatchError);
1806	yield* orchestrator.dispatch({
1807	type: "thread.runtime-mode.set",
1808	commandId: CommandId.make("runtime-layer-lifecycle-runtime"),
1809	threadId,
1810	runtimeMode: "approval-required",
1811	});
1812	yield* orchestrator.dispatch({
1813	type: "thread.interaction-mode.set",
1814	commandId: CommandId.make("runtime-layer-lifecycle-interaction"),
1815	threadId,
1816	interactionMode: "plan",
1817	});
1818	yield* orchestrator.dispatch({
1819	type: "thread.model-selection.set",
1820	commandId: CommandId.make("runtime-layer-lifecycle-model"),
1821	threadId,
1822	modelSelection: { ...modelSelection, model: "gpt-5.5" },
1823	});
1824	yield* orchestrator.dispatch({
1825	type: "thread.active.reorder",
1826	commandId: CommandId.make("runtime-layer-lifecycle-active-order"),
1827	threadId,
1828	orderKey: "a0",
1829	});
1830	assert.equal((yield* orchestrator.getThreadProjection(threadId)).thread.activeOrderKey, "a0");
1831
1832	// Automatic settlement (#8600): a stale snapshot loses to any change
1833	// made after it, and a fresh one settles like a user settle would.
1834	const preAutoProjection = yield* orchestrator.getThreadProjection(threadId);
1835	const staleAutoSettle = yield* orchestrator
1836	.dispatch({
1837	type: "thread.auto-settle",
1838	commandId: CommandId.make("runtime-layer-lifecycle-auto-settle-stale"),
1839	threadId,
1840	snapshotAt: DateTime.makeUnsafe(
1841	DateTime.toEpochMillis(preAutoProjection.thread.updatedAt) - 1,
1842	),
1843	})
1844	.pipe(Effect.flip);
1845	assert.instanceOf(staleAutoSettle, Orchestrator.OrchestratorDispatchError);
1846	yield* orchestrator.dispatch({
1847	type: "thread.auto-settle",
1848	commandId: CommandId.make("runtime-layer-lifecycle-auto-settle"),
1849	threadId,
1850	snapshotAt: preAutoProjection.thread.updatedAt,
1851	});
1852	const autoSettledProjection = yield* orchestrator.getThreadProjection(threadId);
1853	assert.equal(autoSettledProjection.thread.settledOverride, "settled");
1854	assert.isNull(autoSettledProjection.thread.activeOrderKey);
1855	yield* orchestrator.dispatch({
1856	type: "thread.unsettle",
1857	commandId: CommandId.make("runtime-layer-lifecycle-auto-unsettle"),
1858	threadId,
1859	reason: "user",
1860	});
1861	// An explicit un-settle outranks the sweep even with a fresh snapshot.
1862	const postUnsettleProjection = yield* orchestrator.getThreadProjection(threadId);
1863	const overriddenAutoSettle = yield* orchestrator
1864	.dispatch({
1865	type: "thread.auto-settle",
1866	commandId: CommandId.make("runtime-layer-lifecycle-auto-settle-overridden"),
1867	threadId,
1868	snapshotAt: postUnsettleProjection.thread.updatedAt,
1869	})
1870	.pipe(Effect.flip);
1871	assert.instanceOf(overriddenAutoSettle, Orchestrator.OrchestratorDispatchError);
1872
1873	yield* orchestrator.dispatch({
1874	type: "thread.settle",
1875	commandId: CommandId.make("runtime-layer-lifecycle-settle"),
1876	threadId,
1877	});
1878	const settledProjection = yield* orchestrator.getThreadProjection(threadId);
1879	assert.equal(settledProjection.thread.settledOverride, "settled");
1880	assert.isNotNull(settledProjection.thread.settledAt);
1881
1882	yield* orchestrator.dispatch({
1883	type: "thread.unsettle",
1884	commandId: CommandId.make("runtime-layer-lifecycle-unsettle"),
1885	threadId,
1886	reason: "user",
1887	});
1888	const activeProjection = yield* orchestrator.getThreadProjection(threadId);
1889	assert.equal(activeProjection.thread.settledOverride, "active");
1890	assert.isNull(activeProjection.thread.settledAt);
1891	assert.isNotNull(activeProjection.thread.unsettledAt);
1892	const activeShell = (yield* orchestrator.getShellSnapshot()).threads.find(
1893	(thread) => thread.id === threadId,
1894	);
1895	assert.deepEqual(activeShell?.unsettledAt, activeProjection.thread.unsettledAt);
1896
1897	const archive = yield* orchestrator.dispatch({
1898	type: "thread.archive",
1899	commandId: CommandId.make("runtime-layer-lifecycle-archive"),
1900	threadId,
1901	});
1902	const archivedShell = yield* orchestrator.getShellSnapshot();
1903	assert.notInclude(
1904	archivedShell.threads.map((thread) => thread.id),
1905	threadId,
1906	);
1907	assert.include(
1908	archivedShell.archivedThreads.map((thread) => thread.id),
1909	threadId,
1910	);
1911	const activeOnlyShell = yield* orchestrator.getShellSnapshot({ location: "active" });
1912	assert.notInclude(
1913	activeOnlyShell.threads.map((thread) => thread.id),
1914	threadId,
1915	);
1916	assert.lengthOf(activeOnlyShell.archivedThreads, 0);
1917	const archiveOnlyShell = yield* orchestrator.getShellSnapshot({ location: "archive" });
1918	assert.lengthOf(archiveOnlyShell.threads, 0);
1919	assert.include(
1920	archiveOnlyShell.archivedThreads.map((thread) => thread.id),
1921	threadId,
1922	);
1923	assert.deepEqual(
1924	shellStreamItemFromThreadShell({
1925	stored: archive.storedEvents[0]!,
1926	shell: yield* orchestrator.getThreadShell(threadId),
1927	}),
1928	{
1929	kind: "thread.removed",
1930	sequence: archive.sequence,
1931	location: "active",
1932	threadId,
1933	},
1934	);
1935
1936	const remove = yield* orchestrator.dispatch({
1937	type: "thread.delete",
1938	commandId: CommandId.make("runtime-layer-lifecycle-delete"),
1939	threadId,
1940	});
1941	const deletedShell = yield* orchestrator.getShellSnapshot();
1942	assert.notInclude(
1943	deletedShell.threads.map((thread) => thread.id),
1944	threadId,
1945	);
1946	assert.notInclude(
1947	deletedShell.archivedThreads.map((thread) => thread.id),
1948	threadId,
1949	);
1950	assert.deepEqual(
1951	shellStreamItemFromThreadShell({
1952	stored: remove.storedEvents[0]!,
1953	shell: yield* orchestrator.getThreadShell(threadId),
1954	}),
1955	{
1956	kind: "thread.removed",
1957	sequence: remove.sequence,
1958	location: "active",
1959	threadId,
1960	},
1961	);
1962
1963	const projection = yield* orchestrator.getThreadProjection(threadId);
1964	assert.equal(projection.thread.title, "Renamed lifecycle thread");
1965	assert.equal(projection.thread.branch, "feature/v2");
1966	assert.equal(projection.thread.worktreePath, "/tmp/t3-v2-worktree");
1967	assert.equal(projection.thread.runtimeMode, "approval-required");
1968	assert.equal(projection.thread.interactionMode, "plan");
1969	assert.equal(projection.thread.modelSelection.model, "gpt-5.5");
1970	assert.isNotNull(projection.thread.archivedAt);
1971	assert.isNotNull(projection.thread.deletedAt);
1972	}),
1973	);
1974
1975	it.effect("persists linked pull requests through projection rebuilds and unlinking", () =>
1976	Effect.gen(function* () {
1977	const orchestrator = yield* Orchestrator.OrchestratorV2;
1978	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
1979	const threadId = ThreadId.make("runtime-layer-linked-pull-request-thread");
1980	const linkedPullRequest = {
1981	projectId: ProjectId.make("runtime-layer-linked-pull-request-project"),
1982	repository: "pingdotgg/t3code",
1983	number: 8160,
1984	url: "https://github.com/pingdotgg/t3code/pull/8160",
1985	} as const;
1986
1987	yield* orchestrator.dispatch({
1988	type: "thread.create",
1989	createdBy: "user",
1990	creationSource: "web",
1991	commandId: CommandId.make("runtime-layer-linked-pull-request-create"),
1992	threadId,
1993	projectId: linkedPullRequest.projectId,
1994	title: "Linked pull request thread",
1995	modelSelection,
1996	runtimeMode: "full-access",
1997	interactionMode: "default",
1998	branch: null,
1999	worktreePath: null,
2000	});
2001	yield* orchestrator.dispatch({
2002	type: "thread.metadata.update",
2003	commandId: CommandId.make("runtime-layer-linked-pull-request-link"),
2004	threadId,
2005	linkedPullRequest,
2006	});
2007
2008	assert.deepEqual(
2009	(yield* orchestrator.getThreadProjection(threadId)).thread.linkedPullRequest,
2010	linkedPullRequest,
2011	);
2012	const linkedShell = yield* orchestrator.getThreadShell(threadId);
2013	assert.isNotNull(linkedShell);
2014	assert.deepEqual(linkedShell.linkedPullRequest, linkedPullRequest);
2015
2016	const rebuilt = yield* maintenance.rebuild;
2017	assert.isTrue(rebuilt.valid);
2018	assert.deepEqual(
2019	(yield* orchestrator.getThreadProjection(threadId)).thread.linkedPullRequest,
2020	linkedPullRequest,
2021	);
2022
2023	yield* orchestrator.dispatch({
2024	type: "thread.metadata.update",
2025	commandId: CommandId.make("runtime-layer-linked-pull-request-unlink"),
2026	threadId,
2027	linkedPullRequest: null,
2028	});
2029	assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.linkedPullRequest);
2030	const unlinkedShell = yield* orchestrator.getThreadShell(threadId);
2031	assert.isNotNull(unlinkedShell);
2032	assert.isNull(unlinkedShell.linkedPullRequest);
2033	}),
2034	);
2035
2036	it.effect("keeps the branch pull request when linking another pull request", () =>
2037	Effect.gen(function* () {
2038	const orchestrator = yield* Orchestrator.OrchestratorV2;
2039	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
2040	const threadId = ThreadId.make("branch-pr-link");
2041	const projectId = ProjectId.make("branch-pr-project");
2042	yield* seedProject({
2043	projectId,
2044	title: "PR links",
2045	workspaceRoot: "/workspace/pr-links",
2046	defaultModelSelection: null,
2047	createdAt: "2026-09-17T00:00:00.000Z",
2048	});
2049	yield* orchestrator.dispatch({
2050	type: "thread.create",
2051	createdBy: "user",
2052	creationSource: "web",
2053	commandId: CommandId.make("branch-pr-create"),
2054	threadId,
2055	projectId,
2056	title: "PR links",
2057	modelSelection,
2058	runtimeMode: "full-access",
2059	interactionMode: "default",
2060	branch: "feature/pr-links",
2061	worktreePath: null,
2062	});
2063	const snapshot = yield* orchestrator.getShellSnapshot();
2064	yield* orchestrator.dispatch({
2065	type: "thread.pull-request.sync",
2066	commandId: CommandId.make("branch-pr-discover"),
2067	threadId,
2068	projectId,
2069	snapshotSequence: snapshot.snapshotSequence,
2070	expected: {
2071	workspaceRoot: "/workspace/pr-links",
2072	branch: "feature/pr-links",
2073	worktreePath: null,
2074	linkedPullRequest: null,
2075	branchPullRequest: null,
2076	},
2077	branchPullRequest: {
2078	projectId,
2079	repository: "pingdotgg/t3code",
2080	number: 1,
2081	url: "https://github.com/pingdotgg/t3code/pull/1",
2082	},
2083	});
2084	for (const [index, number] of [2, 2, 1, 3].entries()) {
2085	yield* orchestrator.dispatch({
2086	type: "thread.pull-request.link",
2087	commandId: CommandId.make(`branch-pr-link-${index}`),
2088	threadId,
2089	host: "GitHub.com",
2090	repository: "Pingdotgg/T3code",
2091	number,
2092	url: `https://github.com/pingdotgg/t3code/pull/${number}`,
2093	source: "manual",
2094	});
2095	assert.deepEqual(
2096	(yield* orchestrator.getThreadShell(threadId))?.pullRequests?.map((link) => link.number),
2097	number === 3 ? [1, 2, 3] : [1, 2],
2098	);
2099	}
2100	yield* orchestrator.dispatch({
2101	type: "thread.pull-request.unlink",
2102	commandId: CommandId.make("branch-pr-unlink"),
2103	threadId,
2104	host: "github.com",
2105	repository: "pingdotgg/t3code",
2106	number: 1,
2107	});
2108	yield* orchestrator.dispatch({
2109	type: "thread.pull-request.link",
2110	commandId: CommandId.make("branch-pr-link-after-unlink"),
2111	threadId,
2112	host: "github.com",
2113	repository: "pingdotgg/t3code",
2114	number: 4,
2115	url: "https://github.com/pingdotgg/t3code/pull/4",
2116	source: "manual",
2117	});
2118	assert.isTrue((yield* maintenance.rebuild).valid);
2119	assert.deepEqual(
2120	(yield* orchestrator.getThreadShell(threadId))?.pullRequests?.map((link) => link.number),
2121	[2, 3, 4],
2122	);
2123	}),
2124	);
2125
2126	it.effect("retains multiple pull requests and dismissed stack members through rebuilds", () =>
2127	Effect.gen(function* () {
2128	const orchestrator = yield* Orchestrator.OrchestratorV2;
2129	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
2130	const threadId = ThreadId.make("runtime-multiple-pull-requests");
2131	yield* orchestrator.dispatch({
2132	type: "thread.create",
2133	createdBy: "user",
2134	creationSource: "web",
2135	commandId: CommandId.make("multi-pr-create"),
2136	threadId,
2137	projectId: ProjectId.make("multi-pr-project"),
2138	title: "Stack",
2139	modelSelection,
2140	runtimeMode: "full-access",
2141	interactionMode: "default",
2142	branch: null,
2143	worktreePath: null,
2144	});
2145	const key = { host: "GitHub.com", repository: "Pingdotgg/T3code" };
2146	for (const number of [1, 2]) {
2147	yield* orchestrator.dispatch({
2148	type: "thread.pull-request.link",
2149	commandId: CommandId.make(`multi-pr-link-${number}`),
2150	threadId,
2151	...key,
2152	number,
2153	url: `https://github.com/pingdotgg/t3code/pull/${number}`,
2154	source: number === 1 ? "manual" : "stack",
2155	});
2156	}
2157	const linked = yield* orchestrator.getThreadShell(threadId);
2158	assert.deepEqual(
2159	linked?.pullRequests?.map(({ host, repository, number }) => ({ host, repository, number })),
2160	[1, 2].map((number) => ({ host: "github.com", repository: "pingdotgg/t3code", number })),
2161	);
2162	yield* orchestrator.dispatch({
2163	type: "thread.pull-request.unlink",
2164	commandId: CommandId.make("multi-pr-dismiss"),
2165	threadId,
2166	...key,
2167	number: 2,
2168	});
2169	yield* orchestrator.dispatch({
2170	type: "thread.pull-request.unlink",
2171	commandId: CommandId.make("multi-pr-unlink"),
2172	threadId,
2173	...key,
2174	number: 1,
2175	});
2176	assert.deepEqual(
2177	(yield* orchestrator.getThreadShell(threadId))?.pullRequests?.map(({ number, source }) => ({
2178	number,
2179	source,
2180	})),
2181	[{ number: 2, source: "stack-dismissed" }],
2182	);
2183	yield* orchestrator.dispatch({
2184	type: "thread.pull-request.link",
2185	commandId: CommandId.make("multi-pr-rediscover"),
2186	threadId,
2187	...key,
2188	number: 2,
2189	url: "https://github.com/pingdotgg/t3code/pull/2",
2190	source: "stack",
2191	});
2192	assert.equal(
2193	(yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.source,
2194	"stack-dismissed",
2195	);
2196	assert.isTrue((yield* maintenance.rebuild).valid);
2197	assert.equal(
2198	(yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.source,
2199	"stack-dismissed",
2200	);
2201	yield* orchestrator.dispatch({
2202	type: "thread.pull-request.link",
2203	commandId: CommandId.make("multi-pr-restore"),
2204	threadId,
2205	...key,
2206	number: 2,
2207	url: "https://github.com/pingdotgg/t3code/pull/2",
2208	source: "manual",
2209	});
2210	assert.equal(
2211	(yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.source,
2212	"manual",
2213	);
2214	}),
2215	);
2216
2217	it.effect("starts, records, and stops a pull request watch", () =>
2218	Effect.gen(function* () {
2219	const orchestrator = yield* Orchestrator.OrchestratorV2;
2220	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
2221	const threadId = ThreadId.make("runtime-pull-request-watch");
2222	yield* orchestrator.dispatch({
2223	type: "thread.create",
2224	createdBy: "user",
2225	creationSource: "web",
2226	commandId: CommandId.make("pr-watch-create"),
2227	threadId,
2228	projectId: ProjectId.make("pr-watch-project"),
2229	title: "Watch",
2230	modelSelection,
2231	runtimeMode: "full-access",
2232	interactionMode: "default",
2233	branch: null,
2234	worktreePath: null,
2235	});
2236	const key = { host: "github.com", repository: "pingdotgg/t3code", number: 7 };
2237	const url = "https://github.com/pingdotgg/t3code/pull/7";
2238	const watchOf = Effect.map(
2239	orchestrator.getThreadShell(threadId),
2240	(thread) => thread?.pullRequests?.[0]?.watch,
2241	);
2242
2243	// Watching an unlinked pull request links it in the same command.
2244	yield* orchestrator.dispatch({
2245	type: "thread.pull-request.watch",
2246	commandId: CommandId.make("pr-watch-start"),
2247	threadId,
2248	...key,
2249	watching: true,
2250	link: { url, source: "agent" },
2251	});
2252	assert.equal(
2253	(yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.source,
2254	"agent",
2255	);
2256	const started = yield* watchOf;
2257	assert.isDefined(started);
2258	if (started === undefined) return;
2259
2260	// A legacy client re-linking the same pull request keeps its watch.
2261	yield* orchestrator.dispatch({
2262	type: "thread.metadata.update",
2263	commandId: CommandId.make("pr-watch-legacy-relink"),
2264	threadId,
2265	linkedPullRequest: { projectId: ProjectId.make("pr-watch-project"), ...key, url },
2266	});
2267	assert.deepEqual(yield* watchOf, started);
2268
2269	const recorded = { ...started, headSha: "abc123", failedChecks: ["lint"], wakes: 1 };
2270	yield* orchestrator.dispatch({
2271	type: "thread.pull-request-watch.sync",
2272	commandId: CommandId.make("pr-watch-record"),
2273	threadId,
2274	...key,
2275	startedAt: started.startedAt,
2276	watch: recorded,
2277	});
2278	assert.deepEqual(yield* watchOf, recorded);
2279	assert.isTrue((yield* maintenance.rebuild).valid);
2280	assert.deepEqual(yield* watchOf, recorded);
2281
2282	yield* orchestrator.dispatch({
2283	type: "thread.pull-request.watch",
2284	commandId: CommandId.make("pr-watch-stop"),
2285	threadId,
2286	...key,
2287	watching: false,
2288	});
2289	// A wake read before the stop must neither wake the agent nor bring the watch back.
2290	const late = yield* orchestrator
2291	.dispatch({
2292	type: "thread.pull-request-watch.sync",
2293	commandId: CommandId.make("pr-watch-late-record"),
2294	threadId,
2295	...key,
2296	startedAt: started.startedAt,
2297	watch: { ...recorded, wakes: 2 },
2298	wake: {
2299	messageId: MessageId.make("pr-watch-late-wake"),
2300	text: "Update",
2301	notification: { source: { kind: "monitor" }, outcome: "updated", summary: "#7" },
2302	},
2303	})
2304	.pipe(Effect.flip);
2305	assert.equal(late._tag, "OrchestratorDispatchError");
2306	assert.isUndefined(yield* watchOf);
2307	const { messages } = yield* orchestrator.getThreadRecords(threadId, ["messages"]);
2308	assert.deepEqual(messages, []);
2309	}),
2310	);
2311
2312	it.effect.each(["manual", "automatic"])("settling ends every pull request watch: %s", (mode) =>
2313	Effect.gen(function* () {
2314	const orchestrator = yield* Orchestrator.OrchestratorV2;
2315	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
2316	const threadId = ThreadId.make(`pr-watch-settle-${mode}`);
2317	yield* orchestrator.dispatch({
2318	type: "thread.create",
2319	createdBy: "user",
2320	creationSource: "web",
2321	commandId: CommandId.make(`pr-watch-settle-create-${mode}`),
2322	threadId,
2323	projectId: ProjectId.make("pr-watch-settle-project"),
2324	title: "Settle watched PRs",
2325	modelSelection,
2326	runtimeMode: "full-access",
2327	interactionMode: "default",
2328	branch: null,
2329	worktreePath: null,
2330	});
2331	const key = { host: "github.com", repository: "pingdotgg/t3code" };
2332	for (const number of [7, 8]) {
2333	yield* orchestrator.dispatch({
2334	type: "thread.pull-request.watch",
2335	commandId: CommandId.make(`pr-watch-settle-start-${mode}-${number}`),
2336	threadId,
2337	...key,
2338	number,
2339	watching: true,
2340	link: { url: `https://github.com/pingdotgg/t3code/pull/${number}`, source: "agent" },
2341	});
2342	}
2343	const before = yield* orchestrator.getThreadShell(threadId);
2344	const started = before?.pullRequests?.[0]?.watch;
2345	assert.isDefined(started);
2346	if (started === undefined || before === null) return;
2347	yield* orchestrator.dispatch({
2348	...(mode === "manual"
2349	? { type: "thread.settle" as const }
2350	: { type: "thread.auto-settle" as const, snapshotAt: before.updatedAt }),
2351	commandId: CommandId.make(`pr-watch-settle-${mode}`),
2352	threadId,
2353	});
2354	const expected = before.pullRequests?.map(({ watch: _watch, ...link }) => link);
2355	assert.deepEqual((yield* orchestrator.getThreadShell(threadId))?.pullRequests, expected);
2356	assert.isTrue((yield* maintenance.rebuild).valid);
2357	assert.deepEqual((yield* orchestrator.getThreadShell(threadId))?.pullRequests, expected);
2358
2359	for (const wake of [
2360	undefined,
2361	{
2362	messageId: MessageId.make(`pr-watch-settle-late-wake-${mode}`),
2363	text: "Update",
2364	notification: {
2365	source: { kind: "monitor" as const },
2366	outcome: "updated" as const,
2367	summary: "#7",
2368	},
2369	},
2370	]) {
2371	const late = yield* orchestrator
2372	.dispatch({
2373	type: "thread.pull-request-watch.sync",
2374	commandId: CommandId.make(`pr-watch-settle-late-${mode}-${wake !== undefined}`),
2375	threadId,
2376	...key,
2377	number: 7,
2378	startedAt: started.startedAt,
2379	watch: { ...started, headSha: "late", wakes: 1 },
2380	...(wake === undefined ? {} : { wake }),
2381	})
2382	.pipe(Effect.flip);
2383	assert.equal(late._tag, "OrchestratorDispatchError");
2384	}
2385	const restart = {
2386	type: "thread.pull-request.watch" as const,
2387	threadId,
2388	...key,
2389	number: 7,
2390	watching: true,
2391	};
2392	const blocked = yield* orchestrator
2393	.dispatch({
2394	...restart,
2395	commandId: CommandId.make(`pr-watch-settle-blocked-${mode}`),
2396	})
2397	.pipe(Effect.flip);
2398	assert.equal(blocked._tag, "OrchestratorDispatchError");
2399	assert.deepEqual((yield* orchestrator.getThreadRecords(threadId, ["messages"])).messages, []);
2400	yield* orchestrator.dispatch({
2401	type: "thread.unsettle",
2402	commandId: CommandId.make(`pr-watch-settle-unsettle-${mode}`),
2403	threadId,
2404	reason: "user",
2405	});
2406	yield* orchestrator.dispatch({
2407	...restart,
2408	commandId: CommandId.make(`pr-watch-settle-restart-${mode}`),
2409	});
2410	assert.isDefined((yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch);
2411	}),
2412	);
2413
2414	it.effect("archiving ends watches, and an archived thread cannot start one", () =>
2415	Effect.gen(function* () {
2416	const orchestrator = yield* Orchestrator.OrchestratorV2;
2417	const threadId = ThreadId.make("runtime-pull-request-watch-archive");
2418	yield* orchestrator.dispatch({
2419	type: "thread.create",
2420	createdBy: "user",
2421	creationSource: "web",
2422	commandId: CommandId.make("pr-watch-archive-create"),
2423	threadId,
2424	projectId: ProjectId.make("pr-watch-archive-project"),
2425	title: "Archive watched PR",
2426	modelSelection,
2427	runtimeMode: "full-access",
2428	interactionMode: "default",
2429	branch: null,
2430	worktreePath: null,
2431	});
2432	const watch = {
2433	type: "thread.pull-request.watch" as const,
2434	threadId,
2435	host: "github.com",
2436	repository: "pingdotgg/t3code",
2437	number: 7,
2438	watching: true,
2439	link: { url: "https://github.com/pingdotgg/t3code/pull/7", source: "agent" as const },
2440	};
2441	const watched = Effect.map(
2442	orchestrator.getThreadShell(threadId),
2443	(thread) => thread?.pullRequests?.[0]?.watch !== undefined,
2444	);
2445	yield* orchestrator.dispatch({
2446	...watch,
2447	commandId: CommandId.make("pr-watch-archive-start"),
2448	});
2449	assert.isTrue(yield* watched);
2450
2451	yield* orchestrator.dispatch({
2452	type: "thread.archive",
2453	commandId: CommandId.make("pr-watch-archive"),
2454	threadId,
2455	});
2456	assert.isFalse(yield* watched);
2457	const refused = yield* orchestrator
2458	.dispatch({ ...watch, commandId: CommandId.make("pr-watch-archive-restart") })
2459	.pipe(Effect.flip);
2460	assert.equal(refused._tag, "OrchestratorDispatchError");
2461
2462	yield* orchestrator.dispatch({
2463	type: "thread.unarchive",
2464	commandId: CommandId.make("pr-watch-unarchive"),
2465	threadId,
2466	});
2467	assert.isFalse(yield* watched);
2468	}),
2469	);
2470
2471	it.effect("ends a watch it cannot read, and tells the agent", () =>
2472	Effect.gen(function* () {
2473	const orchestrator = yield* Orchestrator.OrchestratorV2;
2474	const threadId = ThreadId.make("runtime-pull-request-watch-unreadable");
2475	const projectId = ProjectId.make("pr-watch-unreadable-project");
2476	yield* seedProject({
2477	projectId,
2478	title: "Watch unreadable",
2479	workspaceRoot: "/workspace/watch-unreadable",
2480	defaultModelSelection: null,
2481	createdAt: "2026-10-01T00:00:00.000Z",
2482	});
2483	yield* orchestrator.dispatch({
2484	type: "thread.create",
2485	createdBy: "user",
2486	creationSource: "web",
2487	commandId: CommandId.make("pr-watch-unreadable-create"),
2488	threadId,
2489	projectId,
2490	title: "Watch unreadable",
2491	modelSelection,
2492	runtimeMode: "full-access",
2493	interactionMode: "default",
2494	branch: null,
2495	worktreePath: null,
2496	});
2497	yield* orchestrator.dispatch({
2498	type: "thread.pull-request.watch",
2499	commandId: CommandId.make("pr-watch-unreadable-start"),
2500	threadId,
2501	host: "github.com",
2502	repository: "pingdotgg/t3code",
2503	number: 8,
2504	watching: true,
2505	link: { url: "https://github.com/pingdotgg/t3code/pull/8", source: "agent" },
2506	});
2507	const reactor = yield* PullRequestWatchReactor.make.pipe(
2508	Effect.provide(
2509	Layer.merge(
2510	Layer.mock(MergebotQueue.MergebotQueue)({
2511	configuration: null,
2512	streamChanges: Stream.empty,
2513	}),
2514	Layer.mergeAll(
2515	NodeServices.layer,
2516	Layer.mock(PullRequestService.PullRequestService)({
2517	detail: () => Effect.die("host unreachable"),
2518	activity: () => Effect.die("host unreachable"),
2519	}),
2520	),
2521	),
2522	),
2523	);
2524	for (let pass = 0; pass < 15; pass += 1) yield* reactor.sweep;
2525
2526	const thread = yield* orchestrator.getThreadShell(threadId);
2527	assert.isUndefined(thread?.pullRequests?.[0]?.watch);
2528	const { messages } = yield* orchestrator.getThreadRecords(threadId, ["messages"]);
2529	assert.deepEqual(
2530	messages.flatMap((message) => message.notification?.summary ?? []),
2531	["#8: stopped watching, could not read it"],
2532	);
2533	}),
2534	);
2535
2536	it.effect(
2537	"reads a watched pull request once a pass, waits out rate limits, and skips quiet passes",
2538	() =>
2539	Effect.gen(function* () {
2540	const orchestrator = yield* Orchestrator.OrchestratorV2;
2541	const projectId = ProjectId.make("pr-watch-shared-project");
2542	yield* seedProject({
2543	projectId,
2544	title: "Watch shared",
2545	workspaceRoot: "/workspace/watch-shared",
2546	defaultModelSelection: null,
2547	createdAt: "2026-10-01T00:00:00.000Z",
2548	});
2549	const threadIds = ["one", "two"].map((name) =>
2550	ThreadId.make(`runtime-pull-request-watch-shared-${name}`),
2551	);
2552	const watchFrom = (threadId: ThreadId) =>
2553	Effect.gen(function* () {
2554	yield* orchestrator.dispatch({
2555	type: "thread.create",
2556	createdBy: "user",
2557	creationSource: "web",
2558	commandId: CommandId.make(`create:${threadId}`),
2559	threadId,
2560	projectId,
2561	title: "Watch shared",
2562	modelSelection,
2563	runtimeMode: "full-access",
2564	interactionMode: "default",
2565	branch: null,
2566	worktreePath: null,
2567	});
2568	yield* orchestrator.dispatch({
2569	type: "thread.pull-request.watch",
2570	commandId: CommandId.make(`watch:${threadId}`),
2571	threadId,
2572	host: "github.com",
2573	repository: "pingdotgg/t3code",
2574	number: 9,
2575	watching: true,
2576	link: { url: "https://github.com/pingdotgg/t3code/pull/9", source: "agent" },
2577	});
2578	});
2579	for (const threadId of threadIds) yield* watchFrom(threadId);
2580	const rateLimited = new PullRequestOperationError({
2581	operation: "getChangeRequest",
2582	detail: "github requests are paused until the rate limit resets",
2583	cause: new PullRequestProviderError({
2584	provider: SourceControlProviderKind.make("github"),
2585	operation: "getChangeRequest",
2586	reason: "rate-limited",
2587	detail: "paused",
2588	}),
2589	});
2590	let host: "rate-limited" | "open" | "closed" = "rate-limited";
2591	let reads = 0;
2592	const reactor = yield* PullRequestWatchReactor.make.pipe(
2593	Effect.provide(
2594	Layer.merge(
2595	Layer.mock(MergebotQueue.MergebotQueue)({
2596	configuration: null,
2597	streamChanges: Stream.empty,
2598	}),
2599	Layer.mergeAll(
2600	NodeServices.layer,
2601	Layer.mock(PullRequestService.PullRequestService)({
2602	detail: () =>
2603	Effect.suspend(() => {
2604	reads += 1;
2605	return host === "rate-limited"
2606	? Effect.fail(rateLimited)
2607	: Effect.succeed({
2608	...watchedPullRequestDetail({
2609	projectId,
2610	number: 9,
2611	at: "2026-10-02T12:00:00.000Z",
2612	}),
2613	state: host,
2614	checks: [
2615	{ name: "lint", status: "success", description: null, url: null },
2616	],
2617	});
2618	}),
2619	activity: () =>
2620	host === "rate-limited"
2621	? Effect.fail(rateLimited)
2622	: Effect.succeed({
2623	comments: [],
2624	commentCount: 0,
2625	commentsTruncated: false,
2626	reviewThreads: [],
2627	commits: [],
2628	}),
2629	}),
2630	),
2631	),
2632	),
2633	);
2634	const summaries = (threadId: ThreadId) =>
2635	Effect.map(orchestrator.getThreadRecords(threadId, ["messages"]), ({ messages }) =>
2636	messages.flatMap((message) => message.notification?.summary ?? []),
2637	);
2638	const watching = (threadId: ThreadId) =>
2639	Effect.map(
2640	orchestrator.getThreadShell(threadId),
2641	(thread) => thread?.pullRequests?.[0]?.watch !== undefined,
2642	);
2643
2644	// However long the host stays rate limited, the watch waits instead of giving up, and
2645	// both threads share one read a pass.
2646	for (let pass = 0; pass < 20; pass += 1) yield* reactor.sweep;
2647	assert.equal(reads, 20);
2648	for (const threadId of threadIds) {
2649	assert.isTrue(yield* watching(threadId));
2650	assert.deepEqual(yield* summaries(threadId), []);
2651	}
2652
2653	host = "open";
2654	yield* reactor.sweep;
2655	assert.equal(reads, 21);
2656	for (const threadId of threadIds) {
2657	assert.deepEqual(yield* summaries(threadId), ["#9: checks passed"]);
2658	}
2659
2660	// Nothing in flight and nothing moved, so the next pass spends no request.
2661	yield* reactor.sweep;
2662	assert.equal(reads, 21);
2663
2664	// A new watch on the quiet pull request takes its first look on the next pass.
2665	const late = ThreadId.make("runtime-pull-request-watch-shared-late");
2666	yield* watchFrom(late);
2667	yield* reactor.sweep;
2668	assert.equal(reads, 22);
2669	assert.deepEqual(yield* summaries(late), ["#9: checks passed"]);
2670	for (const threadId of threadIds) {
2671	assert.deepEqual(yield* summaries(threadId), ["#9: checks passed"]);
2672	}
2673	threadIds.push(late);
2674
2675	// The quiet reread finds the pull request closed, ends both watches, and says so.
2676	host = "closed";
2677	yield* TestClock.adjust("10 minutes");
2678	yield* reactor.sweep;
2679	assert.equal(reads, 23);
2680	for (const threadId of threadIds) {
2681	assert.isFalse(yield* watching(threadId));
2682	assert.deepEqual(yield* summaries(threadId), [
2683	"#9: checks passed",
2684	"#9: closed, stopped watching",
2685	]);
2686	}
2687	}),
2688	);
2689
2690	it.effect("reports how long an ended watch was quiet", () =>
2691	Effect.gen(function* () {
2692	const orchestrator = yield* Orchestrator.OrchestratorV2;
2693	const threadId = ThreadId.make("runtime-pull-request-watch-ended");
2694	const projectId = ProjectId.make("pr-watch-ended-project");
2695	yield* seedProject({
2696	projectId,
2697	title: "Watch ended",
2698	workspaceRoot: "/workspace/watch-ended",
2699	defaultModelSelection: null,
2700	createdAt: "2026-10-01T00:00:00.000Z",
2701	});
2702	yield* orchestrator.dispatch({
2703	type: "thread.create",
2704	createdBy: "user",
2705	creationSource: "web",
2706	commandId: CommandId.make("pr-watch-ended-create"),
2707	threadId,
2708	projectId,
2709	title: "Watch ended",
2710	modelSelection,
2711	runtimeMode: "full-access",
2712	interactionMode: "default",
2713	branch: null,
2714	worktreePath: null,
2715	});
2716	const key = { host: "github.com", repository: "pingdotgg/t3code", number: 11 };
2717	const watching = (on: boolean, id: string) =>
2718	orchestrator.dispatch({
2719	type: "thread.pull-request.watch",
2720	commandId: CommandId.make(`pr-watch-ended-${id}`),
2721	threadId,
2722	...key,
2723	watching: on,
2724	link: { url: "https://github.com/pingdotgg/t3code/pull/11", source: "agent" },
2725	});
2726	yield* watching(true, "start");
2727
2728	let headSha = "aaaaaaa";
2729	const ended: Array<unknown> = [];
2730	const capture = Logger.make(({ message }) => {
2731	const [text, fields] = Array.isArray(message) ? message : [message];
2732	if (text === "pull request watch ended") ended.push(fields);
2733	});
2734	const reactor = yield* PullRequestWatchReactor.make.pipe(
2735	Effect.provide(
2736	Layer.merge(
2737	Layer.mock(MergebotQueue.MergebotQueue)({
2738	configuration: null,
2739	streamChanges: Stream.empty,
2740	}),
2741	Layer.mergeAll(
2742	NodeServices.layer,
2743	Layer.mock(PullRequestService.PullRequestService)({
2744	detail: () =>
2745	Effect.sync(() => ({
2746	...watchedPullRequestDetail({
2747	projectId,
2748	number: key.number,
2749	at: "2026-10-02",
2750	}),
2751	headSha,
2752	checks: [],
2753	})),
2754	activity: () =>
2755	Effect.succeed({
2756	comments: [],
2757	commentCount: 0,
2758	commentsTruncated: false,
2759	reviewThreads: [],
2760	commits: [],
2761	}),
2762	}),
2763	),
2764	),
2765	),
2766	);
2767	const sweep = reactor.sweep.pipe(Effect.provide(Logger.layer([capture])));
2768
2769	// The first read learns the head. A push 6 hours later, then 2 quiet hours.
2770	yield* sweep;
2771	yield* TestClock.adjust("6 hours");
2772	headSha = "bbbbbbb";
2773	yield* sweep;
2774	yield* TestClock.adjust("2 hours");
2775	yield* sweep;
2776	yield* watching(false, "stop");
2777	// The end is reported once.
2778	yield* sweep;
2779	yield* sweep;
2780
2781	assert.deepEqual(ended, [
2782	{
2783	threadId,
2784	pullRequest: "github.com/pingdotgg/t3code#11",
2785	reason: "stopped",
2786	minutes: 480,
2787	quietMinutes: 120,
2788	longestQuietMinutes: 360,
2789	wakes: 0,
2790	reads: 3,
2791	partial: false,
2792	},
2793	]);
2794	}),
2795	);
2796
2797	it.effect("with a host fingerprint, a watch reads only what moved", () =>
2798	Effect.gen(function* () {
2799	const orchestrator = yield* Orchestrator.OrchestratorV2;
2800	const projectId = ProjectId.make("pr-watch-fingerprint-project");
2801	const threadId = ThreadId.make("runtime-pull-request-watch-fingerprint");
2802	yield* seedProject({
2803	projectId,
2804	title: "Watch fingerprint",
2805	workspaceRoot: "/workspace/watch-fingerprint",
2806	defaultModelSelection: null,
2807	createdAt: "2026-10-01T00:00:00.000Z",
2808	});
2809	yield* orchestrator.dispatch({
2810	type: "thread.create",
2811	createdBy: "user",
2812	creationSource: "web",
2813	commandId: CommandId.make("pr-watch-fingerprint-create"),
2814	threadId,
2815	projectId,
2816	title: "Watch fingerprint",
2817	modelSelection,
2818	runtimeMode: "full-access",
2819	interactionMode: "default",
2820	branch: null,
2821	worktreePath: null,
2822	});
2823	yield* orchestrator.dispatch({
2824	type: "thread.pull-request.watch",
2825	commandId: CommandId.make("pr-watch-fingerprint-start"),
2826	threadId,
2827	host: "github.com",
2828	repository: "pingdotgg/t3code",
2829	number: 11,
2830	watching: true,
2831	link: { url: "https://github.com/pingdotgg/t3code/pull/11", source: "agent" },
2832	});
2833
2834	let fingerprint: { status: string; remarks: string } | "rate-limited" = {
2835	status: "OPEN pending",
2836	remarks: "0",
2837	};
2838	type CheckStatus = "pending" | "success" | "failure";
2839	// What the host says now, and what a read answers: a read right after a cache fill can
2840	// still answer from before a change until the cache is invalidated.
2841	let checks: { lint: CheckStatus; test: CheckStatus } = { lint: "pending", test: "pending" };
2842	let cachedChecks: typeof checks | null = null;
2843	let comments: Array<PullRequestComment> = [];
2844	let detailReads = 0;
2845	let activityReads = 0;
2846	const reactor = yield* PullRequestWatchReactor.make.pipe(
2847	Effect.provide(
2848	Layer.merge(
2849	Layer.mock(MergebotQueue.MergebotQueue)({
2850	configuration: null,
2851	streamChanges: Stream.empty,
2852	}),
2853	Layer.mergeAll(
2854	NodeServices.layer,
2855	Layer.mock(PullRequestService.PullRequestService)({
2856	watchFingerprint: () =>
2857	Effect.suspend(() =>
2858	fingerprint === "rate-limited"
2859	? Effect.fail(
2860	new PullRequestOperationError({
2861	operation: "watchFingerprint",
2862	detail: "paused",
2863	cause: new PullRequestProviderError({
2864	provider: SourceControlProviderKind.make("github"),
2865	operation: "getChangeRequestWatchFingerprint",
2866	reason: "rate-limited",
2867	detail: "paused",
2868	}),
2869	}),
2870	)
2871	: Effect.succeed(fingerprint),
2872	),
2873	invalidate: () =>
2874	Effect.sync(() => {
2875	cachedChecks = null;
2876	}),
2877	detail: () =>
2878	Effect.sync(() => {
2879	detailReads += 1;
2880	const answer = cachedChecks ?? checks;
2881	cachedChecks = null;
2882	return {
2883	...watchedPullRequestDetail({
2884	projectId,
2885	number: 11,
2886	at: "2026-10-02T12:00:00.000Z",
2887	}),
2888	checks: Object.entries(answer).map(([name, status]) => ({
2889	name,
2890	status,
2891	description: null,
2892	url: null,
2893	})),
2894	};
2895	}),
2896	activity: () =>
2897	Effect.sync(() => {
2898	activityReads += 1;
2899	return {
2900	comments,
2901	commentCount: comments.length,
2902	commentsTruncated: false,
2903	reviewThreads: [],
2904	commits: [],
2905	};
2906	}),
2907	}),
2908	),
2909	),
2910	),
2911	);
2912	const reads = () => ({ detail: detailReads, activity: activityReads });
2913	const summaries = Effect.map(
2914	orchestrator.getThreadRecords(threadId, ["messages"]),
2915	({ messages }) => messages.flatMap((message) => message.notification?.summary ?? []),
2916	);
2917
2918	// The first look reads everything.
2919	yield* reactor.sweep;
2920	assert.deepEqual(reads(), { detail: 1, activity: 1 });
2921
2922	// While checks run, the cheap detail is read every pass: check counts by state cannot
2923	// tell which check finished, so a failure can arrive with the fingerprint unchanged.
2924	checks = { lint: "failure", test: "pending" };
2925	yield* reactor.sweep;
2926	assert.deepEqual(reads(), { detail: 2, activity: 1 });
2927	assert.deepEqual(yield* summaries, ["#11: checks failed"]);
2928
2929	// When the fingerprint moves, a cached answer from before the move is not trusted.
2930	yield* TestClock.adjust("1 millis");
2931	cachedChecks = checks;
2932	checks = { lint: "failure", test: "success" };
2933	fingerprint = { status: "OPEN settled", remarks: "0" };
2934	yield* reactor.sweep;
2935	assert.deepEqual(reads(), { detail: 3, activity: 1 });
2936
2937	// Nothing in flight and nothing moved, so the passes read nothing.
2938	for (let pass = 0; pass < 5; pass += 1) yield* reactor.sweep;
2939	assert.deepEqual(reads(), { detail: 3, activity: 1 });
2940
2941	// A new comment moves the remarks, which take the activity read too.
2942	yield* TestClock.adjust("1 millis");
2943	comments = [
2944	{
2945	id: "comment-1",
2946	kind: "issue-comment",
2947	author: { login: "reviewer", name: null, avatarUrl: null },
2948	body: "Please rename this.",
2949	createdAt: "2999-01-01T00:00:00.000Z",
2950	url: null,
2951	path: null,
2952	reviewState: null,
2953	},
2954	];
2955	fingerprint = { status: "OPEN settled", remarks: "1" };
2956	yield* reactor.sweep;
2957	assert.deepEqual(reads(), { detail: 4, activity: 2 });
2958	assert.deepEqual(yield* summaries, ["#11: checks failed", "#11: new comments"]);
2959
2960	// While the host is rate limited, a pass reads nothing and the watch stays on.
2961	fingerprint = "rate-limited";
2962	yield* reactor.sweep;
2963	assert.deepEqual(reads(), { detail: 4, activity: 2 });
2964	assert.isDefined((yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch);
2965
2966	// Edits inside review threads leave no trace in the fingerprint, so the activity is read
2967	// again after half an hour regardless.
2968	fingerprint = { status: "OPEN settled", remarks: "1" };
2969	yield* TestClock.adjust("30 minutes");
2970	yield* reactor.sweep;
2971	assert.deepEqual(reads(), { detail: 5, activity: 3 });
2972	assert.deepEqual(yield* summaries, ["#11: checks failed", "#11: new comments"]);
2973	}),
2974	);
2975
2976	it.effect.each([
2977	"single page",
2978	"paginated",
2979	"page failure",
2980	"repeated cursor",
2981	"missing comments",
2982	"thread list truncated",
2983	])("wakes a watched thread once: %s", (mode) =>
2984	Effect.gen(function* () {
2985	const orchestrator = yield* Orchestrator.OrchestratorV2;
2986	const threadId = ThreadId.make(`runtime-pull-request-watch-wake-${mode}`);
2987	const projectId = ProjectId.make(`pr-watch-wake-project-${mode}`);
2988	yield* seedProject({
2989	projectId,
2990	title: "Watch wake",
2991	workspaceRoot: "/workspace/watch",
2992	defaultModelSelection: null,
2993	createdAt: "2026-10-01T00:00:00.000Z",
2994	});
2995	yield* orchestrator.dispatch({
2996	type: "thread.create",
2997	createdBy: "user",
2998	creationSource: "web",
2999	commandId: CommandId.make(`pr-watch-wake-create-${mode}`),
3000	threadId,
3001	projectId,
3002	title: "Watch wake",
3003	modelSelection,
3004	runtimeMode: "full-access",
3005	interactionMode: "default",
3006	branch: null,
3007	worktreePath: null,
3008	});
3009	const key = { host: "github.com", repository: "pingdotgg/t3code", number: 7 };
3010	const url = "https://github.com/pingdotgg/t3code/pull/7";
3011	yield* orchestrator.dispatch({
3012	type: "thread.pull-request.link",
3013	commandId: CommandId.make(`pr-watch-wake-link-${mode}`),
3014	threadId,
3015	...key,
3016	url,
3017	source: "agent",
3018	});
3019	yield* orchestrator.dispatch({
3020	type: "thread.pull-request.watch",
3021	commandId: CommandId.make(`pr-watch-wake-start-${mode}`),
3022	threadId,
3023	...key,
3024	watching: true,
3025	});
3026
3027	const at = "2026-10-02T12:00:00.000Z";
3028	const detail = watchedPullRequestDetail({ projectId, number: key.number, at });
3029	const initialWatch = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch;
3030	assert.isDefined(initialWatch);
3031	const remark: PullRequestComment = {
3032	id: "review-1",
3033	kind: "review-comment",
3034	author: { login: "reviewer", name: null, avatarUrl: null },
3035	body: "One more thing.",
3036	createdAt: "2999-01-01T00:00:03.000Z",
3037	url: null,
3038	path: "src/index.ts",
3039	reviewState: null,
3040	};
3041	const firstTen = Array.from({ length: 10 }, (_, index) => ({
3042	...remark,
3043	id: `old-${index}`,
3044	createdAt: "1900-01-01T00:00:00.000Z",
3045	}));
3046	const eleventh = {
3047	...remark,
3048	id: "reply-11",
3049	body: "Eleventh reply.",
3050	createdAt: "2999-01-01T00:00:01.000Z",
3051	};
3052	const twelfth = {
3053	...remark,
3054	id: "reply-12",
3055	body: "Twelfth reply.",
3056	createdAt: "2999-01-01T00:00:02.000Z",
3057	};
3058	const incomplete = mode !== "single page" && mode !== "paginated";
3059	let recovering = false;
3060	let pagesRead = 0;
3061	const reactor = yield* PullRequestWatchReactor.make.pipe(
3062	Effect.provide(
3063	Layer.merge(
3064	Layer.mock(MergebotQueue.MergebotQueue)({
3065	configuration: null,
3066	streamChanges: Stream.empty,
3067	}),
3068	Layer.mergeAll(
3069	NodeServices.layer,
3070	Layer.mock(PullRequestService.PullRequestService)({
3071	detail: () => Effect.succeed(detail),
3072	activity: () =>
3073	Effect.succeed({
3074	comments:
3075	mode === "single page"
3076	? [remark]
3077	: [...firstTen, { ...remark, kind: "issue-comment" }],
3078	commentCount: mode === "single page" ? 1 : 13,
3079	commentsTruncated: mode !== "single page",
3080	reviewThreadsTruncated: mode === "thread list truncated" && !recovering,
3081	reviewThreads:
3082	mode === "single page"
3083	? []
3084	: [
3085	{
3086	id: "review-thread",
3087	path: "src/index.ts",
3088	line: 1,
3089	side: "right",
3090	isResolved: false,
3091	isOutdated: false,
3092	comments: firstTen,
3093	commentCount: 12,
3094	nextCommentsCursor: "after-10",
3095	},
3096	],
3097	commits: [],
3098	}),
3099	threadComments: (input) => {
3100	pagesRead += 1;
3101	assert.equal(input.threadId, "review-thread");
3102	if (input.cursor === "after-10") {
3103	return Effect.succeed({ comments: [eleventh], nextCursor: "after-11" });
3104	}
3105	assert.equal(input.cursor, "after-11");
3106	if (!recovering && mode === "page failure") {
3107	return Effect.fail(
3108	new PullRequestOperationError({
3109	operation: "threadComments",
3110	detail: "Page unavailable",
3111	}),
3112	);
3113	}
3114	if (!recovering && mode === "repeated cursor") {
3115	return Effect.succeed({ comments: [eleventh], nextCursor: "after-11" });
3116	}
3117	if (!recovering && mode === "missing comments") {
3118	return Effect.succeed({ comments: [], nextCursor: null });
3119	}
3120	// Overlapping pages must not report the same reply twice.
3121	return Effect.succeed({ comments: [eleventh, twelfth], nextCursor: null });
3122	},
3123	}),
3124	),
3125	),
3126	),
3127	);
3128	if (incomplete) {
3129	yield* reactor.sweep;
3130	const held = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch;
3131	assert.equal(held?.remarksThrough, initialWatch?.remarksThrough);
3132	assert.deepEqual(held?.remarkIds, initialWatch?.remarkIds);
3133	assert.deepEqual(held?.failedChecks, ["lint"]);
3134	const { messages } = yield* orchestrator.getThreadRecords(threadId, ["messages"]);
3135	assert.deepEqual(
3136	messages.map((message) => message.notification?.summary),
3137	["#7: checks failed"],
3138	);
3139	// Keep the two notifications ordered independently of their random message IDs.
3140	yield* TestClock.adjust("1 millis");
3141	recovering = true;
3142	}
3143	yield* reactor.sweep;
3144	// A thread whose count has not moved is not paged again.
3145	const pagesBefore = pagesRead;
3146	yield* reactor.sweep;
3147	assert.equal(pagesRead, pagesBefore);
3148
3149	const { messages } = yield* orchestrator.getThreadRecords(threadId, ["messages"]);
3150	assert.deepEqual(
3151	messages.flatMap((message) =>
3152	message.notification === undefined ? [] : [message.notification.summary],
3153	),
3154	incomplete
3155	? ["#7: checks failed", "#7: new comments"]
3156	: ["#7: checks failed, new comments"],
3157	);
3158	if (mode !== "single page") {
3159	const wake = messages.at(-1);
3160	assert.include(wake?.text ?? "", "3 new comments");
3161	assert.include(wake?.text ?? "", "Eleventh reply.");
3162	assert.include(wake?.text ?? "", "Twelfth reply.");
3163	}
3164	const watch = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch;
3165	assert.equal(watch?.remarksThrough, remark.createdAt);
3166	assert.deepEqual(watch?.remarkIds, [remark.id]);
3167	assert.deepEqual(
3168	{ headSha: watch?.headSha, failedChecks: watch?.failedChecks, wakes: watch?.wakes },
3169	{ headSha: "abc1234def", failedChecks: ["lint"], wakes: incomplete ? 1 : 0 },
3170	);
3171	}),
3172	);
3173
3174	it.effect("persists rejected command receipts across retries", () =>
3175	Effect.gen(function* () {
3176	const orchestrator = yield* Orchestrator.OrchestratorV2;
3177	const command = {
3178	type: "thread.archive" as const,
3179	commandId: CommandId.make("runtime-layer-rejected-command"),
3180	threadId: ThreadId.make("runtime-layer-missing-thread"),
3181	};
3182
3183	const first = yield* orchestrator.dispatch(command).pipe(Effect.flip);
3184	const retry = yield* orchestrator.dispatch(command).pipe(Effect.flip);
3185
3186	assert.equal(first._tag, "OrchestratorProjectionError");
3187	assert.equal(retry._tag, "OrchestratorCommandPreviouslyRejectedError");
3188	}),
3189	);
3190
3191	it.effect(
3192	"admits restart continuations once and rejects a stale continuation behind newer work",
3193	() =>
3194	Effect.gen(function* () {
3195	const orchestrator = yield* Orchestrator.OrchestratorV2;
3196	const eventSink = yield* EventSink.EventSinkV2;
3197	const threadId = ThreadId.make("runtime-layer-restart-continuation");
3198	yield* orchestrator.dispatch({
3199	type: "thread.create",
3200	createdBy: "user",
3201	creationSource: "web",
3202	commandId: CommandId.make("restart-create"),
3203	threadId,
3204	projectId: ProjectId.make("restart-project"),
3205	title: "Restart",
3206	modelSelection,
3207	runtimeMode: "full-access",
3208	interactionMode: "default",
3209	branch: null,
3210	worktreePath: "/tmp/runtime-layer-restart",
3211	});
3212	yield* orchestrator.dispatch({
3213	type: "message.dispatch",
3214	createdBy: "user",
3215	creationSource: "web",
3216	commandId: CommandId.make("restart-user-message"),
3217	threadId,
3218	messageId: MessageId.make("restart-user-message"),
3219	text: "Original work",
3220	attachments: [],
3221	modelSelection,
3222	dispatchMode: { type: "start_immediately" },
3223	});
3224	const original = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
3225	const now = yield* DateTime.now;
3226	yield* eventSink.commitCommand({
3227	commandId: CommandId.make("restart-cancel"),
3228	threadId,
3229	commandType: "provider-runtime.reconcile",
3230	acceptedAt: now,
3231	events: [
3232	{
3233	id: EventId.make("restart-cancel-event"),
3234	type: "run.updated",
3235	threadId,
3236	runId: original.id,
3237	occurredAt: now,
3238	payload: { ...original, status: "cancelled", completedAt: now },
3239	},
3240	],
3241	effects: [],
3242	});
3243	const command = {
3244	type: "message.dispatch" as const,
3245	createdBy: "agent" as const,
3246	creationSource: "server" as const,
3247	commandId: CommandId.make("restart-automatic-message"),
3248	threadId,
3249	messageId: MessageId.make("restart-automatic-message"),
3250	text: "Continue where you left off.",
3251	notification: {
3252	source: { kind: "system" as const },
3253	outcome: "updated" as const,
3254	summary: "T3 Code restarted and resumed this turn",
3255	detail: "Continue where you left off.",
3256	},
3257	attachments: [],
3258	modelSelection,
3259	dispatchMode: { type: "start_immediately" as const },
3260	restartContinuationOfRunId: original.id,
3261	};
3262	yield* orchestrator.dispatch(command);
3263	yield* orchestrator.dispatch(command);
3264	const admitted = yield* orchestrator.getThreadProjection(threadId);
3265	assert.lengthOf(admitted.runs, 2);
3266	assert.equal(admitted.runs[1]?.restartContinuationOfRunId, original.id);
3267	// The timeline shows a work log row; the prompt stays in its detail.
3268	const continuationItems = admitted.turnItems.filter(
3269	(item) => item.runId === admitted.runs[1]?.id,
3270	);
3271	assert.isFalse(continuationItems.some((item) => item.type === "user_message"));
3272	assert.deepInclude(
3273	continuationItems.flatMap((item) =>
3274	item.type === "notification" ? [[item.summary, item.detail]] : [],
3275	),
3276	["T3 Code restarted and resumed this turn", "Continue where you left off."],
3277	);
3278	// A differently identified stale delivery still must not create another run.
3279	yield* orchestrator.dispatch({
3280	...command,
3281	commandId: CommandId.make("restart-stale-race"),
3282	messageId: MessageId.make("restart-stale-race"),
3283	});
3284	const raced = yield* orchestrator.getThreadProjection(threadId);
3285	assert.lengthOf(raced.runs, 2);
3286	assert.isFalse(raced.messages.some((message) => message.id === "restart-stale-race"));
3287	}),
3288	);
3289
3290	it.effect.each([
3291	["failed", "shell", false],
3292	["completed", "shell", false],
3293	["waiting", "shell", false],
3294	["cancelled", "shell", false],
3295	// A settled run lost work it was waiting on, so nothing else wakes it.
3296	["failed", "monitor", false],
3297	["completed", "monitor", true],
3298	["waiting", "monitor", true],
3299	["cancelled", "monitor", true],
3300	] as const)(
3301	"admits a restart continuation of a %s run that lost a %s: %s",
3302	([status, kind, admitted]) =>
3303	Effect.gen(function* () {
3304	const orchestrator = yield* Orchestrator.OrchestratorV2;
3305	const eventSink = yield* EventSink.EventSinkV2;
3306	const threadId = ThreadId.make(`runtime-layer-restart-${status}-${kind}-source`);
3307	yield* orchestrator.dispatch({
3308	type: "thread.create",
3309	createdBy: "user",
3310	creationSource: "web",
3311	commandId: CommandId.make(`restart-${status}-${kind}-create`),
3312	threadId,
3313	projectId: ProjectId.make("restart-project"),
3314	title: "Restart",
3315	modelSelection,
3316	runtimeMode: "full-access",
3317	interactionMode: "default",
3318	branch: null,
3319	worktreePath: "/tmp/runtime-layer-restart-failed",
3320	});
3321	yield* orchestrator.dispatch({
3322	type: "message.dispatch",
3323	createdBy: "user",
3324	creationSource: "web",
3325	commandId: CommandId.make(`restart-${status}-${kind}-user-message`),
3326	threadId,
3327	messageId: MessageId.make(`restart-${status}-${kind}-user-message`),
3328	text: "Original work",
3329	attachments: [],
3330	modelSelection,
3331	dispatchMode: { type: "start_immediately" },
3332	});
3333	const original = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
3334	const now = yield* DateTime.now;
3335	yield* eventSink.commitCommand({
3336	commandId: CommandId.make(`restart-${status}-${kind}-reconcile`),
3337	threadId,
3338	commandType: "provider-runtime.reconcile",
3339	acceptedAt: now,
3340	events: [
3341	{
3342	id: EventId.make(`restart-${status}-${kind}-run`),
3343	type: "run.updated",
3344	threadId,
3345	runId: original.id,
3346	occurredAt: now,
3347	payload: { ...original, status, completedAt: now },
3348	},
3349	{
3350	id: EventId.make(`restart-${status}-${kind}-work`),
3351	type: "run.background-work-cancelled",
3352	threadId,
3353	runId: original.id,
3354	occurredAt: now,
3355	payload: {
3356	runId: original.id,
3357	restartCancelledBackgroundWork: [{ kind, label: "sleep 25" }],
3358	},
3359	},
3360	],
3361	effects: [],
3362	});
3363	yield* orchestrator.dispatch({
3364	type: "message.dispatch",
3365	createdBy: "agent",
3366	creationSource: "server",
3367	commandId: CommandId.make(`restart-${status}-${kind}-continuation`),
3368	threadId,
3369	messageId: MessageId.make(`restart-${status}-${kind}-continuation`),
3370	text: "Note: the T3 server restarted.",
3371	attachments: [],
3372	modelSelection,
3373	dispatchMode: { type: "start_immediately" },
3374	restartContinuationOfRunId: original.id,
3375	});
3376	const projection = yield* orchestrator.getThreadProjection(threadId);
3377	assert.equal(projection.runs[0]?.status, status);
3378	assert.deepEqual(
3379	projection.runs.slice(1).map((run) => run.restartContinuationOfRunId),
3380	admitted ? [original.id] : [],
3381	);
3382	}),
3383	);
3384
3385	it.effect("rejects settling a thread while a run is active", () =>
3386	Effect.gen(function* () {
3387	const orchestrator = yield* Orchestrator.OrchestratorV2;
3388	const threadId = ThreadId.make("runtime-layer-active-settle-thread");
3389
3390	yield* orchestrator.dispatch({
3391	type: "thread.create",
3392	createdBy: "user",
3393	creationSource: "web",
3394	commandId: CommandId.make("runtime-layer-active-settle-create"),
3395	threadId,
3396	projectId: ProjectId.make("runtime-layer-active-settle-project"),
3397	title: "Active settle",
3398	modelSelection,
3399	runtimeMode: "full-access",
3400	interactionMode: "default",
3401	branch: null,
3402	worktreePath: "/tmp/runtime-layer-active-settle",
3403	});
3404	yield* orchestrator.dispatch({
3405	type: "thread.settle",
3406	commandId: CommandId.make("runtime-layer-active-settle-initial"),
3407	threadId,
3408	});
3409	yield* orchestrator.dispatch({
3410	type: "message.dispatch",
3411	createdBy: "user",
3412	creationSource: "web",
3413	commandId: CommandId.make("runtime-layer-active-settle-message"),
3414	threadId,
3415	messageId: MessageId.make("runtime-layer-active-settle-message"),
3416	text: "Keep this run active.",
3417	attachments: [],
3418	modelSelection,
3419	dispatchMode: { type: "start_immediately" },
3420	});
3421
3422	const nonEmptyClaim = yield* orchestrator
3423	.dispatch({
3424	type: "thread.metadata.update",
3425	commandId: CommandId.make("runtime-layer-active-settle-empty-claim"),
3426	threadId,
3427	worktreePath: "/tmp/reassigned-after-message",
3428	expectedEmpty: true,
3429	})
3430	.pipe(Effect.flip);
3431	assert.instanceOf(nonEmptyClaim, Orchestrator.OrchestratorDispatchError);
3432
3433	const error = yield* orchestrator
3434	.dispatch({
3435	type: "thread.settle",
3436	commandId: CommandId.make("runtime-layer-active-settle"),
3437	threadId,
3438	})
3439	.pipe(Effect.flip);
3440
3441	assert.equal(error._tag, "OrchestratorDispatchError");
3442	assert.equal(error.cause, `Thread ${threadId} is still running. Stop it before settling.`);
3443	const projection = yield* orchestrator.getThreadProjection(threadId);
3444	assert.equal(projection.runs[0]?.status, "starting");
3445	assert.isNull(projection.thread.settledOverride);
3446	assert.isNull(projection.thread.settledAt);
3447	assert.isNotNull(projection.thread.unsettledAt);
3448	}),
3449	);
3450
3451	it.effect("settles a thread its own agent settled once the turn completes", () =>
3452	Effect.gen(function* () {
3453	const orchestrator = yield* Orchestrator.OrchestratorV2;
3454	const eventSink = yield* EventSink.EventSinkV2;
3455	const threadManagement = yield* ThreadManagementService.ThreadManagementService;
3456	const projectId = ProjectId.make("runtime-layer-settle-after-run-project");
3457	const threadId = ThreadId.make("runtime-layer-settle-after-run-thread");
3458
3459	yield* orchestrator.dispatch({
3460	type: "thread.create",
3461	createdBy: "user",
3462	creationSource: "web",
3463	commandId: CommandId.make("runtime-layer-settle-after-run-create"),
3464	threadId,
3465	projectId,
3466	title: "Settle after run",
3467	modelSelection,
3468	runtimeMode: "full-access",
3469	interactionMode: "default",
3470	branch: null,
3471	worktreePath: "/tmp/runtime-layer-settle-after-run",
3472	});
3473	yield* orchestrator.dispatch({
3474	type: "message.dispatch",
3475	createdBy: "user",
3476	creationSource: "web",
3477	commandId: CommandId.make("runtime-layer-settle-after-run-message"),
3478	threadId,
3479	messageId: MessageId.make("runtime-layer-settle-after-run-message"),
3480	text: "Fix it and then settle this thread.",
3481	attachments: [],
3482	modelSelection,
3483	dispatchMode: { type: "start_immediately" },
3484	});
3485	const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0];
3486	if (run === undefined) return yield* Effect.die(new Error("Run missing."));
3487
3488	const settled = yield* orchestrator
3489	.streamStoredEventsFrom({ threadId, afterSequence: 0, eventType: "thread.settled" })
3490	.pipe(Stream.runHead, Effect.forkChild);
3491	const result = yield* threadManagement.settleThread({
3492	threadId,
3493	commandId: CommandId.make("runtime-layer-settle-after-run-settle"),
3494	byOwnAgent: true,
3495	});
3496	assert.deepEqual(result, { settlesWhenTurnEnds: true });
3497	assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.settledOverride);
3498	const now = yield* DateTime.now;
3499	yield* eventSink.write({
3500	commandId: CommandId.make("runtime-layer-settle-after-run-completed"),
3501	events: [
3502	{
3503	id: EventId.make("runtime-layer-settle-after-run-completed"),
3504	type: "run.updated",
3505	threadId,
3506	runId: run.id,
3507	occurredAt: now,
3508	payload: { ...run, status: "completed", startedAt: now, completedAt: now },
3509	},
3510	],
3511	});
3512	yield* Fiber.join(settled);
3513
3514	const projection = yield* orchestrator.getThreadProjection(threadId);
3515	assert.equal(projection.thread.settledOverride, "settled");
3516	}),
3517	);
3518
3519	it.effect("settles past held automatic runs but not held user messages", () =>
3520	Effect.gen(function* () {
3521	const orchestrator = yield* Orchestrator.OrchestratorV2;
3522	const eventSink = yield* EventSink.EventSinkV2;
3523	const threadId = ThreadId.make("runtime-layer-settle-automatic-queued");
3524
3525	yield* orchestrator.dispatch({
3526	type: "thread.create",
3527	createdBy: "user",
3528	creationSource: "web",
3529	commandId: CommandId.make("settle-automatic-create"),
3530	threadId,
3531	projectId: ProjectId.make("settle-automatic-project"),
3532	title: "Settle automatic queued",
3533	modelSelection,
3534	runtimeMode: "full-access",
3535	interactionMode: "default",
3536	branch: null,
3537	worktreePath: "/tmp/runtime-layer-settle-automatic",
3538	});
3539	yield* orchestrator.dispatch({
3540	type: "message.dispatch",
3541	createdBy: "user",
3542	creationSource: "web",
3543	commandId: CommandId.make("settle-automatic-active"),
3544	threadId,
3545	messageId: MessageId.make("settle-automatic-active"),
3546	text: "Active",
3547	attachments: [],
3548	modelSelection,
3549	dispatchMode: { type: "start_immediately" },
3550	});
3551	const queueMessage = (id: string, automatic: boolean) =>
3552	orchestrator.dispatch({
3553	type: "message.dispatch",
3554	createdBy: automatic ? "agent" : "user",
3555	creationSource: automatic ? "provider" : "web",
3556	...(automatic
3557	? {
3558	notification: {
3559	source: { kind: "background_task" as const },
3560	outcome: "updated" as const,
3561	summary: "Background activity updated",
3562	},
3563	}
3564	: {}),
3565	commandId: CommandId.make(id),
3566	threadId,
3567	messageId: MessageId.make(id),
3568	text: id,
3569	attachments: [],
3570	modelSelection,
3571	dispatchMode: { type: "queue_after_active" },
3572	});
3573	yield* queueMessage("settle-automatic-notification", true);
3574
3575	// Simulate a restart: the active run ends and recovery holds the queue.
3576	const holdQueueAfterRestart = (commandId: string) =>
3577	Effect.gen(function* () {
3578	const projection = yield* orchestrator.getThreadProjection(threadId);
3579	const now = yield* DateTime.now;
3580	yield* eventSink.commitCommand({
3581	commandId: CommandId.make(commandId),
3582	threadId,
3583	commandType: "provider-runtime.reconcile",
3584	acceptedAt: now,
3585	events: projection.runs
3586	.filter((run) => run.status === "starting" || run.status === "queued")
3587	.map((run) => ({
3588	id: EventId.make(`${commandId}:${run.id}`),
3589	type: "run.updated" as const,
3590	threadId,
3591	runId: run.id,
3592	occurredAt: now,
3593	payload:
3594	run.status === "queued"
3595	? { ...run, queueHeld: true }
3596	: { ...run, status: "cancelled" as const, completedAt: now },
3597	})),
3598	effects: [],
3599	});
3600	});
3601	yield* holdQueueAfterRestart("settle-automatic-restart");
3602
3603	yield* orchestrator.dispatch({
3604	type: "thread.settle",
3605	commandId: CommandId.make("settle-automatic-settle"),
3606	threadId,
3607	});
3608	const settled = yield* orchestrator.getThreadProjection(threadId);
3609	assert.isNotNull(settled.thread.settledAt);
3610	assert.isTrue(settled.runs.every((run) => run.status === "cancelled"));
3611
3612	// A held message the user typed still blocks settling.
3613	yield* orchestrator.dispatch({
3614	type: "thread.unsettle",
3615	commandId: CommandId.make("settle-automatic-unsettle"),
3616	threadId,
3617	reason: "user",
3618	});
3619	yield* orchestrator.dispatch({
3620	type: "message.dispatch",
3621	createdBy: "user",
3622	creationSource: "web",
3623	commandId: CommandId.make("settle-automatic-active-2"),
3624	threadId,
3625	messageId: MessageId.make("settle-automatic-active-2"),
3626	text: "Active again",
3627	attachments: [],
3628	modelSelection,
3629	dispatchMode: { type: "start_immediately" },
3630	});
3631	yield* queueMessage("settle-automatic-user-queued", false);
3632	yield* holdQueueAfterRestart("settle-automatic-restart-2");
3633	const error = yield* orchestrator
3634	.dispatch({
3635	type: "thread.settle",
3636	commandId: CommandId.make("settle-automatic-settle-2"),
3637	threadId,
3638	})
3639	.pipe(Effect.flip);
3640	assert.equal(error._tag, "OrchestratorDispatchError");
3641	assert.equal(
3642	error.cause,
3643	`Thread ${threadId} has a queued message. Send it or remove it from the queue before settling.`,
3644	);
3645	}),
3646	);
3647
3648	it.effect("cancels queued work when a thread is archived", () =>
3649	Effect.gen(function* () {
3650	const orchestrator = yield* Orchestrator.OrchestratorV2;
3651	const threadId = ThreadId.make("runtime-layer-archive-queued-thread");
3652
3653	yield* orchestrator.dispatch({
3654	type: "thread.create",
3655	createdBy: "user",
3656	creationSource: "web",
3657	commandId: CommandId.make("runtime-layer-archive-queued-create"),
3658	threadId,
3659	projectId: ProjectId.make("runtime-layer-archive-queued-project"),
3660	title: "Archive queued work",
3661	modelSelection,
3662	runtimeMode: "full-access",
3663	interactionMode: "default",
3664	branch: null,
3665	worktreePath: "/tmp/runtime-layer-archive-queued",
3666	});
3667	yield* orchestrator.dispatch({
3668	type: "message.dispatch",
3669	createdBy: "user",
3670	creationSource: "web",
3671	commandId: CommandId.make("runtime-layer-archive-queued-active-message"),
3672	threadId,
3673	messageId: MessageId.make("runtime-layer-archive-queued-active-message"),
3674	text: "Keep the provider occupied.",
3675	attachments: [],
3676	modelSelection,
3677	dispatchMode: { type: "start_immediately" },
3678	});
3679	yield* orchestrator.dispatch({
3680	type: "message.dispatch",
3681	createdBy: "user",
3682	creationSource: "web",
3683	commandId: CommandId.make("runtime-layer-archive-queued-next-message"),
3684	threadId,
3685	messageId: MessageId.make("runtime-layer-archive-queued-next-message"),
3686	text: "Do not run this after archive.",
3687	attachments: [],
3688	modelSelection,
3689	dispatchMode: { type: "queue_after_active" },
3690	});
3691
3692	const beforeArchive = yield* orchestrator.getThreadProjection(threadId);
3693	const activeRun = beforeArchive.runs.find((run) => run.status === "starting");
3694	const queuedRun = beforeArchive.runs.find((run) => run.status === "queued");
3695	assert.isDefined(activeRun);
3696	assert.isDefined(queuedRun);
3697
3698	yield* orchestrator.dispatch({
3699	type: "thread.archive",
3700	commandId: CommandId.make("runtime-layer-archive-queued-archive"),
3701	threadId,
3702	});
3703
3704	const archived = yield* orchestrator.getThreadProjection(threadId);
3705	assert.isNotNull(archived.thread.archivedAt);
3706	assert.equal(archived.runs.find((run) => run.id === queuedRun.id)?.status, "cancelled");
3707	assert.equal(
3708	archived.attempts.find((attempt) => attempt.runId === queuedRun.id)?.status,
3709	"cancelled",
3710	);
3711	assert.equal(archived.nodes.find((node) => node.runId === queuedRun.id)?.status, "cancelled");
3712	assert.equal(yield* orchestrator.resumeQueuedRuns, 0);
3713
3714	const promoteError = yield* orchestrator
3715	.dispatch({
3716	type: "queued-message.promote-to-steer",
3717	commandId: CommandId.make("runtime-layer-archive-queued-promote"),
3718	threadId,
3719	queuedRunId: queuedRun.id,
3720	targetRunId: activeRun.id,
3721	})
3722	.pipe(Effect.flip);
3723	assert.equal(promoteError._tag, "OrchestratorDispatchError");
3724
3725	const afterPromotion = yield* orchestrator.getThreadProjection(threadId);
3726	assert.equal(afterPromotion.runs.find((run) => run.id === queuedRun.id)?.status, "cancelled");
3727	}),
3728	);
3729
3730	it.effect.each([false, true])(
3731	"promotes only one queued run after each terminal run (notification: %s)",
3732	(automatic) =>
3733	Effect.gen(function* () {
3734	const orchestrator = yield* Orchestrator.OrchestratorV2;
3735	const eventSink = yield* EventSink.EventSinkV2;
3736	const threadId = ThreadId.make(`runtime-layer-serialized-queue-thread-${automatic}`);
3737
3738	yield* orchestrator.dispatch({
3739	type: "thread.create",
3740	createdBy: "user",
3741	creationSource: "web",
3742	commandId: CommandId.make(`runtime-layer-serialized-queue-create-${automatic}`),
3743	threadId,
3744	projectId: ProjectId.make(`runtime-layer-serialized-queue-project-${automatic}`),
3745	title: "Serialized queue",
3746	modelSelection,
3747	runtimeMode: "full-access",
3748	interactionMode: "default",
3749	branch: null,
3750	worktreePath: process.cwd(),
3751	});
3752	yield* orchestrator.dispatch({
3753	type: "message.dispatch",
3754	createdBy: "user",
3755	creationSource: "web",
3756	commandId: CommandId.make(`runtime-layer-serialized-queue-active-${automatic}`),
3757	threadId,
3758	messageId: MessageId.make(`runtime-layer-serialized-queue-active-${automatic}`),
3759	text: "Active",
3760	attachments: [],
3761	modelSelection,
3762	dispatchMode: { type: "start_immediately" },
3763	});
3764	yield* orchestrator.dispatch({
3765	type: "message.dispatch",
3766	createdBy: automatic ? "agent" : "user",
3767	creationSource: automatic ? "provider" : "web",
3768	...(automatic
3769	? {
3770	notification: {
3771	source: { kind: "monitor" as const },
3772	outcome: "updated" as const,
3773	summary: "Monitor updated",
3774	detail: "Build is green",
3775	},
3776	}
3777	: {}),
3778	commandId: CommandId.make(`runtime-layer-serialized-queue-first-${automatic}`),
3779	threadId,
3780	messageId: MessageId.make(`runtime-layer-serialized-queue-first-${automatic}`),
3781	text: "First queued",
3782	attachments: [],
3783	modelSelection,
3784	dispatchMode: { type: "queue_after_active" },
3785	});
3786	yield* orchestrator.dispatch({
3787	type: "message.dispatch",
3788	createdBy: "user",
3789	creationSource: "web",
3790	commandId: CommandId.make(`runtime-layer-serialized-queue-second-${automatic}`),
3791	threadId,
3792	messageId: MessageId.make(`runtime-layer-serialized-queue-second-${automatic}`),
3793	text: "Second queued",
3794	attachments: [],
3795	modelSelection,
3796	dispatchMode: { type: "queue_after_active" },
3797	});
3798
3799	const before = yield* orchestrator.getThreadProjection(threadId);
3800	const activeRun = before.runs.find((run) => run.status === "starting");
3801	const queuedRuns = before.runs
3802	.filter((run) => run.status === "queued")
3803	.toSorted((left, right) => left.ordinal - right.ordinal);
3804	const firstQueuedRun = queuedRuns[0];
3805	const secondQueuedRun = queuedRuns[1];
3806	assert.isDefined(activeRun);
3807	assert.isDefined(firstQueuedRun);
3808	assert.isDefined(secondQueuedRun);
3809	assert.isFalse(
3810	before.turnItems.some(
3811	(item) =>
3812	item.type === "user_message" &&
3813	(item.messageId === firstQueuedRun.userMessageId ||
3814	item.messageId === secondQueuedRun.userMessageId),
3815	),
3816	"queued messages must not exist as turn items before dispatch",
3817	);
3818
3819	const promotedRunIds = yield* Queue.unbounded<RunId>();
3820	const afterSequence = yield* orchestrator.getThreadEventSequence(threadId);
3821	yield* eventSink.stream({ threadId, afterSequence }).pipe(
3822	Stream.runForEach((stored) =>
3823	stored.event.type === "run.updated" && stored.event.payload.status === "starting"
3824	? Queue.offer(promotedRunIds, stored.event.payload.id)
3825	: Effect.void,
3826	),
3827	Effect.forkScoped,
3828	);
3829	yield* Effect.yieldNow;
3830
3831	const activeCompletedAt = yield* DateTime.now;
3832	yield* eventSink.write({
3833	events: [
3834	{
3835	id: EventId.make(`runtime-layer-serialized-queue-active-completed-${automatic}`),
3836	type: "run.updated",
3837	threadId,
3838	runId: activeRun.id,
3839	...(activeRun.rootNodeId === null ? {} : { nodeId: activeRun.rootNodeId }),
3840	providerInstanceId: activeRun.providerInstanceId,
3841	occurredAt: activeCompletedAt,
3842	payload: {
3843	...activeRun,
3844	status: "completed",
3845	completedAt: activeCompletedAt,
3846	},
3847	},
3848	],
3849	});
3850
3851	assert.equal(yield* Queue.take(promotedRunIds), firstQueuedRun.id);
3852	const afterFirstPromotion = yield* orchestrator.getThreadProjection(threadId);
3853	assert.equal(
3854	afterFirstPromotion.runs.find((run) => run.id === firstQueuedRun.id)?.status,
3855	"starting",
3856	);
3857	assert.equal(
3858	afterFirstPromotion.runs.find((run) => run.id === secondQueuedRun.id)?.status,
3859	"queued",
3860	);
3861	const promotedMessageItem = afterFirstPromotion.turnItems.find(
3862	(item) =>
3863	item.runId === firstQueuedRun.id &&
3864	(item.type === "user_message" || item.type === "notification"),
3865	);
3866	assert.isDefined(promotedMessageItem);
3867	if (automatic) {
3868	assert.equal(promotedMessageItem.type, "notification");
3869	assert.equal(
3870	afterFirstPromotion.messages.find(
3871	(message) => message.id === firstQueuedRun.userMessageId,
3872	)?.text,
3873	"First queued",
3874	);
3875	assert.equal(
3876	afterFirstPromotion.messages.find(
3877	(message) => message.id === firstQueuedRun.userMessageId,
3878	)?.notification?.summary,
3879	"Monitor updated",
3880	);
3881	assert.isFalse(
3882	afterFirstPromotion.turnItems.some(
3883	(item) =>
3884	item.type === "user_message" && item.messageId === firstQueuedRun.userMessageId,
3885	),
3886	);
3887	} else {
3888	assert.equal(promotedMessageItem.type, "user_message");
3889	}
3890	assert.isTrue(
3891	promotedMessageItem.startedAt !== null &&
3892	DateTime.toEpochMillis(promotedMessageItem.startedAt) >=
3893	DateTime.toEpochMillis(activeCompletedAt),
3894	);
3895
3896	const promotedFirst = afterFirstPromotion.runs.find((run) => run.id === firstQueuedRun.id);
3897	assert.isDefined(promotedFirst);
3898	const firstCompletedAt = yield* DateTime.now;
3899	yield* eventSink.write({
3900	events: [
3901	{
3902	id: EventId.make(`runtime-layer-serialized-queue-first-completed-${automatic}`),
3903	type: "run.updated",
3904	threadId,
3905	runId: promotedFirst.id,
3906	...(promotedFirst.rootNodeId === null ? {} : { nodeId: promotedFirst.rootNodeId }),
3907	providerInstanceId: promotedFirst.providerInstanceId,
3908	occurredAt: firstCompletedAt,
3909	payload: {
3910	...promotedFirst,
3911	status: "completed",
3912	completedAt: firstCompletedAt,
3913	},
3914	},
3915	],
3916	});
3917
3918	assert.equal(yield* Queue.take(promotedRunIds), secondQueuedRun.id);
3919	const afterSecondPromotion = yield* orchestrator.getThreadProjection(threadId);
3920	assert.equal(
3921	afterSecondPromotion.runs.find((run) => run.id === firstQueuedRun.id)?.status,
3922	"completed",
3923	);
3924	assert.equal(
3925	afterSecondPromotion.runs.find((run) => run.id === secondQueuedRun.id)?.status,
3926	"starting",
3927	);
3928	}),
3929	);
3930
3931	it.effect("starts a wake's work clock from the run that ran before it", () =>
3932	Effect.gen(function* () {
3933	const orchestrator = yield* Orchestrator.OrchestratorV2;
3934	const eventSink = yield* EventSink.EventSinkV2;
3935	const threadId = ThreadId.make("runtime-layer-wake-work-start-thread");
3936	const messageId = (key: string) => MessageId.make(`runtime-layer-wake-work-start-${key}`);
3937
3938	yield* orchestrator.dispatch({
3939	type: "thread.create",
3940	createdBy: "user",
3941	creationSource: "web",
3942	commandId: CommandId.make("runtime-layer-wake-work-start-create"),
3943	threadId,
3944	projectId: ProjectId.make("runtime-layer-wake-work-start-project"),
3945	title: "Wake work start",
3946	modelSelection,
3947	runtimeMode: "full-access",
3948	interactionMode: "default",
3949	branch: null,
3950	worktreePath: process.cwd(),
3951	});
3952	const dispatch = (key: string, wake: boolean) =>
3953	orchestrator.dispatch({
3954	type: "message.dispatch",
3955	createdBy: wake ? "agent" : "user",
3956	creationSource: wake ? "provider" : "web",
3957	...(wake
3958	? {
3959	notification: {
3960	source: { kind: "background_task" as const },
3961	outcome: "updated" as const,
3962	summary: "Background activity updated",
3963	},
3964	}
3965	: {}),
3966	commandId: CommandId.make(`runtime-layer-wake-work-start-${key}`),
3967	threadId,
3968	messageId: messageId(key),
3969	text: key,
3970	attachments: [],
3971	modelSelection,
3972	dispatchMode:
3973	key === "prompt" ? { type: "start_immediately" } : { type: "queue_after_active" },
3974	});
3975	const runFor = (key: string) =>
3976	Effect.map(orchestrator.getThreadProjection(threadId), ({ runs }) => {
3977	const run = runs.find((candidate) => candidate.userMessageId === messageId(key));
3978	assert.isDefined(run);
3979	return run;
3980	});
3981	yield* dispatch("prompt", false);
3982	yield* dispatch("queued", false);
3983	yield* dispatch("early-wake", true);
3984	// The early wake now runs ahead of the older queued prompt, as a
3985	// delegated result does when it jumps the queue.
3986	yield* orchestrator.dispatch({
3987	type: "queued-run.reorder",
3988	commandId: CommandId.make("runtime-layer-wake-work-start-reorder"),
3989	threadId,
3990	runId: (yield* runFor("queued")).id,
3991	beforeRunId: null,
3992	});
3993	yield* dispatch("late-wake", true);
3994	// A queued wake has no clock yet: what runs before it is still unknown.
3995	assert.isUndefined((yield* runFor("early-wake")).workStartedAt);
3996
3997	const startedRunIds = yield* Queue.unbounded<RunId>();
3998	const afterSequence = yield* orchestrator.getThreadEventSequence(threadId);
3999	yield* eventSink.stream({ threadId, afterSequence }).pipe(
4000	Stream.runForEach((stored) =>
4001	stored.event.type === "run.updated" && stored.event.payload.status === "starting"
4002	? Queue.offer(startedRunIds, stored.event.payload.id)
4003	: Effect.void,
4004	),
4005	Effect.forkScoped,
4006	);
4007	yield* Effect.yieldNow;
4008
4009	const now = yield* DateTime.now;
4010	// Runs start and settle the way the provider would report them.
4011	const settle = (key: string, startedAt: DateTime.Utc) =>
4012	Effect.gen(function* () {
4013	const run = yield* runFor(key);
4014	yield* eventSink.write({
4015	events: [
4016	{
4017	id: EventId.make(`runtime-layer-wake-work-start-${key}-completed`),
4018	type: "run.updated",
4019	threadId,
4020	runId: run.id,
4021	...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }),
4022	providerInstanceId: run.providerInstanceId,
4023	occurredAt: startedAt,
4024	payload: { ...run, status: "completed", startedAt, completedAt: startedAt },
4025	},
4026	],
4027	});
4028	});
4029	const millis = (value: DateTime.Utc | undefined) =>
4030	value === undefined ? undefined : DateTime.toEpochMillis(value);
4031
4032	yield* settle("prompt", now);
4033	assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("early-wake")).id);
4034	assert.equal(millis((yield* runFor("early-wake")).workStartedAt), millis(now));
4035
4036	yield* settle("early-wake", DateTime.add(now, { seconds: 1 }));
4037	assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("queued")).id);
4038	assert.isUndefined((yield* runFor("queued")).workStartedAt);
4039
4040	// The queued prompt starts long after it was requested; the wake after it
4041	// counts from that start, not from the request.
4042	const queuedStartedAt = DateTime.add(now, { minutes: 10 });
4043	yield* settle("queued", queuedStartedAt);
4044	assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("late-wake")).id);
4045	assert.equal(millis((yield* runFor("late-wake")).workStartedAt), millis(queuedStartedAt));
4046	}),
4047	);
4048
4049	it.effect.each(["usage_limit", "provider_error"] as const)(
4050	"handles a queued message after a %s failure",
4051	(failureClass) =>
4052	Effect.gen(function* () {
4053	const orchestrator = yield* Orchestrator.OrchestratorV2;
4054	const eventSink = yield* EventSink.EventSinkV2;
4055	const threadId = ThreadId.make(`runtime-layer-failed-queue-${failureClass}`);
4056
4057	yield* orchestrator.dispatch({
4058	type: "thread.create",
4059	createdBy: "user",
4060	creationSource: "web",
4061	commandId: CommandId.make(`${threadId}:create`),
4062	threadId,
4063	projectId: ProjectId.make(`${threadId}:project`),
4064	title: "Failed queue",
4065	modelSelection,
4066	runtimeMode: "full-access",
4067	interactionMode: "default",
4068	branch: null,
4069	worktreePath: process.cwd(),
4070	});
4071	for (const index of [0, 1]) {
4072	yield* orchestrator.dispatch({
4073	type: "message.dispatch",
4074	createdBy: "user",
4075	creationSource: "web",
4076	commandId: CommandId.make(`${threadId}:message:${index}`),
4077	threadId,
4078	messageId: MessageId.make(`${threadId}:message:${index}`),
4079	text: index === 0 ? "Active" : "Queued",
4080	attachments: [],
4081	modelSelection,
4082	dispatchMode: { type: index === 0 ? "start_immediately" : "queue_after_active" },
4083	});
4084	}
4085
4086	const before = yield* orchestrator.getThreadProjection(threadId);
4087	const activeRun = before.runs.find((run) => run.status === "starting");
4088	const queuedRun = before.runs.find((run) => run.status === "queued");
4089	assert.isDefined(activeRun);
4090	assert.isDefined(queuedRun);
4091	assert.isNotNull(activeRun.rootNodeId);
4092
4093	const promotedRunIds = yield* Queue.unbounded<RunId>();
4094	const heldRunIds = yield* Queue.unbounded<RunId>();
4095	const afterSequence = yield* orchestrator.getThreadEventSequence(threadId);
4096	yield* eventSink.stream({ threadId, afterSequence }).pipe(
4097	Stream.runForEach((stored) =>
4098	stored.event.type !== "run.updated"
4099	? Effect.void
4100	: stored.event.payload.status === "starting"
4101	? Queue.offer(promotedRunIds, stored.event.payload.id)
4102	: stored.event.payload.queueHeld === true
4103	? Queue.offer(heldRunIds, stored.event.payload.id)
4104	: Effect.void,
4105	),
4106	Effect.forkScoped,
4107	);
4108	yield* Effect.yieldNow;
4109
4110	const now = yield* DateTime.now;
4111	yield* eventSink.write({
4112	events: [
4113	{
4114	id: EventId.make(`${threadId}:error`),
4115	type: "turn-item.updated",
4116	threadId,
4117	runId: activeRun.id,
4118	nodeId: activeRun.rootNodeId,
4119	providerInstanceId: activeRun.providerInstanceId,
4120	occurredAt: now,
4121	payload: {
4122	id: TurnItemId.make(`${threadId}:error`),
4123	type: "error",
4124	threadId,
4125	runId: activeRun.id,
4126	nodeId: activeRun.rootNodeId,
4127	providerThreadId: activeRun.providerThreadId,
4128	providerTurnId: null,
4129	nativeItemRef: null,
4130	parentItemId: null,
4131	ordinal: 2,
4132	status: "failed",
4133	title: "Provider failure",
4134	startedAt: now,
4135	completedAt: now,
4136	updatedAt: now,
4137	failure: {
4138	class: failureClass,
4139	message: "Provider failed.",
4140	code: "provider_failed",
4141	retryable: null,
4142	...(failureClass === "usage_limit"
4143	? { resetAt: DateTime.formatIso(DateTime.add(now, { hours: 1 })) }
4144	: {}),
4145	},
4146	},
4147	},
4148	{
4149	id: EventId.make(`${threadId}:failed`),
4150	type: "run.updated",
4151	threadId,
4152	runId: activeRun.id,
4153	nodeId: activeRun.rootNodeId,
4154	providerInstanceId: activeRun.providerInstanceId,
4155	occurredAt: now,
4156	payload: { ...activeRun, status: "failed", completedAt: now },
4157	},
4158	],
4159	});
4160
4161	if (failureClass === "provider_error") {
4162	assert.equal(yield* Queue.take(heldRunIds), queuedRun.id);
4163	const held = yield* orchestrator.getThreadProjection(threadId);
4164	assert.equal(held.runs.find((run) => run.id === queuedRun.id)?.status, "queued");
4165	yield* orchestrator.dispatch({
4166	type: "queue.resume",
4167	commandId: CommandId.make(`${threadId}:resume`),
4168	threadId,
4169	});
4170	assert.equal(yield* Queue.take(promotedRunIds), queuedRun.id);
4171	return;
4172	}
4173	yield* orchestrator.resumeQueuedRuns;
4174	const after = yield* orchestrator.getThreadProjection(threadId);
4175	assert.equal(after.runs.find((run) => run.id === queuedRun.id)?.status, "queued");
4176	assert.isFalse(after.turnItems.some((item) => item.runId === queuedRun.id));
4177	const shell = (yield* orchestrator.getShellSnapshot()).threads.find(
4178	(thread) => thread.id === threadId,
4179	);
4180	assert.equal(shell?.latestRunId, activeRun.id);
4181	assert.equal(shell?.status, "failed");
4182	assert.equal(shell?.lastErrorClass, "usage_limit");
4183	}),
4184	);
4185
4186	it.effect("keeps the queue after a user interrupts the active run", () =>
4187	Effect.gen(function* () {
4188	const orchestrator = yield* Orchestrator.OrchestratorV2;
4189	const threadId = ThreadId.make("runtime-layer-interrupted-queue");
4190	yield* orchestrator.dispatch({
4191	type: "thread.create",
4192	createdBy: "user",
4193	creationSource: "web",
4194	commandId: CommandId.make(`${threadId}:create`),
4195	threadId,
4196	projectId: ProjectId.make(`${threadId}:project`),
4197	title: "Interrupted queue",
4198	modelSelection,
4199	runtimeMode: "full-access",
4200	interactionMode: "default",
4201	branch: null,
4202	worktreePath: process.cwd(),
4203	});
4204	for (const [index, text] of ["Active", "Queued"].entries()) {
4205	yield* orchestrator.dispatch({
4206	type: "message.dispatch",
4207	createdBy: "user",
4208	creationSource: "web",
4209	commandId: CommandId.make(`${threadId}:message:${index}`),
4210	threadId,
4211	messageId: MessageId.make(`${threadId}:message:${index}`),
4212	text,
4213	attachments: [],
4214	modelSelection,
4215	dispatchMode: { type: index === 0 ? "start_immediately" : "queue_after_active" },
4216	});
4217	}
4218	const before = yield* orchestrator.getThreadProjection(threadId);
4219	const activeRun = before.runs[0]!;
4220	const queuedRun = before.runs[1]!;
4221	yield* orchestrator.dispatch({
4222	type: "run.interrupt",
4223	commandId: CommandId.make(`${threadId}:interrupt`),
4224	threadId,
4225	runId: activeRun.id,
4226	holdQueue: true,
4227	});
4228
4229	yield* orchestrator.resumeQueuedRuns;
4230	const after = yield* orchestrator.getThreadProjection(threadId);
4231	assert.equal(after.runs.find((run) => run.id === activeRun.id)?.status, "interrupted");
4232	assert.equal(after.runs.find((run) => run.id === queuedRun.id)?.status, "queued");
4233	assert.isTrue(after.runs.find((run) => run.id === queuedRun.id)?.queueHeld);
4234	assert.isFalse(after.turnItems.some((item) => item.runId === queuedRun.id));
4235	}),
4236	);
4237
4238	it.effect.each(["startup", "shutdown"] as const)(
4239	"preserves and holds queued messages across %s until explicitly resumed",
4240	(trigger) =>
4241	Effect.gen(function* () {
4242	const orchestrator = yield* Orchestrator.OrchestratorV2;
4243	const recovery = yield* ProviderRuntimeRecoveryService.ProviderRuntimeRecoveryService;
4244	const threadId = ThreadId.make(`queue-hold-${trigger}`);
4245	yield* orchestrator.dispatch({
4246	type: "thread.create",
4247	createdBy: "user",
4248	creationSource: "web",
4249	commandId: CommandId.make(`${threadId}:create`),
4250	threadId,
4251	projectId: ProjectId.make(`${threadId}:project`),
4252	title: "Recover queue",
4253	modelSelection,
4254	runtimeMode: "full-access",
4255	interactionMode: "default",
4256	branch: null,
4257	worktreePath: process.cwd(),
4258	});
4259	for (const [index, text] of ["Active", "First queued", "Second queued"].entries()) {
4260	yield* orchestrator.dispatch({
4261	type: "message.dispatch",
4262	createdBy: "user",
4263	creationSource: "web",
4264	commandId: CommandId.make(`${threadId}:message:${index}`),
4265	threadId,
4266	messageId: MessageId.make(`${threadId}:message:${index}`),
4267	text,
4268	attachments: [],
4269	modelSelection,
4270	dispatchMode: { type: index === 0 ? "start_immediately" : "queue_after_active" },
4271	});
4272	}
4273	const before = yield* orchestrator.getThreadProjection(threadId);
4274	const queued = before.runs.filter((run) => run.status === "queued");
4275	assert.equal(queued.length, 2);
4276	yield* recovery.reconcile(trigger);
4277	// A second boot must preserve the hold, even when only queued work remains.
4278	yield* recovery.reconcile("startup");
4279	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
4280	assert.isTrue((yield* maintenance.rebuild).valid);
4281	assert.equal(yield* orchestrator.resumeQueuedRuns, 0);
4282	const held = yield* orchestrator.getThreadProjection(threadId);
4283	assert.deepEqual(
4284	held.runs.map((run) => run.status),
4285	["cancelled", "queued", "queued"],
4286	);
4287	for (const run of queued) {
4288	assert.deepEqual(
4289	held.runs.find((row) => row.id === run.id),
4290	{ ...run, queueHeld: true },
4291	);
4292	assert.deepEqual(
4293	held.messages.find((row) => row.id === run.userMessageId),
4294	before.messages.find((row) => row.id === run.userMessageId),
4295	);
4296	assert.equal(
4297	held.attempts.find((row) => row.id === run.activeAttemptId)?.status,
4298	"pending",
4299	);
4300	assert.equal(held.nodes.find((row) => row.id === run.rootNodeId)?.status, "pending");
4301	}
4302	// Editing and reordering are allowed without releasing the hold.
4303	const first = queued[0]!;
4304	const second = queued[1]!;
4305	yield* orchestrator.dispatch({
4306	type: "queued-run.edit",
4307	commandId: CommandId.make(`${threadId}:edit`),
4308	threadId,
4309	runId: second.id,
4310	text: "Edited second message",
4311	});
4312	yield* orchestrator.dispatch({
4313	type: "queued-run.reorder",
4314	commandId: CommandId.make(`${threadId}:reorder`),
4315	threadId,
4316	runId: second.id,
4317	beforeRunId: first.id,
4318	});
4319	assert.equal(yield* orchestrator.resumeQueuedRuns, 0);
4320	const resume = {
4321	type: "queue.resume" as const,
4322	commandId: CommandId.make(`${threadId}:resume`),
4323	threadId,
4324	};
4325	yield* orchestrator.dispatch(resume);
4326	yield* orchestrator.dispatch(resume);
4327	const resumed = yield* orchestrator.getThreadProjection(threadId);
4328	assert.equal(resumed.runs.find((run) => run.id === second.id)?.status, "starting");
4329	assert.equal(resumed.runs.find((run) => run.id === first.id)?.status, "queued");
4330	assert.isFalse(resumed.runs.some((run) => run.status === "queued" && run.queueHeld));
4331	assert.equal(
4332	resumed.messages.find((row) => row.id === second.userMessageId)?.text,
4333	"Edited second message",
4334	);
4335	assert.equal(resumed.runs.length, 3, "resume retries must not duplicate messages or runs");
4336	}),
4337	);
4338
4339	it.effect("edits and removes queued runs", () =>
4340	Effect.gen(function* () {
4341	const orchestrator = yield* Orchestrator.OrchestratorV2;
4342	const threadId = ThreadId.make("runtime-layer-queued-edit-thread");
4343
4344	yield* orchestrator.dispatch({
4345	type: "thread.create",
4346	createdBy: "user",
4347	creationSource: "web",
4348	commandId: CommandId.make("runtime-layer-queued-edit-create"),
4349	threadId,
4350	projectId: ProjectId.make("runtime-layer-queued-edit-project"),
4351	title: "Edit queued work",
4352	modelSelection,
4353	runtimeMode: "full-access",
4354	interactionMode: "default",
4355	branch: null,
4356	worktreePath: "/tmp/runtime-layer-queued-edit",
4357	});
4358	yield* orchestrator.dispatch({
4359	type: "message.dispatch",
4360	createdBy: "user",
4361	creationSource: "web",
4362	commandId: CommandId.make("runtime-layer-queued-edit-active-message"),
4363	threadId,
4364	messageId: MessageId.make("runtime-layer-queued-edit-active-message"),
4365	text: "Keep the provider occupied.",
4366	attachments: [],
4367	modelSelection,
4368	dispatchMode: { type: "start_immediately" },
4369	});
4370	yield* orchestrator.dispatch({
4371	type: "message.dispatch",
4372	createdBy: "user",
4373	creationSource: "web",
4374	commandId: CommandId.make("runtime-layer-queued-edit-queued-message"),
4375	threadId,
4376	messageId: MessageId.make("runtime-layer-queued-edit-queued-message"),
4377	text: "Original queued text.",
4378	attachments: [],
4379	modelSelection,
4380	dispatchMode: { type: "queue_after_active" },
4381	});
4382
4383	const before = yield* orchestrator.getThreadProjection(threadId);
4384	const queuedRun = before.runs.find((run) => run.status === "queued");
4385	assert.isDefined(queuedRun);
4386
4387	yield* orchestrator.dispatch({
4388	type: "queued-run.edit",
4389	commandId: CommandId.make("runtime-layer-queued-edit-edit"),
4390	threadId,
4391	runId: queuedRun.id,
4392	text: "Updated queued text.",
4393	});
4394
4395	const afterEdit = yield* orchestrator.getThreadProjection(threadId);
4396	assert.equal(
4397	afterEdit.messages.find((message) => message.id === queuedRun.userMessageId)?.text,
4398	"Updated queued text.",
4399	);
4400	const editedItem = afterEdit.turnItems.find(
4401	(item) => item.type === "user_message" && item.messageId === queuedRun.userMessageId,
4402	);
4403	assert.isUndefined(editedItem, "editing queue state must not create a timeline turn item");
4404
4405	yield* orchestrator.dispatch({
4406	type: "queued-run.edit",
4407	commandId: CommandId.make("runtime-layer-queued-edit-attachments"),
4408	threadId,
4409	runId: queuedRun.id,
4410	text: "Updated queued text with an attachment.",
4411	attachments: [
4412	{
4413	type: "image",
4414	id: "runtime-layer-queued-edit-attachment",
4415	name: "screenshot.png",
4416	mimeType: "image/png",
4417	sizeBytes: 128,
4418	},
4419	],
4420	});
4421	const afterAttachmentEdit = yield* orchestrator.getThreadProjection(threadId);
4422	assert.deepEqual(
4423	afterAttachmentEdit.messages
4424	.find((message) => message.id === queuedRun.userMessageId)
4425	?.attachments.map((attachment) => attachment.id),
4426	["runtime-layer-queued-edit-attachment"],
4427	);
4428
4429	yield* orchestrator.dispatch({
4430	type: "queued-run.edit",
4431	commandId: CommandId.make("runtime-layer-queued-edit-text-only"),
4432	threadId,
4433	runId: queuedRun.id,
4434	text: "Text-only edit keeps attachments.",
4435	});
4436	const afterTextOnlyEdit = yield* orchestrator.getThreadProjection(threadId);
4437	assert.deepEqual(
4438	afterTextOnlyEdit.messages
4439	.find((message) => message.id === queuedRun.userMessageId)
4440	?.attachments.map((attachment) => attachment.id),
4441	["runtime-layer-queued-edit-attachment"],
4442	"an edit without attachments must leave the stored attachments untouched",
4443	);
4444
4445	const emptyEditError = yield* orchestrator
4446	.dispatch({
4447	type: "queued-run.edit",
4448	commandId: CommandId.make("runtime-layer-queued-edit-empty"),
4449	threadId,
4450	runId: queuedRun.id,
4451	text: " ",
4452	})
4453	.pipe(Effect.flip);
4454	assert.equal(emptyEditError._tag, "OrchestratorCommandRejectedError");
4455
4456	yield* orchestrator.dispatch({
4457	type: "queued-run.cancel",
4458	commandId: CommandId.make("runtime-layer-queued-edit-cancel"),
4459	threadId,
4460	runId: queuedRun.id,
4461	});
4462
4463	const afterCancel = yield* orchestrator.getThreadProjection(threadId);
4464	assert.equal(afterCancel.runs.find((run) => run.id === queuedRun.id)?.status, "cancelled");
4465	assert.equal(
4466	afterCancel.attempts.find((attempt) => attempt.runId === queuedRun.id)?.status,
4467	"cancelled",
4468	);
4469	assert.equal(
4470	afterCancel.nodes.find((node) => node.runId === queuedRun.id)?.status,
4471	"cancelled",
4472	);
4473	assert.isFalse(
4474	afterCancel.visibleTurnItems.some(
4475	(row) => row.item.type === "user_message" && row.item.runId === queuedRun.id,
4476	),
4477	"removed queued message must not surface as a transcript row",
4478	);
4479	assert.equal(yield* orchestrator.resumeQueuedRuns, 0);
4480
4481	const cancelAgainError = yield* orchestrator
4482	.dispatch({
4483	type: "queued-run.cancel",
4484	commandId: CommandId.make("runtime-layer-queued-edit-cancel-again"),
4485	threadId,
4486	runId: queuedRun.id,
4487	})
4488	.pipe(Effect.flip);
4489	assert.equal(cancelAgainError._tag, "OrchestratorDispatchError");
4490	}),
4491	);
4492	});
4493
4494	it.layer(layerSharedApplicationDataPlaneTest)("pending provider interruption", (it) => {
4495	it.effect("interrupts a pending provider start without launching provider work", () =>
4496	Effect.gen(function* () {
4497	const projects = yield* ProjectService.ProjectService;
4498	const orchestrator = yield* Orchestrator.OrchestratorV2;
4499	const threadManagement = yield* ThreadManagementService.ThreadManagementService;
4500	const effectWorker = yield* EffectWorker.OrchestrationEffectWorkerV2;
4501	const projectId = ProjectId.make("runtime-layer-pending-interrupt-project");
4502	const threadId = ThreadId.make("runtime-layer-pending-interrupt-thread");
4503
4504	yield* projects.create({
4505	commandId: CommandId.make("runtime-layer-pending-interrupt-project-create"),
4506	projectId,
4507	title: "Pending interrupt project",
4508	workspaceRoot: "/tmp/runtime-layer-pending-interrupt-project",
4509	});
4510	yield* orchestrator.dispatch({
4511	type: "thread.create",
4512	createdBy: "user",
4513	creationSource: "web",
4514	commandId: CommandId.make("runtime-layer-pending-interrupt-create"),
4515	threadId,
4516	projectId,
4517	title: "Pending interrupt",
4518	modelSelection,
4519	runtimeMode: "full-access",
4520	interactionMode: "default",
4521	branch: null,
4522	worktreePath: null,
4523	});
4524	yield* orchestrator.dispatch({
4525	type: "message.dispatch",
4526	createdBy: "user",
4527	creationSource: "web",
4528	commandId: CommandId.make("runtime-layer-pending-interrupt-message"),
4529	threadId,
4530	messageId: MessageId.make("runtime-layer-pending-interrupt-message"),
4531	text: "Do not reach the provider.",
4532	attachments: [],
4533	modelSelection,
4534	dispatchMode: { type: "start_immediately" },
4535	});
4536
4537	const starting = yield* orchestrator.getThreadProjection(threadId);
4538	const run = starting.runs[0];
4539	assert.isDefined(run);
4540	assert.equal(run.status, "starting");
4541
4542	const interrupt = yield* threadManagement.interruptThread({
4543	projectId,
4544	commandId: CommandId.make("runtime-layer-pending-interrupt-command"),
4545	threadId,
4546	runId: run.id,
4547	reason: "Cancelled before provider start",
4548	});
4549	assert.equal(interrupt.type, "interrupt_requested");
4550
4551	const interrupted = yield* orchestrator.getThreadProjection(threadId);
4552	assert.equal(interrupted.runs[0]?.status, "interrupted");
4553	assert.equal(interrupted.attempts[0]?.status, "interrupted");
4554	assert.equal(
4555	interrupted.nodes.find((node) => node.kind === "root_turn")?.status,
4556	"interrupted",
4557	);
4558	assert.deepEqual(
4559	interrupted.turnItems.filter((item) => item.runId === run.id).map((item) => item.type),
4560	["user_message", "run_interrupt_request", "run_interrupt_result"],
4561	);
4562	assert.deepEqual(interrupted.providerTurns, []);
4563	assert.isFalse(yield* effectWorker.runOnce);
4564	}),
4565	);
4566	});
4567
4568	it.layer(layerTest)("snooze queued runs", (it) => {
4569	it.effect.each(["notification", "delegated-completion", "user"] as const)(
4570	"only user messages block snooze with a held %s run",
4571	(kind) =>
4572	Effect.gen(function* () {
4573	const orchestrator = yield* Orchestrator.OrchestratorV2;
4574	const eventSink = yield* EventSink.EventSinkV2;
4575	const threadId = ThreadId.make(`snooze-queued:${kind}`);
4576	const snoozedUntil = "2099-07-25T09:00:00.000Z";
4577	yield* orchestrator.dispatch({
4578	type: "thread.create",
4579	commandId: CommandId.make(`create:${threadId}`),
4580	threadId,
4581	projectId: ProjectId.make("snooze-queued-project"),
4582	title: "Snooze with held queued work",
4583	modelSelection,
4584	runtimeMode: "full-access",
4585	interactionMode: "default",
4586	branch: null,
4587	worktreePath: "/tmp/snooze-queued",
4588	createdBy: "user",
4589	creationSource: "web",
4590	});
4591	yield* orchestrator.dispatch({
4592	type: "message.dispatch",
4593	commandId: CommandId.make(`active:${threadId}`),
4594	threadId,
4595	messageId: MessageId.make(`active:${threadId}`),
4596	text: "Active run",
4597	attachments: [],
4598	modelSelection,
4599	dispatchMode: { type: "start_immediately" },
4600	createdBy: "user",
4601	creationSource: "web",
4602	});
4603	const activeRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
4604	const now = yield* DateTime.now;
4605	if (kind === "delegated-completion") {
4606	yield* eventSink.write({
4607	events: [
4608	{
4609	id: EventId.make(`completion:${threadId}`),
4610	type: "run.updated",
4611	threadId,
4612	runId: activeRun.id,
4613	occurredAt: now,
4614	payload: {
4615	...activeRun,
4616	delegatedCompletion: {
4617	disposition: "open",
4618	nextGeneration: 2,
4619	delivery: {
4620	messageId: MessageId.make(`queued:${threadId}`),
4621	generation: 1,
4622	taskIds: [NodeId.make("snooze-queued-task")],
4623	},
4624	},
4625	},
4626	},
4627	],
4628	});
4629	}
4630	yield* orchestrator.dispatch({
4631	type: "message.dispatch",
4632	commandId: CommandId.make(`queued:${threadId}`),
4633	threadId,
4634	messageId: MessageId.make(`queued:${threadId}`),
4635	text: "Queued work",
4636	attachments: [],
4637	modelSelection,
4638	dispatchMode: { type: "queue_after_active" },
4639	createdBy: kind === "user" ? "user" : "agent",
4640	creationSource: kind === "user" ? "web" : "server",
4641	...(kind === "notification"
4642	? {
4643	notification: {
4644	source: { kind: "background_task" as const },
4645	outcome: "updated" as const,
4646	summary: "Background activity updated",
4647	},
4648	}
4649	: kind === "delegated-completion"
4650	? {
4651	delegatedCompletion: {
4652	parentRunId: activeRun.id,
4653	generation: 1,
4654	taskIds: [NodeId.make("snooze-queued-task")],
4655	},
4656	}
4657	: {}),
4658	});
4659	// Recovery holds the queue while the thread's active run ends.
4660	const beforeRecovery = yield* orchestrator.getThreadProjection(threadId);
4661	yield* eventSink.commitCommand({
4662	commandId: CommandId.make(`recovery:${threadId}`),
4663	threadId,
4664	commandType: "provider-runtime.reconcile",
4665	acceptedAt: now,
4666	events: beforeRecovery.runs.map((run) => ({
4667	id: EventId.make(`recovery:${run.id}`),
4668	type: "run.updated" as const,
4669	threadId,
4670	runId: run.id,
4671	occurredAt: now,
4672	payload:
4673	run.status === "queued"
4674	? { ...run, queueHeld: true }
4675	: { ...run, status: "completed" as const, completedAt: now },
4676	})),
4677	effects: [],
4678	});
4679	const beforeSnooze = yield* orchestrator.getThreadProjection(threadId);
4680	assert.isTrue(beforeSnooze.runs.some((run) => run.status === "queued" && run.queueHeld));
4681	const snooze = orchestrator.dispatch({
4682	type: "thread.snooze",
4683	commandId: CommandId.make(`snooze:${threadId}`),
4684	threadId,
4685	snoozedUntil,
4686	});
4687	if (kind === "user") {
4688	const error = yield* snooze.pipe(Effect.flip);
4689	assert.equal(error._tag, "OrchestratorDispatchError");
4690	assert.equal(error.cause, `Thread ${threadId} has a queued run and cannot be snoozed.`);
4691	assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.snoozedUntil);
4692	} else {
4693	yield* snooze;
4694	const snoozed = yield* orchestrator.getThreadProjection(threadId);
4695	assert.equal(DateTime.formatIso(snoozed.thread.snoozedUntil!), snoozedUntil);
4696	assert.deepEqual(snoozed.runs, beforeSnooze.runs);
4697	yield* orchestrator.dispatch({
4698	type: "thread.unsnooze",
4699	commandId: CommandId.make(`unsnooze:${threadId}`),
4700	threadId,
4701	reason: "user",
4702	});
4703	const awake = yield* orchestrator.getThreadProjection(threadId);
4704	assert.isNull(awake.thread.snoozedUntil);
4705	assert.deepEqual(awake.runs, beforeSnooze.runs);
4706	}
4707	}),
4708	);
4709	});
4710
4711	it.layer(layerSharedApplicationDataPlaneTest)("snooze projection", (it) => {
4712	it.effect("carries snooze state through the V2 shell projection", () =>
4713	Effect.gen(function* () {
4714	const projects = yield* ProjectService.ProjectService;
4715	const orchestrator = yield* Orchestrator.OrchestratorV2;
4716	const projectId = ProjectId.make("runtime-layer-snoozed-project");
4717	const threadId = ThreadId.make("runtime-layer-snoozed-thread");
4718	const snoozedUntil = "2099-07-25T09:00:00.000Z";
4719
4720	yield* projects.create({
4721	commandId: CommandId.make("runtime-layer-snoozed-project-create"),
4722	projectId,
4723	title: "Snoozed shell projection",
4724	workspaceRoot: "/tmp/runtime-layer-snoozed-project",
4725	});
4726	yield* orchestrator.dispatch({
4727	type: "thread.create",
4728	createdBy: "user",
4729	creationSource: "web",
4730	commandId: CommandId.make("runtime-layer-snoozed-thread-create"),
4731	threadId,
4732	projectId,
4733	title: "Snoozed thread",
4734	modelSelection,
4735	runtimeMode: "full-access",
4736	interactionMode: "default",
4737	branch: null,
4738	worktreePath: null,
4739	});
4740	yield* orchestrator.dispatch({
4741	type: "thread.snooze",
4742	commandId: CommandId.make("runtime-layer-snoozed-thread-snooze"),
4743	threadId,
4744	snoozedUntil,
4745	});
4746
4747	const firstProjection = yield* orchestrator.getThreadProjection(threadId);
4748	const firstSnoozedAt = firstProjection.thread.snoozedAt;
4749	const firstUpdatedAt = firstProjection.thread.updatedAt;
4750	assert.isNotNull(firstSnoozedAt);
4751
4752	yield* orchestrator.dispatch({
4753	type: "thread.snooze",
4754	commandId: CommandId.make("runtime-layer-snoozed-thread-snooze-again"),
4755	threadId,
4756	snoozedUntil,
4757	});
4758
4759	const shell = yield* orchestrator.getShellSnapshot();
4760	const thread = shell.threads.find((candidate) => candidate.id === threadId);
4761	assert.isDefined(thread);
4762	assert.equal(DateTime.formatIso(thread.snoozedUntil!), snoozedUntil);
4763	assert.deepEqual(thread.snoozedAt, firstSnoozedAt);
4764	assert.deepEqual(thread.updatedAt, firstUpdatedAt);
4765
4766	yield* orchestrator.dispatch({
4767	type: "thread.unsnooze",
4768	commandId: CommandId.make("runtime-layer-snoozed-thread-wake"),
4769	threadId,
4770	reason: "user",
4771	});
4772	const projections = yield* ProjectionStore.ProjectionStoreV2.pipe(
4773	Effect.provide(ProjectionStore.layer),
4774	);
4775	const [candidate] = yield* projections.getSettlementCandidates(threadId);
4776	assert.deepEqual(candidate?.lastSnoozeWakeAt, yield* DateTime.now);
4777	yield* orchestrator.dispatch({
4778	type: "thread.snooze",
4779	commandId: CommandId.make("runtime-layer-snoozed-thread-snooze-after-wake"),
4780	threadId,
4781	snoozedUntil,
4782	});
4783
4784	yield* orchestrator.dispatch({
4785	type: "message.dispatch",
4786	createdBy: "user",
4787	creationSource: "web",
4788	commandId: CommandId.make("runtime-layer-snoozed-message"),
4789	threadId,
4790	messageId: MessageId.make("runtime-layer-snoozed-message"),
4791	text: "Wake this thread.",
4792	attachments: [],
4793	modelSelection,
4794	dispatchMode: { type: "start_immediately" },
4795	});
4796
4797	const awakened = yield* orchestrator.getThreadProjection(threadId);
4798	assert.isNull(awakened.thread.snoozedUntil);
4799	assert.isNull(awakened.thread.snoozedAt);
4800	}),
4801	);
4802	});
4803
4804	it.layer(layerSharedApplicationDataPlaneTest)("visited projection", (it) => {
4805	it.effect("carries the visited watermark through the V2 shell projection", () =>
4806	Effect.gen(function* () {
4807	const projects = yield* ProjectService.ProjectService;
4808	const orchestrator = yield* Orchestrator.OrchestratorV2;
4809	const projectId = ProjectId.make("runtime-layer-visited-project");
4810	const threadId = ThreadId.make("runtime-layer-visited-thread");
4811	const visitedAt = "2026-07-24T01:00:00.000Z";
4812
4813	yield* projects.create({
4814	commandId: CommandId.make("runtime-layer-visited-project-create"),
4815	projectId,
4816	title: "Visited shell projection",
4817	workspaceRoot: "/tmp/runtime-layer-visited-project",
4818	});
4819	yield* orchestrator.dispatch({
4820	type: "thread.create",
4821	createdBy: "user",
4822	creationSource: "web",
4823	commandId: CommandId.make("runtime-layer-visited-thread-create"),
4824	threadId,
4825	projectId,
4826	title: "Visited thread",
4827	modelSelection,
4828	runtimeMode: "full-access",
4829	interactionMode: "default",
4830	branch: null,
4831	worktreePath: null,
4832	});
4833	const created = yield* orchestrator.getThreadProjection(threadId);
4834	assert.isNull(created.thread.lastVisitedAt);
4835	const createdUpdatedAt = created.thread.updatedAt;
4836
4837	yield* TestClock.adjust("1 second");
4838	yield* orchestrator.dispatch({
4839	type: "thread.visit",
4840	commandId: CommandId.make("runtime-layer-visited-thread-visit"),
4841	threadId,
4842	visitedAt,
4843	});
4844	const visited = yield* orchestrator.getThreadProjection(threadId);
4845	assert.isNotNull(visited.thread.lastVisitedAt);
4846	assert.equal(DateTime.formatIso(visited.thread.lastVisitedAt!), visitedAt);
4847	// Visiting records read state, not activity: updatedAt must not move.
4848	assert.deepEqual(visited.thread.updatedAt, createdUpdatedAt);
4849
4850	// Monotonic: an older watermark (a replay or a stale device) cannot
4851	// rewind the marker.
4852	yield* orchestrator.dispatch({
4853	type: "thread.visit",
4854	commandId: CommandId.make("runtime-layer-visited-thread-visit-stale"),
4855	threadId,
4856	visitedAt: "2026-07-24T00:30:00.000Z",
4857	});
4858	const afterStaleVisit = yield* orchestrator.getThreadProjection(threadId);
4859	assert.equal(DateTime.formatIso(afterStaleVisit.thread.lastVisitedAt!), visitedAt);
4860
4861	const shell = yield* orchestrator.getShellSnapshot();
4862	const thread = shell.threads.find((candidate) => candidate.id === threadId);
4863	assert.isDefined(thread);
4864	assert.equal(DateTime.formatIso(thread!.lastVisitedAt!), visitedAt);
4865	assert.deepEqual(thread!.updatedAt, createdUpdatedAt);
4866
4867	// No completed run yet → nothing to mark unread against.
4868	const markUnread = yield* orchestrator
4869	.dispatch({
4870	type: "thread.mark-unread",
4871	commandId: CommandId.make("runtime-layer-visited-thread-mark-unread"),
4872	threadId,
4873	})
4874	.pipe(Effect.flip);
4875	assert.instanceOf(markUnread, Orchestrator.OrchestratorDispatchError);
4876
4877	// A read receipt must not decode any transcript, including inherited or
4878	// unreadable historical rows. It only advances the thread's watermark.
4879	const sql = yield* SqlClient.SqlClient;
4880	yield* sql`INSERT INTO orchestration_v2_projection_turn_items (
4881	turn_item_id, thread_id, ordinal, type, status, updated_at, payload_json
4882	) VALUES (
4883	'item:visited:unreadable-history', ${threadId}, 1, 'dynamic_tool', 'completed',
4884	${visitedAt}, '{broken'
4885	)`;
4886	const nextVisitedAt = "2026-07-24T02:00:00.000Z";
4887	yield* orchestrator.dispatch({
4888	type: "thread.visit",
4889	commandId: CommandId.make("runtime-layer-visited-without-history"),
4890	threadId,
4891	visitedAt: nextVisitedAt,
4892	});
4893	const [watermark] = yield* sql<{ readonly visited_at: string; readonly updated_at: string }>`
4894	SELECT json_extract(payload_json, '$.lastVisitedAt') AS visited_at, updated_at
4895	FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId}
4896	`;
4897	assert.equal(watermark!.visited_at, nextVisitedAt);
4898	assert.equal(watermark!.updated_at, DateTime.formatIso(createdUpdatedAt));
4899	const invalidVisit = yield* orchestrator
4900	.dispatch({
4901	type: "thread.visit",
4902	commandId: CommandId.make("runtime-layer-visited-invalid-timestamp"),
4903	threadId,
4904	visitedAt: "invalid-timestamp",
4905	})
4906	.pipe(Effect.flip);
4907	assert.instanceOf(invalidVisit, Orchestrator.OrchestratorDispatchError);
4908	}),
4909	);
4910	});
4911
4912	it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", (it) => {
4913	it.effect("persists only the changed PR link and replays it without losing other links", () =>
4914	Effect.gen(function* () {
4915	const orchestrator = yield* Orchestrator.OrchestratorV2;
4916	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
4917	const sql = yield* SqlClient.SqlClient;
4918	const threadId = ThreadId.make("compact-pr-sync");
4919	yield* orchestrator.dispatch({
4920	type: "thread.create",
4921	createdBy: "user",
4922	creationSource: "web",
4923	commandId: CommandId.make("compact-pr-create"),
4924	threadId,
4925	projectId: ProjectId.make("compact-pr-project"),
4926	title: "PR sync",
4927	modelSelection,
4928	runtimeMode: "full-access",
4929	interactionMode: "default",
4930	branch: null,
4931	worktreePath: null,
4932	});
4933	const key = { host: "github.com", repository: "pingdotgg/t3code" };
4934	for (const number of [1, 2, 3]) {
4935	yield* orchestrator.dispatch({
4936	type: "thread.pull-request.link",
4937	commandId: CommandId.make(`compact-pr-link-${number}`),
4938	threadId,
4939	...key,
4940	number,
4941	url: `https://github.com/pingdotgg/t3code/pull/${number}`,
4942	source: "manual",
4943	});
4944	}
4945	const before = yield* orchestrator.getThreadShell(threadId);
4946	yield* orchestrator.dispatch({
4947	type: "thread.pull-request-link.sync",
4948	commandId: CommandId.make("compact-pr-sync"),
4949	threadId,
4950	...key,
4951	number: 2,
4952	snapshot: {
4953	state: "open",
4954	title: "Updated PR",
4955	headBranch: "feature/pr",
4956	baseBranch: "main",
4957	isDraft: false,
4958	updatedAt: null,
4959	syncedAt: "2026-10-13T12:00:00.000Z",
4960	},
4961	stack: {
4962	kind: "native",
4963	id: "updated",
4964	number: 2,
4965	url: "https://github.com/pingdotgg/t3code/pull/2",
4966	base: "main",
4967	layers: [],
4968	},
4969	});
4970	const rows = yield* sql<{ event_type: string; payload_json: string }>`
4971	SELECT event_type, payload_json FROM orchestration_events
4972	WHERE command_id = ${"compact-pr-sync"}
4973	`;
4974	assert.equal(rows.length, 1);
4975	assert.equal(rows[0]?.event_type, "thread.pull-request-link-updated");
4976	const payload = JSON.parse(rows[0]!.payload_json);
4977	assert.equal(payload.number, 2);
4978	assert.isUndefined(payload.pullRequests);
4979	const after = yield* orchestrator.getThreadShell(threadId);
4980	assert.equal(after?.pullRequests?.length, 3);
4981	assert.deepEqual(after?.pullRequests?.[0], before?.pullRequests?.[0]);
4982	assert.deepEqual(after?.pullRequests?.[2], before?.pullRequests?.[2]);
4983	assert.equal(after?.pullRequests?.[1]?.stack?.id, "updated");
4984	assert.isTrue((yield* maintenance.rebuild).valid);
4985	assert.deepEqual(
4986	(yield* orchestrator.getThreadShell(threadId))?.pullRequests,
4987	after?.pullRequests,
4988	);
4989	yield* maintenance.compactEventStore;
4990	assert.isTrue((yield* maintenance.rebuild).valid);
4991	assert.deepEqual(
4992	(yield* orchestrator.getThreadShell(threadId))?.pullRequests,
4993	after?.pullRequests,
4994	);
4995	}),
4996	);
4997	});
4998
4999	it.layer(layerSharedApplicationDataPlaneTest)("shared application data plane", (it) => {
5000	it.effect("orders retained project transactions and V2 thread transactions in one source", () =>
5001	Effect.gen(function* () {
5002	const projects = yield* ProjectService.ProjectService;
5003	const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
5004	const orchestrator = yield* Orchestrator.OrchestratorV2;
5005	const sql = yield* SqlClient.SqlClient;
5006	const projectId = ProjectId.make("runtime-layer-shared-project");
5007	const threadId = ThreadId.make("runtime-layer-shared-thread");
5008	const projectInput = {
5009	commandId: CommandId.make("runtime-layer-shared-project-create"),
5010	projectId,
5011	title: "Shared application source",
5012	workspaceRoot: "/tmp/runtime-layer-shared-project",
5013	};
5014
5015	const created = yield* projects.create(projectInput);
5016	const retried = yield* projects.create(projectInput);
5017	assert.deepEqual(retried, created);
5018
5019	const delivered = yield* Queue.unbounded<ApplicationStoredEvent>();
5020	yield* applicationEvents.streamApplicationEvents().pipe(
5021	Stream.take(2),
5022	Stream.runForEach((event) => Queue.offer(delivered, event)),
5023	Effect.forkScoped,
5024	);
5025
5026	const projectEvent = yield* Queue.take(delivered);
5027	assert.isTrue("aggregateKind" in projectEvent && projectEvent.aggregateId === projectId);
5028
5029	const threadResult = yield* orchestrator.dispatch({
5030	type: "thread.create",
5031	createdBy: "user",
5032	creationSource: "web",
5033	commandId: CommandId.make("runtime-layer-shared-thread-create"),
5034	threadId,
5035	projectId,
5036	title: "Shared thread",
5037	modelSelection,
5038	runtimeMode: "full-access",
5039	interactionMode: "default",
5040	branch: null,
5041	worktreePath: null,
5042	});
5043	const threadEvent = yield* Queue.take(delivered);
5044
5045	assert.equal(threadEvent.sequence, threadResult.sequence);
5046	assert.isAbove(threadEvent.sequence, projectEvent.sequence);
5047	assert.isTrue("aggregateKind" in projectEvent);
5048	assert.isTrue("event" in threadEvent);
5049	assert.equal((yield* projects.getById(projectId))._tag, "Some");
5050
5051	const retainedReceipts = yield* sql<{
5052	readonly aggregate_kind: string;
5053	readonly aggregate_id: string;
5054	}>`
5055	SELECT aggregate_kind, aggregate_id
5056	FROM orchestration_command_receipts
5057	ORDER BY result_sequence ASC
5058	`;
5059	assert.deepEqual(retainedReceipts, [
5060	{ aggregate_kind: "project", aggregate_id: projectId },
5061	{ aggregate_kind: "thread", aggregate_id: threadId },
5062	]);
5063
5064	const retiredWrites = yield* sql<{ readonly count: number }>`
5065	SELECT
5066	(SELECT COUNT(*) FROM orchestration_v2_events) +
5067	(SELECT COUNT(*) FROM orchestration_v2_command_receipts) AS count
5068	`;
5069	assert.equal(retiredWrites[0]?.count, 0);
5070	}),
5071	);
5072	});
5073
5074	it.layer(layerTest)("usage-limit recovery", (it) => {
5075	it.effect.each(["interrupted", "usage_limit"] as const)(
5076	"manually resumes an %s run ahead of its queued message only once",
5077	(reason) =>
5078	Effect.gen(function* () {
5079	const orchestrator = yield* Orchestrator.OrchestratorV2;
5080	const events = yield* EventSink.EventSinkV2;
5081	const threadId = ThreadId.make(`manual-resume:${reason}`);
5082	const projectId = ProjectId.make(`manual-resume:project:${reason}`);
5083	const now = yield* DateTime.now;
5084	const createdAt = DateTime.formatIso(now);
5085	yield* seedProject({
5086	projectId,
5087	title: "Resume project",
5088	workspaceRoot: process.cwd(),
5089	defaultModelSelection: modelSelection,
5090	createdAt,
5091	});
5092	yield* orchestrator.dispatch({
5093	type: "thread.create",
5094	commandId: CommandId.make(`manual-resume:create:${reason}`),
5095	threadId,
5096	projectId,
5097	title: "Interrupted thread",
5098	modelSelection,
5099	runtimeMode: "full-access",
5100	interactionMode: "default",
5101	branch: null,
5102	worktreePath: null,
5103	createdBy: "user",
5104	creationSource: "web",
5105	});
5106	yield* orchestrator.dispatch({
5107	type: "message.dispatch",
5108	commandId: CommandId.make(`manual-resume:start:${reason}`),
5109	threadId,
5110	messageId: MessageId.make(`manual-resume:start:${reason}`),
5111	text: "Start work.",
5112	attachments: [],
5113	dispatchMode: { type: "defer_start" },
5114	createdBy: "user",
5115	creationSource: "web",
5116	});
5117	yield* orchestrator.dispatch({
5118	type: "message.dispatch",
5119	commandId: CommandId.make(`manual-resume:queue:${reason}`),
5120	threadId,
5121	messageId: MessageId.make(`manual-resume:queue:${reason}`),
5122	text: "Follow up.",
5123	attachments: [],
5124	dispatchMode: { type: "queue_after_active" },
5125	createdBy: "user",
5126	creationSource: "web",
5127	});
5128	const [source, queued] = (yield* orchestrator.getThreadProjection(threadId)).runs as [
5129	OrchestrationV2Run,
5130	OrchestrationV2Run,
5131	];
5132	// A user stop holds the queue as the run ends (thread.turn.interrupt with
5133	// holdQueue). Without the hold, the terminal-run worker may start the
5134	// queued run before the resume below, depending on fiber scheduling.
5135	yield* events.write({
5136	events: [
5137	{
5138	id: EventId.make(`manual-resume:hold:${reason}`),
5139	type: "run.updated",
5140	threadId,
5141	occurredAt: now,
5142	payload: { ...queued, queueHeld: true },
5143	},
5144	{
5145	id: EventId.make(`manual-resume:stop:${reason}`),
5146	type: "run.updated",
5147	threadId,
5148	occurredAt: now,
5149	payload: {
5150	...source,
5151	status: reason === "interrupted" ? "interrupted" : "failed",
5152	completedAt: now,
5153	},
5154	},
5155	],
5156	});
5157	let scheduledResume: ReturnType<typeof limitRecoveryCommand> = null;
5158	if (reason === "usage_limit") {
5159	const resetAt = DateTime.formatIso(DateTime.add(now, { minutes: 1 }));
5160	yield* events.write({
5161	events: [
5162	{
5163	id: EventId.make(`manual-resume:error:${reason}`),
5164	type: "turn-item.updated",
5165	threadId,
5166	occurredAt: now,
5167	payload: {
5168	id: TurnItemId.make(`manual-resume:error:${reason}`),
5169	type: "error",
5170	threadId,
5171	runId: source.id,
5172	nodeId: source.rootNodeId,
5173	providerThreadId: null,
5174	providerTurnId: null,
5175	nativeItemRef: null,
5176	parentItemId: null,
5177	ordinal: 2,
5178	status: "failed",
5179	title: "Usage limit reached",
5180	startedAt: now,
5181	completedAt: now,
5182	updatedAt: now,
5183	failure: {
5184	class: "usage_limit",
5185	message: "Plan limit reached.",
5186	code: "usageLimitExceeded",
5187	retryable: null,
5188	resetAt,
5189	},
5190	},
5191	},
5192	],
5193	});
5194	const shell = (yield* orchestrator.getShellSnapshot()).threads.find(
5195	(thread) => thread.id === threadId,
5196	)!;
5197	yield* orchestrator.dispatch(
5198	limitRecoveryCommand(shell, true, DateTime.toEpochMillis(now))!,
5199	);
5200	const armed = (yield* orchestrator.getShellSnapshot()).threads.find(
5201	(thread) => thread.id === threadId,
5202	)!;
5203	scheduledResume = limitRecoveryCommand(armed, true, Date.parse(resetAt));
5204	assert.isNotNull(scheduledResume);
5205	}
5206	const resume = (suffix: string) => ({
5207	type: "message.dispatch" as const,
5208	commandId: CommandId.make(`manual-resume:${suffix}:${reason}`),
5209	threadId,
5210	messageId: MessageId.make(`manual-resume:${suffix}:${reason}`),
5211	manualContinuationOfRunId: source.id,
5212	text: "Continue where you left off.",
5213	attachments: [],
5214	dispatchMode: { type: "start_immediately" as const },
5215	createdBy: "user" as const,
5216	creationSource: "web" as const,
5217	});
5218	yield* orchestrator.dispatch(resume("first"));
5219	const after = yield* orchestrator.getThreadProjection(threadId);
5220	assert.lengthOf(after.runs, 3);
5221	assert.equal(after.runs[1]?.status, "queued");
5222	assert.equal(after.runs[2]?.status, "starting");
5223	assert.equal(
5224	(yield* orchestrator.dispatch(resume("second")).pipe(Effect.exit))._tag,
5225	"Failure",
5226	);
5227	if (scheduledResume !== null) {
5228	yield* TestClock.adjust("1 minute");
5229	yield* orchestrator.dispatch(scheduledResume);
5230	}
5231	assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 3);
5232	}),
5233	);
5234
5235	it.effect.each([
5236	"resume",
5237	"queued-resume",
5238	"cancel",
5239	"rearm",
5240	"snooze-race",
5241	"new-message",
5242	"archive",
5243	"settle",
5244	"replacement",
5245	"manual-snooze",
5246	"manual-snooze-after-recovery",
5247	"invalid-snooze",
5248	"snooze-only",
5249	"snooze-resume",
5250	"cancel-resume-keep-snooze",
5251	"wake-preserve-resume",
5252	"independent-patches",
5253	"expired-snooze",
5254	"wake",
5255	] as const)("guards a scheduled usage-limit continuation against %s", (scenario) =>
5256	Effect.gen(function* () {
5257	const orchestrator = yield* Orchestrator.OrchestratorV2;
5258	const events = yield* EventSink.EventSinkV2;
5259	const threadId = ThreadId.make(`recovery:${scenario}`);
5260	const projectId = ProjectId.make(`recovery:project:${scenario}`);
5261	yield* seedProject({
5262	projectId,
5263	title: "Recovery project",
5264	workspaceRoot: process.cwd(),
5265	defaultModelSelection: modelSelection,
5266	createdAt: DateTime.formatIso(yield* DateTime.now),
5267	});
5268	yield* orchestrator.dispatch({
5269	type: "thread.create",
5270	commandId: CommandId.make(`recovery:create:${scenario}`),
5271	threadId,
5272	projectId,
5273	title: "Limited thread",
5274	modelSelection,
5275	runtimeMode: "full-access",
5276	interactionMode: "default",
5277	branch: null,
5278	worktreePath: null,
5279	createdBy: "user",
5280	creationSource: "web",
5281	});
5282	yield* orchestrator.dispatch({
5283	type: "message.dispatch",
5284	commandId: CommandId.make(`recovery:message:${scenario}`),
5285	threadId,
5286	messageId: MessageId.make(`recovery:message:${scenario}`),
5287	text: "Work on this.",
5288	attachments: [],
5289	dispatchMode: { type: "defer_start" },
5290	createdBy: "user",
5291	creationSource: "web",
5292	});
5293	if (scenario === "queued-resume") {
5294	yield* orchestrator.dispatch({
5295	type: "message.dispatch",
5296	commandId: CommandId.make(`recovery:queued:${scenario}`),
5297	threadId,
5298	messageId: MessageId.make(`recovery:queued:${scenario}`),
5299	text: "Run after recovery.",
5300	attachments: [],
5301	dispatchMode: { type: "queue_after_active" },
5302	createdBy: "user",
5303	creationSource: "web",
5304	});
5305	}
5306	const projection = yield* orchestrator.getThreadProjection(threadId);
5307	const run = projection.runs[0]!;
5308	const now = yield* DateTime.now;
5309	const resetAt =
5310	scenario === "invalid-snooze"
5311	? "not-a-date"
5312	: DateTime.formatIso(DateTime.add(now, { minutes: 1 })).replace(
5313	"Z",
5314	scenario === "wake" ? "+00:00" : "Z",
5315	);
5316	yield* events.write({
5317	commandId: CommandId.make(`recovery:failure:${scenario}`),
5318	events: [
5319	{
5320	id: EventId.make(`recovery:run:${scenario}`),
5321	type: "run.updated",
5322	threadId,
5323	occurredAt: now,
5324	payload: { ...run, status: "failed", completedAt: now },
5325	},
5326	{
5327	id: EventId.make(`recovery:error:${scenario}`),
5328	type: "turn-item.updated",
5329	threadId,
5330	occurredAt: now,
5331	payload: {
5332	id: TurnItemId.make(`recovery:error:${scenario}`),
5333	type: "error",
5334	threadId,
5335	runId: run.id,
5336	nodeId: run.rootNodeId,
5337	providerThreadId: null,
5338	providerTurnId: null,
5339	nativeItemRef: null,
5340	parentItemId: null,
5341	ordinal: 2,
5342	status: "failed",
5343	title: "Usage limit reached",
5344	startedAt: now,
5345	completedAt: now,
5346	updatedAt: now,
5347	failure: {
5348	class: "usage_limit",
5349	message: "Plan limit reached.",
5350	code: "usageLimitExceeded",
5351	retryable: null,
5352	resetAt,
5353	},
5354	},
5355	},
5356	],
5357	});
5358	if (scenario === "queued-resume") {
5359	const queuedRun = projection.runs[1]!;
5360	yield* events.write({
5361	events: [
5362	{
5363	id: EventId.make("recovery:held:queued-resume"),
5364	type: "run.updated",
5365	threadId,
5366	runId: queuedRun.id,
5367	occurredAt: now,
5368	payload: { ...queuedRun, queueHeld: true },
5369	},
5370	],
5371	});
5372	const resumeHeldQueue = yield* orchestrator
5373	.dispatch({
5374	type: "queue.resume",
5375	commandId: CommandId.make("recovery:resume-held:queued-resume"),
5376	threadId,
5377	})
5378	.pipe(Effect.exit);
5379	assert.equal(resumeHeldQueue._tag, "Failure");
5380	assert.isTrue((yield* orchestrator.getThreadProjection(threadId)).runs[1]?.queueHeld);
5381	}
5382	const shell = (yield* orchestrator.getShellSnapshot()).threads.find(
5383	(thread) => thread.id === threadId,
5384	)!;
5385	assert.isNull(limitRecoveryCommand(shell, false, DateTime.toEpochMillis(now)));
5386	if (scenario === "invalid-snooze") {
5387	const result = yield* orchestrator
5388	.dispatch({
5389	type: "thread.metadata.update",
5390	commandId: CommandId.make("recovery:invalid-snooze"),
5391	threadId,
5392	limitRecovery: { runId: run.id, resetAt, snooze: true },
5393	})
5394	.pipe(Effect.exit);
5395	assert.equal(result._tag, "Failure");
5396	const current = yield* orchestrator.getThreadProjection(threadId);
5397	assert.isNull(current.thread.limitRecovery ?? null);
5398	assert.isNull(current.thread.snoozedUntil);
5399	return;
5400	}
5401	const snooze = [
5402	"snooze-only",
5403	"manual-snooze-after-recovery",
5404	"snooze-resume",
5405	"wake",
5406	"cancel-resume-keep-snooze",
5407	"wake-preserve-resume",
5408	].includes(scenario);
5409	const autoResume = scenario !== "snooze-only" && scenario !== "wake";
5410	const arm = limitRecoveryCommand(shell, autoResume, DateTime.toEpochMillis(now), snooze);
5411	assert.isNotNull(arm);
5412	yield* orchestrator.dispatch(arm!);
5413	let armedShell = (yield* orchestrator.getShellSnapshot()).threads.find(
5414	(thread) => thread.id === threadId,
5415	)!;
5416	assert.deepEqual(armedShell.limitRecovery, {
5417	runId: run.id,
5418	resetAt,
5419	autoResume,
5420	snooze,
5421	requestId: arm!.commandId,
5422	});
5423	if (snooze)
5424	assert.equal(DateTime.toEpochMillis(armedShell.snoozedUntil!), Date.parse(resetAt));
5425	if (scenario === "cancel-resume-keep-snooze" || scenario === "wake-preserve-resume") {
5426	yield* TestClock.adjust("10 seconds");
5427	yield* orchestrator.dispatch({
5428	type: "thread.metadata.update",
5429	commandId: CommandId.make(`recovery:independent-choice:${scenario}`),
5430	threadId,
5431	limitRecovery: {
5432	runId: run.id,
5433	resetAt,
5434	autoResume: scenario === "wake-preserve-resume",
5435	snooze: scenario === "cancel-resume-keep-snooze",
5436	},
5437	});
5438	armedShell = (yield* orchestrator.getShellSnapshot()).threads.find(
5439	(thread) => thread.id === threadId,
5440	)!;
5441	if (scenario === "cancel-resume-keep-snooze") {
5442	assert.equal(DateTime.toEpochMillis(armedShell.snoozedUntil!), Date.parse(resetAt));
5443	// Failed runtime timestamps advance with metadata. Acknowledging the
5444	// same failed run must not turn cancellation into an early wake.
5445	assert.equal(
5446	DateTime.toEpochMillis(armedShell.snoozedAt!),
5447	DateTime.toEpochMillis(armedShell.updatedAt),
5448	);
5449	assert.isFalse(armedShell.limitRecovery!.autoResume);
5450	} else {
5451	assert.isNull(armedShell.snoozedUntil);
5452	assert.isNull(armedShell.snoozedAt);
5453	assert.isTrue(armedShell.limitRecovery!.autoResume);
5454	}
5455	}
5456	if (scenario === "independent-patches") {
5457	yield* TestClock.adjust("10 seconds");
5458	yield* orchestrator.dispatch({
5459	type: "thread.metadata.update",
5460	commandId: CommandId.make("recovery:patch-snooze"),
5461	threadId,
5462	limitRecovery: { runId: run.id, resetAt, snooze: true },
5463	});
5464	let current = yield* orchestrator.getThreadProjection(threadId);
5465	assert.isTrue(current.thread.limitRecovery!.autoResume);
5466	assert.isTrue(current.thread.limitRecovery!.snooze);
5467	// This is also the payload an older auto-resume-only client sends.
5468	yield* TestClock.adjust("10 seconds");
5469	yield* orchestrator.dispatch({
5470	type: "thread.metadata.update",
5471	commandId: CommandId.make("recovery:patch-cancel-resume"),
5472	threadId,
5473	limitRecovery: { runId: run.id, resetAt, autoResume: false },
5474	});
5475	current = yield* orchestrator.getThreadProjection(threadId);
5476	assert.isFalse(current.thread.limitRecovery!.autoResume);
5477	assert.isTrue(current.thread.limitRecovery!.snooze);
5478	assert.equal(DateTime.toEpochMillis(current.thread.snoozedUntil!), Date.parse(resetAt));
5479	assert.equal(
5480	DateTime.toEpochMillis(current.thread.snoozedAt!),
5481	DateTime.toEpochMillis(current.thread.updatedAt),
5482	);
5483	yield* orchestrator.dispatch({
5484	type: "thread.metadata.update",
5485	commandId: CommandId.make("recovery:patch-resume"),
5486	threadId,
5487	limitRecovery: { runId: run.id, resetAt, autoResume: true },
5488	});
5489	armedShell = (yield* orchestrator.getShellSnapshot()).threads.find(
5490	(thread) => thread.id === threadId,
5491	)!;
5492	assert.isTrue(armedShell.limitRecovery!.autoResume);
5493	assert.isTrue(armedShell.limitRecovery!.snooze);
5494	}
5495	if (scenario === "manual-snooze" || scenario === "manual-snooze-after-recovery") {
5496	yield* orchestrator.dispatch({
5497	type: "thread.snooze",
5498	commandId: CommandId.make(`recovery:manual-snooze:${scenario}`),
5499	threadId,
5500	snoozedUntil: resetAt,
5501	});
5502	yield* orchestrator.dispatch({
5503	type: "thread.metadata.update",
5504	commandId: CommandId.make(`recovery:manual-cancel:${scenario}`),
5505	threadId,
5506	limitRecovery: { runId: run.id, resetAt, autoResume: false, snooze: false },
5507	});
5508	assert.equal(
5509	DateTime.toEpochMillis(
5510	(yield* orchestrator.getThreadProjection(threadId)).thread.snoozedUntil!,
5511	),
5512	Date.parse(resetAt),
5513	);
5514	yield* orchestrator.dispatch({
5515	type: "thread.unsnooze",
5516	commandId: CommandId.make(`recovery:manual-wake:${scenario}`),
5517	threadId,
5518	reason: "user",
5519	});
5520	assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.snoozedUntil);
5521	}
5522	if (scenario === "wake") {
5523	yield* orchestrator.dispatch({
5524	type: "thread.metadata.update",
5525	commandId: CommandId.make(`recovery:wake:${scenario}`),
5526	threadId,
5527	limitRecovery: { runId: run.id, resetAt, autoResume: false, snooze: false },
5528	});
5529	assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.snoozedUntil);
5530	}
5531	assert.isNull(limitRecoveryCommand(armedShell, true, DateTime.toEpochMillis(now)));
5532	yield* orchestrator.dispatch({
5533	type: "message.dispatch",
5534	commandId: CommandId.make(`recovery:early:${scenario}`),
5535	messageId: MessageId.make(`recovery:early:${scenario}`),
5536	threadId,
5537	usageLimitContinuationOfRunId: run.id,
5538	text: "Continue where you left off.",
5539	attachments: [],
5540	dispatchMode: { type: "start_immediately" },
5541	createdBy: "user",
5542	creationSource: "server",
5543	});
5544	assert.lengthOf(
5545	(yield* orchestrator.getThreadProjection(threadId)).runs,
5546	scenario === "queued-resume" ? 2 : 1,
5547	);
5548	yield* TestClock.adjust("1 minute");
5549	const resume = limitRecoveryCommand(
5550	armedShell,
5551	true,
5552	DateTime.toEpochMillis(yield* DateTime.now),
5553	);
5554	if (autoResume && scenario !== "cancel-resume-keep-snooze") assert.isNotNull(resume);
5555	else assert.isNull(resume);
5556	if (scenario === "snooze-race") {
5557	const wakeAt = DateTime.formatIso(DateTime.add(yield* DateTime.now, { minutes: 1 }));
5558	yield* orchestrator.dispatch({
5559	type: "thread.snooze",
5560	commandId: CommandId.make("recovery:raced-snooze"),
5561	threadId,
5562	snoozedUntil: wakeAt,
5563	});
5564	yield* orchestrator.dispatch(resume!);
5565	assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1);
5566	yield* TestClock.adjust("1 minute");
5567	const current = (yield* orchestrator.getShellSnapshot()).threads.find(
5568	(thread) => thread.id === threadId,
5569	)!;
5570	const freshResume = limitRecoveryCommand(
5571	current,
5572	true,
5573	DateTime.toEpochMillis(yield* DateTime.now),
5574	);
5575	assert.isNotNull(freshResume);
5576	assert.notEqual(freshResume!.commandId, resume!.commandId);
5577	yield* orchestrator.dispatch(freshResume!);
5578	yield* orchestrator.dispatch(freshResume!);
5579	assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 2);
5580	}
5581	if (scenario === "expired-snooze") {
5582	const staleSnooze = yield* orchestrator
5583	.dispatch({
5584	type: "thread.metadata.update",
5585	commandId: CommandId.make("recovery:expired-snooze"),
5586	threadId,
5587	limitRecovery: { runId: run.id, resetAt, snooze: true },
5588	})
5589	.pipe(Effect.exit);
5590	assert.equal(staleSnooze._tag, "Failure");
5591	const current = yield* orchestrator.getThreadProjection(threadId);
5592	assert.isNull(current.thread.snoozedUntil);
5593	assert.isFalse(current.thread.limitRecovery!.snooze);
5594	assert.isTrue(current.thread.limitRecovery!.autoResume);
5595	}
5596
5597	if (scenario === "cancel" || scenario === "rearm")
5598	yield* orchestrator.dispatch({
5599	type: "thread.metadata.update",
5600	commandId: CommandId.make(`recovery:cancel:${scenario}`),
5601	threadId,
5602	limitRecovery: { runId: run.id, resetAt, autoResume: false },
5603	});
5604	if (scenario === "archive")
5605	yield* orchestrator.dispatch({
5606	type: "thread.archive",
5607	commandId: CommandId.make(`recovery:archive:${scenario}`),
5608	threadId,
5609	});
5610	if (scenario === "new-message")
5611	yield* orchestrator.dispatch({
5612	type: "message.dispatch",
5613	commandId: CommandId.make(`recovery:new-message:${scenario}`),
5614	threadId,
5615	messageId: MessageId.make(`recovery:new-message:${scenario}`),
5616	text: "I will continue manually.",
5617	attachments: [],
5618	dispatchMode: { type: "defer_start" },
5619	createdBy: "user",
5620	creationSource: "web",
5621	});
5622	if (scenario === "settle")
5623	yield* orchestrator.dispatch({
5624	type: "thread.settle",
5625	commandId: CommandId.make(`recovery:settle:${scenario}`),
5626	threadId,
5627	});
5628	if (scenario === "replacement") {
5629	const current = yield* orchestrator.getThreadProjection(threadId);
5630	const error = current.turnItems.find((item) => item.type === "error")!;
5631	if (error.type !== "error") throw new Error("Expected provider error");
5632	yield* events.write({
5633	commandId: CommandId.make(`recovery:replacement:${scenario}`),
5634	events: [
5635	{
5636	id: EventId.make(`recovery:replacement:${scenario}`),
5637	type: "turn-item.updated",
5638	threadId,
5639	occurredAt: yield* DateTime.now,
5640	payload: {
5641	...error,
5642	failure: {
5643	...error.failure,
5644	class: "provider_error",
5645	message: "A replacement failure.",
5646	},
5647	},
5648	},
5649	],
5650	});
5651	}
5652	if (scenario === "rearm") {
5653	yield* orchestrator.dispatch(resume!);
5654	yield* orchestrator.dispatch({
5655	type: "thread.metadata.update",
5656	commandId: CommandId.make(`recovery:rearm:${scenario}`),
5657	threadId,
5658	limitRecovery: { runId: run.id, resetAt, autoResume: true },
5659	});
5660	yield* orchestrator.dispatch(resume!);
5661	assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1);
5662	const rearmedShell = (yield* orchestrator.getShellSnapshot()).threads.find(
5663	(thread) => thread.id === threadId,
5664	)!;
5665	const freshResume = limitRecoveryCommand(
5666	rearmedShell,
5667	true,
5668	DateTime.toEpochMillis(yield* DateTime.now),
5669	);
5670	assert.isNotNull(freshResume);
5671	assert.notEqual(freshResume!.commandId, resume!.commandId);
5672	yield* orchestrator.dispatch(freshResume!);
5673	yield* orchestrator.dispatch(freshResume!);
5674	assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 2);
5675	}
5676	const before = yield* orchestrator.getThreadProjection(threadId);
5677	if (resume !== null) {
5678	yield* orchestrator.dispatch(resume);
5679	yield* orchestrator.dispatch(resume);
5680	}
5681	const after = yield* orchestrator.getThreadProjection(threadId);
5682	assert.lengthOf(
5683	after.runs,
5684	before.runs.length +
5685	(scenario === "resume" ||
5686	scenario === "queued-resume" ||
5687	scenario === "snooze-resume" ||
5688	scenario === "wake-preserve-resume" ||
5689	scenario === "independent-patches" ||
5690	scenario === "expired-snooze"
5691	? 1
5692	: 0),
5693	);
5694	assert.lengthOf(
5695	after.messages,
5696	before.messages.length +
5697	(scenario === "resume" ||
5698	scenario === "queued-resume" ||
5699	scenario === "snooze-resume" ||
5700	scenario === "wake-preserve-resume" ||
5701	scenario === "independent-patches" ||
5702	scenario === "expired-snooze"
5703	? 1
5704	: 0),
5705	);
5706	if (scenario === "queued-resume") {
5707	assert.equal(after.runs[1]?.status, "queued");
5708	assert.isTrue(after.runs[1]?.queueHeld);
5709	const continuation = after.runs[2]!;
5710	const completedAt = yield* DateTime.now;
5711	yield* events.write({
5712	events: [
5713	{
5714	id: EventId.make("recovery:continuation-completed:queued-resume"),
5715	type: "run.updated",
5716	threadId,
5717	runId: continuation.id,
5718	occurredAt: completedAt,
5719	payload: { ...continuation, status: "completed", completedAt },
5720	},
5721	],
5722	});
5723	yield* orchestrator.dispatch({
5724	type: "queue.resume",
5725	commandId: CommandId.make("recovery:resume-held-after-limit:queued-resume"),
5726	threadId,
5727	});
5728	const resumed = yield* orchestrator.getThreadProjection(threadId);
5729	assert.equal(resumed.runs[1]?.status, "starting");
5730	assert.isFalse(resumed.runs[1]?.queueHeld);
5731	}
5732	}),
5733	);
5734	});
5735
5736	const queueResumeDisabledInstanceProjectId = ProjectId.make(
5737	"project:queue-resume-disabled-instance",
5738	);
5739
5740	const layerQueueResumeDisabledInstanceTest = Layer.mergeAll(
5741	RuntimeLayer.layer,
5742	RuntimeLayer.layerEventSink,
5743	ProjectStore.layer,
5744	ProjectionStore.layer,
5745	EffectOutbox.layer,
5746	ThreadCommandExecutor.layer,
5747	).pipe(
5748	Layer.provide(McpSessionRegistryTestkit.layer),
5749	Layer.provide(Layer.merge(McpProviderSessions.layer, ScriptProgress.layerUnavailable)),
5750	Layer.provide(SqlitePersistence.layerMemory),
5751	Layer.provide(layerCheckpointStoreTest),
5752	Layer.provide(layerServerConfig),
5753	Layer.provide(
5754	ServerSettings.layerTest({
5755	providerInstances: {
5756	[alternateInstanceId]: { driver: "codex", enabled: true },
5757	},
5758	projectSettingsOverrides: {
5759	[queueResumeDisabledInstanceProjectId]: {
5760	providerInstances: { [alternateInstanceId]: { enabled: false } },
5761	},
5762	},
5763	}),
5764	),
5765	Layer.provide(layerTestProviderInstanceRegistry),
5766	Layer.provide(layerGitWorkflowTest),
5767	Layer.provide(layerProjectServiceTest),
5768	Layer.provide(layerPlatformTest),
5769	);
5770
5771	it.layer(layerQueueResumeDisabledInstanceTest)(
5772	"queue.resume on a project-disabled instance",
5773	(it) => {
5774	it.effect(
5775	"holds a queued run instead of starting it on an instance the project turned off",
5776	() =>
5777	Effect.gen(function* () {
5778	const orchestrator = yield* Orchestrator.OrchestratorV2;
5779	const events = yield* EventSink.EventSinkV2;
5780	const threadId = ThreadId.make("queue-resume-disabled-instance-thread");
5781	const queuedModelSelection = {
5782	instanceId: alternateInstanceId,
5783	model: "gpt-5.5",
5784	} satisfies ModelSelection;
5785
5786	yield* orchestrator.dispatch({
5787	type: "thread.create",
5788	createdBy: "user",
5789	creationSource: "web",
5790	commandId: CommandId.make(`${threadId}:create`),
5791	threadId,
5792	projectId: queueResumeDisabledInstanceProjectId,
5793	title: "Queue resume disabled instance",
5794	modelSelection,
5795	runtimeMode: "full-access",
5796	interactionMode: "default",
5797	branch: null,
5798	worktreePath: null,
5799	});
5800	yield* orchestrator.dispatch({
5801	type: "message.dispatch",
5802	createdBy: "user",
5803	creationSource: "web",
5804	commandId: CommandId.make(`${threadId}:message:active`),
5805	threadId,
5806	messageId: MessageId.make(`${threadId}:message:active`),
5807	text: "Active",
5808	attachments: [],
5809	dispatchMode: { type: "defer_start" },
5810	});
5811	yield* orchestrator.dispatch({
5812	type: "message.dispatch",
5813	createdBy: "user",
5814	creationSource: "web",
5815	commandId: CommandId.make(`${threadId}:message:queued`),
5816	threadId,
5817	messageId: MessageId.make(`${threadId}:message:queued`),
5818	text: "Queued",
5819	attachments: [],
5820	modelSelection: queuedModelSelection,
5821	dispatchMode: { type: "queue_after_active" },
5822	});
5823
5824	const before = yield* orchestrator.getThreadProjection(threadId);
5825	const activeRun = before.runs.find((run) => run.status !== "queued")!;
5826	const queuedRun = before.runs.find((run) => run.status === "queued")!;
5827	const now = yield* DateTime.now;
5828	// A raw event ends the active run without going through dispatch,
5829	// so nothing has yet tried to advance the queue onto the disabled
5830	// instance.
5831	yield* events.write({
5832	events: [
5833	{
5834	id: EventId.make(`${threadId}:active-completed`),
5835	type: "run.updated",
5836	threadId,
5837	runId: activeRun.id,
5838	occurredAt: now,
5839	payload: { ...activeRun, status: "completed", completedAt: now },
5840	},
5841	],
5842	});
5843
5844	yield* orchestrator.dispatch({
5845	type: "queue.resume",
5846	commandId: CommandId.make(`${threadId}:resume`),
5847	threadId,
5848	});
5849
5850	const resumed = yield* orchestrator.getThreadProjection(threadId);
5851	const resumedQueuedRun = resumed.runs.find((run) => run.id === queuedRun.id);
5852	assert.equal(
5853	resumedQueuedRun?.status,
5854	"queued",
5855	"a disabled instance must not start the queued run",
5856	);
5857	assert.isTrue(
5858	resumedQueuedRun?.queueHeld,
5859	"a refused resume re-holds the queue instead of leaving it looking resumable",
5860	);
5861	}),
5862	);
5863	},
5864	);