1import type {
2 OrchestrationV2Command,
3 OrchestrationV2DomainEvent,
4 OrchestrationV2RuntimeRequest,
5 OrchestrationV2Run,
6 OrchestrationV2ThreadShellSnapshot,
7 OrchestrationV2StoredEvent,
8 OrchestrationV2ThreadProjection,
9 OrchestrationV2TurnItem,
10 ProviderApprovalDecision,
11 ProviderUserInputAnswers,
12 CommandId,
13 ThreadId,
14} from "@t3tools/contracts";
15import * as Clock from "effect/Clock";
16import * as Duration from "effect/Duration";
17import * as Effect from "effect/Effect";
18import * as Fiber from "effect/Fiber";
19import * as Option from "effect/Option";
20import * as Ref from "effect/Ref";
21import * as Schema from "effect/Schema";
22import * as Stream from "effect/Stream";
23import { TestClock } from "effect/testing";
24
25import * as Orchestrator from "../Orchestrator.ts";
26import type { ProviderReplayGate } from "@t3tools/provider-testing/replayGate";
27
28export type OrchestratorV2ScenarioStep =
29 | {
30 readonly type: "dispatch";
31 readonly command: OrchestrationV2Command;
32 readonly await?: boolean;
33 readonly key?: string;
34 }
35 | {
36 readonly type: "advance_clock";
37 readonly duration: Duration.Input;
38 }
39 | {
40 readonly type: "await";
41 readonly key: string;
42 }
43 | {
44 readonly type: "await_all";
45 }
46 | {
47 readonly type: "await_thread_idle";
48 readonly threadId: ThreadId;
49 }
50 | {
51 /** Waits until the thread's Waiting strip lists no background work. */
52 readonly type: "await_no_background_work";
53 readonly threadId: ThreadId;
54 }
55 | {
56 readonly type: "await_run_steerable";
57 readonly threadId: ThreadId;
58 readonly runId: OrchestrationV2Run["id"];
59 }
60 | {
61 readonly type: "await_run_status";
62 readonly threadId: ThreadId;
63 readonly runId: OrchestrationV2Run["id"];
64 readonly status: OrchestrationV2Run["status"];
65 }
66 | {
67 /**
68 * Wait for a run held open for background work to finish: each time the
69 * adapter arms its finish debounce (a replay gate receipt), advance the
70 * test clock by exactly that debounce, until the run reaches `status`.
71 */
72 readonly type: "finish_held_run";
73 readonly threadId: ThreadId;
74 readonly runId: OrchestrationV2Run["id"];
75 readonly status: OrchestrationV2Run["status"];
76 }
77 | {
78 readonly type: "await_run_turn_item";
79 readonly threadId: ThreadId;
80 readonly runId: OrchestrationV2Run["id"];
81 readonly itemType: OrchestrationV2TurnItem["type"];
82 }
83 | {
84 readonly type: "release_replay_gate_after_waiting";
85 readonly label: string;
86 readonly threadId: ThreadId;
87 readonly runId: OrchestrationV2Run["id"];
88 }
89 | {
90 readonly type: "release_replay_gate";
91 readonly label: string;
92 }
93 | {
94 readonly type: "capture_shell_snapshot";
95 readonly key: string;
96 }
97 | {
98 readonly type: "respond_to_next_runtime_request";
99 readonly threadId: ThreadId;
100 readonly commandId: CommandId;
101 readonly decision?: ProviderApprovalDecision;
102 readonly answers?: ProviderUserInputAnswers;
103 /** Captures the shell snapshot under this key while the request is pending. */
104 readonly shellSnapshotKeyWhilePending?: string;
105 };
106
107export interface OrchestratorV2Scenario {
108 readonly name: string;
109 readonly commands: ReadonlyArray<OrchestrationV2Command>;
110 readonly steps?: ReadonlyArray<OrchestratorV2ScenarioStep>;
111 readonly projectionThreadIds?: ReadonlyArray<ThreadId>;
112}
113
114export interface OrchestratorV2ScenarioResult {
115 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
116 readonly domainEvents: ReadonlyArray<OrchestrationV2DomainEvent>;
117 readonly projections: ReadonlyMap<ThreadId, OrchestrationV2ThreadProjection>;
118 readonly shellSnapshot: OrchestrationV2ThreadShellSnapshot;
119 readonly capturedShellSnapshots: ReadonlyMap<string, OrchestrationV2ThreadShellSnapshot>;
120}
121
122export class OrchestratorV2ScenarioStepError extends Schema.TaggedError<OrchestratorV2ScenarioStepError>()(
123 "OrchestratorV2ScenarioStepError",
124 {
125 scenario: Schema.String,
126 step: Schema.String,
127 },
128) {
129 override get message(): string {
130 return `Invalid orchestrator scenario step ${this.step} in ${this.scenario}.`;
131 }
132}
133
134function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray<ThreadId> {
135 switch (command.type) {
136 case "thread.create":
137 case "thread.archive":
138 case "thread.unarchive":
139 case "thread.delete":
140 case "thread.settle":
141 case "thread.auto-settle":
142 case "thread.unsettle":
143 case "thread.snooze":
144 case "thread.unsnooze":
145 case "thread.auto-settle.set":
146 case "thread.pin":
147 case "thread.unpin":
148 case "thread.pin.reorder":
149 case "thread.active.reorder":
150 case "thread.visit":
151 case "thread.mark-unread":
152 case "thread.metadata.update":
153 case "thread.pull-request.link":
154 case "thread.pull-request.unlink":
155 case "thread.pull-request-link.sync":
156 case "thread.pull-request.watch":
157 case "thread.pull-request.sync":
158 case "thread.title.regeneration.complete":
159 case "thread.runtime-mode.set":
160 case "thread.interaction-mode.set":
161 case "thread.model-selection.set":
162 case "provider-session.detach":
163 case "message.dispatch":
164 case "notification.delivery.accept":
165 case "prepared-run.release":
166 case "prepared-run.progress":
167 case "prepared-run.fail":
168 case "prepared-run.retry":
169 case "subagent.stop":
170 case "run.interrupt":
171 case "queued-message.promote-to-steer":
172 case "queue.resume":
173 case "queued-run.reorder":
174 case "queued-run.cancel":
175 case "queued-run.edit":
176 case "runtime-request.respond":
177 case "thread.user-input.dismiss":
178 case "checkpoint.rollback":
179 case "provider.switch":
180 return [command.threadId];
181 case "delegated_task.request":
182 case "delegated_task.wake-policy":
183 case "delegated_task.completion-delivery.acknowledge":
184 case "delegated_task.completion-delivery.dispose":
185 case "thread.created.record":
186 return [command.parentThreadId];
187 case "thread.fork":
188 case "thread.merge_back":
189 return [command.sourceThreadId, command.targetThreadId];
190 }
191}
192
193function scenarioSteps(
194 scenario: OrchestratorV2Scenario,
195): ReadonlyArray<OrchestratorV2ScenarioStep> {
196 return (
197 scenario.steps ??
198 scenario.commands.map((command) => ({
199 type: "dispatch" as const,
200 command,
201 await: true,
202 }))
203 );
204}
205
206function scenarioCommands(scenario: OrchestratorV2Scenario): ReadonlyArray<OrchestrationV2Command> {
207 return scenarioSteps(scenario).flatMap((step) =>
208 step.type === "dispatch" ? [step.command] : [],
209 );
210}
211
212const findPendingRuntimeRequest = (projection: OrchestrationV2ThreadProjection) =>
213 projection.runtimeRequests.find((request) => request.status === "pending");
214
215const hasActiveRun = (projection: OrchestrationV2ThreadProjection) =>
216 projection.runs.some((run) =>
217 ["preparing", "queued", "starting", "running", "waiting"].includes(run.status),
218 );
219
220const SCENARIO_WAIT_ATTEMPTS = 10_000;
221// Iterations count event-loop turns, not time: while async work (git
222// subprocesses, fixture IO) is in flight the loop is idle and the counter
223// burns at full speed on slow runners. A wait is exhausted only once BOTH
224// the iteration budget and this wall-clock deadline are spent.
225const SCENARIO_WAIT_DEADLINE_MS = 60_000;
226const scenarioWaitDeadline = () => performance.now() + SCENARIO_WAIT_DEADLINE_MS;
227const scenarioWaitExhausted = (attemptsRemaining: number, deadlineAt: number) =>
228 attemptsRemaining <= 0 && performance.now() >= deadlineAt;
229
230const yieldToRuntime = Effect.yieldNow.pipe(
231 Effect.andThen(
232 Effect.promise(
233 () =>
234 new Promise<void>((resolve) => {
235 setImmediate(resolve);
236 }),
237 ),
238 ),
239);
240
241function collectProjectionThreadIds(scenario: OrchestratorV2Scenario): ReadonlyArray<ThreadId> {
242 if (scenario.projectionThreadIds) {
243 return scenario.projectionThreadIds;
244 }
245
246 const ids = new Set<ThreadId>();
247 for (const command of scenarioCommands(scenario)) {
248 for (const threadId of commandThreadIds(command)) {
249 ids.add(threadId);
250 }
251 }
252 return Array.from(ids);
253}
254
255export function runOrchestratorV2Scenario(
256 scenario: OrchestratorV2Scenario,
257 options: {
258 readonly replayGate?: ProviderReplayGate;
259 /** Runs after the steps, while the provider session is still open. */
260 readonly afterSteps?: Effect.Effect<void>;
261 } = {},
262): Effect.Effect<
263 OrchestratorV2ScenarioResult,
264 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
265 Orchestrator.OrchestratorV2
266> {
267 return Effect.scoped(
268 Effect.gen(function* () {
269 const orchestrator = yield* Orchestrator.OrchestratorV2;
270 const storedEventGroups: Array<ReadonlyArray<OrchestrationV2StoredEvent>> = [];
271 const observedStoredEvents = yield* Ref.make<Array<OrchestrationV2StoredEvent>>([]);
272 yield* orchestrator.streamStoredEvents.pipe(
273 Stream.runForEach((event) =>
274 Ref.update(observedStoredEvents, (existing) => [...existing, event]),
275 ),
276 Effect.forkScoped,
277 );
278 const backgroundDispatches = new Map<
279 string,
280 Fiber.Fiber<ReadonlyArray<OrchestrationV2StoredEvent>, Orchestrator.OrchestratorV2Error>
281 >();
282 const capturedShellSnapshots = new Map<string, OrchestrationV2ThreadShellSnapshot>();
283 let anonymousBackgroundDispatchIndex = 0;
284
285 const awaitDispatch = (key: string) =>
286 Effect.gen(function* () {
287 const fiber = backgroundDispatches.get(key);
288 if (!fiber) {
289 return yield* new OrchestratorV2ScenarioStepError({
290 scenario: scenario.name,
291 step: `await:${key}`,
292 });
293 }
294 const events = yield* Fiber.join(fiber);
295 backgroundDispatches.delete(key);
296 storedEventGroups.push(events);
297 });
298
299 const waitForPendingRuntimeRequest = (
300 threadId: ThreadId,
301 attemptsRemaining = SCENARIO_WAIT_ATTEMPTS,
302 deadlineAt = scenarioWaitDeadline(),
303 ): Effect.Effect<
304 OrchestrationV2RuntimeRequest,
305 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
306 never
307 > =>
308 Effect.gen(function* () {
309 const projection = yield* orchestrator.getThreadProjection(threadId);
310 const request = findPendingRuntimeRequest(projection);
311 if (request !== undefined) {
312 return request;
313 }
314 if (scenarioWaitExhausted(attemptsRemaining, deadlineAt)) {
315 const runState = projection.runs.map((run) => `${run.id}:${run.status}`).join(",");
316 return yield* new OrchestratorV2ScenarioStepError({
317 scenario: scenario.name,
318 step: `respond_to_next_runtime_request:${threadId}:runs=${runState}:providerTurns=${projection.providerTurns.length}`,
319 });
320 }
321 yield* yieldToRuntime;
322 return yield* waitForPendingRuntimeRequest(threadId, attemptsRemaining - 1, deadlineAt);
323 });
324
325 const waitForThreadIdle = (
326 threadId: ThreadId,
327 attemptsRemaining = SCENARIO_WAIT_ATTEMPTS,
328 deadlineAt = scenarioWaitDeadline(),
329 ): Effect.Effect<
330 void,
331 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
332 never
333 > =>
334 Effect.gen(function* () {
335 const projection = yield* orchestrator.getThreadProjection(threadId);
336 if (!hasActiveRun(projection)) {
337 return;
338 }
339 if (scenarioWaitExhausted(attemptsRemaining, deadlineAt)) {
340 const activeRuns = projection.runs
341 .filter((run) =>
342 ["preparing", "queued", "starting", "running", "waiting"].includes(run.status),
343 )
344 .map((run) => `${run.id}:${run.status}`)
345 .join(",");
346 const pendingRequests = projection.runtimeRequests
347 .filter((request) => request.status === "pending")
348 .map((request) => `${request.id}:${request.kind}`)
349 .join(",");
350 return yield* new OrchestratorV2ScenarioStepError({
351 scenario: scenario.name,
352 step: `await_thread_idle:${threadId}:runs=${activeRuns}:requests=${pendingRequests}`,
353 });
354 }
355 yield* yieldToRuntime;
356 return yield* waitForThreadIdle(threadId, attemptsRemaining - 1, deadlineAt);
357 });
358
359 const waitForNoBackgroundWork = (
360 threadId: ThreadId,
361 attemptsRemaining = SCENARIO_WAIT_ATTEMPTS,
362 deadlineAt = scenarioWaitDeadline(),
363 ): Effect.Effect<
364 void,
365 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
366 never
367 > =>
368 Effect.gen(function* () {
369 const pending = (yield* orchestrator.getThreadShell(threadId))?.pendingBackgroundTasks;
370 if ((pending?.length ?? 0) === 0) {
371 return;
372 }
373 if (scenarioWaitExhausted(attemptsRemaining, deadlineAt)) {
374 return yield* new OrchestratorV2ScenarioStepError({
375 scenario: scenario.name,
376 step: `await_no_background_work:${threadId}:pending=${pending?.map((task) => task.taskId).join(",")}`,
377 });
378 }
379 yield* yieldToRuntime;
380 return yield* waitForNoBackgroundWork(threadId, attemptsRemaining - 1, deadlineAt);
381 });
382
383 const waitForRunSteerable = (
384 threadId: ThreadId,
385 runId: OrchestrationV2Run["id"],
386 attemptsRemaining = SCENARIO_WAIT_ATTEMPTS,
387 deadlineAt = scenarioWaitDeadline(),
388 ): Effect.Effect<
389 void,
390 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
391 never
392 > =>
393 Effect.gen(function* () {
394 const projection = yield* orchestrator.getThreadProjection(threadId);
395 const run = projection.runs.find((candidate) => candidate.id === runId);
396 const providerTurn = projection.providerTurns.find(
397 (candidate) =>
398 run?.activeAttemptId !== null &&
399 candidate.runAttemptId === run?.activeAttemptId &&
400 candidate.status === "running",
401 );
402 if (run?.status === "running" && providerTurn !== undefined) {
403 return;
404 }
405 if (scenarioWaitExhausted(attemptsRemaining, deadlineAt)) {
406 return yield* new OrchestratorV2ScenarioStepError({
407 scenario: scenario.name,
408 step: `await_run_steerable:${runId}`,
409 });
410 }
411 yield* yieldToRuntime;
412 return yield* waitForRunSteerable(threadId, runId, attemptsRemaining - 1, deadlineAt);
413 });
414
415 const waitForRunStatus = (
416 threadId: ThreadId,
417 runId: OrchestrationV2Run["id"],
418 status: OrchestrationV2Run["status"],
419 attemptsRemaining = SCENARIO_WAIT_ATTEMPTS,
420 deadlineAt = scenarioWaitDeadline(),
421 ): Effect.Effect<
422 void,
423 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
424 never
425 > =>
426 Effect.gen(function* () {
427 const projection = yield* orchestrator.getThreadProjection(threadId);
428 const run = projection.runs.find((candidate) => candidate.id === runId);
429 if (run?.status === status) {
430 return;
431 }
432 if (scenarioWaitExhausted(attemptsRemaining, deadlineAt)) {
433 return yield* new OrchestratorV2ScenarioStepError({
434 scenario: scenario.name,
435 step: `await_run_status:${runId}:${status}:actual=${run?.status ?? "missing"}`,
436 });
437 }
438 yield* yieldToRuntime;
439 return yield* waitForRunStatus(
440 threadId,
441 runId,
442 status,
443 attemptsRemaining - 1,
444 deadlineAt,
445 );
446 });
447
448 const waitForRunTurnItem = (
449 threadId: ThreadId,
450 runId: OrchestrationV2Run["id"],
451 itemType: OrchestrationV2TurnItem["type"],
452 attemptsRemaining = SCENARIO_WAIT_ATTEMPTS,
453 deadlineAt = scenarioWaitDeadline(),
454 ): Effect.Effect<
455 void,
456 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
457 never
458 > =>
459 Effect.gen(function* () {
460 const projection = yield* orchestrator.getThreadProjection(threadId);
461 const hasTurnItem = projection.turnItems.some(
462 (item) => item.runId === runId && item.type === itemType,
463 );
464 if (hasTurnItem) {
465 return;
466 }
467 if (scenarioWaitExhausted(attemptsRemaining, deadlineAt)) {
468 return yield* new OrchestratorV2ScenarioStepError({
469 scenario: scenario.name,
470 step: `await_run_turn_item:${runId}:${itemType}`,
471 });
472 }
473 yield* yieldToRuntime;
474 return yield* waitForRunTurnItem(
475 threadId,
476 runId,
477 itemType,
478 attemptsRemaining - 1,
479 deadlineAt,
480 );
481 });
482
483 const releaseReplayGateAfterWaiting = (
484 label: string,
485 threadId: ThreadId,
486 runId: OrchestrationV2Run["id"],
487 attemptsRemaining = SCENARIO_WAIT_ATTEMPTS,
488 ): Effect.Effect<
489 void,
490 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
491 never
492 > =>
493 Effect.gen(function* () {
494 const projection = yield* orchestrator.getThreadProjection(threadId);
495 const run = projection.runs.find((candidate) => candidate.id === runId);
496 const providerThread = projection.providerThreads.find(
497 (candidate) => candidate.id === run?.providerThreadId,
498 );
499 const pendingTaskCount = providerThread?.pendingBackgroundTasks?.length ?? 0;
500 const gateReached = options.replayGate?.hasReached(label) ?? false;
501 if (
502 gateReached &&
503 run?.status === "completed" &&
504 providerThread !== undefined &&
505 pendingTaskCount > 0
506 ) {
507 options.replayGate?.release(label);
508 yield* waitForProviderBackgroundTasksCleared(threadId, providerThread.id);
509 return;
510 }
511 if (attemptsRemaining <= 0) {
512 options.replayGate?.release(label);
513 return yield* new OrchestratorV2ScenarioStepError({
514 scenario: scenario.name,
515 step: `release_replay_gate_after_waiting:${label}:reached=${gateReached}:run=${run?.status ?? "missing"}:providerThread=${run?.providerThreadId ?? "missing"}:pending=${pendingTaskCount}`,
516 });
517 }
518 yield* yieldToRuntime;
519 return yield* releaseReplayGateAfterWaiting(
520 label,
521 threadId,
522 runId,
523 attemptsRemaining - 1,
524 );
525 });
526
527 const waitForProviderBackgroundTasksCleared = (
528 threadId: ThreadId,
529 providerThreadId: NonNullable<OrchestrationV2Run["providerThreadId"]>,
530 attemptsRemaining = SCENARIO_WAIT_ATTEMPTS,
531 ): Effect.Effect<
532 void,
533 Orchestrator.OrchestratorV2Error | OrchestratorV2ScenarioStepError,
534 never
535 > =>
536 Effect.gen(function* () {
537 const projection = yield* orchestrator.getThreadProjection(threadId);
538 const providerThread = projection.providerThreads.find(
539 (candidate) => candidate.id === providerThreadId,
540 );
541 const hasPendingTasks = (providerThread?.pendingBackgroundTasks?.length ?? 0) > 0;
542 if (!hasPendingTasks && providerThread?.status === "idle") {
543 return;
544 }
545 if (attemptsRemaining <= 0) {
546 const providerState = projection.providerThreads
547 .map(
548 (candidate) =>
549 `${candidate.id}:${candidate.status}:pending=${candidate.pendingBackgroundTasks?.length ?? 0}`,
550 )
551 .join(",");
552 return yield* new OrchestratorV2ScenarioStepError({
553 scenario: scenario.name,
554 step: `await_provider_background_tasks_cleared:${threadId}:target=${providerThreadId}:providers=${providerState}`,
555 });
556 }
557 yield* yieldToRuntime;
558 return yield* waitForProviderBackgroundTasksCleared(
559 threadId,
560 providerThreadId,
561 attemptsRemaining - 1,
562 );
563 });
564
565 // Finish receipts earlier held runs already consumed.
566 let consumedFinishReceipts = 0;
567 const finishHeldRun = Effect.fn("scenario.finishHeldRun")(function* (
568 step: Extract<OrchestratorV2ScenarioStep, { readonly type: "finish_held_run" }>,
569 ) {
570 const gate = options.replayGate;
571 if (gate === undefined) {
572 return yield* new OrchestratorV2ScenarioStepError({
573 scenario: scenario.name,
574 step: `finish_held_run:${step.runId}:no_replay_gate`,
575 });
576 }
577 while (true) {
578 // The run settles only through the latest armed debounce; a rearm
579 // (a late frame) supersedes it with a new receipt. Advance past the
580 // latest one, then either the run gets there or it rearmed again.
581 const armed = yield* Effect.promise(() =>
582 gate.waitForFinishArmed(consumedFinishReceipts),
583 ).pipe(
584 Effect.timeoutOption(Duration.millis(SCENARIO_WAIT_DEADLINE_MS)),
585 Effect.provideService(Clock.Clock, Clock.Clock.defaultValue()),
586 );
587 if (Option.isNone(armed)) {
588 const projection = yield* orchestrator.getThreadProjection(step.threadId);
589 const run = projection.runs.find((candidate) => candidate.id === step.runId);
590 return yield* new OrchestratorV2ScenarioStepError({
591 scenario: scenario.name,
592 step: `finish_held_run:${step.runId}:${step.status}:actual=${run?.status ?? "missing"}:no_finish_armed`,
593 });
594 }
595 consumedFinishReceipts = gate.finishArmedCount();
596 yield* TestClock.adjust(armed.value);
597 const settled = yield* Effect.raceFirst(
598 waitForRunStatus(step.threadId, step.runId, step.status).pipe(Effect.as(true)),
599 Effect.promise(() => gate.waitForFinishArmed(consumedFinishReceipts)).pipe(
600 Effect.as(false),
601 ),
602 );
603 // A finalized turn arms no further receipts, so the count is final.
604 if (settled) {
605 consumedFinishReceipts = gate.finishArmedCount();
606 return;
607 }
608 }
609 });
610
611 const releaseReplayGate = Effect.fn("scenario.releaseReplayGate")(function* (label: string) {
612 const gate = options.replayGate;
613 const reached =
614 gate === undefined ? false : yield* Effect.promise(() => gate.waitForReached(label));
615 if (!reached) {
616 return yield* new OrchestratorV2ScenarioStepError({
617 scenario: scenario.name,
618 step: `release_replay_gate:${label}:not_configured`,
619 });
620 }
621 gate?.release(label);
622 });
623
624 for (const step of scenarioSteps(scenario)) {
625 switch (step.type) {
626 case "dispatch": {
627 if (step.await ?? true) {
628 const result = yield* orchestrator.dispatch(step.command);
629 storedEventGroups.push(result.storedEvents);
630 break;
631 }
632
633 anonymousBackgroundDispatchIndex += 1;
634 const key = step.key ?? `dispatch:${anonymousBackgroundDispatchIndex}`;
635 backgroundDispatches.set(
636 key,
637 yield* orchestrator.dispatch(step.command).pipe(
638 Effect.map((result) => result.storedEvents),
639 Effect.forkScoped,
640 ),
641 );
642 break;
643 }
644 case "advance_clock":
645 yield* TestClock.adjust(step.duration);
646 break;
647 case "await":
648 yield* awaitDispatch(step.key);
649 break;
650 case "await_all":
651 for (const key of Array.from(backgroundDispatches.keys())) {
652 yield* awaitDispatch(key);
653 }
654 break;
655 case "await_thread_idle":
656 yield* waitForThreadIdle(step.threadId);
657 break;
658 case "await_no_background_work":
659 yield* waitForNoBackgroundWork(step.threadId);
660 break;
661 case "await_run_steerable":
662 yield* waitForRunSteerable(step.threadId, step.runId);
663 break;
664 case "await_run_status":
665 yield* waitForRunStatus(step.threadId, step.runId, step.status);
666 break;
667 case "finish_held_run":
668 yield* finishHeldRun(step);
669 break;
670 case "await_run_turn_item":
671 yield* waitForRunTurnItem(step.threadId, step.runId, step.itemType);
672 break;
673 case "release_replay_gate_after_waiting":
674 yield* releaseReplayGateAfterWaiting(step.label, step.threadId, step.runId);
675 break;
676 case "release_replay_gate":
677 yield* releaseReplayGate(step.label);
678 break;
679 case "capture_shell_snapshot":
680 capturedShellSnapshots.set(step.key, yield* orchestrator.getShellSnapshot());
681 break;
682 case "respond_to_next_runtime_request": {
683 const request = yield* waitForPendingRuntimeRequest(step.threadId);
684 if (step.shellSnapshotKeyWhilePending !== undefined) {
685 capturedShellSnapshots.set(
686 step.shellSnapshotKeyWhilePending,
687 yield* orchestrator.getShellSnapshot(),
688 );
689 }
690 const result = yield* orchestrator.dispatch({
691 type: "runtime-request.respond",
692 commandId: step.commandId,
693 threadId: step.threadId,
694 requestId: request.id,
695 ...(step.decision === undefined ? {} : { decision: step.decision }),
696 ...(step.answers === undefined ? {} : { answers: step.answers }),
697 });
698 storedEventGroups.push(result.storedEvents);
699 break;
700 }
701 }
702 }
703
704 for (const key of Array.from(backgroundDispatches.keys())) {
705 yield* awaitDispatch(key);
706 }
707 if (options.afterSteps !== undefined) {
708 yield* options.afterSteps;
709 }
710
711 const shellSnapshot = yield* orchestrator.getShellSnapshot();
712 const projectionThreadIds = new Set(collectProjectionThreadIds(scenario));
713 for (const thread of shellSnapshot.threads) {
714 projectionThreadIds.add(thread.id);
715 }
716 const projections = new Map<ThreadId, OrchestrationV2ThreadProjection>();
717 for (const threadId of projectionThreadIds) {
718 projections.set(threadId, yield* orchestrator.getThreadProjection(threadId));
719 }
720
721 yield* Effect.yieldNow;
722
723 const observedEvents = yield* Ref.get(observedStoredEvents);
724 const storedEvents = (
725 observedEvents.length > 0 ? observedEvents : storedEventGroups.flat()
726 ).toSorted((left, right) => left.sequence - right.sequence);
727 return {
728 storedEvents,
729 domainEvents: storedEvents.map((stored) => stored.event),
730 projections,
731 shellSnapshot,
732 capturedShellSnapshots,
733 };
734 }),
735 ).pipe(
736 // A failed step can leave provider frames held at a gate. Release them
737 // before the harness shuts the provider down, or teardown waits on them.
738 Effect.ensuring(Effect.sync(() => options.replayGate?.releaseAll())),
739 );
740}