clover's git

the-knife

1	import * as ServerSettings from "../serverSettings.ts";
2	import { assert, it } from "@effect/vitest";
3	import {
4	CheckpointId,
5	CheckpointRef,
6	CheckpointScopeId,
7	CommandId,
8	ContextTransferId,
9	EventId,
10	MessageId,
11	type ModelSelection,
12	NodeId,
13	type OrchestrationV2AppThread,
14	type OrchestrationV2DomainEvent,
15	type OrchestrationV2Run,
16	type OrchestrationV2ThreadProjection,
17	type OrchestrationV2TurnItem,
18	ProjectId,
19	ProviderDriverKind,
20	ProviderInstanceId,
21	ProviderSessionId,
22	ProviderThreadId,
23	RunAttemptId,
24	RunId,
25	ThreadId,
26	TurnItemId,
27	} from "@t3tools/contracts";
28	import * as DateTime from "effect/DateTime";
29	import * as Deferred from "effect/Deferred";
30	import * as Duration from "effect/Duration";
31	import * as Effect from "effect/Effect";
32	import * as Fiber from "effect/Fiber";
33	import * as Layer from "effect/Layer";
34	import * as Option from "effect/Option";
35	import * as Queue from "effect/Queue";
36	import * as Ref from "effect/Ref";
37	import * as Scheduler from "effect/Scheduler";
38	import * as Schema from "effect/Schema";
39	import * as Stream from "effect/Stream";
40	import * as TestClock from "effect/testing/TestClock";
41	import * as Tracer from "effect/Tracer";
42	import * as SqlClient from "effect/sql/SqlClient";
43	import * as Statement from "effect/sql/Statement";
44
45	import { LIVE_STREAM_MAX_ITEMS, LiveStreamBufferError } from "./LiveStreamBudget.ts";
46	import * as SqlitePersistence from "../persistence/Sqlite.ts";
47	import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
48	import * as CommandReceiptStore from "./CommandReceiptStore.ts";
49	import * as EffectOutbox from "./EffectOutbox.ts";
50	import * as EffectWorker from "./EffectWorker.ts";
51	import * as EventSink from "./EventSink.ts";
52	import * as EventStore from "./EventStore.ts";
53	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
54	import * as ProjectionMaintenance from "./ProjectionMaintenance.ts";
55	import * as ProjectionStore from "./ProjectionStore.ts";
56	import * as ProjectStore from "./ProjectStore.ts";
57	import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
58	import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
59
60	const isLiveStreamBufferError = Schema.is(LiveStreamBufferError);
61	const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));
62
63	const layerDatabase = SqlitePersistence.layerMemory;
64	const layerEventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(layerDatabase));
65	const layerProjectionStoreProvided = ProjectionStore.layer.pipe(Layer.provideMerge(layerDatabase));
66	const layerStoresProvided = Layer.mergeAll(
67	layerDatabase,
68	layerEventStoreProvided,
69	layerProjectionStoreProvided,
70	);
71	const layerEventSinkProvided = EventSink.layer.pipe(Layer.provide(layerStoresProvided));
72	const layerEffectOutboxProvided = EffectOutbox.layer.pipe(Layer.provide(layerDatabase));
73	const layerCommandReceiptStoreProvided = CommandReceiptStore.layer.pipe(
74	Layer.provide(layerDatabase),
75	);
76	// Its own database, for tests that act on every row in the outbox table.
77	const layerIsolatedOutbox = Layer.fresh(EffectOutbox.layer.pipe(Layer.provideMerge(layerDatabase)));
78	const layerProjectionMaintenanceProvided = ProjectionMaintenance.layer.pipe(
79	Layer.provide(layerStoresProvided),
80	);
81	const layerTest = Layer.mergeAll(
82	layerStoresProvided,
83	layerEventSinkProvided,
84	layerEffectOutboxProvided,
85	layerCommandReceiptStoreProvided,
86	IdAllocator.layer,
87	layerProjectionMaintenanceProvided,
88	);
89
90	const providerInstanceId = ProviderInstanceId.make("codex");
91	const providerDriver = ProviderDriverKind.make("codex");
92	const modelSelection = {
93	instanceId: providerInstanceId,
94	model: "gpt-5.4",
95	} satisfies ModelSelection;
96
97	function makeThread(threadId: ThreadId, now: DateTime.Utc): OrchestrationV2AppThread {
98	return {
99	createdBy: "user",
100	creationSource: "web",
101	id: threadId,
102	projectId: ProjectId.make(`project:${threadId}`),
103	title: `Thread ${threadId}`,
104	providerInstanceId,
105	modelSelection,
106	runtimeMode: "full-access",
107	interactionMode: "default",
108	branch: null,
109	worktreePath: null,
110	activeProviderThreadId: null,
111	lineage: {
112	parentThreadId: null,
113	relationshipToParent: null,
114	rootThreadId: threadId,
115	},
116	forkedFrom: null,
117	createdAt: now,
118	updatedAt: now,
119	archivedAt: null,
120	settledOverride: null,
121	settledAt: null,
122	lastVisitedAt: null,
123	deletedAt: null,
124	};
125	}
126
127	function threadCreatedEvent(input: {
128	readonly id: string;
129	readonly thread: OrchestrationV2AppThread;
130	readonly now: DateTime.Utc;
131	}): OrchestrationV2DomainEvent {
132	return {
133	id: EventId.make(input.id),
134	type: "thread.created",
135	threadId: input.thread.id,
136	providerInstanceId,
137	occurredAt: input.now,
138	payload: input.thread,
139	};
140	}
141
142	it.effect("rebuilds event history one bounded page at a time", () =>
143	Effect.gen(function* () {
144	const eventStore = yield* EventStore.EventStoreV2;
145	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
146	const now = yield* DateTime.now;
147	const threadId = ThreadId.make("thread:foundation-paged-rebuild");
148	const thread = makeThread(threadId, now);
149	const eventCount = 1_005;
150	yield* eventStore.append({
151	events: [
152	threadCreatedEvent({ id: "event:paged-rebuild:0", thread, now }),
153	...Array.from({ length: eventCount - 1 }, (_, index) => ({
154	id: EventId.make(`event:paged-rebuild:${index + 1}`),
155	type: "thread.metadata-updated" as const,
156	threadId,
157	occurredAt: now,
158	payload: { ...thread, title: `Rebuilt update ${index + 1}` },
159	})),
160	],
161	});
162	let applied = 0;
163	let failAfterFirstPage = false;
164	const appliedAtRead: Array<number> = [];
165	const layerObservedStores = Layer.mergeAll(
166	Layer.succeed(EventStore.EventStoreV2, {
167	...eventStore,
168	read: (input) =>
169	Stream.suspend(() => {
170	appliedAtRead.push(applied);
171	if (failAfterFirstPage && applied >= 500) {
172	return Stream.fail(
173	new EventStore.EventStoreReadEventsError({ afterSequence: input?.afterSequence }),
174	);
175	}
176	return eventStore.read(input);
177	}),
178	}),
179	Layer.succeed(ProjectionStore.ProjectionStoreV2, {
180	...projectionStore,
181	apply: (event) =>
182	projectionStore.apply(event).pipe(
183	Effect.tap(() =>
184	Effect.sync(() => {
185	applied += 1;
186	}),
187	),
188	),
189	}),
190	);
191	const rebuild = Effect.gen(function* () {
192	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
193	return yield* maintenance.rebuild;
194	}).pipe(
195	Effect.provide(
196	Layer.fresh(ProjectionMaintenance.layer.pipe(Layer.provide(layerObservedStores))),
197	),
198	);
199	const rebuilt = yield* rebuild;
200	assert.isTrue(rebuilt.valid);
201	assert.equal(applied, eventCount);
202	assert.deepEqual(appliedAtRead, [0, 500, 1_000]);
203	assert.equal(
204	(yield* projectionStore.getThreadProjection(threadId)).thread.title,
205	"Rebuilt update 1004",
206	);
207	applied = 0;
208	failAfterFirstPage = true;
209	const failed = yield* Effect.exit(rebuild);
210	assert.equal(failed._tag, "Failure");
211	assert.equal(applied, 500);
212	assert.equal(
213	(yield* projectionStore.getThreadProjection(threadId)).thread.title,
214	"Rebuilt update 1004",
215	);
216	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
217	assert.isTrue((yield* maintenance.verify).valid);
218	}).pipe(Effect.provide(layerTest)),
219	);
220
221	it.effect("verifies thread membership using only the thread-created partial index", () =>
222	Effect.gen(function* () {
223	const sql = yield* SqlClient.SqlClient;
224	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
225	const now = DateTime.formatIso(yield* DateTime.now);
226	yield* sql`
227	WITH RECURSIVE history(n) AS (
228	SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 25000
229	)
230	INSERT INTO orchestration_events (
231	event_id, aggregate_kind, stream_id, stream_version, event_type,
232	occurred_at, actor_kind, payload_json, metadata_json, application_event_version
233	)
234	SELECT 'history:' || n, 'thread', 'thread:history', n,
235	'provider-session.detached', ${now}, 'server', '{}', '{}', 2
236	FROM history
237	`;
238	let membershipQuery: string | undefined;
239	const tracer = Tracer.make({
240	span(options) {
241	const span = new Tracer.NativeSpan(options);
242	const end = span.end.bind(span);
243	span.end = (endTime, exit) => {
244	end(endTime, exit);
245	const query = span.attributes.get("db.query.text");
246	if (
247	typeof query === "string" &&
248	query.includes("SELECT DISTINCT stream_id AS thread_id")
249	) {
250	membershipQuery = query;
251	}
252	};
253	return span;
254	},
255	});
256	yield* maintenance.verify.pipe(Effect.withTracer(tracer));
257	assert.isDefined(membershipQuery);
258	const plan = yield* sql.unsafe<{ readonly detail: string }>(
259	`EXPLAIN QUERY PLAN ${membershipQuery}`,
260	);
261	const details = plan.map((row) => row.detail).join("\n");
262	assert.match(details, /USING (?:COVERING )?INDEX orchestration_events_v2_created_threads_idx/);
263	assert.notMatch(details, /idx_orch_events_stream_sequence|TEMP B-TREE/);
264	}).pipe(Effect.provide(layerTest)),
265	);
266
267	it.effect("keeps other database work runnable while discovering compaction candidates", () =>
268	Effect.scoped(
269	Effect.gen(function* () {
270	const sql = yield* SqlClient.SqlClient;
271	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
272	const now = DateTime.formatIso(yield* DateTime.now);
273	yield* sql`
274	WITH RECURSIVE history(n) AS (
275	SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 2001
276	)
277	INSERT INTO orchestration_events (
278	event_id, aggregate_kind, stream_id, stream_version, event_type,
279	occurred_at, actor_kind, payload_json, metadata_json, application_event_version
280	)
281	SELECT 'retained:' || n, 'thread', 'thread:retained', n,
282	'provider-session.detached', ${now}, 'server', '{}', '{}', 2
283	FROM history
284	`;
285	yield* sql`
286	WITH RECURSIVE history(n) AS (
287	SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 2001
288	)
289	INSERT INTO orchestration_command_receipts (
290	command_id, aggregate_kind, aggregate_id, accepted_at, result_sequence, status, command_type
291	)
292	SELECT 'retained:' || n, 'thread', 'thread:retained', ${now}, n, 'accepted', 'thread.create'
293	FROM history
294	`;
295	const discoveryStarted = {
296	events: yield* Deferred.make<void>(),
297	receipts: yield* Deferred.make<void>(),
298	};
299	const queries = { events: 0, receipts: 0 };
300	let finished = false;
301	const tracer = Tracer.make({
302	span(options) {
303	const span = new Tracer.NativeSpan(options);
304	const end = span.end.bind(span);
305	span.end = (endTime, exit) => {
306	end(endTime, exit);
307	const query = span.attributes.get("db.query.text");
308	if (typeof query !== "string" || !query.trimStart().startsWith("SELECT")) return;
309	if (query.includes("MAX(")) return;
310	const table = query.includes("FROM orchestration_command_receipts")
311	? "receipts"
312	: query.includes("FROM orchestration_events")
313	? "events"
314	: undefined;
315	if (table === undefined) return;
316	queries[table] += 1;
317	Deferred.doneUnsafe(discoveryStarted[table], Effect.void);
318	};
319	return span;
320	},
321	});
322	const probes = yield* Effect.forEach(["events", "receipts"] as const, (table) =>
323	Effect.gen(function* () {
324	yield* Deferred.await(discoveryStarted[table]);
325	const result = yield* sql<{ readonly responsive: number }>`SELECT 1 AS responsive`;
326	assert.equal(result[0]?.responsive, 1);
327	return { table, queriesAtProbe: queries[table], finished };
328	}).pipe(Effect.forkScoped),
329	);
330
331	const summary = yield* maintenance.compactEventStore.pipe(
332	Effect.withTracer(tracer),
333	// Exercise the explicit page yields, independent of Effect's operation budget.
334	Effect.provideService(Scheduler.MaxOpsBeforeYield, Number.POSITIVE_INFINITY),
335	Effect.tap(() =>
336	Effect.sync(() => {
337	finished = true;
338	}),
339	),
340	);
341	assert.equal(summary.deletedEventCount, 0);
342	assert.equal(summary.deletedReceiptCount, 0);
343	for (const probe of probes) {
344	const result = yield* Fiber.join(probe);
345	assert.isFalse(result.finished);
346	assert.isAtLeast(result.queriesAtProbe, 1);
347	assert.isBelow(result.queriesAtProbe, queries[result.table]);
348	}
349	}),
350	).pipe(Effect.provide(layerTest)),
351	);
352
353	it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
354	it.effect("projects oversized tool bodies before both replay and live RPC retention", () =>
355	Effect.scoped(
356	Effect.gen(function* () {
357	const sink = yield* EventSink.EventSinkV2;
358	const store = yield* EventStore.EventStoreV2;
359	const projections = yield* ProjectionStore.ProjectionStoreV2;
360	const now = yield* DateTime.now;
361	const thread = makeThread(ThreadId.make("thread:large-stream-body"), now);
362	const output = {
363	content: [{ type: "text", text: "first line\n" + "x".repeat(9 * 1024 * 1024) }],
364	};
365	const tool: OrchestrationV2TurnItem = {
366	id: TurnItemId.make("tool:large-stream-body"),
367	type: "dynamic_tool",
368	threadId: thread.id,
369	runId: null,
370	nodeId: null,
371	providerThreadId: null,
372	providerTurnId: null,
373	nativeItemRef: null,
374	parentItemId: null,
375	ordinal: 1,
376	status: "running",
377	title: "Large tool",
378	toolName: "mcp__large_tool",
379	input: { query: "details" },
380	output,
381	startedAt: now,
382	completedAt: null,
383	updatedAt: now,
384	};
385	const [created] = yield* sink.write({
386	events: [
387	threadCreatedEvent({ id: "event:large-stream-body:created", thread, now }),
388	{
389	id: EventId.make("event:large-stream-body:replay"),
390	type: "turn-item.updated",
391	threadId: thread.id,
392	occurredAt: now,
393	payload: tool,
394	},
395	],
396	});
397	const pull = yield* Stream.toPull(
398	sink.stream({
399	threadId: thread.id,
400	afterSequence: created!.sequence,
401	bounded: true,
402	}),
403	);
404	const replay = yield* pull;
405	assert.lengthOf(replay, 1);
406	assert.equal(replay[0]!.event.type, "turn-item.updated");
407	assert.isBelow(Buffer.byteLength(yield* encodeJson(replay)), 4_000);
408	const [completed] = yield* sink.write({
409	events: [
410	{
411	id: EventId.make("event:large-stream-body:live"),
412	type: "turn-item.updated",
413	threadId: thread.id,
414	occurredAt: now,
415	payload: { ...tool, status: "completed", completedAt: now },
416	},
417	],
418	});
419	const live = yield* pull;
420	assert.deepEqual(
421	live.map((stored) => stored.sequence),
422	[completed!.sequence],
423	);
424	assert.isBelow(Buffer.byteLength(yield* encodeJson(live)), 4_000);
425	for (const stored of [...replay, ...live]) {
426	assert.equal(stored.event.type, "turn-item.updated");
427	if (stored.event.type !== "turn-item.updated") return;
428	assert.equal(stored.event.payload.type, "dynamic_tool");
429	if (stored.event.payload.type !== "dynamic_tool") return;
430	assert.notProperty(stored.event.payload, "output");
431	}
432	const persisted = yield* store.read({ threadId: thread.id }).pipe(Stream.runCollect);
433	const fullEvent = persisted.find(
434	(stored) => stored.sequence === completed!.sequence,
435	)!.event;
436	assert.equal(fullEvent.type, "turn-item.updated");
437	if (fullEvent.type !== "turn-item.updated") return;
438	assert.equal(fullEvent.payload.type, "dynamic_tool");
439	if (fullEvent.payload.type !== "dynamic_tool") return;
440	assert.deepEqual(fullEvent.payload.output, output);
441	const detail = (yield* projections.getThreadProjection(thread.id)).turnItems.find(
442	(item) => item.id === tool.id,
443	)!;
444	assert.equal(detail.type, "dynamic_tool");
445	if (detail.type === "dynamic_tool") assert.deepEqual(detail.output, output);
446	}),
447	),
448	);
449
450	it.effect("traces an event write and its transaction without its SQL statements", () =>
451	Effect.gen(function* () {
452	const sink = yield* EventSink.EventSinkV2;
453	const now = yield* DateTime.now;
454	const thread = makeThread(ThreadId.make("thread:untraced-write-statements"), now);
455	const names: Array<string> = [];
456	const tracer = Tracer.make({
457	span: (options) => {
458	names.push(options.name);
459	return new Tracer.NativeSpan(options);
460	},
461	});
462
463	yield* sink
464	.write({
465	events: [
466	threadCreatedEvent({ id: "event:untraced-write-statements:created", thread, now }),
467	],
468	})
469	.pipe(Effect.withTracer(tracer));
470
471	// These two spans time a slow commit; the statements inside are not traced.
472	assert.deepEqual(names, ["orchestrationV2.EventSink.write", "sql.transaction"]);
473	}),
474	);
475
476	it.effect.each(["high-water", "replay"] as const)(
477	"bounds live events while the V2 %s query is blocked",
478	(phase) =>
479	Effect.scoped(
480	Effect.gen(function* () {
481	const sink = yield* EventSink.EventSinkV2;
482	const now = yield* DateTime.now;
483	const thread = makeThread(ThreadId.make(`thread:blocked-${phase}`), now);
484	const [created] = yield* sink.write({
485	events: [threadCreatedEvent({ id: `event:blocked-${phase}:created`, thread, now })],
486	});
487	const readStarted = yield* Deferred.make<void>();
488	const readClosed = yield* Deferred.make<void>();
489	const blockRead: Statement.Transformer = (statement) => {
490	const [query] = statement.compile();
491	if (
492	query.includes("FROM orchestration_events") &&
493	query.includes("MAX(sequence)") === (phase === "high-water")
494	) {
495	return Deferred.succeed(readStarted, undefined).pipe(
496	Effect.andThen(Effect.never),
497	Effect.ensuring(Deferred.succeed(readClosed, undefined)),
498	Effect.as(statement),
499	);
500	}
501	return Effect.succeed(statement);
502	};
503	const reader = yield* sink
504	.stream({ threadId: thread.id, afterSequence: created!.sequence, bounded: true })
505	.pipe(
506	Stream.provideService(Statement.CurrentTransformer, blockRead),
507	Stream.runDrain,
508	Effect.result,
509	Effect.forkScoped,
510	);
511	yield* Deferred.await(readStarted);
512	yield* sink.write({
513	events: Array.from({ length: LIVE_STREAM_MAX_ITEMS + 1 }, (_, index) => ({
514	id: EventId.make(`event:blocked-${phase}:${index}`),
515	type: "thread.metadata-updated" as const,
516	threadId: thread.id,
517	occurredAt: now,
518	payload: { ...thread, title: `Updated ${index}` },
519	})),
520	});
521	yield* Deferred.await(readClosed);
522	const result = yield* Fiber.join(reader);
523	assert.equal(result._tag, "Failure");
524	if (result._tag === "Failure") {
525	assert.equal(result.failure._tag, "EventSinkStreamError");
526	assert.isTrue(isLiveStreamBufferError(result.failure.cause));
527	}
528	}),
529	),
530	);
531
532	it.effect(
533	"keeps internal streams subscribed while replay is blocked beyond the RPC buffer cap",
534	() =>
535	Effect.scoped(
536	Effect.gen(function* () {
537	const sink = yield* EventSink.EventSinkV2;
538	const now = yield* DateTime.now;
539	const thread = makeThread(ThreadId.make("thread:internal-stream-burst"), now);
540	const [created] = yield* sink.write({
541	events: [
542	threadCreatedEvent({ id: "event:internal-stream-burst:created", thread, now }),
543	],
544	});
545	const readStarted = yield* Deferred.make<void>();
546	const releaseRead = yield* Deferred.make<void>();
547	const blockHighWater: Statement.Transformer = (statement) => {
548	const [query] = statement.compile();
549	return query.includes("MAX(sequence)") && query.includes("FROM orchestration_events")
550	? Deferred.succeed(readStarted, undefined).pipe(
551	Effect.andThen(Deferred.await(releaseRead)),
552	Effect.as(statement),
553	)
554	: Effect.succeed(statement);
555	};
556	const eventCount = LIVE_STREAM_MAX_ITEMS + 1;
557	const reader = yield* sink
558	.stream({ threadId: thread.id, afterSequence: created!.sequence })
559	.pipe(
560	Stream.provideService(Statement.CurrentTransformer, blockHighWater),
561	Stream.take(eventCount),
562	Stream.runCollect,
563	Effect.forkScoped,
564	);
565	yield* Deferred.await(readStarted);
566	const written = yield* sink.write({
567	events: Array.from({ length: eventCount }, (_, index) => ({
568	id: EventId.make(`event:internal-stream-burst:${index}`),
569	type: "thread.metadata-updated" as const,
570	threadId: thread.id,
571	occurredAt: now,
572	payload: { ...thread, title: `Updated ${index}` },
573	})),
574	});
575	yield* Deferred.succeed(releaseRead, undefined);
576	assert.deepEqual(
577	(yield* Fiber.join(reader)).map((event) => event.sequence),
578	written.map((event) => event.sequence),
579	);
580	}),
581	),
582	);
583
584	it.effect("filters worker replay and live queues without losing matching events", () =>
585	Effect.scoped(
586	Effect.gen(function* () {
587	const sink = yield* EventSink.EventSinkV2;
588	const sql = yield* SqlClient.SqlClient;
589	const now = yield* DateTime.now;
590	const thread = makeThread(ThreadId.make("thread:filtered-worker"), now);
591	const other = makeThread(ThreadId.make("thread:filtered-worker-other"), now);
592	const run: OrchestrationV2Run = {
593	id: RunId.make("run:filtered-worker"),
594	threadId: thread.id,
595	ordinal: 1,
596	providerInstanceId,
597	modelSelection,
598	providerThreadId: null,
599	userMessageId: MessageId.make("message:filtered-worker"),
600	rootNodeId: null,
601	activeAttemptId: null,
602	status: "completed",
603	queuePosition: null,
604	requestedAt: now,
605	startedAt: now,
606	completedAt: now,
607	checkpointId: null,
608	contextHandoffId: null,
609	};
610	const runEvent = (id: string, payload: OrchestrationV2Run): OrchestrationV2DomainEvent => ({
611	id: EventId.make(id),
612	type: "run.updated",
613	threadId: payload.threadId,
614	runId: payload.id,
615	occurredAt: now,
616	payload,
617	});
618	const history = yield* sink.write({
619	events: [
620	threadCreatedEvent({ id: "event:filtered-worker:created", thread, now }),
621	threadCreatedEvent({ id: "event:filtered-worker:other", thread: other, now }),
622	runEvent("event:filtered-worker:history", run),
623	runEvent("event:filtered-worker:other-history", {
624	...run,
625	id: RunId.make("run:filtered-worker-other"),
626	threadId: other.id,
627	}),
628	],
629	});
630	// Unrelated payloads must be skipped in SQL, before decoding or
631	// allocating their bodies, even when a retained row is unreadable.
632	const original = yield* sql<{ readonly payload_json: string }>`
633	SELECT payload_json FROM orchestration_events
634	WHERE sequence = ${history[0]!.sequence}
635	`;
636	yield* Effect.acquireRelease(
637	sql`
638	UPDATE orchestration_events SET payload_json = 'unreadable unrelated payload'
639	WHERE sequence = ${history[0]!.sequence}
640	`,
641	() =>
642	sql`
643	UPDATE orchestration_events SET payload_json = ${original[0]!.payload_json}
644	WHERE sequence = ${history[0]!.sequence}
645	`.pipe(Effect.orDie),
646	);
647	const pull = yield* Stream.toPull(
648	sink.stream({ threadId: thread.id, eventType: "run.updated" }),
649	);
650	assert.deepEqual(
651	(yield* pull).map((stored) => stored.sequence),
652	[history[2]!.sequence],
653	);
654	// The worker is occupied with the previous batch while the thread
655	// publishes output. Its live queue must receive just run updates.
656	yield* sink.write({
657	events: Array.from({ length: LIVE_STREAM_MAX_ITEMS + 1 }, (_, index) => ({
658	id: EventId.make(`event:filtered-worker:output:${index}`),
659	type: "thread.metadata-updated" as const,
660	threadId: thread.id,
661	occurredAt: now,
662	payload: { ...thread, title: `Output ${index}` },
663	})),
664	});
665	const live = yield* sink.write({
666	events: [
667	runEvent("event:filtered-worker:live:1", { ...run, status: "interrupted" }),
668	runEvent("event:filtered-worker:live:2", { ...run, status: "failed" }),
669	],
670	});
671	assert.deepEqual(
672	(yield* pull).map((stored) => stored.sequence),
673	live.map((stored) => stored.sequence),
674	);
675	}),
676	),
677	);
678
679	it.effect("paginates catch-up beyond the event-store read limit", () =>
680	Effect.gen(function* () {
681	const eventSink = yield* EventSink.EventSinkV2;
682	const now = yield* DateTime.now;
683	const threadId = ThreadId.make("thread:foundation-large-catch-up");
684	const thread = makeThread(threadId, now);
685	const eventCount = 1_005;
686	const events: Array<OrchestrationV2DomainEvent> = [
687	threadCreatedEvent({ id: "event:foundation-catch-up:0", thread, now }),
688	...Array.from({ length: eventCount - 1 }, (_, index) => ({
689	id: EventId.make(`event:foundation-catch-up:${index + 1}`),
690	type: "thread.metadata-updated" as const,
691	threadId,
692	providerInstanceId,
693	occurredAt: now,
694	payload: {
695	...thread,
696	title: `Catch-up update ${index + 1}`,
697	},
698	})),
699	];
700
701	yield* eventSink.write({ events });
702	const replayed = yield* eventSink.stream({ afterSequence: 0 }).pipe(
703	Stream.take(eventCount),
704	Stream.runCollect,
705	Effect.map((events) => Array.from(events)),
706	);
707
708	assert.lengthOf(replayed, eventCount);
709	assert.deepEqual(
710	replayed.map((stored) => stored.sequence),
711	Array.from({ length: eventCount }, (_, index) => index + 1),
712	);
713	}),
714	);
715
716	it.effect("does not lose or duplicate events while transitioning from catch-up to live", () =>
717	Effect.gen(function* () {
718	const eventSink = yield* EventSink.EventSinkV2;
719	const now = yield* DateTime.now;
720	const threadId = ThreadId.make("thread:foundation-stream-race");
721	const thread = makeThread(threadId, now);
722	const created = yield* eventSink.write({
723	events: [threadCreatedEvent({ id: "event:foundation-stream-race:0", thread, now })],
724	});
725
726	let afterSequence = created[0]!.sequence;
727	for (let index = 1; index <= 32; index += 1) {
728	const nextEvent = {
729	id: EventId.make(`event:foundation-stream-race:${index}`),
730	type: "thread.metadata-updated" as const,
731	threadId,
732	providerInstanceId,
733	occurredAt: now,
734	payload: { ...thread, title: `Race update ${index}` },
735	} satisfies OrchestrationV2DomainEvent;
736	const reader = yield* eventSink
737	.stream({ threadId, afterSequence })
738	.pipe(Stream.runHead, Effect.forkChild);
739	yield* Effect.yieldNow;
740	const written = yield* eventSink.write({ events: [nextEvent] });
741	const received = yield* Fiber.join(reader);
742	if (Option.isNone(received)) {
743	return yield* Effect.die("The event stream ended before delivering the live event.");
744	}
745	assert.equal(received.value.sequence, written[0]?.sequence);
746	assert.equal(received.value.event.id, nextEvent.id);
747	afterSequence = received.value.sequence;
748	}
749	}),
750	);
751
752	it.effect("replays shared provider-session payloads across every bound thread", () =>
753	Effect.gen(function* () {
754	const eventSink = yield* EventSink.EventSinkV2;
755	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
756	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
757	const now = yield* DateTime.now;
758	const firstThreadId = ThreadId.make("thread:foundation-shared-session:first");
759	const secondThreadId = ThreadId.make("thread:foundation-shared-session:second");
760	const providerSessionId = ProviderSessionId.make("provider-session:foundation:shared");
761	const firstSession = {
762	id: providerSessionId,
763	driver: providerDriver,
764	providerInstanceId,
765	status: "ready" as const,
766	cwd: "/workspace/first",
767	model: modelSelection.model,
768	capabilities: CodexProviderCapabilitiesV2,
769	createdAt: now,
770	updatedAt: now,
771	lastError: null,
772	};
773	const secondSession = { ...firstSession, cwd: "/workspace/second" };
774
775	yield* eventSink.write({
776	events: [
777	threadCreatedEvent({
778	id: "event:foundation-shared-session:first-thread",
779	thread: makeThread(firstThreadId, now),
780	now,
781	}),
782	{
783	id: EventId.make("event:foundation-shared-session:first-attachment"),
784	type: "provider-session.attached",
785	threadId: firstThreadId,
786	driver: providerDriver,
787	providerInstanceId,
788	occurredAt: now,
789	payload: firstSession,
790	},
791	threadCreatedEvent({
792	id: "event:foundation-shared-session:second-thread",
793	thread: makeThread(secondThreadId, now),
794	now,
795	}),
796	{
797	id: EventId.make("event:foundation-shared-session:second-attachment"),
798	type: "provider-session.attached",
799	threadId: secondThreadId,
800	driver: providerDriver,
801	providerInstanceId,
802	occurredAt: now,
803	payload: secondSession,
804	},
805	],
806	});
807
808	assert.equal(
809	(yield* projectionStore.getThreadProjection(firstThreadId)).providerSessions[0]?.cwd,
810	secondSession.cwd,
811	);
812	assert.isTrue((yield* maintenance.verify).valid);
813	assert.isTrue((yield* maintenance.rebuild).valid);
814	}),
815	);
816
817	it.effect("compacts superseded state events, imported v1 events, and legacy receipts", () =>
818	Effect.gen(function* () {
819	const eventSink = yield* EventSink.EventSinkV2;
820	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
821	const sql = yield* SqlClient.SqlClient;
822	const projections = yield* ProjectionStore.ProjectionStoreV2;
823	const now = yield* DateTime.now;
824	const nowIso = DateTime.formatIso(now);
825	const threadId = ThreadId.make("thread:foundation-compact");
826	const thread = makeThread(threadId, now);
827	const messageId = MessageId.make("message:foundation-compact");
828	const threadStateEvent = (
829	suffix: string,
830	type: "thread.visited" | "thread.metadata-updated",
831	): OrchestrationV2DomainEvent => ({
832	id: EventId.make(`event:foundation-compact:${suffix}`),
833	type,
834	threadId,
835	providerInstanceId,
836	occurredAt: now,
837	payload: { ...thread, lastVisitedAt: now },
838	});
839	const messageEvent = (suffix: string, text: string): OrchestrationV2DomainEvent => ({
840	id: EventId.make(`event:foundation-compact:${suffix}`),
841	type: "message.updated",
842	threadId,
843	occurredAt: now,
844	payload: {
845	createdBy: "user",
846	creationSource: "web",
847	id: messageId,
848	threadId,
849	runId: null,
850	nodeId: null,
851	role: "user",
852	text,
853	attachments: [],
854	streaming: false,
855	createdAt: now,
856	updatedAt: now,
857	},
858	});
859	const nodeId = NodeId.make("node:foundation-compact");
860	const nodeEvent = (
861	suffix: string,
862	status: "running" | "completed",
863	): OrchestrationV2DomainEvent => ({
864	id: EventId.make(`event:foundation-compact:${suffix}`),
865	type: "node.updated",
866	threadId,
867	occurredAt: now,
868	payload: {
869	id: nodeId,
870	threadId,
871	runId: null,
872	parentNodeId: null,
873	rootNodeId: nodeId,
874	kind: "assistant_message",
875	status,
876	countsForRun: false,
877	providerThreadId: null,
878	providerTurnId: null,
879	nativeItemRef: null,
880	runtimeRequestId: null,
881	checkpointScopeId: null,
882	startedAt: now,
883	completedAt: status === "completed" ? now : null,
884	},
885	});
886	const itemEvent = (suffix: string, text: string): OrchestrationV2DomainEvent => ({
887	id: EventId.make(`event:foundation-compact:${suffix}`),
888	type: "turn-item.updated",
889	threadId,
890	occurredAt: now,
891	payload: {
892	id: TurnItemId.make("turn-item:foundation-compact"),
893	threadId,
894	runId: null,
895	nodeId: null,
896	providerThreadId: null,
897	providerTurnId: null,
898	nativeItemRef: null,
899	parentItemId: null,
900	ordinal: 1,
901	status: "completed",
902	title: null,
903	startedAt: now,
904	completedAt: now,
905	updatedAt: now,
906	type: "assistant_message",
907	messageId,
908	text,
909	streaming: false,
910	},
911	});
912
913	yield* eventSink.write({
914	events: [
915	threadCreatedEvent({ id: "event:foundation-compact:create", thread, now }),
916	threadStateEvent("meta", "thread.metadata-updated"),
917	threadStateEvent("visit-1", "thread.visited"),
918	messageEvent("message-1", "streaming"),
919	nodeEvent("node-1", "running"),
920	itemEvent("item-1", "streaming"),
921	...Array.from({ length: 501 }, (_, index) =>
922	threadStateEvent(`history-${index}`, "thread.visited"),
923	),
924	threadStateEvent("visit-2", "thread.visited"),
925	messageEvent("message-2", "final"),
926	nodeEvent("node-2", "completed"),
927	itemEvent("item-2", "final"),
928	],
929	});
930	const beforeCompaction = yield* projections.getThreadProjection(threadId);
931
932	// A fully imported legacy thread: its v1 events and pre-migration
933	// receipts are dead weight; a still-pending import keeps its rows.
934	const importedV1ThreadId = "thread:foundation-compact-v1-imported";
935	const pendingV1ThreadId = "thread:foundation-compact-v1-pending";
936	const insertV1Event = (threadIdValue: string, version: number) => sql`
937	INSERT INTO orchestration_events (
938	event_id, aggregate_kind, stream_id, stream_version, event_type,
939	occurred_at, actor_kind, payload_json, metadata_json, application_event_version
940	)
941	VALUES (
942	${`event:v1:${threadIdValue}:${version}`}, 'thread', ${threadIdValue}, ${version},
943	'thread.message-appended', ${nowIso}, 'user', '{}', '{}', 1
944	)
945	`;
946	yield* insertV1Event(importedV1ThreadId, 1);
947	yield* insertV1Event(importedV1ThreadId, 2);
948	yield* insertV1Event(pendingV1ThreadId, 1);
949	yield* sql`
950	INSERT INTO orchestration_v2_legacy_imports (
951	thread_id, source_updated_at, shell_imported_at, transcript_imported_at,
952	imported_message_count, last_error
953	)
954	VALUES
955	(${importedV1ThreadId}, ${nowIso}, ${nowIso}, ${nowIso}, 2, NULL),
956	(${pendingV1ThreadId}, ${nowIso}, ${nowIso}, NULL, 0, NULL)
957	ON CONFLICT(thread_id) DO NOTHING
958	`;
959	yield* sql`
960	INSERT INTO orchestration_command_receipts (
961	command_id, aggregate_kind, aggregate_id, accepted_at, result_sequence, status, command_type
962	)
963	VALUES
964	('command:foundation-compact:legacy', 'thread', ${importedV1ThreadId}, ${nowIso}, 1, 'accepted', 'legacy'),
965	('command:foundation-compact:pending', 'thread', ${pendingV1ThreadId}, ${nowIso}, 1, 'accepted', 'legacy')
966	ON CONFLICT(command_id) DO NOTHING
967	`;
968
969	const summary = yield* maintenance.compactEventStore;
970	// Superseded state spans several discovery pages. Both turn-item updates stay.
971	assert.isAtLeast(summary.deletedEventCount, 507);
972	assert.isAtLeast(summary.deletedReceiptCount, 1);
973
974	const remaining = yield* sql<{ readonly event_id: string }>`
975	SELECT event_id
976	FROM orchestration_events
977	WHERE aggregate_kind = 'thread'
978	AND stream_id = ${threadId}
979	ORDER BY sequence ASC
980	`;
981	assert.deepEqual(
982	remaining.map((row) => row.event_id),
983	[
984	"event:foundation-compact:create",
985	"event:foundation-compact:item-1",
986	"event:foundation-compact:visit-2",
987	"event:foundation-compact:message-2",
988	"event:foundation-compact:node-2",
989	"event:foundation-compact:item-2",
990	],
991	);
992
993	const remainingV1 = yield* sql<{ readonly stream_id: string }>`
994	SELECT stream_id
995	FROM orchestration_events
996	WHERE application_event_version = 1
997	AND stream_id IN (${importedV1ThreadId}, ${pendingV1ThreadId})
998	`;
999	assert.deepEqual(
1000	remainingV1.map((row) => row.stream_id),
1001	[pendingV1ThreadId],
1002	);
1003
1004	const remainingReceipts = yield* sql<{ readonly command_id: string }>`
1005	SELECT command_id
1006	FROM orchestration_command_receipts
1007	WHERE command_id IN ('command:foundation-compact:legacy', 'command:foundation-compact:pending')
1008	`;
1009	assert.deepEqual(
1010	remainingReceipts.map((row) => row.command_id),
1011	["command:foundation-compact:pending"],
1012	);
1013
1014	// Replay across the deletion gaps must still produce a valid projection.
1015	assert.isTrue((yield* maintenance.verify).valid);
1016	assert.isTrue((yield* maintenance.rebuild).valid);
1017	assert.deepEqual(yield* projections.getThreadProjection(threadId), beforeCompaction);
1018	}),
1019	);
1020
1021	it.effect("verifies and rebuilds projections with cross-thread subagent relations", () =>
1022	Effect.gen(function* () {
1023	const eventSink = yield* EventSink.EventSinkV2;
1024	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1025	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
1026	const sql = yield* SqlClient.SqlClient;
1027	const now = yield* DateTime.now;
1028	const parentThreadId = ThreadId.make("thread:foundation-cross-thread:parent");
1029	const childThreadId = ThreadId.make("thread:foundation-cross-thread:child");
1030	const childProviderThreadId = ProviderThreadId.make(
1031	"provider-thread:foundation-cross-thread:child",
1032	);
1033	const subagentId = NodeId.make("subagent:foundation-cross-thread");
1034	const spawnTransferId = ContextTransferId.make("transfer:foundation-cross-thread:spawn");
1035	const resultTransferId = ContextTransferId.make("transfer:foundation-cross-thread:result");
1036	const parentThread = makeThread(parentThreadId, now);
1037	const childThread = {
1038	...makeThread(childThreadId, now),
1039	createdBy: "agent" as const,
1040	creationSource: "provider" as const,
1041	lineage: {
1042	parentThreadId,
1043	relationshipToParent: "subagent" as const,
1044	rootThreadId: parentThreadId,
1045	},
1046	};
1047
1048	yield* eventSink.write({
1049	events: [
1050	threadCreatedEvent({
1051	id: "event:foundation-cross-thread:parent",
1052	thread: parentThread,
1053	now,
1054	}),
1055	threadCreatedEvent({
1056	id: "event:foundation-cross-thread:child",
1057	thread: childThread,
1058	now,
1059	}),
1060	{
1061	id: EventId.make("event:foundation-cross-thread:subagent"),
1062	type: "subagent.updated",
1063	threadId: parentThreadId,
1064	nodeId: subagentId,
1065	providerInstanceId,
1066	occurredAt: now,
1067	payload: {
1068	id: subagentId,
1069	threadId: parentThreadId,
1070	runId: null,
1071	parentNodeId: NodeId.make("node:foundation-cross-thread:parent"),
1072	origin: "app_owned",
1073	createdBy: "agent",
1074	driver: providerDriver,
1075	providerInstanceId,
1076	providerThreadId: childProviderThreadId,
1077	childThreadId,
1078	nativeTaskRef: null,
1079	prompt: "Inspect the child flow",
1080	title: "Cross-thread child",
1081	model: modelSelection.model,
1082	status: "completed",
1083	result: "done",
1084	startedAt: now,
1085	completedAt: now,
1086	updatedAt: now,
1087	},
1088	},
1089	{
1090	id: EventId.make("event:foundation-cross-thread:spawn-transfer"),
1091	type: "context-transfer.created",
1092	threadId: childThreadId,
1093	providerInstanceId,
1094	occurredAt: now,
1095	payload: {
1096	id: spawnTransferId,
1097	type: "subagent_spawn",
1098	sourceThreadId: parentThreadId,
1099	targetThreadId: childThreadId,
1100	sourcePoint: { threadId: parentThreadId },
1101	basePoint: null,
1102	sourceProviderInstanceId: providerInstanceId,
1103	targetProviderInstanceId: providerInstanceId,
1104	targetRunId: null,
1105	status: "consumed",
1106	resolution: null,
1107	createdBy: "agent",
1108	error: null,
1109	createdAt: now,
1110	updatedAt: now,
1111	consumedAt: now,
1112	},
1113	},
1114	{
1115	id: EventId.make("event:foundation-cross-thread:provider-thread"),
1116	type: "provider-thread.updated",
1117	threadId: parentThreadId,
1118	driver: providerDriver,
1119	providerInstanceId,
1120	occurredAt: now,
1121	payload: {
1122	id: childProviderThreadId,
1123	driver: providerDriver,
1124	providerInstanceId,
1125	providerSessionId: null,
1126	appThreadId: childThreadId,
1127	ownerNodeId: null,
1128	nativeThreadRef: null,
1129	nativeConversationHeadRef: null,
1130	status: "idle",
1131	firstRunOrdinal: 1,
1132	lastRunOrdinal: 1,
1133	handoffIds: [],
1134	forkedFrom: null,
1135	createdAt: now,
1136	updatedAt: now,
1137	},
1138	},
1139	{
1140	id: EventId.make("event:foundation-cross-thread:result-transfer"),
1141	type: "context-transfer.created",
1142	threadId: parentThreadId,
1143	providerInstanceId,
1144	occurredAt: now,
1145	payload: {
1146	id: resultTransferId,
1147	type: "subagent_result",
1148	sourceThreadId: childThreadId,
1149	targetThreadId: parentThreadId,
1150	sourcePoint: { threadId: childThreadId },
1151	basePoint: null,
1152	sourceProviderInstanceId: providerInstanceId,
1153	targetProviderInstanceId: providerInstanceId,
1154	targetRunId: null,
1155	status: "consumed",
1156	resolution: null,
1157	createdBy: "system",
1158	error: null,
1159	createdAt: now,
1160	updatedAt: now,
1161	consumedAt: now,
1162	},
1163	},
1164	],
1165	});
1166
1167	const assertCrossThreadProjection = Effect.gen(function* () {
1168	const parent = yield* projectionStore.getThreadProjection(parentThreadId);
1169	const child = yield* projectionStore.getThreadProjection(childThreadId);
1170	const expectedTransferIds = [spawnTransferId, resultTransferId].toSorted();
1171	assert.deepEqual(
1172	parent.contextTransfers.map((transfer) => transfer.id).toSorted(),
1173	expectedTransferIds,
1174	);
1175	assert.deepEqual(
1176	child.contextTransfers.map((transfer) => transfer.id).toSorted(),
1177	expectedTransferIds,
1178	);
1179	assert.deepEqual(
1180	parent.providerThreads.map((providerThread) => providerThread.id),
1181	[childProviderThreadId],
1182	);
1183	assert.equal(child.thread.activeProviderThreadId, childProviderThreadId);
1184	});
1185
1186	yield* assertCrossThreadProjection;
1187	assert.isTrue((yield* maintenance.verify).valid);
1188	yield* sql`
1189	UPDATE orchestration_v2_projection_provider_threads SET payload_json = '{}'
1190	WHERE provider_thread_id = ${childProviderThreadId}
1191	`;
1192	assert.deepEqual(
1193	new Set((yield* maintenance.verify).unreadableThreadIds),
1194	new Set([parentThreadId, childThreadId]),
1195	);
1196	assert.isTrue((yield* maintenance.rebuild).valid);
1197	yield* assertCrossThreadProjection;
1198	}),
1199	);
1200
1201	it.effect(
1202	"rolls back events, projections, receipts, and effects after a projection failure",
1203	() =>
1204	Effect.gen(function* () {
1205	const eventSink = yield* EventSink.EventSinkV2;
1206	const eventStore = yield* EventStore.EventStoreV2;
1207	const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
1208	const outbox = yield* EffectOutbox.EffectOutboxV2;
1209	const sql = yield* SqlClient.SqlClient;
1210	const now = yield* DateTime.now;
1211	const commandId = CommandId.make("command:foundation-atomic-failure");
1212	const threadId = ThreadId.make("thread:foundation-atomic-failure");
1213	const scopeId = CheckpointScopeId.make("scope:foundation-atomic-failure");
1214	const checkpoint = (index: number) => ({
1215	id: CheckpointId.make(`checkpoint:foundation-atomic-failure:${index}`),
1216	threadId,
1217	scopeId,
1218	runId: null,
1219	nodeId: NodeId.make("node:foundation-atomic-failure"),
1220	parentCheckpointId: null,
1221	ordinalWithinScope: 1,
1222	appRunOrdinal: null,
1223	ref: CheckpointRef.make(`checkpoint-ref:foundation-atomic-failure:${index}`),
1224	status: "ready" as const,
1225	files: [],
1226	capturedAt: now,
1227	});
1228	const events = [1, 2].map(
1229	(index) =>
1230	({
1231	id: EventId.make(`event:foundation-atomic-failure:${index}`),
1232	type: "checkpoint.captured",
1233	threadId,
1234	occurredAt: now,
1235	payload: checkpoint(index),
1236	}) satisfies OrchestrationV2DomainEvent,
1237	);
1238
1239	const exit = yield* Effect.exit(
1240	eventSink.commitCommand({
1241	commandId,
1242	threadId,
1243	commandType: "checkpoint.atomicity-test",
1244	acceptedAt: now,
1245	events,
1246	effects: [
1247	{
1248	id: "effect:foundation-atomic-failure",
1249	commandId,
1250	threadId,
1251	request: {
1252	type: "provider-turn.start",
1253	runId: RunId.make("run:foundation-atomic-failure"),
1254	},
1255	},
1256	],
1257	}),
1258	);
1259	assert.equal(exit._tag, "Failure");
1260	assert.isTrue(Option.isNone(yield* receipts.getByCommandId(commandId)));
1261	assert.deepEqual(yield* outbox.listByCommandId(commandId), []);
1262	assert.deepEqual(
1263	yield* eventStore.readByCommandId({ commandId }).pipe(
1264	Stream.runCollect,
1265	Effect.map((events) => Array.from(events)),
1266	),
1267	[],
1268	);
1269	const checkpointRows = yield* sql<{ readonly count: number }>`
1270	SELECT COUNT(*) AS count
1271	FROM orchestration_v2_projection_checkpoints
1272	WHERE thread_id = ${threadId}
1273	`;
1274	assert.equal(checkpointRows[0]?.count, 0);
1275	}),
1276	);
1277
1278	it.effect("replays every command event across bounded persistence pages", () =>
1279	Effect.gen(function* () {
1280	const sql = yield* SqlClient.SqlClient;
1281	const eventStore = yield* EventStore.EventStoreV2;
1282	const eventSink = yield* EventSink.EventSinkV2;
1283	const now = yield* DateTime.now;
1284	const occurredAt = DateTime.formatIso(now);
1285	const commandId = CommandId.make("command:paged-command-replay");
1286	const exactPageCommandId = CommandId.make("command:paged-command-replay:exact");
1287	const emptyCommandId = CommandId.make("command:paged-command-replay:empty");
1288	const threadId = "thread:paged-command-replay";
1289
1290	const eventRow = (
1291	ordinal: number,
1292	input: {
1293	readonly commandId?: string | null;
1294	readonly aggregateKind?: "project" | "thread";
1295	readonly streamId?: string;
1296	readonly version?: number;
1297	} = {},
1298	) => ({
1299	event_id: `event:paged-command-replay:${ordinal}`,
1300	aggregate_kind: input.aggregateKind ?? "thread",
1301	stream_id: input.streamId ?? threadId,
1302	stream_version: ordinal,
1303	event_type: "provider-session.detached",
1304	occurred_at: occurredAt,
1305	command_id: input.commandId ?? null,
1306	causation_event_id: null,
1307	correlation_id: null,
1308	actor_kind: "server",
1309	payload_json: JSON.stringify({
1310	providerSessionId: `session:paged-command-replay:${ordinal}`,
1311	detachedAt: occurredAt,
1312	}),
1313	metadata_json: "{}",
1314	application_event_version: input.version ?? 2,
1315	});
1316
1317	const matchingCount = 1_001;
1318	const rows: Array<ReturnType<typeof eventRow>> = [];
1319	let ordinal = 0;
1320	for (let index = 0; index < matchingCount; index += 1) {
1321	rows.push(eventRow(++ordinal, { commandId }));
1322	if (index % 2 === 0) {
1323	rows.push(eventRow(++ordinal, { commandId: "command:unrelated" }));
1324	}
1325	if (index % 5 === 0) {
1326	rows.push(eventRow(++ordinal));
1327	rows.push(eventRow(++ordinal, { commandId, version: 1 }));
1328	rows.push(
1329	eventRow(++ordinal, {
1330	commandId,
1331	aggregateKind: "project",
1332	streamId: `project:paged-command-replay:${index}`,
1333	}),
1334	);
1335	}
1336	}
1337	for (let index = 0; index < 500; index += 1) {
1338	rows.push(eventRow(++ordinal, { commandId: exactPageCommandId }));
1339	}
1340	const inserted = yield* Effect.forEach(
1341	Array.from({ length: Math.ceil(rows.length / 400) }, (_, chunk) =>
1342	rows.slice(chunk * 400, (chunk + 1) * 400),
1343	),
1344	(chunk) =>
1345	sql<{
1346	readonly sequence: number;
1347	readonly command_id: string | null;
1348	readonly aggregate_kind: string;
1349	readonly application_event_version: number;
1350	}>`
1351	INSERT INTO orchestration_events ${sql.insert(chunk)}
1352	RETURNING sequence, command_id, aggregate_kind, application_event_version
1353	`,
1354	{ concurrency: 1 },
1355	).pipe(Effect.map((chunks) => chunks.flat()));
1356	const expectedSequences = inserted
1357	.filter(
1358	(row) =>
1359	row.command_id === commandId &&
1360	row.aggregate_kind === "thread" &&
1361	row.application_event_version === 2,
1362	)
1363	.map((row) => row.sequence);
1364	assert.lengthOf(expectedSequences, matchingCount);
1365
1366	yield* sql`
1367	INSERT INTO orchestration_command_receipts (
1368	command_id, aggregate_kind, aggregate_id, accepted_at, result_sequence, status, command_type
1369	)
1370	VALUES (
1371	${commandId}, 'thread', ${threadId}, ${occurredAt},
1372	${expectedSequences.at(-1)!}, 'accepted', 'thread.create'
1373	)
1374	`;
1375
1376	const pageLimits: Array<number> = [];
1377	const recordReads: Statement.Transformer = (statement) => {
1378	const [query, params] = statement.compile();
1379	if (query.includes("FROM orchestration_events") && query.includes("command_id")) {
1380	const limit = params.at(-1);
1381	if (typeof limit === "number") {
1382	pageLimits.push(limit);
1383	}
1384	}
1385	return Effect.succeed(statement);
1386	};
1387	const collectByCommandId = (id: CommandId) =>
1388	eventStore.readByCommandId({ commandId: id }).pipe(
1389	Stream.provideService(Statement.CurrentTransformer, recordReads),
1390	Stream.runCollect,
1391	Effect.map((events) => Array.from(events)),
1392	);
1393
1394	const replayed = yield* collectByCommandId(commandId);
1395	assert.deepEqual(
1396	replayed.map((stored) => stored.sequence),
1397	expectedSequences,
1398	);
1399	assert.deepEqual(pageLimits, [500, 500, 500]);
1400
1401	pageLimits.length = 0;
1402	const exactPage = yield* collectByCommandId(exactPageCommandId);
1403	assert.lengthOf(exactPage, 500);
1404	assert.deepEqual(pageLimits, [500, 500]);
1405
1406	pageLimits.length = 0;
1407	const empty = yield* collectByCommandId(emptyCommandId);
1408	assert.lengthOf(empty, 0);
1409	assert.deepEqual(pageLimits, [500]);
1410
1411	pageLimits.length = 0;
1412	const retried = yield* eventSink
1413	.commitCommand({
1414	commandId,
1415	threadId: ThreadId.make(threadId),
1416	commandType: "thread.create",
1417	acceptedAt: now,
1418	events: [
1419	threadCreatedEvent({
1420	id: "event:paged-command-replay:retry",
1421	thread: makeThread(ThreadId.make(threadId), now),
1422	now,
1423	}),
1424	],
1425	effects: [],
1426	})
1427	.pipe(Effect.provideService(Statement.CurrentTransformer, recordReads));
1428	assert.isFalse(retried.committed);
1429	assert.equal(retried.receipt.resultSequence, expectedSequences.at(-1));
1430	assert.deepEqual(
1431	retried.storedEvents.map((stored) => stored.sequence),
1432	expectedSequences,
1433	);
1434	assert.deepEqual(pageLimits, [500, 500, 500]);
1435	}),
1436	);
1437
1438	it.effect("keeps one durable effect across command retries and executes it after recovery", () =>
1439	Effect.gen(function* () {
1440	const eventSink = yield* EventSink.EventSinkV2;
1441	const outbox = yield* EffectOutbox.EffectOutboxV2;
1442	const now = yield* DateTime.now;
1443	const commandId = CommandId.make("command:foundation-effect-recovery");
1444	const threadId = ThreadId.make("thread:foundation-effect-recovery");
1445	const thread = makeThread(threadId, now);
1446	const event = threadCreatedEvent({
1447	id: "event:foundation-effect-recovery",
1448	thread,
1449	now,
1450	});
1451	const effect = {
1452	id: "effect:foundation-effect-recovery",
1453	commandId,
1454	threadId,
1455	request: {
1456	type: "provider-turn.start" as const,
1457	runId: RunId.make("run:foundation-effect-recovery"),
1458	},
1459	};
1460
1461	const first = yield* eventSink.commitCommand({
1462	commandId,
1463	threadId,
1464	commandType: "foundation.effect-recovery",
1465	acceptedAt: now,
1466	events: [event],
1467	effects: [effect],
1468	});
1469	const retry = yield* eventSink.commitCommand({
1470	commandId,
1471	threadId,
1472	commandType: "foundation.effect-recovery",
1473	acceptedAt: now,
1474	events: [event],
1475	effects: [effect],
1476	});
1477
1478	assert.isTrue(first.committed);
1479	assert.isFalse(retry.committed);
1480	assert.equal(retry.receipt.resultSequence, first.receipt.resultSequence);
1481	assert.lengthOf(retry.storedEvents, 1);
1482	assert.lengthOf(yield* outbox.listByCommandId(commandId), 1);
1483
1484	const executionCount = yield* Ref.make(0);
1485	const layerExecutor = Layer.succeed(
1486	EffectWorker.OrchestrationEffectExecutorV2,
1487	EffectWorker.OrchestrationEffectExecutorV2.of({
1488	execute: () => Ref.update(executionCount, (count) => count + 1),
1489	}),
1490	);
1491	const layerWorker = EffectWorker.layerWithOptions({ workerId: "recovery-worker" }).pipe(
1492	Layer.provide(
1493	Layer.merge(Layer.succeed(EffectOutbox.EffectOutboxV2, outbox), layerExecutor),
1494	),
1495	);
1496	yield* Effect.gen(function* () {
1497	const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
1498	assert.isTrue(yield* worker.runOnce);
1499	assert.isFalse(yield* worker.runOnce);
1500	}).pipe(Effect.provide(layerWorker));
1501
1502	assert.equal(yield* Ref.get(executionCount), 1);
1503	const storedEffect = yield* outbox.get(effect.id);
1504	assert.isTrue(Option.isSome(storedEffect));
1505	if (Option.isSome(storedEffect)) {
1506	assert.equal(storedEffect.value.status, "succeeded");
1507	}
1508	}),
1509	);
1510
1511	it.effect("does not wake claimers for effects from an idempotent command retry", () =>
1512	Effect.gen(function* () {
1513	const eventSink = yield* EventSink.EventSinkV2;
1514	const outbox = yield* EffectOutbox.EffectOutboxV2;
1515	const now = yield* DateTime.now;
1516	const commandId = CommandId.make("command:foundation-idempotent-wakeup");
1517	const threadId = ThreadId.make("thread:foundation-idempotent-wakeup");
1518	const input = {
1519	commandId,
1520	threadId,
1521	commandType: "foundation.idempotent-wakeup",
1522	acceptedAt: now,
1523	events: [
1524	threadCreatedEvent({
1525	id: "event:foundation-idempotent-wakeup",
1526	thread: makeThread(threadId, now),
1527	now,
1528	}),
1529	],
1530	effects: [
1531	{
1532	id: "effect:foundation-idempotent-wakeup",
1533	commandId,
1534	threadId,
1535	request: { type: "terminal.cleanup" as const },
1536	},
1537	],
1538	};
1539
1540	assert.isTrue((yield* eventSink.commitCommand(input)).committed);
1541	yield* outbox.awaitAvailable;
1542	assert.isFalse((yield* eventSink.commitCommand(input)).committed);
1543
1544	const unexpectedWake = yield* outbox.awaitAvailable.pipe(Effect.forkChild);
1545	yield* Effect.yieldNow;
1546	assert.isUndefined(unexpectedWake.pollUnsafe());
1547	}).pipe(Effect.provide(Layer.fresh(layerTest))),
1548	);
1549
1550	it.effect("does not publish a stale provider start after an interrupt wins", () =>
1551	Effect.gen(function* () {
1552	const eventSink = yield* EventSink.EventSinkV2;
1553	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1554	const outbox = yield* EffectOutbox.EffectOutboxV2;
1555	const now = yield* DateTime.now;
1556	const threadId = ThreadId.make("thread:foundation-stale-provider-start");
1557	const runId = RunId.make("run:foundation-stale-provider-start");
1558	const attemptId = RunAttemptId.make("run-attempt:foundation-stale-provider-start");
1559	const thread = makeThread(threadId, now);
1560	const startingRun: OrchestrationV2Run = {
1561	id: runId,
1562	threadId,
1563	ordinal: 1,
1564	providerInstanceId,
1565	modelSelection,
1566	providerThreadId: null,
1567	userMessageId: MessageId.make("message:foundation-stale-provider-start"),
1568	rootNodeId: null,
1569	activeAttemptId: attemptId,
1570	status: "starting",
1571	queuePosition: null,
1572	requestedAt: now,
1573	startedAt: null,
1574	completedAt: null,
1575	checkpointId: null,
1576	contextHandoffId: null,
1577	};
1578	yield* eventSink.write({
1579	events: [
1580	threadCreatedEvent({
1581	id: "event:foundation-stale-provider-start:thread",
1582	thread,
1583	now,
1584	}),
1585	{
1586	id: EventId.make("event:foundation-stale-provider-start:run"),
1587	type: "run.created",
1588	threadId,
1589	runId,
1590	providerInstanceId,
1591	occurredAt: now,
1592	payload: startingRun,
1593	},
1594	],
1595	});
1596
1597	const captureEffect = {
1598	id: "effect:foundation-current-capture",
1599	commandId: CommandId.make("command:foundation-current-capture"),
1600	threadId,
1601	request: {
1602	type: "checkpoint.capture" as const,
1603	runId,
1604	scopeId: CheckpointScopeId.make("scope:foundation-current-capture"),
1605	},
1606	};
1607	assert.isTrue(
1608	(yield* eventSink.writeIfRunCurrent({
1609	threadId,
1610	runId,
1611	activeAttemptId: attemptId,
1612	expectedStatus: "starting",
1613	events: [],
1614	effects: [captureEffect],
1615	})).committed,
1616	);
1617	assert.isTrue(Option.isSome(yield* outbox.get(captureEffect.id)));
1618	yield* outbox.awaitAvailable;
1619	const staleCaptureEffect = { ...captureEffect, id: "effect:foundation-stale-capture" };
1620
1621	const reachedPrecommitGap = yield* Deferred.make<void>();
1622	const releaseStaleStart = yield* Deferred.make<void>();
1623	const providerStartCount = yield* Ref.make(0);
1624	const staleStartFiber = yield* Effect.gen(function* () {
1625	yield* Deferred.succeed(reachedPrecommitGap, undefined);
1626	yield* Deferred.await(releaseStaleStart);
1627	const result = yield* eventSink.writeIfRunCurrent({
1628	threadId,
1629	runId,
1630	activeAttemptId: attemptId,
1631	expectedStatus: "starting",
1632	effects: [staleCaptureEffect],
1633	events: [
1634	{
1635	id: EventId.make("event:foundation-stale-provider-start:running"),
1636	type: "run.updated",
1637	threadId,
1638	runId,
1639	providerInstanceId,
1640	occurredAt: now,
1641	payload: { ...startingRun, status: "running", startedAt: now },
1642	},
1643	],
1644	});
1645	if (result.committed) {
1646	yield* Ref.update(providerStartCount, (count) => count + 1);
1647	}
1648	return result;
1649	}).pipe(Effect.forkChild);
1650
1651	yield* Deferred.await(reachedPrecommitGap);
1652	const interruptedAt = yield* DateTime.now;
1653	yield* eventSink.write({
1654	events: [
1655	{
1656	id: EventId.make("event:foundation-stale-provider-start:cancelled"),
1657	type: "run.updated",
1658	threadId,
1659	runId,
1660	providerInstanceId,
1661	occurredAt: interruptedAt,
1662	payload: {
1663	...startingRun,
1664	status: "cancelled",
1665	completedAt: interruptedAt,
1666	},
1667	},
1668	],
1669	});
1670	yield* Deferred.succeed(releaseStaleStart, undefined);
1671
1672	const staleResult = yield* Fiber.join(staleStartFiber);
1673	assert.isFalse(staleResult.committed);
1674	assert.isTrue(Option.isNone(yield* outbox.get(staleCaptureEffect.id)));
1675	assert.deepEqual(staleResult.storedEvents, []);
1676	assert.equal(yield* Ref.get(providerStartCount), 0);
1677	const projection = yield* projectionStore.getThreadProjection(threadId);
1678	assert.equal(projection.runs[0]?.status, "cancelled");
1679	}).pipe(Effect.provide(Layer.fresh(layerTest))),
1680	);
1681
1682	it.effect("guards post-terminal provider-thread writes by attempt and run ordinal", () =>
1683	Effect.gen(function* () {
1684	const eventSink = yield* EventSink.EventSinkV2;
1685	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1686	const now = yield* DateTime.now;
1687	const threadId = ThreadId.make("thread:foundation-provider-thread-owner");
1688	const runId = RunId.make("run:foundation-provider-thread-owner");
1689	const attemptId = RunAttemptId.make("attempt:foundation-provider-thread-owner");
1690	const replacementAttemptId = RunAttemptId.make(
1691	"attempt:foundation-provider-thread-owner:replacement",
1692	);
1693	const rootNodeId = NodeId.make("node:foundation-provider-thread-owner");
1694	const providerThreadId = ProviderThreadId.make(
1695	"provider-thread:foundation-provider-thread-owner",
1696	);
1697	const thread = makeThread(threadId, now);
1698	const run: OrchestrationV2Run = {
1699	id: runId,
1700	threadId,
1701	ordinal: 1,
1702	providerInstanceId,
1703	modelSelection,
1704	providerThreadId,
1705	userMessageId: MessageId.make("message:foundation-provider-thread-owner"),
1706	rootNodeId,
1707	activeAttemptId: attemptId,
1708	status: "completed",
1709	queuePosition: null,
1710	requestedAt: now,
1711	startedAt: now,
1712	completedAt: now,
1713	checkpointId: null,
1714	contextHandoffId: null,
1715	};
1716	const baseProviderThread = {
1717	id: providerThreadId,
1718	driver: providerDriver,
1719	providerInstanceId,
1720	providerSessionId: null,
1721	appThreadId: threadId,
1722	ownerNodeId: null,
1723	nativeThreadRef: null,
1724	nativeConversationHeadRef: null,
1725	status: "idle" as const,
1726	firstRunOrdinal: 1,
1727	lastRunOrdinal: 1,
1728	handoffIds: [] as const,
1729	forkedFrom: null,
1730	pendingBackgroundTasks: [
1731	{ taskId: "bg-owner", description: "sleep 20", kind: "command" as const },
1732	],
1733	createdAt: now,
1734	updatedAt: now,
1735	};
1736
1737	yield* eventSink.write({
1738	events: [
1739	threadCreatedEvent({
1740	id: "event:foundation-provider-thread-owner:thread",
1741	thread,
1742	now,
1743	}),
1744	{
1745	id: EventId.make("event:foundation-provider-thread-owner:run"),
1746	type: "run.created",
1747	threadId,
1748	runId,
1749	nodeId: rootNodeId,
1750	providerInstanceId,
1751	occurredAt: now,
1752	payload: run,
1753	},
1754	{
1755	id: EventId.make("event:foundation-provider-thread-owner:provider-thread"),
1756	type: "provider-thread.updated",
1757	threadId,
1758	driver: providerDriver,
1759	providerInstanceId,
1760	occurredAt: now,
1761	payload: baseProviderThread,
1762	},
1763	],
1764	});
1765
1766	const ownedClear = yield* eventSink.writeIfProviderThreadOwner({
1767	providerThreadId,
1768	runId,
1769	activeAttemptId: attemptId,
1770	expectedLastRunOrdinal: 1,
1771	events: [
1772	{
1773	id: EventId.make("event:foundation-provider-thread-owner:clear"),
1774	type: "provider-thread.updated",
1775	threadId,
1776	driver: providerDriver,
1777	providerInstanceId,
1778	occurredAt: now,
1779	payload: {
1780	...baseProviderThread,
1781	pendingBackgroundTasks: [],
1782	updatedAt: now,
1783	},
1784	},
1785	],
1786	});
1787	assert.isTrue(ownedClear.committed);
1788	assert.equal(ownedClear.storedEvents.length, 1);
1789
1790	const afterReplacement = yield* DateTime.now;
1791	yield* eventSink.write({
1792	events: [
1793	{
1794	id: EventId.make("event:foundation-provider-thread-owner:replacement-attempt"),
1795	type: "run.updated",
1796	threadId,
1797	runId,
1798	nodeId: rootNodeId,
1799	providerInstanceId,
1800	occurredAt: afterReplacement,
1801	payload: {
1802	...run,
1803	activeAttemptId: replacementAttemptId,
1804	status: "running",
1805	},
1806	},
1807	],
1808	});
1809
1810	const supersededAttemptWrite = yield* eventSink.writeIfProviderThreadOwner({
1811	providerThreadId,
1812	runId,
1813	activeAttemptId: attemptId,
1814	expectedLastRunOrdinal: 1,
1815	events: [
1816	{
1817	id: EventId.make("event:foundation-provider-thread-owner:superseded-attempt"),
1818	type: "provider-thread.updated",
1819	threadId,
1820	driver: providerDriver,
1821	providerInstanceId,
1822	occurredAt: afterReplacement,
1823	payload: {
1824	...baseProviderThread,
1825	status: "active",
1826	pendingBackgroundTasks: [
1827	{
1828	taskId: "bg-superseded",
1829	description: "should not land",
1830	kind: "command" as const,
1831	},
1832	],
1833	updatedAt: afterReplacement,
1834	},
1835	},
1836	],
1837	});
1838	assert.isFalse(supersededAttemptWrite.committed);
1839	assert.deepEqual(supersededAttemptWrite.storedEvents, []);
1840
1841	yield* eventSink.write({
1842	events: [
1843	{
1844	id: EventId.make("event:foundation-provider-thread-owner:replacement-completed"),
1845	type: "run.updated",
1846	threadId,
1847	runId,
1848	nodeId: rootNodeId,
1849	providerInstanceId,
1850	occurredAt: afterReplacement,
1851	payload: {
1852	...run,
1853	activeAttemptId: replacementAttemptId,
1854	},
1855	},
1856	{
1857	id: EventId.make("event:foundation-provider-thread-owner:newer-run"),
1858	type: "provider-thread.updated",
1859	threadId,
1860	driver: providerDriver,
1861	providerInstanceId,
1862	occurredAt: afterReplacement,
1863	payload: {
1864	...baseProviderThread,
1865	lastRunOrdinal: 2,
1866	status: "active",
1867	pendingBackgroundTasks: [],
1868	updatedAt: afterReplacement,
1869	},
1870	},
1871	],
1872	});
1873
1874	const staleOrdinalWrite = yield* eventSink.writeIfProviderThreadOwner({
1875	providerThreadId,
1876	runId,
1877	activeAttemptId: replacementAttemptId,
1878	expectedLastRunOrdinal: 1,
1879	events: [
1880	{
1881	id: EventId.make("event:foundation-provider-thread-owner:stale-ordinal"),
1882	type: "provider-thread.updated",
1883	threadId,
1884	driver: providerDriver,
1885	providerInstanceId,
1886	occurredAt: afterReplacement,
1887	payload: baseProviderThread,
1888	},
1889	],
1890	});
1891	assert.isFalse(staleOrdinalWrite.committed);
1892	assert.deepEqual(staleOrdinalWrite.storedEvents, []);
1893
1894	const projection = yield* projectionStore.getThreadProjection(threadId);
1895	const providerThread = projection.providerThreads.find(
1896	(candidate) => candidate.id === providerThreadId,
1897	);
1898	assert.isDefined(providerThread);
1899	assert.equal(providerThread?.lastRunOrdinal, 2);
1900	assert.equal(providerThread?.status, "active");
1901	assert.deepEqual(providerThread?.pendingBackgroundTasks ?? [], []);
1902	}),
1903	);
1904
1905	it.effect("interrupts a running process-bound effect when it is cancelled", () =>
1906	Effect.gen(function* () {
1907	const outbox = yield* EffectOutbox.EffectOutboxV2;
1908	const commandId = CommandId.make("command:foundation-cancel-running-effect");
1909	const threadId = ThreadId.make("thread:foundation-cancel-running-effect");
1910	const effectId = "effect:foundation-cancel-running-effect";
1911	const started = yield* Deferred.make<void>();
1912	const interrupted = yield* Deferred.make<void>();
1913	yield* outbox.enqueue([
1914	{
1915	id: effectId,
1916	commandId,
1917	threadId,
1918	request: {
1919	type: "provider-turn.start",
1920	runId: RunId.make("run:foundation-cancel-running-effect"),
1921	},
1922	},
1923	]);
1924
1925	const layerExecutor = Layer.succeed(
1926	EffectWorker.OrchestrationEffectExecutorV2,
1927	EffectWorker.OrchestrationEffectExecutorV2.of({
1928	execute: () =>
1929	Deferred.succeed(started, undefined).pipe(
1930	Effect.andThen(Effect.never),
1931	Effect.onInterrupt(() =>
1932	Deferred.succeed(interrupted, undefined).pipe(Effect.ignore),
1933	),
1934	),
1935	}),
1936	);
1937	const layerWorker = EffectWorker.layerWithOptions({
1938	workerId: "cancellation-worker",
1939	}).pipe(
1940	Layer.provide(
1941	Layer.merge(Layer.succeed(EffectOutbox.EffectOutboxV2, outbox), layerExecutor),
1942	),
1943	);
1944
1945	yield* Effect.gen(function* () {
1946	const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
1947	const workerFiber = yield* worker.runOnce.pipe(Effect.forkChild);
1948	yield* Deferred.await(started);
1949	const cancelledEffectIds = yield* outbox.cancelUnsettled({
1950	threadId,
1951	effectTypes: ["provider-turn.start"],
1952	reason: "The owning run was interrupted.",
1953	});
1954	assert.deepEqual(cancelledEffectIds, [effectId]);
1955	yield* outbox.signalCancellations(cancelledEffectIds);
1956	assert.isTrue(yield* Fiber.join(workerFiber));
1957	yield* Deferred.await(interrupted);
1958	}).pipe(Effect.provide(layerWorker));
1959
1960	const cancelled = yield* outbox.get(effectId);
1961	assert.isTrue(Option.isSome(cancelled));
1962	if (Option.isSome(cancelled)) assert.equal(cancelled.value.status, "cancelled");
1963	}),
1964	);
1965
1966	it.effect("treats cancellation between execution and settlement as a normal outcome", () =>
1967	Effect.gen(function* () {
1968	const effectId = "effect:foundation-cancel-before-settlement";
1969	const threadId = ThreadId.make("thread:foundation-cancel-before-settlement");
1970	const commandId = CommandId.make("command:foundation-cancel-before-settlement");
1971	const now = DateTime.formatIso(yield* DateTime.now);
1972	const claimedEffect = {
1973	id: effectId,
1974	commandId,
1975	threadId,
1976	request: { type: "terminal.cleanup" as const },
1977	status: "running" as const,
1978	attemptCount: 1,
1979	availableAt: now,
1980	leaseOwner: "settlement-race-worker",
1981	leaseExpiresAt: now,
1982	createdAt: now,
1983	updatedAt: now,
1984	completedAt: null,
1985	lastError: null,
1986	};
1987	const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({
1988	claimNext: () => Effect.succeed(Option.some(claimedEffect)),
1989	awaitCancellation: () => Effect.never,
1990	clearCancellation: () => Effect.void,
1991	succeed: () => Effect.succeed(false),
1992	get: () =>
1993	Effect.succeed(
1994	Option.some({
1995	...claimedEffect,
1996	status: "cancelled" as const,
1997	leaseOwner: null,
1998	leaseExpiresAt: null,
1999	completedAt: now,
2000	}),
2001	),
2002	});
2003	const layerExecutor = Layer.succeed(
2004	EffectWorker.OrchestrationEffectExecutorV2,
2005	EffectWorker.OrchestrationEffectExecutorV2.of({ execute: () => Effect.void }),
2006	);
2007	const layerWorker = EffectWorker.layerWithOptions({
2008	workerId: "settlement-race-worker",
2009	}).pipe(Layer.provide(Layer.merge(layerOutbox, layerExecutor)));
2010
2011	assert.isTrue(
2012	yield* EffectWorker.OrchestrationEffectWorkerV2.pipe(
2013	Effect.flatMap((worker) => worker.runOnce),
2014	Effect.provide(layerWorker),
2015	),
2016	);
2017	}),
2018	);
2019
2020	it.effect("does not start an effect that was cancelled during claim registration", () =>
2021	Effect.gen(function* () {
2022	const effectId = "effect:foundation-cancelled-during-claim";
2023	const threadId = ThreadId.make("thread:foundation-cancelled-during-claim");
2024	const commandId = CommandId.make("command:foundation-cancelled-during-claim");
2025	const now = DateTime.formatIso(yield* DateTime.now);
2026	const claimedEffect = {
2027	id: effectId,
2028	commandId,
2029	threadId,
2030	request: { type: "terminal.cleanup" as const },
2031	status: "running" as const,
2032	attemptCount: 1,
2033	availableAt: now,
2034	leaseOwner: "claim-cancellation-worker",
2035	leaseExpiresAt: now,
2036	createdAt: now,
2037	updatedAt: now,
2038	completedAt: null,
2039	lastError: null,
2040	};
2041	const executionCount = yield* Ref.make(0);
2042	const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({
2043	claimNext: () => Effect.succeed(Option.some(claimedEffect)),
2044	get: () =>
2045	Effect.succeed(
2046	Option.some({
2047	...claimedEffect,
2048	status: "cancelled" as const,
2049	leaseOwner: null,
2050	leaseExpiresAt: null,
2051	completedAt: now,
2052	}),
2053	),
2054	clearCancellation: () => Effect.void,
2055	awaitCancellation: () => Effect.never,
2056	});
2057	const layerExecutor = Layer.succeed(
2058	EffectWorker.OrchestrationEffectExecutorV2,
2059	EffectWorker.OrchestrationEffectExecutorV2.of({
2060	execute: () => Ref.update(executionCount, (count) => count + 1),
2061	}),
2062	);
2063	const layerWorker = EffectWorker.layerWithOptions({
2064	workerId: "claim-cancellation-worker",
2065	}).pipe(Layer.provide(Layer.merge(layerOutbox, layerExecutor)));
2066
2067	assert.isTrue(
2068	yield* EffectWorker.OrchestrationEffectWorkerV2.pipe(
2069	Effect.flatMap((worker) => worker.runOnce),
2070	Effect.provide(layerWorker),
2071	),
2072	);
2073	assert.equal(yield* Ref.get(executionCount), 0);
2074	}),
2075	);
2076
2077	it.effect("allows only one worker to claim an available effect", () =>
2078	Effect.gen(function* () {
2079	const outbox = yield* EffectOutbox.EffectOutboxV2;
2080	const commandId = CommandId.make("command:foundation-exclusive-claim");
2081	yield* outbox.enqueue([
2082	{
2083	id: "effect:foundation-exclusive-claim",
2084	commandId,
2085	threadId: ThreadId.make("thread:foundation-exclusive-claim"),
2086	request: {
2087	type: "provider-turn.start",
2088	runId: RunId.make("run:foundation-exclusive-claim"),
2089	},
2090	},
2091	]);
2092
2093	const claims = yield* Effect.all(
2094	[
2095	outbox.claimNext({ workerId: "worker-a", leaseDurationMs: 30_000 }),
2096	outbox.claimNext({ workerId: "worker-b", leaseDurationMs: 30_000 }),
2097	],
2098	{ concurrency: "unbounded" },
2099	);
2100	assert.equal(claims.filter(Option.isSome).length, 1);
2101	assert.equal(claims.filter(Option.isNone).length, 1);
2102	const claimedByA = claims[0];
2103	const claimedByB = claims[1];
2104	if (Option.isSome(claimedByA)) {
2105	yield* outbox.succeed({ effectId: claimedByA.value.id, workerId: "worker-a" });
2106	}
2107	if (Option.isSome(claimedByB)) {
2108	yield* outbox.succeed({ effectId: claimedByB.value.id, workerId: "worker-b" });
2109	}
2110	}),
2111	);
2112
2113	it.effect("wakes a settle waiter when its effect finishes", () =>
2114	Effect.gen(function* () {
2115	const outbox = yield* EffectOutbox.EffectOutboxV2;
2116	const effectId = "effect:foundation-await-settled";
2117	yield* outbox.enqueue([
2118	{
2119	id: effectId,
2120	commandId: CommandId.make("command:foundation-await-settled"),
2121	threadId: ThreadId.make("thread:foundation-await-settled"),
2122	request: {
2123	type: "provider-runtime.continue",
2124	sourceRunId: RunId.make("run:foundation-await-settled"),
2125	},
2126	},
2127	]);
2128	const waiter = yield* outbox.awaitSettled(effectId).pipe(Effect.forkChild);
2129	const claimed = yield* outbox.claimNext({
2130	workerId: "settle-worker",
2131	leaseDurationMs: 30_000,
2132	});
2133	assert.isTrue(Option.isSome(claimed));
2134	yield* Effect.yieldNow;
2135	assert.isUndefined(waiter.pollUnsafe());
2136
2137	yield* outbox.succeed({ effectId, workerId: "settle-worker" });
2138	yield* Fiber.join(waiter);
2139	// A settled effect does not wait.
2140	yield* outbox.awaitSettled(effectId);
2141	}),
2142	);
2143
2144	it.effect("runs title generation beside critical work while serializing each lane", () =>
2145	Effect.gen(function* () {
2146	const outbox = yield* EffectOutbox.EffectOutboxV2;
2147	const commandId = CommandId.make("command:foundation-title-effect-lane");
2148	const threadId = ThreadId.make("thread:foundation-title-effect-lane");
2149	const titleEffectId = "effect:foundation-title-effect-lane:a-title";
2150	const providerEffectId = "effect:foundation-title-effect-lane:b-provider";
2151	const nextTitleEffectId = "effect:foundation-title-effect-lane:c-title";
2152	const nextCriticalEffectId = "effect:foundation-title-effect-lane:d-critical";
2153	yield* outbox.enqueue([
2154	{
2155	id: titleEffectId,
2156	commandId,
2157	threadId,
2158	request: {
2159	type: "thread-title.generate",
2160	kind: {
2161	type: "initial",
2162	messageId: MessageId.make("message:foundation-title-effect-lane"),
2163	},
2164	},
2165	},
2166	{
2167	id: providerEffectId,
2168	commandId,
2169	threadId,
2170	request: {
2171	type: "provider-turn.start",
2172	runId: RunId.make("run:foundation-title-effect-lane"),
2173	},
2174	},
2175	{
2176	id: nextTitleEffectId,
2177	commandId,
2178	threadId,
2179	request: { type: "thread-title.generate", kind: { type: "regenerate" } },
2180	},
2181	{
2182	id: nextCriticalEffectId,
2183	commandId,
2184	threadId,
2185	request: { type: "terminal.cleanup" },
2186	},
2187	]);
2188
2189	const title = yield* outbox.claimNext({
2190	workerId: "title-lane-worker",
2191	leaseDurationMs: 30_000,
2192	});
2193	assert.isTrue(Option.isSome(title));
2194	if (Option.isNone(title)) return;
2195	assert.equal(title.value.id, titleEffectId);
2196
2197	const provider = yield* outbox.claimNext({
2198	workerId: "critical-lane-worker",
2199	leaseDurationMs: 30_000,
2200	});
2201	assert.isTrue(Option.isSome(provider));
2202	if (Option.isNone(provider)) return;
2203	assert.equal(provider.value.id, providerEffectId);
2204
2205	assert.isTrue(
2206	Option.isNone(
2207	yield* outbox.claimNext({
2208	workerId: "blocked-lanes-worker",
2209	leaseDurationMs: 30_000,
2210	}),
2211	),
2212	);
2213
2214	yield* outbox.succeed({
2215	effectId: provider.value.id,
2216	workerId: "critical-lane-worker",
2217	});
2218	const nextCritical = yield* outbox.claimNext({
2219	workerId: "critical-lane-worker",
2220	leaseDurationMs: 30_000,
2221	});
2222	assert.isTrue(Option.isSome(nextCritical));
2223	if (Option.isNone(nextCritical)) return;
2224	assert.equal(nextCritical.value.id, nextCriticalEffectId);
2225	yield* outbox.succeed({
2226	effectId: nextCritical.value.id,
2227	workerId: "critical-lane-worker",
2228	});
2229
2230	yield* outbox.succeed({
2231	effectId: title.value.id,
2232	workerId: "title-lane-worker",
2233	});
2234	const nextTitle = yield* outbox.claimNext({
2235	workerId: "title-lane-worker",
2236	leaseDurationMs: 30_000,
2237	});
2238	assert.isTrue(Option.isSome(nextTitle));
2239	if (Option.isSome(nextTitle)) {
2240	assert.equal(nextTitle.value.id, nextTitleEffectId);
2241	yield* outbox.succeed({
2242	effectId: nextTitle.value.id,
2243	workerId: "title-lane-worker",
2244	});
2245	}
2246	}),
2247	);
2248
2249	it.effect("ignores deadlines blocked by a running effect on the same thread", () =>
2250	Effect.gen(function* () {
2251	const outbox = yield* EffectOutbox.EffectOutboxV2;
2252	const now = yield* DateTime.now;
2253	const future = DateTime.add(now, { seconds: 10 });
2254	const commandId = CommandId.make("command:foundation-next-claimable");
2255	const blockedThreadId = ThreadId.make("thread:foundation-next-claimable:blocked");
2256	yield* outbox.enqueue([
2257	{
2258	id: "effect:foundation-next-claimable:a1",
2259	commandId,
2260	threadId: blockedThreadId,
2261	request: { type: "terminal.cleanup" },
2262	},
2263	{
2264	id: "effect:foundation-next-claimable:a2",
2265	commandId,
2266	threadId: blockedThreadId,
2267	request: { type: "terminal.cleanup" },
2268	},
2269	{
2270	id: "effect:foundation-next-claimable:b1",
2271	commandId,
2272	threadId: ThreadId.make("thread:foundation-next-claimable:future"),
2273	request: { type: "terminal.cleanup" },
2274	availableAt: future,
2275	},
2276	]);
2277
2278	const claimed = yield* outbox.claimNext({
2279	workerId: "next-claimable-worker",
2280	leaseDurationMs: 30_000,
2281	});
2282	assert.isTrue(Option.isSome(claimed));
2283	if (Option.isNone(claimed)) return;
2284	assert.equal(claimed.value.id, "effect:foundation-next-claimable:a1");
2285
2286	const whileBlocked = yield* outbox.nextClaimableAt;
2287	assert.isTrue(Option.isSome(whileBlocked));
2288	if (Option.isSome(whileBlocked)) {
2289	assert.equal(DateTime.toEpochMillis(whileBlocked.value), DateTime.toEpochMillis(future));
2290	}
2291
2292	yield* outbox.succeed({
2293	effectId: claimed.value.id,
2294	workerId: "next-claimable-worker",
2295	});
2296	const afterCompletion = yield* outbox.nextClaimableAt;
2297	assert.isTrue(Option.isSome(afterCompletion));
2298	if (Option.isSome(afterCompletion)) {
2299	assert.isAtMost(DateTime.toEpochMillis(afterCompletion.value), DateTime.toEpochMillis(now));
2300	}
2301
2302	const unblocked = yield* outbox.claimNext({
2303	workerId: "next-claimable-worker",
2304	leaseDurationMs: 30_000,
2305	});
2306	assert.isTrue(Option.isSome(unblocked));
2307	if (Option.isSome(unblocked)) {
2308	assert.equal(unblocked.value.id, "effect:foundation-next-claimable:a2");
2309	yield* outbox.succeed({
2310	effectId: unblocked.value.id,
2311	workerId: "next-claimable-worker",
2312	});
2313	}
2314	yield* outbox.cancelUnsettled({
2315	threadId: ThreadId.make("thread:foundation-next-claimable:future"),
2316	effectTypes: ["terminal.cleanup"],
2317	reason: "Test cleanup.",
2318	});
2319	}),
2320	);
2321
2322	it.effect("prunes settled effects past retention in batches and keeps the rest", () =>
2323	Effect.gen(function* () {
2324	const outbox = yield* EffectOutbox.EffectOutboxV2;
2325	const sql = yield* SqlClient.SqlClient;
2326	const now = yield* DateTime.now;
2327	const insert = (
2328	prefix: string,
2329	count: number,
2330	status: EffectOutbox.OrchestrationEffectStatusV2,
2331	completedAgo: Duration.Duration | null,
2332	) => {
2333	const completedAt =
2334	completedAgo === null
2335	? null
2336	: DateTime.formatIso(DateTime.subtractDuration(now, completedAgo));
2337	const createdAt = DateTime.formatIso(now);
2338	return sql`
2339	WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${count})
2340	INSERT INTO orchestration_v2_effect_outbox (
2341	effect_id, command_id, thread_id, effect_type, payload_json, status,
2342	available_at, created_at, updated_at, completed_at
2343	)
2344	SELECT ${prefix} || i, 'command:prune', 'thread:prune', 'terminal.cleanup',
2345	'{"type":"terminal.cleanup"}', ${status}, ${createdAt}, ${createdAt}, ${createdAt},
2346	${completedAt}
2347	FROM n
2348	`;
2349	};
2350	const old = Duration.sum(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(1));
2351	const recent = Duration.subtract(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(1));
2352	// More expired rows than one delete batch.
2353	yield* insert("succeeded-old:", 1_201, "succeeded", old);
2354	yield* insert("cancelled-old:", 2, "cancelled", old);
2355	yield* insert("succeeded-recent:", 1, "succeeded", recent);
2356	yield* insert("failed-old:", 1, "failed", old);
2357	yield* insert("pending:", 1, "pending", null);
2358	yield* insert("running:", 1, "running", null);
2359
2360	assert.equal(yield* outbox.pruneSettled, 1_203);
2361
2362	const remaining = yield* sql<{ readonly effect_id: string }>`
2363	SELECT effect_id FROM orchestration_v2_effect_outbox ORDER BY effect_id
2364	`;
2365	assert.deepEqual(
2366	remaining.map((row) => row.effect_id),
2367	["failed-old:1", "pending:1", "running:1", "succeeded-recent:1"],
2368	);
2369	assert.equal(yield* outbox.pruneSettled, 0);
2370	}).pipe(Effect.provide(layerIsolatedOutbox)),
2371	);
2372
2373	it.effect("prunes settled effects hourly from the layer-owned worker", () =>
2374	Effect.gen(function* () {
2375	const outbox = yield* EffectOutbox.EffectOutboxV2;
2376	const commandId = CommandId.make("command:foundation-prune-worker");
2377	const threadId = ThreadId.make("thread:foundation-prune-worker");
2378	const workerId = "prune-worker";
2379	const request = { type: "terminal.cleanup" } as const;
2380	yield* outbox.enqueue([{ id: "effect:prune-worker:done", commandId, threadId, request }]);
2381	yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2382	assert.isTrue(yield* outbox.succeed({ effectId: "effect:prune-worker:done", workerId }));
2383	yield* outbox.enqueue([{ id: "effect:prune-worker:pending", commandId, threadId, request }]);
2384	// Observe each run so the test waits for it instead of racing the clock.
2385	const runs = yield* Queue.unbounded<number>();
2386	const observed = EffectOutbox.EffectOutboxV2.of({
2387	...outbox,
2388	pruneSettled: outbox.pruneSettled.pipe(Effect.tap((pruned) => Queue.offer(runs, pruned))),
2389	});
2390	const ids = Effect.map(outbox.listByCommandId(commandId), (rows) =>
2391	rows.map((row) => row.id).toSorted(),
2392	);
2393
2394	yield* TestClock.adjust(
2395	Duration.subtract(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(30)),
2396	);
2397	yield* Layer.build(
2398	EffectOutbox.layerPruneWorker.pipe(
2399	Layer.provide(Layer.succeed(EffectOutbox.EffectOutboxV2, observed)),
2400	),
2401	);
2402	assert.equal(yield* Queue.take(runs), 0);
2403	assert.deepEqual(yield* ids, ["effect:prune-worker:done", "effect:prune-worker:pending"]);
2404
2405	yield* TestClock.adjust("1 hour");
2406	assert.equal(yield* Queue.take(runs), 1);
2407	assert.deepEqual(yield* ids, ["effect:prune-worker:pending"]);
2408	}).pipe(Effect.scoped, Effect.provide(layerIsolatedOutbox)),
2409	);
2410
2411	it.effect("does not emit a SQL span for an empty safety claim", () =>
2412	Effect.gen(function* () {
2413	const outbox = yield* EffectOutbox.EffectOutboxV2;
2414	const statements: Array<unknown> = [];
2415	const tracer = Tracer.make({
2416	span: (options) => {
2417	const span = new Tracer.NativeSpan(options);
2418	const end = span.end.bind(span);
2419	span.end = (endTime, exit) => {
2420	end(endTime, exit);
2421	const query = span.attributes.get("db.query.text");
2422	if (query !== undefined) statements.push(query);
2423	};
2424	return span;
2425	},
2426	});
2427
2428	const claim = yield* outbox
2429	.claimNext({ workerId: "idle-safety-worker", leaseDurationMs: 30_000 })
2430	.pipe(Effect.withTracer(tracer));
2431
2432	assert.isTrue(Option.isNone(claim));
2433	assert.deepEqual(statements, []);
2434	}).pipe(Effect.provide(Layer.fresh(layerEffectOutboxProvided))),
2435	);
2436
2437	it.effect("wakes claimers when cancellation unblocks same-thread work", () =>
2438	Effect.gen(function* () {
2439	const outbox = yield* EffectOutbox.EffectOutboxV2;
2440	const commandId = CommandId.make("command:foundation-cancellation-wakeup");
2441	const threadId = ThreadId.make("thread:foundation-cancellation-wakeup");
2442	yield* outbox.enqueue([
2443	{
2444	id: "effect:foundation-cancellation-wakeup:a-running",
2445	commandId,
2446	threadId,
2447	request: {
2448	type: "provider-turn.start",
2449	runId: RunId.make("run:foundation-cancellation-wakeup"),
2450	},
2451	},
2452	{
2453	id: "effect:foundation-cancellation-wakeup:b-pending",
2454	commandId,
2455	threadId,
2456	request: { type: "terminal.cleanup" },
2457	},
2458	]);
2459
2460	const running = yield* outbox.claimNext({
2461	workerId: "cancellation-wakeup-worker",
2462	leaseDurationMs: 30_000,
2463	});
2464	assert.isTrue(Option.isSome(running));
2465	if (Option.isNone(running)) return;
2466	assert.equal(running.value.request.type, "provider-turn.start");
2467
2468	const cancelledEffectIds = yield* outbox.cancelUnsettled({
2469	threadId,
2470	effectTypes: ["provider-turn.start"],
2471	reason: "Test cancellation wakeup.",
2472	});
2473	yield* outbox.signalCancellations(cancelledEffectIds);
2474
2475	const wake = yield* outbox.awaitAvailable.pipe(Effect.forkChild);
2476	yield* Effect.yieldNow;
2477	assert.isDefined(wake.pollUnsafe());
2478
2479	const unblocked = yield* outbox.claimNext({
2480	workerId: "cancellation-wakeup-worker",
2481	leaseDurationMs: 30_000,
2482	});
2483	assert.isTrue(Option.isSome(unblocked));
2484	if (Option.isSome(unblocked)) {
2485	assert.equal(unblocked.value.request.type, "terminal.cleanup");
2486	yield* outbox.succeed({
2487	effectId: unblocked.value.id,
2488	workerId: "cancellation-wakeup-worker",
2489	});
2490	}
2491	}).pipe(Effect.provide(Layer.fresh(layerEffectOutboxProvided))),
2492	);
2493
2494	it.effect("keeps later thread effects behind an earlier effect waiting to retry", () =>
2495	Effect.gen(function* () {
2496	const outbox = yield* EffectOutbox.EffectOutboxV2;
2497	const workerId = "retry-order-worker";
2498	const commandId = CommandId.make("command:foundation-retry-order");
2499	const threadId = ThreadId.make("thread:foundation-retry-order");
2500	yield* outbox.enqueue([
2501	{
2502	id: "effect:foundation-retry-order:z-rollback",
2503	commandId,
2504	threadId,
2505	request: {
2506	type: "provider-thread.rollback",
2507	providerThreadId: ProviderThreadId.make("provider-thread:foundation-retry-order"),
2508	checkpointId: CheckpointId.make("checkpoint:foundation-retry-order"),
2509	scopeId: CheckpointScopeId.make("scope:foundation-retry-order"),
2510	},
2511	},
2512	]);
2513	const rollback = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2514	assert.isTrue(Option.isSome(rollback));
2515	if (Option.isNone(rollback)) return;
2516	yield* outbox.retry({
2517	effectId: rollback.value.id,
2518	workerId,
2519	error: "rollback failed once",
2520	delayMs: 60_000,
2521	});
2522
2523	// A turn the user starts during the rollback's backoff must not run first,
2524	// even when its timestamp ties and its id sorts first.
2525	yield* outbox.enqueue([
2526	{
2527	id: "effect:foundation-retry-order:a-start",
2528	commandId: CommandId.make("command:foundation-retry-order:start"),
2529	threadId,
2530	request: { type: "provider-turn.start", runId: RunId.make("run:foundation-retry-order") },
2531	},
2532	{
2533	id: "effect:foundation-retry-order:b-title",
2534	commandId: CommandId.make("command:foundation-retry-order:title"),
2535	threadId,
2536	request: { type: "thread-title.generate", kind: { type: "regenerate" } },
2537	},
2538	]);
2539	const title = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2540	assert.equal(Option.getOrUndefined(title)?.id, "effect:foundation-retry-order:b-title");
2541	const blocked = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2542	assert.isTrue(Option.isNone(blocked));
2543	const nextClaimable = yield* outbox.nextClaimableAt;
2544	assert.isTrue(Option.isSome(nextClaimable));
2545	if (Option.isSome(nextClaimable)) {
2546	assert.equal(
2547	DateTime.formatIso(nextClaimable.value),
2548	(yield* outbox.get(rollback.value.id)).pipe(Option.getOrThrow).availableAt,
2549	);
2550	}
2551
2552	yield* outbox.cancelUnsettled({
2553	threadId,
2554	effectTypes: ["provider-thread.rollback"],
2555	reason: "Test cleanup.",
2556	});
2557	const unblocked = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2558	assert.equal(Option.getOrUndefined(unblocked)?.id, "effect:foundation-retry-order:a-start");
2559	yield* outbox.succeed({ effectId: "effect:foundation-retry-order:a-start", workerId });
2560	yield* outbox.succeed({ effectId: "effect:foundation-retry-order:b-title", workerId });
2561	}).pipe(Effect.provide(Layer.fresh(layerEffectOutboxProvided))),
2562	);
2563
2564	it.effect("executes a retry at its durable deadline instead of the liveness interval", () =>
2565	Effect.gen(function* () {
2566	const outbox = yield* EffectOutbox.EffectOutboxV2;
2567	const effectId = "effect:foundation-durable-retry-deadline";
2568	const commandId = CommandId.make("command:foundation-durable-retry-deadline");
2569	const threadId = ThreadId.make("thread:foundation-durable-retry-deadline");
2570	const executions = yield* Ref.make(0);
2571	const completed = yield* Deferred.make<void>();
2572	const layerExecutor = Layer.succeed(
2573	EffectWorker.OrchestrationEffectExecutorV2,
2574	EffectWorker.OrchestrationEffectExecutorV2.of({
2575	execute: () =>
2576	Effect.gen(function* () {
2577	const attempt = yield* Ref.updateAndGet(executions, (count) => count + 1);
2578	if (attempt === 1) {
2579	return yield* new EffectWorker.OrchestrationEffectExecutionError({
2580	effectId,
2581	effectType: "terminal.cleanup",
2582	cause: "simulated retry",
2583	});
2584	}
2585	yield* Deferred.succeed(completed, undefined);
2586	}),
2587	}),
2588	);
2589	const layerWorker = EffectWorker.layerWithOptions({
2590	workerId: "durable-retry-deadline-worker",
2591	}).pipe(
2592	Layer.provide(
2593	Layer.merge(Layer.succeed(EffectOutbox.EffectOutboxV2, outbox), layerExecutor),
2594	),
2595	);
2596
2597	yield* Effect.gen(function* () {
2598	yield* EffectWorker.runDaemonWithOptions({
2599	concurrency: 1,
2600	livenessPollIntervalMs: 30_000,
2601	}).pipe(Effect.forkScoped);
2602	yield* outbox.enqueue([
2603	{
2604	id: effectId,
2605	commandId,
2606	threadId,
2607	request: { type: "terminal.cleanup" },
2608	},
2609	]);
2610	yield* outbox.notifyAvailable();
2611
2612	let retryScheduled = false;
2613	while (!retryScheduled) {
2614	const effect = yield* outbox.get(effectId);
2615	retryScheduled =
2616	Option.isSome(effect) &&
2617	effect.value.status === "pending" &&
2618	effect.value.attemptCount === 1;
2619	if (!retryScheduled) yield* Effect.yieldNow;
2620	}
2621
2622	yield* TestClock.adjust("99 millis");
2623	assert.equal(yield* Ref.get(executions), 1);
2624	yield* TestClock.adjust("1 millis");
2625	yield* Deferred.await(completed);
2626	assert.equal(yield* Ref.get(executions), 2);
2627	}).pipe(Effect.provide(layerWorker), Effect.scoped);
2628	}).pipe(Effect.provide(TestClock.layer())),
2629	);
2630
2631	it.effect("runs distinct threads concurrently while serializing effects within a thread", () =>
2632	Effect.gen(function* () {
2633	const outbox = yield* EffectOutbox.EffectOutboxV2;
2634	const commandId = CommandId.make("command:foundation-concurrent-effects");
2635	const threadA = ThreadId.make("thread:foundation-concurrent-effects:a");
2636	const threadB = ThreadId.make("thread:foundation-concurrent-effects:b");
2637	const effectA1 = "effect:foundation-concurrent-effects:a1";
2638	const effectA2 = "effect:foundation-concurrent-effects:a2";
2639	const effectB1 = "effect:foundation-concurrent-effects:b1";
2640	const startedA1 = yield* Deferred.make<void>();
2641	const startedA2 = yield* Deferred.make<void>();
2642	const startedB1 = yield* Deferred.make<void>();
2643	const releaseA1 = yield* Deferred.make<void>();
2644	const releaseA2 = yield* Deferred.make<void>();
2645	const releaseB1 = yield* Deferred.make<void>();
2646	const gates = new Map([
2647	[effectA1, { started: startedA1, release: releaseA1 }],
2648	[effectA2, { started: startedA2, release: releaseA2 }],
2649	[effectB1, { started: startedB1, release: releaseB1 }],
2650	]);
2651	const layerExecutor = Layer.succeed(
2652	EffectWorker.OrchestrationEffectExecutorV2,
2653	EffectWorker.OrchestrationEffectExecutorV2.of({
2654	execute: (effect) => {
2655	const gate = gates.get(effect.id);
2656	if (gate === undefined) return Effect.die(`Missing gate for ${effect.id}`);
2657	return Deferred.succeed(gate.started, undefined).pipe(
2658	Effect.andThen(Deferred.await(gate.release)),
2659	);
2660	},
2661	}),
2662	);
2663	const layerWorker = EffectWorker.layerWithOptions({
2664	workerId: "concurrency-worker",
2665	}).pipe(
2666	Layer.provide(
2667	Layer.merge(Layer.succeed(EffectOutbox.EffectOutboxV2, outbox), layerExecutor),
2668	),
2669	);
2670
2671	yield* Effect.gen(function* () {
2672	yield* EffectWorker.runDaemonWithOptions({ concurrency: 2 }).pipe(Effect.forkScoped);
2673	// Let both slots reach the idle wait before work becomes available.
2674	yield* Effect.yieldNow;
2675	yield* Effect.yieldNow;
2676	yield* outbox.enqueue([
2677	{
2678	id: effectA1,
2679	commandId,
2680	threadId: threadA,
2681	request: {
2682	type: "provider-turn.start",
2683	runId: RunId.make("run:foundation-concurrent-effects:a1"),
2684	},
2685	},
2686	{
2687	id: effectA2,
2688	commandId,
2689	threadId: threadA,
2690	request: {
2691	type: "provider-turn.start",
2692	runId: RunId.make("run:foundation-concurrent-effects:a2"),
2693	},
2694	},
2695	{
2696	id: effectB1,
2697	commandId,
2698	threadId: threadB,
2699	request: {
2700	type: "provider-turn.start",
2701	runId: RunId.make("run:foundation-concurrent-effects:b1"),
2702	},
2703	},
2704	]);
2705	yield* outbox.notifyAvailable(3);
2706
2707	yield* Effect.all([Deferred.await(startedA1), Deferred.await(startedB1)]);
2708	assert.isFalse(yield* Deferred.isDone(startedA2));
2709
2710	yield* Deferred.succeed(releaseA1, undefined);
2711	yield* Deferred.succeed(releaseB1, undefined);
2712	yield* Deferred.await(startedA2);
2713	yield* Deferred.succeed(releaseA2, undefined);
2714	let settled = false;
2715	while (!settled) {
2716	settled = (yield* outbox.listByCommandId(commandId)).every(
2717	(effect) => effect.status === "succeeded",
2718	);
2719	if (!settled) yield* Effect.yieldNow;
2720	}
2721	}).pipe(Effect.provide(layerWorker), Effect.scoped);
2722	}),
2723	);
2724
2725	it.effect("does not reclaim a running effect after its process-local lease expires", () =>
2726	Effect.gen(function* () {
2727	const outbox = yield* EffectOutbox.EffectOutboxV2;
2728	const sql = yield* SqlClient.SqlClient;
2729	const commandId = CommandId.make("command:foundation-no-live-reclaim");
2730	const threadId = ThreadId.make("thread:foundation-no-live-reclaim");
2731	const firstEffectId = "effect:foundation-no-live-reclaim:first";
2732	const secondEffectId = "effect:foundation-no-live-reclaim:second";
2733	const firstStarted = yield* Deferred.make<void>();
2734	const releaseFirst = yield* Deferred.make<void>();
2735	const executions = yield* Ref.make<ReadonlyArray<string>>([]);
2736	yield* outbox.enqueue([
2737	{
2738	id: firstEffectId,
2739	commandId,
2740	threadId,
2741	request: { type: "terminal.cleanup" },
2742	},
2743	{
2744	id: secondEffectId,
2745	commandId,
2746	threadId,
2747	request: { type: "terminal.cleanup" },
2748	},
2749	]);
2750
2751	const layerExecutor = Layer.succeed(
2752	EffectWorker.OrchestrationEffectExecutorV2,
2753	EffectWorker.OrchestrationEffectExecutorV2.of({
2754	execute: (effect) =>
2755	Ref.update(executions, (current) => [...current, effect.id]).pipe(
2756	Effect.andThen(
2757	effect.id === firstEffectId
2758	? Deferred.succeed(firstStarted, undefined).pipe(
2759	Effect.andThen(Deferred.await(releaseFirst)),
2760	)
2761	: Effect.void,
2762	),
2763	),
2764	}),
2765	);
2766	const layerWorker = EffectWorker.layerWithOptions({
2767	workerId: "no-live-reclaim-worker",
2768	leaseDurationMs: 1,
2769	}).pipe(
2770	Layer.provide(
2771	Layer.merge(Layer.succeed(EffectOutbox.EffectOutboxV2, outbox), layerExecutor),
2772	),
2773	);
2774
2775	yield* Effect.gen(function* () {
2776	const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
2777	const firstFiber = yield* worker.runOnce.pipe(Effect.forkChild);
2778	yield* Deferred.await(firstStarted);
2779	yield* sql`
2780	UPDATE orchestration_v2_effect_outbox
2781	SET lease_expires_at = '1970-01-01T00:00:00.000Z'
2782	WHERE effect_id = ${firstEffectId}
2783	`;
2784
2785	assert.isFalse(yield* worker.runOnce);
2786	assert.deepEqual(yield* Ref.get(executions), [firstEffectId]);
2787
2788	yield* Deferred.succeed(releaseFirst, undefined);
2789	assert.isTrue(yield* Fiber.join(firstFiber));
2790	assert.isTrue(yield* worker.runOnce);
2791	assert.deepEqual(yield* Ref.get(executions), [firstEffectId, secondEffectId]);
2792	}).pipe(Effect.provide(layerWorker));
2793	}),
2794	);
2795
2796	it.effect(
2797	"persists shutdown continuation intent through the real event sink without domain events",
2798	() =>
2799	Effect.gen(function* () {
2800	const outbox = yield* EffectOutbox.EffectOutboxV2;
2801	const now = yield* DateTime.now;
2802	const threadId = ThreadId.make("thread:shutdown-prepare");
2803	const runId = RunId.make("run:shutdown-prepare");
2804	const providerThreadId = ProviderThreadId.make("provider-thread:shutdown-prepare");
2805	const sessionId = ProviderSessionId.make("session:shutdown-prepare");
2806	const attemptId = RunAttemptId.make("attempt:shutdown-prepare");
2807	const projection = {
2808	thread: makeThread(threadId, now),
2809	runs: [
2810	{
2811	id: runId,
2812	ordinal: 1,
2813	status: "running",
2814	providerInstanceId,
2815	providerThreadId,
2816	activeAttemptId: attemptId,
2817	},
2818	],
2819	providerThreads: [
2820	{
2821	id: providerThreadId,
2822	appThreadId: threadId,
2823	ownerNodeId: null,
2824	driver: "codex",
2825	providerInstanceId,
2826	providerSessionId: sessionId,
2827	status: "active",
2828	nativeThreadRef: {
2829	driver: "codex",
2830	nativeId: "saved-native-thread",
2831	strength: "strong",
2832	},
2833	},
2834	],
2835	providerSessions: [
2836	{ id: sessionId, driver: "codex", providerInstanceId, status: "running" },
2837	],
2838	providerTurns: [{ providerThreadId, runAttemptId: attemptId, status: "running" }],
2839	runtimeRequests: [],
2840	turnItems: [],
2841	} as unknown as OrchestrationV2ThreadProjection;
2842	const recovery = yield* ProviderRuntimeRecovery.make.pipe(
2843	Effect.provide(
2844	Layer.mergeAll(
2845	ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
2846	Layer.mock(ProjectionStore.ProjectionStoreV2)({
2847	getRecoveryThreadIds: () => Effect.succeed([threadId]),
2848	getRuntimeRecoveryProjection: () => Effect.succeed(projection),
2849	}),
2850	Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
2851	),
2852	),
2853	);
2854	yield* recovery.prepareForShutdown;
2855	yield* recovery.prepareForShutdown;
2856	const effects = yield* outbox.listByCommandId(
2857	CommandId.make(`command:restart-prepare:${runId}`),
2858	);
2859	assert.lengthOf(effects, 1);
2860	assert.equal(effects[0]?.status, "pending");
2861	assert.deepEqual(effects[0]?.request, {
2862	type: "provider-runtime.continue",
2863	sourceRunId: runId,
2864	});
2865	// This fixture shares the database; settle its intent before later claim tests.
2866	yield* outbox.cancelUnsettled({
2867	threadId,
2868	effectTypes: ["provider-runtime.continue"],
2869	reason: "fixture complete",
2870	});
2871	}),
2872	);
2873
2874	it.effect("leaves restart continuation pending until normal worker claims are enabled", () =>
2875	Effect.gen(function* () {
2876	const outbox = yield* EffectOutbox.EffectOutboxV2;
2877	const threadId = ThreadId.make("thread:activation-restart");
2878	const commandId = CommandId.make("command:activation-restart");
2879	yield* outbox.enqueue([
2880	{
2881	id: "effect:activation-a-restart",
2882	commandId,
2883	threadId,
2884	request: { type: "provider-runtime.continue", sourceRunId: RunId.make("run:activation") },
2885	},
2886	{
2887	id: "effect:activation-b-cleanup",
2888	commandId,
2889	threadId,
2890	request: { type: "terminal.cleanup" },
2891	},
2892	]);
2893	const cleanup = yield* outbox.claimNext({
2894	workerId: "recovery",
2895	leaseDurationMs: 30_000,
2896	excludeRestartContinuations: true,
2897	});
2898	assert.isTrue(Option.isSome(cleanup));
2899	if (Option.isSome(cleanup)) {
2900	assert.equal(cleanup.value.request.type, "terminal.cleanup");
2901	yield* outbox.succeed({ effectId: cleanup.value.id, workerId: "recovery" });
2902	}
2903	assert.isTrue(
2904	Option.isNone(
2905	yield* outbox.claimNext({
2906	workerId: "recovery",
2907	leaseDurationMs: 30_000,
2908	excludeRestartContinuations: true,
2909	}),
2910	),
2911	);
2912	const pending = yield* outbox.get("effect:activation-a-restart");
2913	assert.isTrue(Option.isSome(pending));
2914	if (Option.isSome(pending)) assert.equal(pending.value.status, "pending");
2915	const resumed = yield* outbox.claimNext({ workerId: "activated", leaseDurationMs: 30_000 });
2916	assert.isTrue(Option.isSome(resumed));
2917	if (Option.isSome(resumed)) {
2918	assert.equal(resumed.value.request.type, "provider-runtime.continue");
2919	yield* outbox.succeed({ effectId: resumed.value.id, workerId: "activated" });
2920	}
2921	}),
2922	);
2923
2924	it.effect.each([
2925	{ type: "terminal.cleanup" },
2926	{ type: "provider-runtime.continue", sourceRunId: RunId.make("run:restart-replay") },
2927	] as const)(
2928	"retires live provider effects and requeues $type after process loss",
2929	(replayRequest) =>
2930	Effect.gen(function* () {
2931	const outbox = yield* EffectOutbox.EffectOutboxV2;
2932	const commandId = CommandId.make(
2933	`command:foundation-reclaim-running:${replayRequest.type}`,
2934	);
2935	yield* outbox.enqueue([
2936	{
2937	id: `effect:a-foundation-cancel-provider-turn:${replayRequest.type}`,
2938	commandId,
2939	threadId: ThreadId.make(`thread:foundation-reclaim-running:${replayRequest.type}`),
2940	request: {
2941	type: "provider-turn.start",
2942	runId: RunId.make(`run:foundation-reclaim-running:${replayRequest.type}`),
2943	},
2944	},
2945	{
2946	id: `effect:b-foundation-requeue-cleanup:${replayRequest.type}`,
2947	commandId,
2948	threadId: ThreadId.make(`thread:foundation-reclaim-cleanup:${replayRequest.type}`),
2949	request: replayRequest,
2950	},
2951	]);
2952	assert.isTrue(
2953	Option.isSome(
2954	yield* outbox.claimNext({ workerId: "crashed-worker", leaseDurationMs: 30_000 }),
2955	),
2956	);
2957	assert.isTrue(
2958	Option.isSome(
2959	yield* outbox.claimNext({ workerId: "crashed-worker", leaseDurationMs: 30_000 }),
2960	),
2961	);
2962	assert.deepEqual(yield* outbox.reconcileAfterProcessLoss, {
2963	cancelled: 1,
2964	requeued: 1,
2965	});
2966	const cancelled = yield* outbox.get(
2967	`effect:a-foundation-cancel-provider-turn:${replayRequest.type}`,
2968	);
2969	assert.isTrue(Option.isSome(cancelled));
2970	if (Option.isSome(cancelled)) assert.equal(cancelled.value.status, "cancelled");
2971
2972	const reclaimed = yield* outbox.claimNext({
2973	workerId: "recovery-worker",
2974	leaseDurationMs: 30_000,
2975	});
2976	assert.isTrue(Option.isSome(reclaimed));
2977	if (Option.isSome(reclaimed)) {
2978	assert.equal(reclaimed.value.request.type, replayRequest.type);
2979	assert.equal(reclaimed.value.attemptCount, 2);
2980	yield* outbox.succeed({ effectId: reclaimed.value.id, workerId: "recovery-worker" });
2981	}
2982	}),
2983	);
2984
2985	it.effect("atomically cancels stale runs and their process-bound effects", () =>
2986	Effect.gen(function* () {
2987	const eventSink = yield* EventSink.EventSinkV2;
2988	const outbox = yield* EffectOutbox.EffectOutboxV2;
2989	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2990	const now = yield* DateTime.now;
2991	const threadId = ThreadId.make("thread:foundation-process-loss");
2992	const runId = RunId.make("run:foundation-process-loss");
2993	const commandId = CommandId.make("command:foundation-process-loss");
2994	const thread = makeThread(threadId, now);
2995	yield* eventSink.commitCommand({
2996	commandId,
2997	threadId,
2998	commandType: "foundation.process-loss",
2999	acceptedAt: now,
3000	events: [
3001	threadCreatedEvent({ id: "event:foundation-process-loss:thread", thread, now }),
3002	{
3003	id: EventId.make("event:foundation-process-loss:run"),
3004	type: "run.created",
3005	threadId,
3006	runId,
3007	providerInstanceId,
3008	occurredAt: now,
3009	payload: {
3010	id: runId,
3011	threadId,
3012	ordinal: 1,
3013	providerInstanceId,
3014	modelSelection,
3015	providerThreadId: null,
3016	userMessageId: MessageId.make("message:foundation-process-loss"),
3017	rootNodeId: null,
3018	activeAttemptId: null,
3019	status: "starting",
3020	queuePosition: null,
3021	requestedAt: now,
3022	startedAt: null,
3023	completedAt: null,
3024	checkpointId: null,
3025	contextHandoffId: null,
3026	},
3027	},
3028	],
3029	effects: [
3030	{
3031	id: "effect:foundation-process-loss",
3032	commandId,
3033	threadId,
3034	request: { type: "provider-turn.start", runId },
3035	},
3036	],
3037	});
3038	assert.isTrue(
3039	Option.isSome(
3040	yield* outbox.claimNext({ workerId: "crashed-worker", leaseDurationMs: 30_000 }),
3041	),
3042	);
3043
3044	const recovery = yield* ProviderRuntimeRecovery.make.pipe(
3045	Effect.provide(ServerSettings.layerTest()),
3046	Effect.provideService(
3047	EffectWorker.OrchestrationEffectWorkerV2,
3048	EffectWorker.OrchestrationEffectWorkerV2.of({
3049	awaitWork: Effect.void,
3050	runRecoveryOnce: Effect.succeed(false),
3051	runOnce: Effect.succeed(false),
3052	nextClaimableAt: Effect.succeed(Option.none()),
3053	drain: () => Effect.succeed(0),
3054	}),
3055	),
3056	);
3057	const first = yield* recovery.recover;
3058	assert.equal(first.terminalizedRuns, 1);
3059	assert.equal(first.retiredEffects, 1);
3060	const projection = yield* projectionStore.getThreadProjection(threadId);
3061	assert.equal(projection.runs[0]?.status, "cancelled");
3062	const effect = yield* outbox.get("effect:foundation-process-loss");
3063	assert.isTrue(Option.isSome(effect));
3064	if (Option.isSome(effect)) assert.equal(effect.value.status, "cancelled");
3065
3066	const second = yield* recovery.recover;
3067	assert.equal(second.terminalizedRuns, 0);
3068	assert.equal(second.retiredEffects, 0);
3069	}),
3070	);
3071
3072	it.effect("settles a native subagent's child thread when its provider process is gone", () =>
3073	Effect.gen(function* () {
3074	const eventSink = yield* EventSink.EventSinkV2;
3075	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3076	const now = yield* DateTime.now;
3077	const parentId = ThreadId.make("thread:foundation-native-subagent-parent");
3078	const childId = ThreadId.make("thread:foundation-native-subagent-child");
3079	const runId = RunId.make("run:foundation-native-subagent");
3080	const subagentId = NodeId.make("node:foundation-native-subagent");
3081	const childRootId = NodeId.make("node:foundation-native-subagent-child-root");
3082	const parent = makeThread(parentId, now);
3083	const child: OrchestrationV2AppThread = {
3084	...makeThread(childId, now),
3085	createdBy: "agent",
3086	creationSource: "provider",
3087	lineage: {
3088	parentThreadId: parentId,
3089	relationshipToParent: "subagent",
3090	rootThreadId: parentId,
3091	},
3092	forkedFrom: { type: "node", nodeId: subagentId },
3093	};
3094	const node = (input: {
3095	readonly id: NodeId;
3096	readonly threadId: ThreadId;
3097	readonly runId: RunId | null;
3098	readonly kind: "root_turn" | "subagent";
3099	}) => ({
3100	...input,
3101	parentNodeId: null,
3102	rootNodeId: input.id,
3103	status: "running" as const,
3104	countsForRun: false,
3105	providerThreadId: null,
3106	providerTurnId: null,
3107	nativeItemRef: null,
3108	runtimeRequestId: null,
3109	checkpointScopeId: null,
3110	startedAt: now,
3111	completedAt: null,
3112	});
3113	// The parent run settled while its background subagent kept working,
3114	// Stop ended the subagent's item, then the server died. Recovery cancels
3115	// the parent's subagent entity and node; the child's runless root turn
3116	// lives on another thread and must be settled too.
3117	yield* eventSink.commitCommand({
3118	commandId: CommandId.make("command:foundation-native-subagent"),
3119	threadId: parentId,
3120	commandType: "foundation.native-subagent",
3121	acceptedAt: now,
3122	events: [
3123	threadCreatedEvent({
3124	id: "event:foundation-native-subagent:parent",
3125	thread: parent,
3126	now,
3127	}),
3128	threadCreatedEvent({ id: "event:foundation-native-subagent:child", thread: child, now }),
3129	{
3130	id: EventId.make("event:foundation-native-subagent:run"),
3131	type: "run.created",
3132	threadId: parentId,
3133	runId,
3134	providerInstanceId,
3135	occurredAt: now,
3136	payload: {
3137	id: runId,
3138	threadId: parentId,
3139	ordinal: 1,
3140	providerInstanceId,
3141	modelSelection,
3142	providerThreadId: null,
3143	userMessageId: MessageId.make("message:foundation-native-subagent"),
3144	rootNodeId: null,
3145	activeAttemptId: null,
3146	status: "completed",
3147	queuePosition: null,
3148	requestedAt: now,
3149	startedAt: now,
3150	completedAt: now,
3151	checkpointId: null,
3152	contextHandoffId: null,
3153	},
3154	},
3155	{
3156	id: EventId.make("event:foundation-native-subagent:subagent-node"),
3157	type: "node.updated",
3158	threadId: parentId,
3159	runId,
3160	nodeId: subagentId,
3161	occurredAt: now,
3162	payload: node({ id: subagentId, threadId: parentId, runId, kind: "subagent" }),
3163	},
3164	{
3165	id: EventId.make("event:foundation-native-subagent:child-root"),
3166	type: "node.updated",
3167	threadId: childId,
3168	nodeId: childRootId,
3169	occurredAt: now,
3170	payload: node({ id: childRootId, threadId: childId, runId: null, kind: "root_turn" }),
3171	},
3172	{
3173	// The subagent's live thinking in the child, still streaming.
3174	id: EventId.make("event:foundation-native-subagent:child-progress"),
3175	type: "turn-item.updated",
3176	threadId: childId,
3177	nodeId: childRootId,
3178	occurredAt: now,
3179	payload: {
3180	id: TurnItemId.make("item:foundation-native-subagent:progress"),
3181	threadId: childId,
3182	runId: null,
3183	nodeId: childRootId,
3184	providerThreadId: null,
3185	providerTurnId: null,
3186	nativeItemRef: null,
3187	parentItemId: null,
3188	ordinal: 101,
3189	type: "reasoning",
3190	status: "running",
3191	title: "Thinking",
3192	startedAt: now,
3193	completedAt: null,
3194	updatedAt: now,
3195	text: "Checking the diff.",
3196	streaming: true,
3197	},
3198	},
3199	{
3200	id: EventId.make("event:foundation-native-subagent:subagent"),
3201	type: "subagent.updated",
3202	threadId: parentId,
3203	runId,
3204	nodeId: subagentId,
3205	driver: providerDriver,
3206	providerInstanceId,
3207	occurredAt: now,
3208	payload: {
3209	id: subagentId,
3210	threadId: parentId,
3211	runId,
3212	parentNodeId: subagentId,
3213	origin: "provider_native",
3214	createdBy: "agent",
3215	driver: providerDriver,
3216	providerInstanceId,
3217	providerThreadId: null,
3218	childThreadId: childId,
3219	nativeTaskRef: null,
3220	prompt: "Audit the adapters",
3221	title: null,
3222	model: null,
3223	status: "running",
3224	result: null,
3225	startedAt: now,
3226	completedAt: null,
3227	updatedAt: now,
3228	},
3229	},
3230	{
3231	id: EventId.make("event:foundation-native-subagent:item"),
3232	type: "turn-item.updated",
3233	threadId: parentId,
3234	runId,
3235	nodeId: subagentId,
3236	occurredAt: now,
3237	payload: {
3238	id: TurnItemId.make("item:foundation-native-subagent"),
3239	threadId: parentId,
3240	runId,
3241	nodeId: subagentId,
3242	providerThreadId: null,
3243	providerTurnId: null,
3244	nativeItemRef: null,
3245	parentItemId: null,
3246	ordinal: 1,
3247	type: "subagent",
3248	status: "interrupted",
3249	title: null,
3250	startedAt: now,
3251	completedAt: null,
3252	updatedAt: now,
3253	subagentId,
3254	origin: "provider_native",
3255	driver: providerDriver,
3256	providerInstanceId,
3257	childThreadId: childId,
3258	prompt: "Audit the adapters",
3259	result: null,
3260	},
3261	},
3262	],
3263	effects: [],
3264	});
3265
3266	const recovery = yield* ProviderRuntimeRecovery.make.pipe(
3267	Effect.provide(ServerSettings.layerTest()),
3268	Effect.provideService(
3269	EffectWorker.OrchestrationEffectWorkerV2,
3270	EffectWorker.OrchestrationEffectWorkerV2.of({
3271	awaitWork: Effect.void,
3272	runRecoveryOnce: Effect.succeed(false),
3273	runOnce: Effect.succeed(false),
3274	nextClaimableAt: Effect.succeed(Option.none()),
3275	drain: () => Effect.succeed(0),
3276	}),
3277	),
3278	);
3279	assert.include(yield* projectionStore.getRecoveryThreadIds("runtime"), childId);
3280	yield* recovery.recover;
3281
3282	const parentProjection = yield* projectionStore.getThreadProjection(parentId);
3283	assert.equal(parentProjection.subagents[0]?.status, "cancelled");
3284	assert.equal(
3285	parentProjection.nodes.find((node) => node.id === subagentId)?.status,
3286	"cancelled",
3287	);
3288	const childProjection = yield* projectionStore.getThreadProjection(childId);
3289	const childRoot = childProjection.nodes.find((candidate) => candidate.id === childRootId);
3290	assert.equal(childRoot?.status, "cancelled");
3291	assert.isNotNull(childRoot?.completedAt ?? null);
3292	// Nothing inside the child keeps reading as live work either.
3293	const progress = childProjection.turnItems.find((item) => item.type === "reasoning");
3294	assert.equal(progress?.status, "cancelled");
3295	assert.isFalse(progress?.type === "reasoning" && progress.streaming);
3296	assert.isNotNull(progress?.completedAt ?? null);
3297	assert.notInclude(yield* projectionStore.getRecoveryThreadIds("runtime"), childId);
3298	}),
3299	);
3300
3301	it.effect("allocates collision-free positions beyond 100 items and rebuilds equivalently", () =>
3302	Effect.gen(function* () {
3303	const eventSink = yield* EventSink.EventSinkV2;
3304	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3305	const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
3306	const sql = yield* SqlClient.SqlClient;
3307	const now = yield* DateTime.now;
3308	const threadId = ThreadId.make("thread:foundation-many-items");
3309	const runId = RunId.make("run:foundation-many-items");
3310	const providerThreadId = ProviderThreadId.make("provider-thread:foundation-many-items");
3311	const thread = makeThread(threadId, now);
3312	const providerThreadEvent = {
3313	id: EventId.make("event:foundation-many-items:provider-thread"),
3314	type: "provider-thread.updated" as const,
3315	threadId,
3316	providerInstanceId,
3317	occurredAt: now,
3318	payload: {
3319	id: providerThreadId,
3320	driver: providerDriver,
3321	providerInstanceId,
3322	providerSessionId: null,
3323	appThreadId: threadId,
3324	ownerNodeId: null,
3325	nativeThreadRef: null,
3326	nativeConversationHeadRef: null,
3327	status: "active" as const,
3328	firstRunOrdinal: 1,
3329	lastRunOrdinal: 1,
3330	handoffIds: [],
3331	forkedFrom: null,
3332	createdAt: now,
3333	updatedAt: now,
3334	},
3335	} satisfies OrchestrationV2DomainEvent;
3336	const runEvent = {
3337	id: EventId.make("event:foundation-many-items:run"),
3338	type: "run.created" as const,
3339	threadId,
3340	runId,
3341	providerInstanceId,
3342	occurredAt: now,
3343	payload: {
3344	id: runId,
3345	threadId,
3346	ordinal: 1,
3347	providerInstanceId,
3348	modelSelection,
3349	providerThreadId: null,
3350	userMessageId: MessageId.make("message:foundation-many-items"),
3351	rootNodeId: null,
3352	activeAttemptId: null,
3353	status: "completed" as const,
3354	queuePosition: null,
3355	requestedAt: now,
3356	startedAt: now,
3357	completedAt: now,
3358	checkpointId: null,
3359	contextHandoffId: null,
3360	},
3361	} satisfies OrchestrationV2DomainEvent;
3362	const items = Array.from({ length: 151 }, (_, index) => ({
3363	id: TurnItemId.make(`turn-item:foundation-many-items:${index}`),
3364	threadId,
3365	runId,
3366	nodeId: null,
3367	providerThreadId: null,
3368	providerTurnId: null,
3369	nativeItemRef: null,
3370	parentItemId: null,
3371	ordinal: index % 3,
3372	status: "completed" as const,
3373	title: null,
3374	startedAt: now,
3375	completedAt: now,
3376	updatedAt: now,
3377	type: "dynamic_tool" as const,
3378	toolName: `tool-${index}`,
3379	input: { index },
3380	output: { completed: true },
3381	}));
3382	const itemEvents = items.map(
3383	(item, index) =>
3384	({
3385	id: EventId.make(`event:foundation-many-items:item:${index}`),
3386	type: "turn-item.updated",
3387	threadId,
3388	runId,
3389	providerInstanceId,
3390	occurredAt: now,
3391	payload: item,
3392	}) satisfies OrchestrationV2DomainEvent,
3393	);
3394
3395	yield* eventSink.write({
3396	events: [
3397	threadCreatedEvent({ id: "event:foundation-many-items:thread", thread, now }),
3398	providerThreadEvent,
3399	runEvent,
3400	...itemEvents,
3401	],
3402	});
3403	const beforeUpdate = yield* projectionStore.getThreadProjection(threadId);
3404	assert.equal(beforeUpdate.thread.activeProviderThreadId, providerThreadId);
3405	const ordinals = beforeUpdate.turnItems.map((item) => item.ordinal);
3406	assert.lengthOf(ordinals, 151);
3407	assert.equal(new Set(ordinals).size, 151);
3408	assert.isTrue(ordinals.every((ordinal) => ordinal > 1_000_000));
3409	assert.isTrue(
3410	ordinals.every((ordinal, index) => index === 0 || ordinal > ordinals[index - 1]!),
3411	);
3412
3413	yield* eventSink.write({
3414	events: [
3415	{
3416	id: EventId.make("event:foundation-many-items:update"),
3417	type: "turn-item.updated",
3418	threadId,
3419	runId,
3420	providerInstanceId,
3421	occurredAt: now,
3422	payload: { ...items[0]!, ordinal: 99_999_999, title: "Updated" },
3423	},
3424	],
3425	});
3426	const afterUpdate = yield* projectionStore.getThreadProjection(threadId);
3427	assert.equal(afterUpdate.turnItems[0]?.ordinal, ordinals[0]);
3428	assert.equal((yield* maintenance.verify).valid, true);
3429
3430	yield* sql`
3431	UPDATE orchestration_v2_projection_turn_items
3432	SET payload_json = '{}'
3433	WHERE turn_item_id = ${items[75]!.id}
3434	`;
3435	const broken = yield* maintenance.verify;
3436	assert.isFalse(broken.valid);
3437	assert.deepEqual(broken.unreadableThreadIds, [threadId]);
3438
3439	const rebuilt = yield* maintenance.rebuild;
3440	assert.isTrue(rebuilt.valid);
3441	assert.lengthOf((yield* projectionStore.getThreadProjection(threadId)).turnItems, 151);
3442	}),
3443	);
3444	});
3445
3446	it.live("keeps claiming new work after repeated idle periods", () =>
3447	Effect.gen(function* () {
3448	const outbox = yield* EffectOutbox.EffectOutboxV2;
3449	const completed = new Map<string, Deferred.Deferred<void>>();
3450	const layerExecutor = Layer.succeed(
3451	EffectWorker.OrchestrationEffectExecutorV2,
3452	EffectWorker.OrchestrationEffectExecutorV2.of({
3453	execute: (effect) => {
3454	const completion = completed.get(effect.id);
3455	return completion === undefined
3456	? Effect.die(`Missing completion signal for ${effect.id}`)
3457	: Deferred.succeed(completion, undefined).pipe(Effect.asVoid);
3458	},
3459	}),
3460	);
3461	const layerWorker = EffectWorker.layerWithOptions({
3462	workerId: "idle-wave-worker",
3463	}).pipe(
3464	Layer.provide(Layer.merge(Layer.succeed(EffectOutbox.EffectOutboxV2, outbox), layerExecutor)),
3465	);
3466
3467	yield* Effect.gen(function* () {
3468	yield* EffectWorker.runDaemonWithOptions({ concurrency: 2 }).pipe(Effect.forkScoped);
3469	for (let wave = 1; wave <= 6; wave += 1) {
3470	yield* Effect.sleep("125 millis");
3471	const effectId = `effect:foundation-idle-wave:${wave}`;
3472	const completion = yield* Deferred.make<void>();
3473	completed.set(effectId, completion);
3474	yield* outbox.enqueue([
3475	{
3476	id: effectId,
3477	commandId: CommandId.make(`command:foundation-idle-wave:${wave}`),
3478	threadId: ThreadId.make(`thread:foundation-idle-wave:${wave}`),
3479	request: { type: "terminal.cleanup" },
3480	},
3481	]);
3482	yield* outbox.notifyAvailable();
3483	const observed = yield* Deferred.await(completion).pipe(Effect.timeoutOption("2 seconds"));
3484	assert.isTrue(Option.isSome(observed), `worker stopped before idle wave ${wave}`);
3485	}
3486	}).pipe(Effect.provide(layerWorker), Effect.scoped);
3487	}).pipe(Effect.provide(layerTest)),
3488	);
3489
3490	it.effect("publishes live events in commit order across concurrent writers", () =>
3491	Effect.gen(function* () {
3492	const firstCommitted = yield* Deferred.make<void>();
3493	const releaseFirst = yield* Deferred.make<void>();
3494	// The first writer's post-commit wakeup stands in for any scheduler yield
3495	// between its commit and its publish.
3496	const layerPausingOutbox = Layer.effect(
3497	EffectOutbox.EffectOutboxV2,
3498	Effect.gen(function* () {
3499	const delegate = yield* EffectOutbox.EffectOutboxV2;
3500	return EffectOutbox.EffectOutboxV2.of({
3501	...delegate,
3502	notifyAvailable: (count) =>
3503	Deferred.succeed(firstCommitted, undefined).pipe(
3504	Effect.andThen(Deferred.await(releaseFirst)),
3505	Effect.andThen(delegate.notifyAvailable(count)),
3506	),
3507	});
3508	}),
3509	).pipe(Layer.provide(layerEffectOutboxProvided));
3510	const layerEventSink = EventSink.layerFromStores.pipe(
3511	Layer.provide(
3512	Layer.mergeAll(
3513	layerStoresProvided,
3514	layerPausingOutbox,
3515	layerCommandReceiptStoreProvided,
3516	ProjectStore.layer.pipe(Layer.provide(layerDatabase)),
3517	TurnItemPositionStore.layer.pipe(Layer.provide(layerDatabase)),
3518	),
3519	),
3520	);
3521
3522	yield* Effect.gen(function* () {
3523	const eventSink = yield* EventSink.EventSinkV2;
3524	const now = yield* DateTime.now;
3525	const first = makeThread(ThreadId.make("thread:foundation-publish-order:first"), now);
3526	const second = makeThread(ThreadId.make("thread:foundation-publish-order:second"), now);
3527	const published = yield* eventSink
3528	.stream({ afterSequence: yield* eventSink.latestSequence() })
3529	.pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped({ startImmediately: true }));
3530
3531	const firstWrite = yield* eventSink
3532	.writeWithEffects({
3533	events: [
3534	threadCreatedEvent({ id: "event:foundation-publish-order:first", thread: first, now }),
3535	],
3536	effects: [
3537	{
3538	id: "effect:foundation-publish-order:first",
3539	commandId: CommandId.make("command:foundation-publish-order:first"),
3540	threadId: first.id,
3541	request: { type: "terminal.cleanup" },
3542	},
3543	],
3544	})
3545	.pipe(Effect.forkScoped);
3546	yield* Deferred.await(firstCommitted);
3547	// The second writer commits after the first. It may run as far as it can
3548	// before the first writer resumes.
3549	const secondWrite = yield* eventSink
3550	.write({
3551	events: [
3552	threadCreatedEvent({
3553	id: "event:foundation-publish-order:second",
3554	thread: second,
3555	now,
3556	}),
3557	],
3558	})
3559	.pipe(
3560	Effect.provideService(Scheduler.MaxOpsBeforeYield, Number.POSITIVE_INFINITY),
3561	Effect.forkScoped,
3562	);
3563	yield* Effect.yieldNow;
3564	yield* Deferred.succeed(releaseFirst, undefined);
3565	yield* Fiber.join(firstWrite);
3566	yield* Fiber.join(secondWrite);
3567
3568	const sequences = Array.from(yield* Fiber.join(published), (stored) => stored.sequence);
3569	assert.deepEqual(
3570	sequences,
3571	[...sequences].sort((left, right) => left - right),
3572	);
3573	}).pipe(Effect.provide(layerEventSink));
3574	}).pipe(Effect.provide(layerDatabase)),
3575	);