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