1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 MessageId,
5 EventId,
6 NodeId,
7 ProjectId,
8 ProviderDriverKind,
9 ProviderInstanceId,
10 ProviderThreadId,
11 ProviderTurnId,
12 ThreadId,
13 type ModelSelection,
14 type OrchestrationV2DomainEvent,
15} from "@t3tools/contracts";
16import * as DateTime from "effect/DateTime";
17import * as Deferred from "effect/Deferred";
18import * as Effect from "effect/Effect";
19import * as Fiber from "effect/Fiber";
20import * as Queue from "effect/Queue";
21import * as Stream from "effect/Stream";
22import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
23import * as EffectWorker from "./EffectWorker.ts";
24import * as EventSink from "./EventSink.ts";
25import * as Orchestrator from "./Orchestrator.ts";
26import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
27import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
28import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
29import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
30
31const driver = ProviderDriverKind.make("codex");
32const instanceId = ProviderInstanceId.make("codex");
33const modelSelection = { instanceId, model: "test-model" };
34
35it.effect.each(
36 [false, true]
37 .flatMap((mailbox) =>
38 (
39 [
40 "before delivery",
41 "during delivery",
42 "before dispatch",
43 "after delivery",
44 "without native steering",
45 "with interrupting native steering",
46 "settled only",
47 ] as const
48 ).map((timing) => ({
49 mailbox,
50 timing,
51 label: mailbox ? "mailbox notification" : "steering",
52 })),
53 )
54 .filter(
55 ({ mailbox, timing }) =>
56 mailbox ||
57 (timing !== "without native steering" &&
58 timing !== "with interrupting native steering" &&
59 timing !== "settled only"),
60 ),
61)("delivers $label when completion wins $timing", ({ mailbox, timing }) =>
62 Effect.scoped(
63 Effect.gen(function* () {
64 const cwd = yield* checkpointWorkspace(`steering-completion-${timing.replaceAll(" ", "-")}`);
65 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
66 const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = [];
67 const steerEntered = yield* Deferred.make<void>();
68 const rejectSteer = yield* Deferred.make<void>();
69 let steerCalls = 0;
70 const capabilities = {
71 ...CodexProviderCapabilitiesV2,
72 turns: {
73 ...CodexProviderCapabilitiesV2.turns,
74 supportsActiveSteering: timing !== "without native steering",
75 activeSteeringInterruptsTools: timing === "with interrupting native steering",
76 },
77 };
78 const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
79 instanceId,
80 driver,
81 getCapabilities: () => Effect.succeed(capabilities),
82 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
83 openSession: (input) =>
84 Effect.gen(function* () {
85 const now = yield* DateTime.now;
86 return {
87 instanceId,
88 driver,
89 providerSessionId: input.providerSessionId,
90 providerSession: {
91 id: input.providerSessionId,
92 driver,
93 providerInstanceId: instanceId,
94 status: "ready",
95 cwd,
96 model: modelSelection.model,
97 capabilities,
98 createdAt: now,
99 updatedAt: now,
100 lastError: null,
101 },
102 events: Stream.fromQueue(events),
103 ensureThread: ({ threadId }) =>
104 Effect.succeed({
105 id: ProviderThreadId.make(`provider-thread:${threadId}`),
106 driver,
107 providerInstanceId: instanceId,
108 providerSessionId: input.providerSessionId,
109 appThreadId: threadId,
110 ownerNodeId: null,
111 nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
112 nativeConversationHeadRef: null,
113 status: "idle",
114 firstRunOrdinal: null,
115 lastRunOrdinal: null,
116 handoffIds: [],
117 forkedFrom: null,
118 createdAt: now,
119 updatedAt: now,
120 }),
121 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
122 startTurn: (turn) =>
123 Effect.gen(function* () {
124 started.push(turn);
125 yield* Queue.offer(events, {
126 type: "provider_turn.updated",
127 driver,
128 providerTurn: {
129 id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`),
130 providerThreadId: turn.providerThread.id,
131 nodeId: turn.rootNodeId,
132 runAttemptId: turn.attemptId,
133 nativeTurnRef: {
134 driver,
135 nativeId: `native:${turn.attemptId}`,
136 strength: "strong",
137 },
138 ordinal: turn.providerTurnOrdinal,
139 status: "running",
140 startedAt: now,
141 completedAt: null,
142 },
143 });
144 }),
145 steerTurn: (turn) =>
146 Effect.gen(function* () {
147 steerCalls += 1;
148 if (timing === "after delivery") return;
149 yield* Deferred.succeed(steerEntered, undefined);
150 yield* Deferred.await(rejectSteer);
151 return yield* new ProviderAdapter.ProviderAdapterSteerRunError({
152 driver,
153 providerThreadId: turn.providerThread.id,
154 providerTurnId: turn.providerTurnId,
155 cause: "turn already completed",
156 });
157 }),
158 interruptTurn: () => Effect.void,
159 respondToRuntimeRequest: () => Effect.void,
160 readThreadSnapshot: () => Effect.die("unused"),
161 rollbackThread: () => Effect.die("unused"),
162 forkThread: () => Effect.die("unused"),
163 };
164 }),
165 };
166 yield* Effect.gen(function* () {
167 const orchestrator = yield* Orchestrator.OrchestratorV2;
168 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
169 const threadId = ThreadId.make("thread:steering-completion");
170 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
171 orchestrator.streamDomainEvents.pipe(
172 Stream.filter(predicate),
173 Stream.take(1),
174 Stream.runDrain,
175 Effect.forkScoped,
176 );
177 yield* orchestrator.dispatch({
178 type: "thread.create",
179 commandId: CommandId.make("create"),
180 threadId,
181 projectId: ProjectId.make("project:steering-completion"),
182 title: "Steering race",
183 modelSelection,
184 runtimeMode: "full-access",
185 interactionMode: "default",
186 branch: null,
187 worktreePath: cwd,
188 createdBy: "user",
189 creationSource: "web",
190 });
191 yield* orchestrator.dispatch({
192 type: "message.dispatch",
193 commandId: CommandId.make("first"),
194 threadId,
195 messageId: MessageId.make("message:first"),
196 text: "first",
197 attachments: [],
198 dispatchMode: { type: "start_immediately" },
199 createdBy: "user",
200 creationSource: "web",
201 });
202 const running = yield* watch(
203 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
204 );
205 yield* worker.drain();
206 yield* Fiber.join(running);
207 const first = started[0]!;
208 const messageId = MessageId.make("message:steering");
209 const taskId = NodeId.make("task:mailbox");
210 if (mailbox) {
211 const sink = yield* EventSink.EventSinkV2;
212 const current = yield* orchestrator.getThreadProjection(threadId);
213 const parentRun = current.runs.find((run) => run.id === first.runId)!;
214 const now = yield* DateTime.now;
215 yield* sink.write({
216 events: [
217 {
218 id: EventId.make("mailbox:cohort"),
219 type: "run.updated",
220 threadId,
221 runId: first.runId,
222 occurredAt: now,
223 payload: {
224 ...parentRun,
225 delegatedCompletion: {
226 disposition: "open",
227 nextGeneration: 2,
228 delivery: { generation: 1, messageId, taskIds: [taskId] },
229 },
230 },
231 },
232 {
233 id: EventId.make("mailbox:task"),
234 type: "subagent.updated",
235 threadId,
236 runId: first.runId,
237 nodeId: taskId,
238 occurredAt: now,
239 payload: {
240 id: taskId,
241 threadId,
242 runId: first.runId,
243 parentNodeId: first.rootNodeId,
244 origin: "app_owned",
245 createdBy: "agent",
246 driver,
247 providerInstanceId: instanceId,
248 providerThreadId: null,
249 childThreadId: null,
250 nativeTaskRef: null,
251 prompt: "Do background work",
252 title: "Background test",
253 model: null,
254 completionWake: timing === "settled only" ? "settled_only" : "always",
255 completionDelivery: { state: "claimed", observedByRunId: null },
256 status: "completed",
257 result: "done",
258 startedAt: now,
259 completedAt: now,
260 updatedAt: now,
261 },
262 },
263 ],
264 });
265 }
266 const dispatchSteer = orchestrator.dispatch({
267 type: "message.dispatch",
268 commandId: CommandId.make("steer"),
269 threadId,
270 messageId,
271 text: "fix the popover",
272 attachments: [
273 {
274 type: "image",
275 id: "steering-screenshot",
276 name: "image.png",
277 mimeType: "image/png",
278 sizeBytes: 10,
279 },
280 ],
281 dispatchMode: mailbox
282 ? { type: "queue_after_active" }
283 : { type: "steer_active", targetRunId: first.runId },
284 createdBy: mailbox ? "agent" : "user",
285 creationSource: mailbox ? "server" : "web",
286 ...(mailbox
287 ? {
288 delegatedCompletion: {
289 parentRunId: first.runId,
290 generation: 1,
291 taskIds: [taskId],
292 },
293 }
294 : {}),
295 });
296 if (timing !== "before dispatch") yield* dispatchSteer;
297 if (timing === "after delivery") yield* worker.drain();
298 const delivery =
299 timing === "during delivery" ? yield* worker.runOnce.pipe(Effect.forkScoped) : null;
300 if (delivery !== null) yield* Deferred.await(steerEntered);
301 const completed = yield* watch(
302 (event) =>
303 event.type === "run.updated" &&
304 event.payload.id === first.runId &&
305 event.payload.status === "waiting",
306 );
307 const projection = yield* orchestrator.getThreadProjection(threadId);
308 const turn = projection.providerTurns[0]!;
309 yield* Queue.offer(events, {
310 type: "provider_turn.updated",
311 driver,
312 providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now },
313 });
314 yield* Queue.offer(events, {
315 type: "turn.terminal",
316 driver,
317 providerThreadId: turn.providerThreadId,
318 providerTurnId: turn.id,
319 runOrdinal: first.runOrdinal,
320 status: "completed",
321 failure: null,
322 threadDisposition: "reusable",
323 });
324 yield* Fiber.join(completed);
325 if (delivery !== null) {
326 yield* Deferred.succeed(rejectSteer, undefined);
327 yield* Fiber.join(delivery);
328 }
329 if (timing === "before dispatch") yield* dispatchSteer;
330 yield* worker.drain();
331 yield* orchestrator.resumeQueuedRuns;
332 yield* worker.drain();
333 if (timing === "after delivery") {
334 assert.equal(steerCalls, 1);
335 assert.equal(started.length, 1);
336 if (mailbox) {
337 const delivered = yield* orchestrator.getThreadProjection(threadId);
338 assert.equal(delivered.subagents[0]?.completionDelivery?.state, "delivered");
339 assert.equal(delivered.subagents[0]?.completionDelivery?.observedByRunId, null);
340 assert.equal(delivered.runs[0]?.delegatedCompletion?.delivery, null);
341 assert.equal(
342 delivered.turnItems.filter((item) => item.type === "notification").length,
343 1,
344 );
345 yield* orchestrator.dispatch({
346 type: "notification.delivery.accept",
347 commandId: CommandId.make("duplicate-acceptance"),
348 threadId,
349 messageId,
350 });
351 yield* worker.drain();
352 assert.equal(steerCalls, 1);
353 yield* orchestrator.dispatch({
354 type: "delegated_task.completion-delivery.acknowledge",
355 commandId: CommandId.make("read-result"),
356 parentThreadId: threadId,
357 taskId,
358 observedByRunId: first.runId,
359 });
360 const acknowledged = yield* orchestrator.getThreadProjection(threadId);
361 assert.equal(acknowledged.subagents[0]?.completionDelivery?.state, "acknowledged");
362 }
363 return;
364 }
365 assert.equal(started.length, 2);
366 assert.equal(started[1]?.message.messageId, messageId);
367 if (mailbox) assert.include(started[1]?.message.text ?? "", String(taskId));
368 else assert.equal(started[1]?.message.text, "fix the popover");
369 assert.deepEqual(started[1]?.message.attachments, [
370 {
371 type: "image",
372 id: "steering-screenshot",
373 name: "image.png",
374 mimeType: "image/png",
375 sizeBytes: 10,
376 },
377 ]);
378 assert.equal(steerCalls, timing === "during delivery" ? 1 : 0);
379 const final = yield* orchestrator.getThreadProjection(threadId);
380 assert.equal(final.messages.filter((message) => message.id === messageId).length, 1);
381 assert.equal(
382 final.messages.find((message) => message.id === messageId)?.runId,
383 started[1]?.runId,
384 );
385 assert.equal(
386 final.turnItems.filter((item) =>
387 mailbox
388 ? item.type === "notification"
389 : item.type === "user_message" && item.messageId === messageId,
390 ).length,
391 1,
392 );
393 yield* worker.drain();
394 assert.equal(started.length, 2);
395 }).pipe(
396 Effect.provide(
397 ProviderReplayHarness.layerWithRegistry(
398 { name: `steering-completion-${timing}` },
399 ProviderAdapterRegistry.layerSingle(adapter),
400 { runEffectWorker: false },
401 ),
402 ),
403 );
404 }),
405 ),
406);
407
408// Claude and Pi steer live but cannot interrupt-and-restart, so a changed
409// selection waits for the next turn as the thread's saved selection.
410const runSelection = {
411 instanceId,
412 model: "test-model",
413 options: [{ id: "effort", value: "xhigh" }],
414};
415const composerSelection = {
416 ...runSelection,
417 options: [...runSelection.options, { id: "fastMode", value: false }],
418};
419
420const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* (name: string) {
421 const cwd = yield* checkpointWorkspace(name);
422 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
423 const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = [];
424 const steered: string[] = [];
425 const capabilities = {
426 ...CodexProviderCapabilitiesV2,
427 turns: {
428 ...CodexProviderCapabilitiesV2.turns,
429 supportsActiveSteering: true,
430 supportsSteeringByInterruptRestart: false,
431 },
432 };
433 const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
434 instanceId,
435 driver,
436 getCapabilities: () => Effect.succeed(capabilities),
437 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
438 openSession: (input) =>
439 Effect.gen(function* () {
440 const now = yield* DateTime.now;
441 return {
442 instanceId,
443 driver,
444 providerSessionId: input.providerSessionId,
445 providerSession: {
446 id: input.providerSessionId,
447 driver,
448 providerInstanceId: instanceId,
449 status: "ready",
450 cwd,
451 model: runSelection.model,
452 capabilities,
453 createdAt: now,
454 updatedAt: now,
455 lastError: null,
456 },
457 events: Stream.fromQueue(events),
458 ensureThread: ({ threadId }) =>
459 Effect.succeed({
460 id: ProviderThreadId.make(`provider-thread:${threadId}`),
461 driver,
462 providerInstanceId: instanceId,
463 providerSessionId: input.providerSessionId,
464 appThreadId: threadId,
465 ownerNodeId: null,
466 nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
467 nativeConversationHeadRef: null,
468 status: "idle",
469 firstRunOrdinal: null,
470 lastRunOrdinal: null,
471 handoffIds: [],
472 forkedFrom: null,
473 createdAt: now,
474 updatedAt: now,
475 }),
476 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
477 startTurn: (turn) =>
478 Effect.gen(function* () {
479 started.push(turn);
480 yield* Queue.offer(events, {
481 type: "provider_turn.updated",
482 driver,
483 providerTurn: {
484 id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`),
485 providerThreadId: turn.providerThread.id,
486 nodeId: turn.rootNodeId,
487 runAttemptId: turn.attemptId,
488 nativeTurnRef: {
489 driver,
490 nativeId: `native:${turn.attemptId}`,
491 strength: "strong",
492 },
493 ordinal: turn.providerTurnOrdinal,
494 status: "running",
495 startedAt: now,
496 completedAt: null,
497 },
498 });
499 }),
500 steerTurn: (turn) =>
501 Effect.sync(() => {
502 steered.push(turn.message.text);
503 }),
504 interruptTurn: () => Effect.void,
505 respondToRuntimeRequest: () => Effect.void,
506 readThreadSnapshot: () => Effect.die("unused"),
507 rollbackThread: () => Effect.die("unused"),
508 forkThread: () => Effect.die("unused"),
509 };
510 }),
511 };
512 const layer = ProviderReplayHarness.layerWithRegistry(
513 { name },
514 ProviderAdapterRegistry.layerSingle(adapter),
515 { runEffectWorker: false },
516 );
517 // Creates the thread and starts its first turn on `runSelection`.
518 const startFirstTurn = Effect.gen(function* () {
519 const orchestrator = yield* Orchestrator.OrchestratorV2;
520 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
521 const threadId = ThreadId.make(`thread:${name}`);
522 yield* orchestrator.dispatch({
523 type: "thread.create",
524 commandId: CommandId.make("create"),
525 threadId,
526 projectId: ProjectId.make(`project:${name}`),
527 title: "Steer with changed options",
528 modelSelection: runSelection,
529 runtimeMode: "full-access",
530 interactionMode: "default",
531 branch: null,
532 worktreePath: cwd,
533 createdBy: "user",
534 creationSource: "web",
535 });
536 const running = yield* orchestrator.streamDomainEvents.pipe(
537 Stream.filter(
538 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
539 ),
540 Stream.take(1),
541 Stream.runDrain,
542 Effect.forkScoped,
543 );
544 yield* orchestrator.dispatch({
545 type: "message.dispatch",
546 commandId: CommandId.make("first"),
547 threadId,
548 messageId: MessageId.make("message:first"),
549 text: "first",
550 attachments: [],
551 modelSelection: runSelection,
552 dispatchMode: { type: "start_immediately" },
553 createdBy: "user",
554 creationSource: "web",
555 });
556 yield* worker.drain();
557 yield* Fiber.join(running);
558 return threadId;
559 });
560 return { events, started, steered, layer, startFirstTurn };
561});
562
563it.effect("steers a changed turn-scoped selection into a provider that cannot restart", () =>
564 Effect.scoped(
565 Effect.gen(function* () {
566 const { started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness(
567 "steering-selection-change",
568 );
569 yield* Effect.gen(function* () {
570 const orchestrator = yield* Orchestrator.OrchestratorV2;
571 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
572 const threadId = yield* startFirstTurn;
573 const steer = (id: string, selection: ModelSelection) =>
574 orchestrator
575 .dispatch({
576 type: "message.dispatch",
577 commandId: CommandId.make(id),
578 threadId,
579 messageId: MessageId.make(`message:${id}`),
580 text: id,
581 attachments: [],
582 modelSelection: selection,
583 dispatchMode: { type: "steer_active", targetRunId: started[0]!.runId },
584 createdBy: "user",
585 creationSource: "web",
586 })
587 .pipe(Effect.andThen(worker.drain()));
588
589 yield* steer("steer-changed", composerSelection);
590 const changed = yield* orchestrator.getThreadProjection(threadId);
591 assert.deepEqual(steered, ["steer-changed"]);
592 assert.equal(started.length, 1);
593 assert.lengthOf(changed.attempts, 1);
594 assert.equal(changed.runs[0]?.status, "running");
595 assert.deepEqual(changed.runs[0]?.modelSelection, runSelection);
596 assert.deepEqual(changed.thread.modelSelection, composerSelection);
597
598 // Choosing the running run's selection again replaces the saved choice.
599 yield* steer("steer-reverted", runSelection);
600 const reverted = yield* orchestrator.getThreadProjection(threadId);
601 assert.deepEqual(steered, ["steer-changed", "steer-reverted"]);
602 assert.lengthOf(reverted.attempts, 1);
603 assert.deepEqual(reverted.thread.modelSelection, runSelection);
604
605 // The saved choice moves to another instance while the run keeps going.
606 // Steering with the run's selection brings the thread's instance back too.
607 const otherSelection = {
608 instanceId: ProviderInstanceId.make("codex-work"),
609 model: "other",
610 };
611 const sink = yield* EventSink.EventSinkV2;
612 yield* sink.write({
613 events: [
614 {
615 id: EventId.make("switched-away"),
616 type: "thread.provider-switched",
617 threadId,
618 providerInstanceId: otherSelection.instanceId,
619 occurredAt: yield* DateTime.now,
620 payload: {
621 ...reverted.thread,
622 providerInstanceId: otherSelection.instanceId,
623 modelSelection: otherSelection,
624 },
625 },
626 ],
627 });
628 yield* steer("steer-back", runSelection);
629 const back = yield* orchestrator.getThreadProjection(threadId);
630 assert.deepEqual(steered, ["steer-changed", "steer-reverted", "steer-back"]);
631 assert.equal(back.thread.providerInstanceId, instanceId);
632 assert.deepEqual(back.thread.modelSelection, runSelection);
633 }).pipe(Effect.provide(layer));
634 }),
635 ),
636);
637
638it.effect("starts a steer that missed the turn on the saved next-turn selection", () =>
639 Effect.scoped(
640 Effect.gen(function* () {
641 const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness(
642 "steering-selection-follow-up",
643 );
644 yield* Effect.gen(function* () {
645 const orchestrator = yield* Orchestrator.OrchestratorV2;
646 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
647 const threadId = yield* startFirstTurn;
648 const first = started[0]!;
649 yield* orchestrator.dispatch({
650 type: "message.dispatch",
651 commandId: CommandId.make("steer"),
652 threadId,
653 messageId: MessageId.make("message:steer"),
654 text: "late steer",
655 attachments: [],
656 modelSelection: composerSelection,
657 dispatchMode: { type: "steer_active", targetRunId: first.runId },
658 createdBy: "user",
659 creationSource: "web",
660 });
661 // The turn ends before the worker delivers the steer, so the steer
662 // becomes a follow-up turn.
663 const settled = yield* orchestrator.streamDomainEvents.pipe(
664 Stream.filter(
665 (event) =>
666 event.type === "run.updated" &&
667 event.payload.id === first.runId &&
668 event.payload.status === "waiting",
669 ),
670 Stream.take(1),
671 Stream.runDrain,
672 Effect.forkScoped,
673 );
674 const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!;
675 yield* Queue.offer(events, {
676 type: "provider_turn.updated",
677 driver,
678 providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now },
679 });
680 yield* Queue.offer(events, {
681 type: "turn.terminal",
682 driver,
683 providerThreadId: turn.providerThreadId,
684 providerTurnId: turn.id,
685 runOrdinal: first.runOrdinal,
686 status: "completed",
687 failure: null,
688 threadDisposition: "reusable",
689 });
690 yield* Fiber.join(settled);
691 yield* worker.drain();
692 yield* orchestrator.resumeQueuedRuns;
693 yield* worker.drain();
694
695 const projection = yield* orchestrator.getThreadProjection(threadId);
696 assert.deepEqual(steered, []);
697 assert.equal(started.length, 2);
698 assert.equal(started[1]?.message.messageId, MessageId.make("message:steer"));
699 assert.deepEqual(started[1]?.modelSelection, composerSelection);
700 assert.deepEqual(projection.thread.modelSelection, composerSelection);
701 }).pipe(Effect.provide(layer));
702 }),
703 ),
704);