1import { assert, it } from "@effect/vitest";
2import {
3 CheckpointScopeId,
4 CommandId,
5 EventId,
6 MessageId,
7 NodeId,
8 type OrchestrationV2AppThread,
9 type OrchestrationV2DomainEvent,
10 type OrchestrationV2Run,
11 ProjectId,
12 ProviderDriverKind,
13 ProviderInstanceId,
14 ProviderSessionId,
15 ProviderThreadId,
16 ProviderTurnId,
17 RunAttemptId,
18 RunId,
19 ThreadId,
20 TurnItemId,
21} from "@t3tools/contracts";
22import * as Console from "effect/Console";
23import * as DateTime from "effect/DateTime";
24import * as Effect from "effect/Effect";
25import * as Layer from "effect/Layer";
26import * as SqlClient from "effect/sql/SqlClient";
27
28import * as SqlitePersistence from "../persistence/Sqlite.ts";
29import * as ServerSettings from "../serverSettings.ts";
30import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
31import * as EffectOutbox from "./EffectOutbox.ts";
32import * as EventSink from "./EventSink.ts";
33import * as EventStore from "./EventStore.ts";
34import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
35import * as ProjectionStore from "./ProjectionStore.ts";
36import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
37
38// Startup recovery cost for restart continuation. The default case is small
39// enough for CI and asserts correctness; T3_BENCH_RECOVERY=1 adds the matrix.
40// `it.live` keeps a real clock: each reconcile gets a fresh command id, so the
41// second recover cannot hide behind command receipt dedup.
42
43const layerStores = Layer.mergeAll(
44 EventStore.layer,
45 ProjectionStore.layer,
46 EffectOutbox.layer,
47 IdAllocator.layer,
48).pipe(Layer.provideMerge(SqlitePersistence.layerMemory));
49const layerTest = ProviderRuntimeRecovery.layer.pipe(
50 Layer.provideMerge(EventSink.layer.pipe(Layer.provideMerge(layerStores))),
51 Layer.provideMerge(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true })),
52);
53
54const providerInstanceId = ProviderInstanceId.make("codex");
55const driver = ProviderDriverKind.make("codex");
56const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" };
57const projectId = ProjectId.make("project:recovery-bench");
58const ITEMS_PER_SETTLED_THREAD = 10;
59const OPEN_ITEMS_PER_ACTIVE_THREAD = 3;
60const WAITING_THREADS = 3;
61const SEED_BATCH = 500;
62
63type UnstampedEvent = OrchestrationV2DomainEvent extends infer Event
64 ? Event extends OrchestrationV2DomainEvent
65 ? Omit<Event, "id" | "occurredAt">
66 : never
67 : never;
68
69interface Scenario {
70 readonly settled: number;
71 readonly active: number;
72}
73
74interface Seeded {
75 readonly unheldQueuedThreads: number;
76 readonly waitingThreads: number;
77}
78
79const seedScenario = Effect.fn(function* (scenario: Scenario) {
80 const projections = yield* ProjectionStore.ProjectionStoreV2;
81 const outbox = yield* EffectOutbox.EffectOutboxV2;
82 const sql = yield* SqlClient.SqlClient;
83 const now = yield* DateTime.now;
84 let eventIndex = 0;
85 const apply = (event: UnstampedEvent) =>
86 projections.apply({
87 ...event,
88 id: EventId.make(`event:bench:${eventIndex++}`),
89 occurredAt: now,
90 } as OrchestrationV2DomainEvent);
91
92 const thread = (id: ThreadId, overrides: Partial<OrchestrationV2AppThread> = {}) =>
93 apply({
94 type: "thread.created",
95 threadId: id,
96 payload: {
97 createdBy: "user",
98 creationSource: "web",
99 id,
100 projectId,
101 title: id,
102 providerInstanceId,
103 modelSelection,
104 runtimeMode: "full-access",
105 interactionMode: "default",
106 branch: null,
107 worktreePath: null,
108 activeProviderThreadId: null,
109 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: id },
110 forkedFrom: null,
111 createdAt: now,
112 updatedAt: now,
113 archivedAt: null,
114 settledOverride: null,
115 settledAt: null,
116 lastVisitedAt: null,
117 deletedAt: null,
118 ...overrides,
119 },
120 });
121
122 const run = (
123 threadId: ThreadId,
124 ordinal: number,
125 status: OrchestrationV2Run["status"],
126 overrides: Partial<OrchestrationV2Run> = {},
127 ) => {
128 const runId = RunId.make(`run:${threadId}:${ordinal}`);
129 return apply({
130 type: "run.created",
131 threadId,
132 runId,
133 payload: {
134 id: runId,
135 threadId,
136 ordinal,
137 providerInstanceId,
138 modelSelection,
139 providerThreadId: null,
140 userMessageId: MessageId.make(`message:${runId}`),
141 rootNodeId: null,
142 activeAttemptId: null,
143 status,
144 requestedAt: now,
145 startedAt: status === "queued" ? null : now,
146 completedAt: status === "completed" ? now : null,
147 checkpointId: null,
148 contextHandoffId: null,
149 ...overrides,
150 },
151 }).pipe(Effect.as(runId));
152 };
153
154 const commandItem = (input: {
155 readonly threadId: ThreadId;
156 readonly runId: RunId;
157 readonly ordinal: number;
158 readonly status: "running" | "completed";
159 readonly nodeId?: NodeId;
160 readonly providerThreadId?: ProviderThreadId;
161 readonly providerTurnId?: ProviderTurnId;
162 }) =>
163 apply({
164 type: "turn-item.updated",
165 threadId: input.threadId,
166 runId: input.runId,
167 payload: {
168 id: TurnItemId.make(`item:${input.runId}:${input.ordinal}`),
169 threadId: input.threadId,
170 runId: input.runId,
171 nodeId: input.nodeId ?? null,
172 providerThreadId: input.providerThreadId ?? null,
173 providerTurnId: input.providerTurnId ?? null,
174 nativeItemRef: null,
175 parentItemId: null,
176 ordinal: input.ordinal,
177 type: "command_execution",
178 status: input.status,
179 title: `Command ${input.ordinal}`,
180 input: `echo ${input.ordinal}`,
181 output: input.status === "completed" ? `${input.ordinal}\n` : "",
182 ...(input.status === "completed" ? { exitCode: 0 } : {}),
183 startedAt: now,
184 completedAt: input.status === "completed" ? now : null,
185 updatedAt: now,
186 },
187 });
188
189 const settledThread = Effect.fn(function* (index: number) {
190 const threadId = ThreadId.make(`thread:bench:settled:${index}`);
191 yield* thread(threadId);
192 const runId = yield* run(threadId, 1, "completed");
193 for (let ordinal = 1; ordinal <= ITEMS_PER_SETTLED_THREAD; ordinal += 1) {
194 yield* commandItem({ threadId, runId, ordinal, status: "completed" });
195 }
196 // ~5% of threads carry a queued follow-up; half are already held, and an
197 // earlier recovery already handled those, so only unheld ones are candidates.
198 if (index % 20 === 0) {
199 const queueHeld = index % 40 === 0;
200 yield* run(threadId, 2, "queued", { queueHeld });
201 return !queueHeld;
202 }
203 return false;
204 });
205
206 const activeThread = Effect.fn(function* (index: number) {
207 const threadId = ThreadId.make(`thread:bench:active:${index}`);
208 const providerSessionId = ProviderSessionId.make(`session:bench:${index}`);
209 const providerThreadId = ProviderThreadId.make(`provider-thread:bench:${index}`);
210 const runId = RunId.make(`run:${threadId}:1`);
211 const attemptId = RunAttemptId.make(`attempt:bench:${index}`);
212 const nodeId = NodeId.make(`node:bench:${index}`);
213 const providerTurnId = ProviderTurnId.make(`provider-turn:bench:${index}`);
214 yield* thread(threadId, { activeProviderThreadId: providerThreadId });
215 yield* apply({
216 type: "provider-session.attached",
217 threadId,
218 driver,
219 providerInstanceId,
220 payload: {
221 id: providerSessionId,
222 driver,
223 providerInstanceId,
224 status: "ready",
225 cwd: "/workspace",
226 model: modelSelection.model,
227 capabilities: CodexProviderCapabilitiesV2,
228 createdAt: now,
229 updatedAt: now,
230 lastError: null,
231 },
232 });
233 yield* apply({
234 type: "provider-thread.updated",
235 threadId,
236 driver,
237 providerInstanceId,
238 payload: {
239 id: providerThreadId,
240 appThreadId: threadId,
241 ownerNodeId: null,
242 driver,
243 providerInstanceId,
244 providerSessionId,
245 nativeThreadRef: { driver, nativeId: `native:bench:${index}`, strength: "strong" },
246 nativeConversationHeadRef: null,
247 status: "active",
248 firstRunOrdinal: 1,
249 lastRunOrdinal: 1,
250 handoffIds: [],
251 forkedFrom: null,
252 createdAt: now,
253 updatedAt: now,
254 pendingBackgroundTasks: [],
255 },
256 });
257 yield* run(threadId, 1, "running", {
258 providerThreadId,
259 rootNodeId: nodeId,
260 activeAttemptId: attemptId,
261 });
262 yield* apply({
263 type: "run-attempt.created",
264 threadId,
265 runId,
266 payload: {
267 id: attemptId,
268 runId,
269 attemptOrdinal: 1,
270 rootNodeId: nodeId,
271 providerInstanceId,
272 providerThreadId,
273 providerTurnId,
274 reason: "initial",
275 status: "running",
276 startedAt: now,
277 completedAt: null,
278 },
279 });
280 yield* apply({
281 type: "node.updated",
282 threadId,
283 runId,
284 nodeId,
285 providerInstanceId,
286 payload: {
287 id: nodeId,
288 threadId,
289 runId,
290 parentNodeId: null,
291 rootNodeId: nodeId,
292 kind: "root_turn",
293 status: "running",
294 countsForRun: true,
295 providerThreadId,
296 providerTurnId,
297 nativeItemRef: null,
298 runtimeRequestId: null,
299 checkpointScopeId: null,
300 startedAt: now,
301 completedAt: null,
302 },
303 });
304 yield* apply({
305 type: "provider-turn.updated",
306 threadId,
307 runId,
308 nodeId,
309 providerInstanceId,
310 payload: {
311 id: providerTurnId,
312 providerThreadId,
313 nodeId,
314 runAttemptId: attemptId,
315 nativeTurnRef: { driver, nativeId: `native-turn:bench:${index}`, strength: "strong" },
316 ordinal: 1,
317 status: "running",
318 startedAt: now,
319 completedAt: null,
320 },
321 });
322 for (let ordinal = 1; ordinal <= OPEN_ITEMS_PER_ACTIVE_THREAD; ordinal += 1) {
323 yield* commandItem({
324 threadId,
325 runId,
326 ordinal,
327 status: "running",
328 nodeId,
329 providerThreadId,
330 providerTurnId,
331 });
332 }
333 });
334
335 const waitingThread = Effect.fn(function* (index: number) {
336 const threadId = ThreadId.make(`thread:bench:waiting:${index}`);
337 yield* thread(threadId);
338 yield* run(threadId, 1, "completed");
339 const runId = yield* run(threadId, 2, "waiting");
340 yield* outbox.enqueue([
341 {
342 id: `effect:bench:checkpoint:${runId}`,
343 commandId: CommandId.make(`command:effect:checkpoint.capture:${runId}`),
344 threadId,
345 request: {
346 type: "checkpoint.capture",
347 runId,
348 scopeId: CheckpointScopeId.make(`scope:bench:${index}`),
349 },
350 },
351 ]);
352 });
353
354 let unheldQueuedThreads = 0;
355 for (let start = 0; start < scenario.settled; start += SEED_BATCH) {
356 const end = Math.min(scenario.settled, start + SEED_BATCH);
357 yield* sql.withTransaction(
358 Effect.gen(function* () {
359 for (let index = start; index < end; index += 1) {
360 if (yield* settledThread(index)) unheldQueuedThreads += 1;
361 }
362 }),
363 );
364 }
365 yield* sql.withTransaction(
366 Effect.gen(function* () {
367 for (let index = 0; index < scenario.active; index += 1) yield* activeThread(index);
368 for (let index = 0; index < WAITING_THREADS; index += 1) yield* waitingThread(index);
369 }),
370 );
371 return { unheldQueuedThreads, waitingThreads: WAITING_THREADS } satisfies Seeded;
372});
373
374const timed = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
375 Effect.gen(function* () {
376 const start = performance.now();
377 const value = yield* effect;
378 return [value, performance.now() - start] as const;
379 });
380
381const continuationEffectCount = Effect.gen(function* () {
382 const sql = yield* SqlClient.SqlClient;
383 const rows = yield* sql<{ readonly count: number }>`
384 SELECT COUNT(*) AS count FROM orchestration_v2_effect_outbox
385 WHERE effect_id LIKE 'effect:restart-continuation:%' AND status = 'pending'
386 `;
387 return Number(rows[0]?.count ?? 0);
388});
389
390interface Measurement {
391 readonly settled: number;
392 readonly active: number;
393 readonly seedMs: number;
394 readonly candidates: number;
395 readonly selectMs: number;
396 readonly prepareMs: number;
397 readonly shutdownMs: number;
398 readonly startupAfterShutdownMs: number;
399 readonly crashRecoverMs: number;
400 readonly secondRecoverMs: number;
401}
402
403/** A graceful restart: prepare intent, reconcile on shutdown, recover on boot. */
404const measureGraceful = Effect.fn(function* (scenario: Scenario) {
405 const [seeded, seedMs] = yield* timed(seedScenario(scenario));
406 const projections = yield* ProjectionStore.ProjectionStoreV2;
407 const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
408 const [candidates, selectMs] = yield* timed(projections.getRecoveryThreadIds("runtime"));
409 assert.equal(
410 candidates.length,
411 scenario.active + seeded.unheldQueuedThreads + seeded.waitingThreads,
412 "runtime recovery candidates",
413 );
414 const [, prepareMs] = yield* timed(recovery.prepareForShutdown);
415 assert.equal(yield* continuationEffectCount, scenario.active, "prepared continuations");
416 const [, shutdownMs] = yield* timed(recovery.reconcile("shutdown"));
417 const [, startupAfterShutdownMs] = yield* timed(recovery.recover);
418 assert.equal(yield* continuationEffectCount, scenario.active, "continuations after boot");
419 return {
420 seedMs,
421 candidates: candidates.length,
422 selectMs,
423 prepareMs,
424 shutdownMs,
425 startupAfterShutdownMs,
426 };
427});
428
429/** A crash: no shutdown hook ran, so boot recovery records the continuations. */
430const measureCrash = Effect.fn(function* (scenario: Scenario) {
431 yield* seedScenario(scenario);
432 const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
433 const [summary, crashRecoverMs] = yield* timed(recovery.recover);
434 assert.equal(summary.terminalizedRuns, scenario.active, "terminalized active runs");
435 assert.equal(summary.stoppedSessions, scenario.active, "stopped active sessions");
436 assert.equal(yield* continuationEffectCount, scenario.active, "recorded continuations");
437 const [, secondRecoverMs] = yield* timed(recovery.recover);
438 assert.equal(yield* continuationEffectCount, scenario.active, "second recover is idempotent");
439 return { crashRecoverMs, secondRecoverMs };
440});
441
442const measure = Effect.fn(function* (scenario: Scenario) {
443 const graceful = yield* measureGraceful(scenario).pipe(Effect.provide(Layer.fresh(layerTest)));
444 const crash = yield* measureCrash(scenario).pipe(Effect.provide(Layer.fresh(layerTest)));
445 return { ...scenario, ...graceful, ...crash } satisfies Measurement;
446});
447
448const formatTable = (rows: ReadonlyArray<Measurement>) => {
449 const ms = (value: number) => value.toFixed(1);
450 const perActive = (value: number, active: number) => (value / Math.max(active, 1)).toFixed(2);
451 const header = [
452 "N",
453 "A",
454 "seed",
455 "cand",
456 "select",
457 "prepare",
458 "prep/A",
459 "shutdown",
460 "boot-after",
461 "crash-recover",
462 "recover/A",
463 "recover-2",
464 ];
465 const lines = rows.map((row) => [
466 String(row.settled),
467 String(row.active),
468 ms(row.seedMs),
469 String(row.candidates),
470 ms(row.selectMs),
471 ms(row.prepareMs),
472 perActive(row.prepareMs, row.active),
473 ms(row.shutdownMs),
474 ms(row.startupAfterShutdownMs),
475 ms(row.crashRecoverMs),
476 perActive(row.crashRecoverMs, row.active),
477 ms(row.secondRecoverMs),
478 ]);
479 const widths = header.map((cell, column) =>
480 Math.max(cell.length, ...lines.map((line) => line[column]!.length)),
481 );
482 return [header, ...lines]
483 .map((line) => line.map((cell, column) => cell.padStart(widths[column]!)).join(" "))
484 .join("\n");
485};
486
487it.live(
488 "restart recovery records one continuation per active thread and is idempotent",
489 () =>
490 Effect.gen(function* () {
491 const row = yield* measure({ settled: 500, active: 20 });
492 yield* Console.log(`provider runtime recovery (ms)\n${formatTable([row])}`);
493 }),
494 60_000,
495);
496
497it.live.skipIf(process.env.T3_BENCH_RECOVERY !== "1")(
498 "restart recovery scales with active threads, not settled history",
499 () =>
500 Effect.gen(function* () {
501 const rows: Array<Measurement> = [];
502 for (const settled of [5_000, 50_000]) {
503 for (const active of [10, 100, 1_000]) {
504 rows.push(yield* measure({ settled, active }));
505 yield* Console.log(`provider runtime recovery (ms)\n${formatTable(rows)}`);
506 }
507 }
508 }),
509 3_600_000,
510);