1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 EventId,
5 MessageId,
6 type ModelSelection,
7 type OrchestrationV2ProviderCapabilities,
8 type OrchestrationV2ProviderSession,
9 type OrchestrationV2ProviderThread,
10 ProjectId,
11 ProviderDriverKind,
12 ProviderInstanceId,
13 ProviderSessionId,
14 ProviderThreadId,
15 ProviderTurnId,
16 type RunId,
17 ThreadId,
18} from "@t3tools/contracts";
19import * as DateTime from "effect/DateTime";
20import * as Effect from "effect/Effect";
21import * as Fiber from "effect/Fiber";
22import * as Layer from "effect/Layer";
23import * as Queue from "effect/Queue";
24import * as Ref from "effect/Ref";
25import * as Stream from "effect/Stream";
26
27import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
28import * as EffectWorker from "./EffectWorker.ts";
29import * as EventSink from "./EventSink.ts";
30import * as Orchestrator from "./Orchestrator.ts";
31import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
32import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
33import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
34import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
35
36const driver = ProviderDriverKind.make("codex");
37const providerInstanceId = ProviderInstanceId.make("codex-restart-test");
38const initialSelection = {
39 instanceId: providerInstanceId,
40 model: "restart-model-a",
41} satisfies ModelSelection;
42const replacementSelection = {
43 instanceId: providerInstanceId,
44 model: "restart-model-b",
45} satisfies ModelSelection;
46const seedSelection = {
47 instanceId: providerInstanceId,
48 model: "seed-model",
49} satisfies ModelSelection;
50const handoffDriver = ProviderDriverKind.make("claudeAgent");
51const handoffProviderInstanceId = ProviderInstanceId.make("claude-handoff-test");
52const handoffSelection = {
53 instanceId: handoffProviderInstanceId,
54 model: "handoff-model",
55} satisfies ModelSelection;
56const pooledCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2;
57const exclusiveCapabilities: OrchestrationV2ProviderCapabilities = {
58 ...CodexProviderCapabilitiesV2,
59 sessions: {
60 ...CodexProviderCapabilitiesV2.sessions,
61 supportsMultipleProviderThreadsPerSession: false,
62 supportsModelSwitchInSession: false,
63 },
64};
65
66interface ActiveTurn {
67 readonly input: ProviderAdapter.ProviderAdapterV2TurnInput;
68 readonly providerTurnId: ProviderTurnId;
69}
70
71interface RestartAdapterState {
72 readonly activeTurn: ActiveTurn | null;
73 readonly opened: ReadonlyArray<{
74 readonly model: string | null;
75 readonly cwd: string | null;
76 }>;
77 readonly started: ReadonlyArray<{
78 readonly model: string;
79 readonly cwd: string | null;
80 readonly attemptId: string;
81 }>;
82 readonly closedSessionCount: number;
83 readonly failedReplacementOpen: boolean;
84}
85
86function makeRestartAdapter(
87 state: Ref.Ref<RestartAdapterState>,
88 sessionCapabilities: OrchestrationV2ProviderCapabilities = pooledCapabilities,
89 providerInstanceId = initialSelection.instanceId,
90): ProviderAdapter.ProviderAdapterV2["Service"] {
91 return {
92 instanceId: providerInstanceId,
93 driver,
94 getCapabilities: () => Effect.succeed(sessionCapabilities),
95 planSelectionTransition: ({ current, target }) =>
96 Effect.succeed(
97 current.model === target.model
98 ? ({ type: "apply_on_next_turn" } as const)
99 : ({ type: "restart_session" } as const),
100 ),
101 openSession: (sessionInput) =>
102 Effect.gen(function* () {
103 const failThisOpen = yield* Ref.modify(state, (current) => {
104 const shouldFail =
105 sessionInput.modelSelection.model === replacementSelection.model &&
106 !current.failedReplacementOpen;
107 return [
108 shouldFail,
109 {
110 ...current,
111 failedReplacementOpen: current.failedReplacementOpen || shouldFail,
112 opened: [
113 ...current.opened,
114 {
115 model: sessionInput.modelSelection.model,
116 cwd: sessionInput.runtimePolicy.cwd,
117 },
118 ],
119 },
120 ] as const;
121 });
122 if (failThisOpen) {
123 return yield* new ProviderAdapter.ProviderAdapterOpenSessionError({
124 driver,
125 providerSessionId: sessionInput.providerSessionId,
126 cause: "simulated replacement open failure",
127 });
128 }
129
130 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
131 const now = yield* DateTime.now;
132 const providerSession: OrchestrationV2ProviderSession = {
133 id: sessionInput.providerSessionId,
134 driver,
135 providerInstanceId,
136 status: "ready",
137 cwd: sessionInput.runtimePolicy.cwd ?? "/fallback",
138 model: sessionInput.modelSelection.model,
139 capabilities: sessionCapabilities,
140 createdAt: now,
141 updatedAt: now,
142 lastError: null,
143 };
144 yield* Effect.addFinalizer(() =>
145 Ref.update(state, (current) => ({
146 ...current,
147 closedSessionCount: current.closedSessionCount + 1,
148 })),
149 );
150
151 const publishTerminal = (active: ActiveTurn, status: "completed" | "interrupted") =>
152 Effect.gen(function* () {
153 const occurredAt = yield* DateTime.now;
154 yield* Queue.offer(events, {
155 type: "provider_turn.updated",
156 driver,
157 providerTurn: {
158 id: active.providerTurnId,
159 providerThreadId: active.input.providerThread.id,
160 nodeId: active.input.rootNodeId,
161 runAttemptId: active.input.attemptId,
162 nativeTurnRef: {
163 driver,
164 nativeId: `native:${active.providerTurnId}`,
165 strength: "strong",
166 },
167 ordinal: active.input.providerTurnOrdinal,
168 status,
169 startedAt: occurredAt,
170 completedAt: occurredAt,
171 },
172 });
173 yield* Queue.offer(events, {
174 type: "turn.terminal",
175 driver,
176 providerThreadId: active.input.providerThread.id,
177 providerTurnId: active.providerTurnId,
178 runOrdinal: active.input.runOrdinal,
179 status,
180 failure: null,
181 threadDisposition: "reusable",
182 });
183 });
184
185 return {
186 instanceId: providerInstanceId,
187 driver,
188 providerSessionId: sessionInput.providerSessionId,
189 providerSession,
190 events: Stream.fromQueue(events),
191 ensureThread: (threadInput) =>
192 Effect.gen(function* () {
193 const createdAt = yield* DateTime.now;
194 return {
195 id: ProviderThreadId.make(`provider-thread:${threadInput.threadId}`),
196 driver,
197 providerInstanceId,
198 providerSessionId: sessionInput.providerSessionId,
199 appThreadId: threadInput.threadId,
200 ownerNodeId: null,
201 nativeThreadRef: {
202 driver,
203 nativeId: `native-thread:${threadInput.threadId}`,
204 strength: "strong",
205 },
206 nativeConversationHeadRef: null,
207 status: "idle",
208 firstRunOrdinal: null,
209 lastRunOrdinal: null,
210 handoffIds: [],
211 forkedFrom: null,
212 createdAt,
213 updatedAt: createdAt,
214 } satisfies OrchestrationV2ProviderThread;
215 }),
216 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
217 startTurn: (input) =>
218 Effect.gen(function* () {
219 yield* Ref.update(state, (current) => ({
220 ...current,
221 started: [
222 ...current.started,
223 {
224 model: input.modelSelection.model,
225 cwd: input.runtimePolicy.cwd,
226 attemptId: input.attemptId,
227 },
228 ],
229 }));
230 const active = {
231 input,
232 providerTurnId: ProviderTurnId.make(`provider-turn:${input.attemptId}`),
233 } satisfies ActiveTurn;
234 if (input.modelSelection.model === initialSelection.model) {
235 const occurredAt = yield* DateTime.now;
236 yield* Ref.update(state, (current) => ({ ...current, activeTurn: active }));
237 yield* Queue.offer(events, {
238 type: "provider_turn.updated",
239 driver,
240 providerTurn: {
241 id: active.providerTurnId,
242 providerThreadId: input.providerThread.id,
243 nodeId: input.rootNodeId,
244 runAttemptId: input.attemptId,
245 nativeTurnRef: {
246 driver,
247 nativeId: `native:${active.providerTurnId}`,
248 strength: "strong",
249 },
250 ordinal: input.providerTurnOrdinal,
251 status: "running",
252 startedAt: occurredAt,
253 completedAt: null,
254 },
255 });
256 return;
257 }
258 yield* publishTerminal(active, "completed");
259 }),
260 steerTurn: () => Effect.void,
261 interruptTurn: () =>
262 Effect.gen(function* () {
263 const active = (yield* Ref.get(state)).activeTurn;
264 if (active !== null) {
265 const updatedAt = yield* DateTime.now;
266 yield* Queue.offer(events, {
267 type: "provider_thread.updated",
268 driver,
269 providerThread: {
270 ...active.input.providerThread,
271 status: "idle",
272 updatedAt,
273 },
274 });
275 yield* publishTerminal(active, "interrupted");
276 yield* Ref.update(state, (current) => ({ ...current, activeTurn: null }));
277 }
278 }),
279 respondToRuntimeRequest: () => Effect.void,
280 readThreadSnapshot: () => Effect.die("unused readThreadSnapshot"),
281 rollbackThread: () => Effect.die("unused rollbackThread"),
282 forkThread: () => Effect.die("unused forkThread"),
283 };
284 }),
285 };
286}
287
288function makeCompletingHandoffAdapter(
289 startCount: Ref.Ref<number>,
290): ProviderAdapter.ProviderAdapterV2["Service"] {
291 return {
292 instanceId: handoffProviderInstanceId,
293 driver: handoffDriver,
294 getCapabilities: () => Effect.succeed(exclusiveCapabilities),
295 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
296 openSession: (sessionInput) =>
297 Effect.gen(function* () {
298 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
299 const now = yield* DateTime.now;
300 return {
301 instanceId: handoffProviderInstanceId,
302 driver: handoffDriver,
303 providerSessionId: sessionInput.providerSessionId,
304 providerSession: {
305 id: sessionInput.providerSessionId,
306 driver: handoffDriver,
307 providerInstanceId: handoffProviderInstanceId,
308 status: "ready",
309 cwd: sessionInput.runtimePolicy.cwd ?? "/fallback",
310 model: sessionInput.modelSelection.model,
311 capabilities: exclusiveCapabilities,
312 createdAt: now,
313 updatedAt: now,
314 lastError: null,
315 },
316 events: Stream.fromQueue(events),
317 ensureThread: (threadInput) =>
318 Effect.gen(function* () {
319 const createdAt = yield* DateTime.now;
320 return {
321 id: ProviderThreadId.make(`provider-thread:handoff:${threadInput.threadId}`),
322 driver: handoffDriver,
323 providerInstanceId: handoffProviderInstanceId,
324 providerSessionId: sessionInput.providerSessionId,
325 appThreadId: threadInput.threadId,
326 ownerNodeId: null,
327 nativeThreadRef: {
328 driver: handoffDriver,
329 nativeId: `native-thread:handoff:${threadInput.threadId}`,
330 strength: "strong",
331 },
332 nativeConversationHeadRef: null,
333 status: "idle",
334 firstRunOrdinal: null,
335 lastRunOrdinal: null,
336 handoffIds: [],
337 forkedFrom: null,
338 createdAt,
339 updatedAt: createdAt,
340 } satisfies OrchestrationV2ProviderThread;
341 }),
342 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
343 startTurn: (input) =>
344 Effect.gen(function* () {
345 yield* Ref.update(startCount, (count) => count + 1);
346 const occurredAt = yield* DateTime.now;
347 const providerTurnId = ProviderTurnId.make(
348 `provider-turn:handoff:${input.attemptId}`,
349 );
350 yield* Queue.offer(events, {
351 type: "provider_turn.updated",
352 driver: handoffDriver,
353 providerTurn: {
354 id: providerTurnId,
355 providerThreadId: input.providerThread.id,
356 nodeId: input.rootNodeId,
357 runAttemptId: input.attemptId,
358 nativeTurnRef: {
359 driver: handoffDriver,
360 nativeId: `native:${providerTurnId}`,
361 strength: "strong",
362 },
363 ordinal: input.providerTurnOrdinal,
364 status: "completed",
365 startedAt: occurredAt,
366 completedAt: occurredAt,
367 },
368 });
369 yield* Queue.offer(events, {
370 type: "turn.terminal",
371 driver: handoffDriver,
372 providerThreadId: input.providerThread.id,
373 providerTurnId,
374 runOrdinal: input.runOrdinal,
375 status: "completed",
376 failure: null,
377 threadDisposition: "reusable",
378 });
379 }),
380 steerTurn: () => Effect.void,
381 interruptTurn: () => Effect.void,
382 respondToRuntimeRequest: () => Effect.void,
383 readThreadSnapshot: () => Effect.die("unused readThreadSnapshot"),
384 rollbackThread: () => Effect.die("unused rollbackThread"),
385 forkThread: () => Effect.die("unused forkThread"),
386 };
387 }),
388 };
389}
390
391it.live("retries a failed selection restart session open before completing", () =>
392 Effect.scoped(
393 Effect.gen(function* () {
394 const cwd = yield* checkpointWorkspace("selection-restart-lifecycle");
395 const threadId = ThreadId.make("thread:selection-restart-lifecycle");
396 const state = yield* Ref.make<RestartAdapterState>({
397 activeTurn: null,
398 opened: [],
399 started: [],
400 closedSessionCount: 0,
401 failedReplacementOpen: false,
402 });
403 const layerRegistry = ProviderAdapterRegistry.layerSingle(makeRestartAdapter(state));
404
405 const result = yield* Effect.gen(function* () {
406 const orchestrator = yield* Orchestrator.OrchestratorV2;
407 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
408 const eventSink = yield* EventSink.EventSinkV2;
409
410 yield* orchestrator.dispatch({
411 type: "thread.create",
412 createdBy: "user",
413 creationSource: "web",
414 commandId: CommandId.make("command:selection-restart:create"),
415 threadId,
416 projectId: ProjectId.make("project:selection-restart"),
417 title: "Selection restart",
418 modelSelection: initialSelection,
419 runtimeMode: "full-access",
420 interactionMode: "default",
421 branch: null,
422 worktreePath: cwd,
423 });
424 const running = yield* eventSink
425 .stream({
426 threadId,
427 afterSequence: yield* eventSink.latestSequence({ threadId }),
428 })
429 .pipe(
430 Stream.map((stored) => stored.event),
431 Stream.filter(
432 (event) =>
433 event.type === "provider-turn.updated" && event.payload.status === "running",
434 ),
435 Stream.take(1),
436 Stream.runDrain,
437 Effect.forkScoped,
438 );
439 yield* orchestrator.dispatch({
440 type: "message.dispatch",
441 createdBy: "user",
442 creationSource: "web",
443 commandId: CommandId.make("command:selection-restart:first"),
444 threadId,
445 messageId: MessageId.make("message:selection-restart:first"),
446 text: "first",
447 attachments: [],
448 modelSelection: initialSelection,
449 dispatchMode: { type: "start_immediately" },
450 });
451 yield* Fiber.join(running);
452 yield* worker.drain();
453 const activeProjection = yield* orchestrator.getThreadProjection(threadId);
454 assert.isTrue(activeProjection.providerTurns.some((turn) => turn.status === "running"));
455 const activeRunId = activeProjection.runs[0]?.id;
456 if (activeRunId === undefined) {
457 return yield* Effect.die("active restart test run is missing");
458 }
459
460 const completed = yield* eventSink
461 .stream({
462 threadId,
463 afterSequence: yield* eventSink.latestSequence({ threadId }),
464 })
465 .pipe(
466 Stream.map((stored) => stored.event),
467 Stream.filter(
468 (event) => event.type === "run.updated" && event.payload.status === "completed",
469 ),
470 Stream.take(1),
471 Stream.runDrain,
472 Effect.forkScoped,
473 );
474 yield* orchestrator.dispatch({
475 type: "message.dispatch",
476 createdBy: "user",
477 creationSource: "web",
478 commandId: CommandId.make("command:selection-restart:second"),
479 threadId,
480 messageId: MessageId.make("message:selection-restart:second"),
481 text: "second",
482 attachments: [],
483 modelSelection: replacementSelection,
484 dispatchMode: { type: "restart_active", targetRunId: activeRunId },
485 });
486 yield* Fiber.join(completed);
487 yield* worker.drain();
488 return {
489 projection: yield* orchestrator.getThreadProjection(threadId),
490 captured: yield* Ref.get(state),
491 };
492 }).pipe(
493 Effect.provide(
494 ProviderReplayHarness.layerWithRegistry(
495 { name: "selection-restart-lifecycle" },
496 layerRegistry,
497 ),
498 ),
499 );
500 const { projection, captured } = result;
501
502 assert.lengthOf(projection.runs, 1);
503 assert.lengthOf(projection.attempts, 2);
504 assert.deepEqual(
505 projection.attempts.map((attempt) => attempt.status),
506 ["superseded", "completed"],
507 );
508 assert.deepEqual(
509 projection.providerTurns.map((turn) => turn.status),
510 ["interrupted", "completed"],
511 );
512 assert.isFalse(
513 projection.turnItems.some(
514 (item) => item.type === "run_interrupt_request" || item.type === "run_interrupt_result",
515 ),
516 "selection restart supersede must not project hard-Stop interrupt items",
517 );
518 assert.equal(projection.runs.at(-1)?.modelSelection.model, replacementSelection.model);
519 assert.equal(projection.runs.at(-1)?.status, "completed");
520 assert.isTrue(captured.failedReplacementOpen);
521 // The old pooled process remains available to its other threads; this
522 // thread moved to a freshly allocated replacement session.
523 assert.equal(captured.closedSessionCount, 0);
524 assert.deepEqual(
525 captured.opened.map((open) => [open.model, open.cwd]),
526 [
527 [initialSelection.model, cwd],
528 [replacementSelection.model, cwd],
529 [replacementSelection.model, cwd],
530 ],
531 );
532 assert.deepEqual(
533 captured.started.map((turn) => [turn.model, turn.cwd]),
534 [
535 [initialSelection.model, cwd],
536 [replacementSelection.model, cwd],
537 ],
538 );
539 assert.notEqual(
540 projection.providerSessions[0]?.id,
541 projection.providerThreads[0]?.providerSessionId,
542 );
543 assert.equal(
544 projection.providerThreads[0]?.providerSessionId,
545 projection.providerSessions.find((session) => session.model === replacementSelection.model)
546 ?.id,
547 );
548 }),
549 ),
550);
551
552it.live.each(["stopped", "error"] as const)(
553 "restarts the live session on a model change when a newer %s session record exists",
554 (deadStatus) =>
555 Effect.scoped(
556 Effect.gen(function* () {
557 const name = `selection-restart-dead-${deadStatus}`;
558 const cwd = yield* checkpointWorkspace(name);
559 const threadId = ThreadId.make(`thread:${name}`);
560 const state = yield* Ref.make<RestartAdapterState>({
561 activeTurn: null,
562 opened: [],
563 started: [],
564 closedSessionCount: 0,
565 // The dead record is seeded directly, so the adapter's one-shot
566 // simulated replacement-open failure is skipped.
567 failedReplacementOpen: true,
568 });
569 const layerRegistry = ProviderAdapterRegistry.layerSingle(
570 makeRestartAdapter(state, exclusiveCapabilities),
571 );
572
573 const result = yield* Effect.gen(function* () {
574 const orchestrator = yield* Orchestrator.OrchestratorV2;
575 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
576 const eventSink = yield* EventSink.EventSinkV2;
577 const dispatch = (step: string, modelSelection: ModelSelection) =>
578 Effect.gen(function* () {
579 const terminal = yield* orchestrator.streamDomainEvents.pipe(
580 Stream.filter(
581 (event) => event.type === "run.updated" && event.payload.status === "completed",
582 ),
583 Stream.take(1),
584 Stream.runDrain,
585 Effect.forkScoped,
586 );
587 yield* orchestrator.dispatch({
588 type: "message.dispatch",
589 createdBy: "user",
590 creationSource: "web",
591 commandId: CommandId.make(`${name}:${step}`),
592 threadId,
593 messageId: MessageId.make(`${name}:${step}`),
594 text: step,
595 attachments: [],
596 modelSelection,
597 dispatchMode: { type: "start_immediately" },
598 });
599 yield* worker.drain();
600 yield* Fiber.join(terminal);
601 yield* worker.drain();
602 return yield* orchestrator.getThreadProjection(threadId);
603 });
604 yield* orchestrator.dispatch({
605 type: "thread.create",
606 createdBy: "user",
607 creationSource: "web",
608 commandId: CommandId.make(`${name}:create`),
609 threadId,
610 projectId: ProjectId.make(`project:${name}`),
611 title: name,
612 modelSelection: seedSelection,
613 runtimeMode: "full-access",
614 interactionMode: "default",
615 branch: null,
616 worktreePath: cwd,
617 });
618 const first = yield* dispatch("first", seedSelection);
619 const liveSession = first.providerSessions.find(
620 (session) => session.status !== "stopped" && session.status !== "error",
621 );
622 assert.isDefined(liveSession);
623
624 // Dead session records stay bound in the projection until
625 // detachment. A stale stopped/error record written after the
626 // live session attached must not hide it.
627 const deadAt = yield* DateTime.now;
628 const deadSession: OrchestrationV2ProviderSession = {
629 id: ProviderSessionId.make(`session:${name}:dead`),
630 driver,
631 providerInstanceId,
632 status: "ready",
633 cwd,
634 model: seedSelection.model,
635 capabilities: exclusiveCapabilities,
636 createdAt: deadAt,
637 updatedAt: deadAt,
638 lastError: null,
639 };
640 yield* eventSink.write({
641 events: [
642 {
643 id: EventId.make(`event:${name}:dead-attached`),
644 type: "provider-session.attached",
645 threadId,
646 driver,
647 providerInstanceId,
648 occurredAt: deadAt,
649 payload: deadSession,
650 },
651 {
652 id: EventId.make(`event:${name}:dead-updated`),
653 type: "provider-session.updated",
654 threadId,
655 driver,
656 providerInstanceId,
657 occurredAt: deadAt,
658 payload: {
659 ...deadSession,
660 status: deadStatus,
661 updatedAt: deadAt,
662 lastError: deadStatus === "error" ? "Simulated session failure." : null,
663 },
664 },
665 ],
666 });
667
668 const switchCommandId = CommandId.make(`${name}:switch`);
669 yield* orchestrator.dispatch({
670 type: "thread.model-selection.set",
671 commandId: switchCommandId,
672 threadId,
673 modelSelection: replacementSelection,
674 });
675 yield* worker.drain();
676 const storedSwitchEvents = yield* eventSink
677 .readByCommandId({ commandId: switchCommandId })
678 .pipe(Stream.runCollect);
679 const detachedSessionIds = [...storedSwitchEvents].flatMap((stored) =>
680 stored.event.type === "provider-session.detached"
681 ? [stored.event.payload.providerSessionId]
682 : [],
683 );
684
685 const second = yield* dispatch("second", replacementSelection);
686 return {
687 projection: second,
688 captured: yield* Ref.get(state),
689 liveSessionId: liveSession.id,
690 detachedSessionIds,
691 };
692 }).pipe(Effect.provide(ProviderReplayHarness.layerWithRegistry({ name }, layerRegistry)));
693
694 const { projection, captured } = result;
695 assert.lengthOf(projection.runs, 2);
696 assert.equal(projection.runs[1]?.modelSelection.model, replacementSelection.model);
697 // The exact released session is the older live one, never the newer
698 // dead record.
699 assert.deepEqual(result.detachedSessionIds, [result.liveSessionId]);
700 assert.equal(captured.closedSessionCount, 1);
701 assert.deepEqual(
702 captured.opened.map((open) => open.model),
703 [seedSelection.model, replacementSelection.model],
704 );
705 assert.deepEqual(
706 captured.started.map((turn) => turn.model),
707 [seedSelection.model, replacementSelection.model],
708 );
709 const servingSession = projection.providerSessions.find(
710 (session) =>
711 session.id ===
712 projection.providerThreads.find(
713 (providerThread) => providerThread.id === projection.thread.activeProviderThreadId,
714 )?.providerSessionId,
715 );
716 assert.equal(servingSession?.model, replacementSelection.model);
717 }),
718 ),
719);
720
721it.live("detaches the old provider session after an active provider handoff", () =>
722 Effect.scoped(
723 Effect.gen(function* () {
724 const cwd = yield* checkpointWorkspace("selection-provider-handoff-lifecycle");
725 const threadId = ThreadId.make("thread:selection-provider-handoff-lifecycle");
726 const state = yield* Ref.make<RestartAdapterState>({
727 activeTurn: null,
728 opened: [],
729 started: [],
730 closedSessionCount: 0,
731 failedReplacementOpen: false,
732 });
733 const targetStartCount = yield* Ref.make(0);
734 const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([
735 makeRestartAdapter(state, exclusiveCapabilities),
736 makeCompletingHandoffAdapter(targetStartCount),
737 ]);
738
739 const result = yield* Effect.gen(function* () {
740 const orchestrator = yield* Orchestrator.OrchestratorV2;
741 yield* orchestrator.dispatch({
742 type: "thread.create",
743 createdBy: "user",
744 creationSource: "web",
745 commandId: CommandId.make("command:selection-handoff:create"),
746 threadId,
747 projectId: ProjectId.make("project:selection-handoff"),
748 title: "Selection provider handoff",
749 modelSelection: initialSelection,
750 runtimeMode: "full-access",
751 interactionMode: "default",
752 branch: null,
753 worktreePath: cwd,
754 });
755 yield* orchestrator.dispatch({
756 type: "message.dispatch",
757 createdBy: "user",
758 creationSource: "web",
759 commandId: CommandId.make("command:selection-handoff:first"),
760 threadId,
761 messageId: MessageId.make("message:selection-handoff:first"),
762 text: "first",
763 attachments: [],
764 modelSelection: initialSelection,
765 dispatchMode: { type: "start_immediately" },
766 });
767 let activeRunId: RunId | null = null;
768 for (let index = 0; index < 1_000; index += 1) {
769 const current = yield* orchestrator.getThreadProjection(threadId);
770 if (current.providerTurns.some((turn) => turn.status === "running")) {
771 activeRunId = current.runs[0]?.id ?? null;
772 break;
773 }
774 yield* Effect.sleep("5 millis");
775 }
776 if (activeRunId === null) {
777 return yield* Effect.die("active provider-handoff run is missing");
778 }
779
780 yield* orchestrator.dispatch({
781 type: "message.dispatch",
782 createdBy: "user",
783 creationSource: "web",
784 commandId: CommandId.make("command:selection-handoff:second"),
785 threadId,
786 messageId: MessageId.make("message:selection-handoff:second"),
787 text: "second",
788 attachments: [],
789 modelSelection: handoffSelection,
790 dispatchMode: { type: "restart_active", targetRunId: activeRunId },
791 });
792 for (let index = 0; index < 1_000; index += 1) {
793 const current = yield* orchestrator.getThreadProjection(threadId);
794 if (current.attempts.length === 2 && current.attempts[1]?.status === "completed") {
795 const captured = yield* Ref.get(state);
796 return { projection: current, captured };
797 }
798 yield* Effect.sleep("5 millis");
799 }
800 const current = yield* orchestrator.getThreadProjection(threadId);
801 const adapterState = yield* Ref.get(state);
802 const capturedTargetStartCount = yield* Ref.get(targetStartCount);
803 yield* Effect.logError("active provider handoff did not complete", {
804 runs: current.runs.map((run) => [run.status, run.activeAttemptId]),
805 attempts: current.attempts.map((attempt) => [attempt.id, attempt.status]),
806 providerTurns: current.providerTurns.map((turn) => [turn.id, turn.status]),
807 providerThreads: current.providerThreads.map((thread) => [
808 thread.providerInstanceId,
809 thread.providerSessionId,
810 thread.status,
811 ]),
812 targetStartCount: capturedTargetStartCount,
813 adapterState,
814 });
815 return yield* Effect.die("active provider handoff did not complete");
816 }).pipe(
817 Effect.provide(
818 ProviderReplayHarness.layerWithRegistry(
819 { name: "selection-provider-handoff-lifecycle" },
820 layerRegistry,
821 ),
822 ),
823 );
824
825 assert.lengthOf(result.projection.runs, 1);
826 assert.lengthOf(result.projection.attempts, 2);
827 assert.equal(result.projection.runs[0]?.providerInstanceId, handoffProviderInstanceId);
828 assert.equal(result.projection.contextHandoffs.length, 1);
829 assert.equal(result.captured.closedSessionCount, 1);
830 assert.equal(yield* Ref.get(targetStartCount), 1);
831 }),
832 ),
833);
834
835it.live.each(["active", "idle", "selection-command", "pooled", "separate-home"] as const)(
836 "preserves native history only for compatible account switches (%s)",
837 (mode) =>
838 Effect.scoped(
839 Effect.gen(function* () {
840 const name = `shared-home-${mode}`;
841 const cwd = yield* checkpointWorkspace(name);
842 const threadId = ThreadId.make(`thread:${name}`);
843 const targetId = ProviderInstanceId.make("codex-shadow-account");
844 const state = yield* Ref.make<RestartAdapterState>({
845 activeTurn: null,
846 opened: [],
847 started: [],
848 closedSessionCount: 0,
849 failedReplacementOpen: true,
850 });
851 const resumes: Array<{
852 instanceId: ProviderInstanceId;
853 nativeId: string | null | undefined;
854 }> = [];
855 const messages: string[] = [];
856 const capabilities = mode === "pooled" ? pooledCapabilities : exclusiveCapabilities;
857 const adapters = [providerInstanceId, targetId].map((instanceId) => {
858 const base = makeRestartAdapter(state, capabilities, instanceId);
859 return {
860 ...base,
861 openSession: (input) =>
862 base.openSession(input).pipe(
863 Effect.map((session) => ({
864 ...session,
865 resumeThread: (input) =>
866 Effect.gen(function* () {
867 resumes.push({
868 instanceId,
869 nativeId: input.providerThread.nativeThreadRef?.nativeId,
870 });
871 return yield* session.resumeThread(input);
872 }),
873 startTurn: (input) =>
874 Effect.gen(function* () {
875 messages.push(input.message.text);
876 yield* session.startTurn(input);
877 }),
878 })),
879 ),
880 } satisfies ProviderAdapter.ProviderAdapterV2["Service"];
881 });
882 const layerRegistry = Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistryV2, {
883 get: (instanceId) =>
884 Effect.succeed(adapters.find((adapter) => adapter.instanceId === instanceId)!),
885 list: () => Effect.succeed([providerInstanceId, targetId]),
886 getMetadata: (instanceId) =>
887 Effect.succeed({
888 driver,
889 continuationKey:
890 mode === "separate-home" ? `codex:home:/${instanceId}` : "codex:home:/shared",
891 enabled: true,
892 capabilities,
893 }),
894 });
895 yield* Effect.gen(function* () {
896 const orchestrator = yield* Orchestrator.OrchestratorV2;
897 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
898 const selection = {
899 ...initialSelection,
900 model: mode === "active" ? initialSelection.model : "complete",
901 };
902 yield* orchestrator.dispatch({
903 type: "thread.create",
904 commandId: CommandId.make(`${name}:create`),
905 threadId,
906 projectId: ProjectId.make(`project:${name}`),
907 title: name,
908 modelSelection: selection,
909 runtimeMode: "full-access",
910 interactionMode: "default",
911 branch: null,
912 worktreePath: cwd,
913 createdBy: "user",
914 creationSource: "web",
915 });
916 const dispatch = (step: string, modelSelection: ModelSelection, targetRunId?: RunId) =>
917 Effect.gen(function* () {
918 const terminal = yield* orchestrator.streamDomainEvents.pipe(
919 Stream.filter((event) =>
920 mode === "active" && step === "first"
921 ? event.type === "provider-turn.updated" && event.payload.status === "running"
922 : event.type === "run.updated" && event.payload.status === "completed",
923 ),
924 Stream.take(1),
925 Stream.runDrain,
926 Effect.forkScoped,
927 );
928 yield* orchestrator.dispatch({
929 type: "message.dispatch",
930 commandId: CommandId.make(`${name}:${step}`),
931 threadId,
932 messageId: MessageId.make(`${name}:${step}`),
933 text: step,
934 attachments: [],
935 modelSelection,
936 dispatchMode:
937 targetRunId === undefined
938 ? { type: "start_immediately" }
939 : { type: "restart_active", targetRunId },
940 createdBy: "user",
941 creationSource: "web",
942 });
943 yield* worker.drain();
944 yield* Fiber.join(terminal);
945 yield* worker.drain();
946 return yield* orchestrator.getThreadProjection(threadId);
947 });
948 const first = yield* dispatch("first", selection);
949 const originalThread = first.providerThreads.find(
950 (thread) => thread.id === first.thread.activeProviderThreadId,
951 )!;
952 const targetSelection = { instanceId: targetId, model: "complete" };
953 if (mode === "selection-command") {
954 yield* orchestrator.dispatch({
955 type: "provider.switch",
956 commandId: CommandId.make(`${name}:switch`),
957 threadId,
958 modelSelection: targetSelection,
959 });
960 yield* worker.drain();
961 }
962 const second = yield* dispatch(
963 "second",
964 targetSelection,
965 mode === "active" ? first.runs[0]!.id : undefined,
966 );
967 const targetThread = second.providerThreads.find(
968 (thread) => thread.id === second.thread.activeProviderThreadId,
969 )!;
970 if (mode === "separate-home") {
971 assert.notEqual(targetThread.id, originalThread.id);
972 assert.lengthOf(second.contextHandoffs, 1);
973 assert.lengthOf(second.contextTransfers, 1);
974 assert.isEmpty(resumes);
975 assert.notEqual(messages[1], "second");
976 return;
977 }
978 assert.equal(targetThread.id, originalThread.id);
979 assert.deepEqual(targetThread.nativeThreadRef, originalThread.nativeThreadRef);
980 assert.equal(targetThread.providerInstanceId, targetId);
981 assert.notEqual(targetThread.providerSessionId, originalThread.providerSessionId);
982 assert.isEmpty(second.contextHandoffs);
983 assert.isEmpty(second.contextTransfers);
984 assert.deepEqual(resumes, [
985 { instanceId: targetId, nativeId: originalThread.nativeThreadRef?.nativeId },
986 ]);
987 assert.deepEqual(messages, ["first", "second"]);
988 assert.equal((yield* Ref.get(state)).closedSessionCount, mode === "pooled" ? 0 : 1);
989 if (mode === "pooled") {
990 assert.isFalse(
991 second.providerSessions.some(
992 (session) => session.id === originalThread.providerSessionId,
993 ),
994 );
995 }
996 const third = yield* dispatch("third", { ...selection, model: "complete" });
997 const returnedThread = third.providerThreads.find(
998 (thread) => thread.id === third.thread.activeProviderThreadId,
999 )!;
1000 assert.equal(returnedThread.id, originalThread.id);
1001 assert.deepEqual(returnedThread.nativeThreadRef, originalThread.nativeThreadRef);
1002 assert.equal(returnedThread.providerInstanceId, providerInstanceId);
1003 assert.isEmpty(third.contextHandoffs);
1004 assert.deepEqual(messages, ["first", "second", "third"]);
1005 }).pipe(Effect.provide(ProviderReplayHarness.layerWithRegistry({ name }, layerRegistry)));
1006 }),
1007 ),
1008);