1import { vi } from "vite-plus/test";
2import { assert, it } from "@effect/vitest";
3import {
4 CommandId,
5 ProjectId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 ThreadId,
9} from "@t3tools/contracts";
10import * as Cause from "effect/Cause";
11import * as Deferred from "effect/Deferred";
12import * as Effect from "effect/Effect";
13import * as Exit from "effect/Exit";
14import * as Layer from "effect/Layer";
15import * as Ref from "effect/Ref";
16import * as SqlitePersistence from "../persistence/Sqlite.ts";
17import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
18import * as Orchestrator from "./Orchestrator.ts";
19import * as ProjectionStore from "./ProjectionStore.ts";
20import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
21import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
22import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
23
24const instanceId = ProviderInstanceId.make("codex");
25const modelSelection = { instanceId, model: "gpt-5.1-codex" };
26// Holds model-selection planning until both racing commands have read no receipt.
27let capabilityGate: Effect.Effect<void> | undefined;
28const adapter = {
29 instanceId,
30 driver: ProviderDriverKind.make("codex"),
31 getCapabilities: () =>
32 (capabilityGate ?? Effect.void).pipe(Effect.as(CodexProviderCapabilitiesV2)),
33 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
34 openSession: () => Effect.die("No provider process needed for metadata controls"),
35} as ProviderAdapter.ProviderAdapterV2["Service"];
36const layerDatabase = SqlitePersistence.layerMemory;
37const layerTest = Layer.mergeAll(
38 layerDatabase,
39 ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
40 ProviderReplayHarness.layerWithRegistry(
41 { name: "receipt-race" },
42 ProviderAdapterRegistry.layerFromAdapters([adapter]),
43 { databaseLayer: layerDatabase, runEffectWorker: false },
44 ),
45);
46
47it.effect("a command id reused concurrently for two threads succeeds for only one", () =>
48 Effect.gen(function* () {
49 const orchestrator = yield* Orchestrator.OrchestratorV2;
50 const projections = yield* ProjectionStore.ProjectionStoreV2;
51 const threadIds = [ThreadId.make("thread:race-a"), ThreadId.make("thread:race-b")];
52 for (const threadId of threadIds) {
53 yield* orchestrator.dispatch({
54 type: "thread.create",
55 commandId: CommandId.make(`create:${threadId}`),
56 threadId,
57 projectId: ProjectId.make("project:receipt-race"),
58 title: threadId,
59 modelSelection,
60 runtimeMode: "full-access",
61 interactionMode: "default",
62 branch: null,
63 worktreePath: null,
64 createdBy: "user",
65 creationSource: "web",
66 });
67 }
68
69 const arrived = yield* Ref.make(0);
70 const bothArrived = yield* Deferred.make<void>();
71 capabilityGate = Effect.gen(function* () {
72 if ((yield* Ref.updateAndGet(arrived, (count) => count + 1)) === 2) {
73 yield* Deferred.succeed(bothArrived, undefined);
74 }
75 yield* Deferred.await(bothArrived);
76 });
77 const setModel = (threadId: ThreadId) =>
78 orchestrator.dispatch({
79 type: "thread.model-selection.set",
80 commandId: CommandId.make("command:shared"),
81 threadId,
82 modelSelection: { ...modelSelection, model: "gpt-6" },
83 });
84 const exits = yield* Effect.forEach(threadIds, (threadId) => Effect.exit(setModel(threadId)), {
85 concurrency: "unbounded",
86 }).pipe(Effect.ensuring(Effect.sync(() => (capabilityGate = undefined))));
87
88 const winner = exits.findIndex(Exit.isSuccess);
89 const loser = exits.findIndex(Exit.isFailure);
90 assert.isAtLeast(winner, 0);
91 assert.isAtLeast(loser, 0);
92 const winnerExit = exits[winner]!;
93 const loserExit = exits[loser]!;
94 assert.isTrue(Exit.isSuccess(winnerExit) && Exit.isFailure(loserExit));
95 if (Exit.isFailure(loserExit)) {
96 const error = Cause.findErrorOption(loserExit.cause);
97 assert.isTrue(error._tag === "Some");
98 if (error._tag === "Some") {
99 assert.equal(error.value._tag, "OrchestratorCommandIdConflictError");
100 if (error.value._tag === "OrchestratorCommandIdConflictError") {
101 assert.equal(error.value.receiptThreadId, threadIds[winner]);
102 assert.equal(error.value.commandThreadId, threadIds[loser]);
103 }
104 }
105 }
106
107 const models = yield* Effect.forEach(threadIds, (threadId) =>
108 projections.getThreadShell(threadId).pipe(Effect.map((shell) => shell?.modelSelection.model)),
109 );
110 assert.equal(models[winner], "gpt-6");
111 assert.equal(models[loser], "gpt-5.1-codex");
112
113 const noOpArrived = yield* Ref.make(0);
114 const bothNoOpsArrived = yield* Deferred.make<void>();
115 const getThreadRecords = projections.getThreadRecords;
116 using _records = vi.spyOn(projections, "getThreadRecords").mockImplementation((...args) =>
117 Effect.gen(function* () {
118 if ((yield* Ref.updateAndGet(noOpArrived, (count) => count + 1)) === 2) {
119 yield* Deferred.succeed(bothNoOpsArrived, undefined);
120 }
121 yield* Deferred.await(bothNoOpsArrived);
122 return yield* getThreadRecords(...args);
123 }),
124 );
125 const closeShare = (threadId: ThreadId) =>
126 orchestrator.dispatch({
127 type: "thread.guest-share.close",
128 commandId: CommandId.make("command:shared-noop"),
129 threadId,
130 });
131 const noOpExits = yield* Effect.forEach(
132 threadIds,
133 (threadId) => Effect.exit(closeShare(threadId)),
134 {
135 concurrency: "unbounded",
136 },
137 );
138 const noOpWinner = noOpExits.findIndex(Exit.isSuccess);
139 const noOpLoser = noOpExits.findIndex(Exit.isFailure);
140 assert.isAtLeast(noOpWinner, 0);
141 assert.isAtLeast(noOpLoser, 0);
142 const noOpFailure = noOpExits[noOpLoser]!;
143 if (Exit.isFailure(noOpFailure)) {
144 const error = Cause.findErrorOption(noOpFailure.cause);
145 assert.isTrue(error._tag === "Some");
146 if (error._tag === "Some") {
147 assert.equal(error.value._tag, "OrchestratorCommandIdConflictError");
148 }
149 }
150 const noOpReplay = yield* closeShare(threadIds[noOpWinner]!);
151 const noOpSuccess = noOpExits[noOpWinner]!;
152 if (Exit.isSuccess(noOpSuccess)) {
153 assert.equal(noOpReplay.sequence, noOpSuccess.value.sequence);
154 assert.deepEqual(noOpReplay.storedEvents, []);
155 }
156
157 // The winner's own thread still replays its receipt.
158 const replay = yield* setModel(threadIds[winner]!);
159 if (Exit.isSuccess(winnerExit)) {
160 assert.equal(replay.sequence, winnerExit.value.sequence);
161 assert.isAbove(replay.storedEvents.length, 0);
162 assert.deepEqual(
163 replay.storedEvents.map((stored) => stored.sequence),
164 winnerExit.value.storedEvents.map((stored) => stored.sequence),
165 );
166 }
167 }).pipe(Effect.provide(layerTest)),
168);