1import { assert, it } from "@effect/vitest";
2import {
3 ORCHESTRATION_PROTOCOL_VERSION,
4 ProjectId,
5 ThreadId,
6 type OrchestrationV2StoredEvent,
7 type OrchestrationProjectShell,
8 type ServerConfig,
9 type ServerConfigStreamEvent,
10} from "@t3tools/contracts";
11import * as HostProcess from "@t3tools/shared/HostProcess";
12import * as ConfigProvider from "effect/ConfigProvider";
13import * as Deferred from "effect/Deferred";
14import * as Duration from "effect/Duration";
15import * as Effect from "effect/Effect";
16import * as Fiber from "effect/Fiber";
17import * as FileSystem from "effect/FileSystem";
18import * as Layer from "effect/Layer";
19import * as Path from "effect/Path";
20import * as PubSub from "effect/PubSub";
21import * as Queue from "effect/Queue";
22import * as Stream from "effect/Stream";
23import * as TestClock from "effect/testing/TestClock";
24import { ChildProcessSpawner } from "effect/process";
25import * as SqlClient from "effect/sql/SqlClient";
26
27import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts";
28import * as ProjectStore from "./orchestration-v2/ProjectStore.ts";
29import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts";
30import * as SqlitePersistence from "./persistence/Sqlite.ts";
31import * as ExternalLauncher from "./process/externalLauncher.ts";
32import * as ProjectEnrichment from "./project/ProjectEnrichmentService.ts";
33import * as ProjectService from "./project/ProjectService.ts";
34import {
35 hasCompatibleOrchestrationProtocol,
36 resolveAvailableEditorsForConfig,
37 shouldUseBoundedThreadSnapshot,
38 subscribeOrchestrationV2Shell,
39 subscribeOrchestrationV2Thread,
40 withLateEditorConfig,
41} from "./ws.ts";
42
43it("accepts only the current orchestration protocol before websocket RPC setup", () => {
44 assert.isTrue(
45 hasCompatibleOrchestrationProtocol(
46 new URL(`https://host.test/ws?orchestrationProtocol=${ORCHESTRATION_PROTOCOL_VERSION}`),
47 ),
48 );
49 assert.isFalse(hasCompatibleOrchestrationProtocol(new URL("https://host.test/ws")));
50 assert.isFalse(
51 hasCompatibleOrchestrationProtocol(
52 new URL(`https://host.test/ws?orchestrationProtocol=${ORCHESTRATION_PROTOCOL_VERSION - 1}`),
53 ),
54 );
55});
56
57it("keeps full thread snapshot fallback unless the client opts into bounded history", () => {
58 assert.isFalse(shouldUseBoundedThreadSnapshot({}));
59 assert.isFalse(shouldUseBoundedThreadSnapshot({ acceptBoundedSnapshot: false }));
60 assert.isTrue(shouldUseBoundedThreadSnapshot({ acceptBoundedSnapshot: true }));
61});
62
63it.effect("does not block server config when editor discovery never resolves", () =>
64 Effect.gen(function* () {
65 const discoveryInterrupted = yield* Deferred.make<void>();
66 const responseFiber = yield* resolveAvailableEditorsForConfig(
67 Effect.never.pipe(
68 Effect.onInterrupt(() => Deferred.succeed(discoveryInterrupted, undefined)),
69 ),
70 ).pipe(Effect.forkChild);
71
72 yield* TestClock.adjust(Duration.seconds(5));
73
74 const availableEditors = yield* Fiber.join(responseFiber);
75 yield* Deferred.await(discoveryInterrupted);
76 assert.deepEqual(availableEditors, []);
77 }),
78);
79
80// Only the fields the late-editor fold reads or rewrites.
81const snapshotConfig = (fields: Partial<ServerConfig>) =>
82 ({ availableEditors: [], settings: {}, ...fields }) as unknown as ServerConfig;
83
84const settingsUpdated = (settings: object): ServerConfigStreamEvent => ({
85 version: 1,
86 type: "settingsUpdated",
87 payload: { settings: settings as ServerConfig["settings"] },
88});
89
90it.effect("resends late editors without rolling back updates already sent", () =>
91 Effect.gen(function* () {
92 const settingsSent = yield* Deferred.make<void>();
93 const events = yield* withLateEditorConfig(
94 snapshotConfig({ settings: { enableProviderUpdateChecks: true } as never }),
95 Stream.make(settingsUpdated({ enableProviderUpdateChecks: false })),
96 {
97 resolveAvailableEditors: () => Effect.succeed(["file-manager"]),
98 // Holds the late snapshot until the settings change has gone out.
99 resolveFileManagerRevealKind: () =>
100 Deferred.await(settingsSent).pipe(Effect.as("file-explorer" as const)),
101 },
102 ).pipe(
103 Stream.tap((event) =>
104 event.type === "settingsUpdated" ? Deferred.succeed(settingsSent, undefined) : Effect.void,
105 ),
106 Stream.runCollect,
107 );
108
109 const [first, second] = Array.from(events);
110 assert.equal(events.length, 2);
111 assert.equal(first?.type, "settingsUpdated");
112 assert.equal(second?.type, "snapshot");
113 if (second?.type === "snapshot") {
114 assert.deepEqual(second.config.availableEditors, ["file-manager"]);
115 assert.equal(second.config.shellRevealInFileManagerKind, "file-explorer");
116 assert.deepEqual(second.config.settings, { enableProviderUpdateChecks: false } as never);
117 }
118 }),
119);
120
121it.effect("sends no late snapshot when the scan matches the snapshot", () =>
122 Effect.gen(function* () {
123 const events = yield* withLateEditorConfig(
124 snapshotConfig({ availableEditors: ["vscode"] }),
125 Stream.empty,
126 {
127 resolveAvailableEditors: () => Effect.succeed(["vscode"]),
128 resolveFileManagerRevealKind: () => Effect.succeed(undefined),
129 },
130 ).pipe(Stream.runCollect);
131
132 assert.equal(events.length, 0);
133 }),
134);
135
136it.effect("resends a file manager reveal kind that missed the snapshot", () =>
137 Effect.gen(function* () {
138 const events = yield* withLateEditorConfig(
139 snapshotConfig({ availableEditors: ["file-manager"] }),
140 Stream.empty,
141 {
142 resolveAvailableEditors: () => Effect.succeed(["file-manager"]),
143 resolveFileManagerRevealKind: () => Effect.succeed("file-explorer"),
144 },
145 ).pipe(Stream.runCollect);
146
147 const [late] = Array.from(events);
148 assert.equal(events.length, 1);
149 assert.equal(late?.type, "snapshot");
150 if (late?.type === "snapshot") {
151 assert.equal(late.config.shellRevealInFileManagerKind, "file-explorer");
152 }
153 }),
154);
155
156// The real launcher on Windows over a filesystem whose probes park until
157// released, like a host too busy to finish discovery inside the snapshot timeout.
158const makeParkedWindowsLauncher = Effect.gen(function* () {
159 const parkedProbes = yield* Queue.unbounded<void>();
160 const release = yield* Deferred.make<void>();
161 const launcher = yield* ExternalLauncher.make.pipe(
162 Effect.provide(
163 Layer.mergeAll(
164 FileSystem.layerNoop({
165 stat: () =>
166 Queue.offer(parkedProbes, undefined).pipe(
167 Effect.andThen(Deferred.await(release)),
168 Effect.as({ type: "File" } as FileSystem.File.Info),
169 ),
170 }),
171 Path.layer,
172 Layer.succeed(
173 ChildProcessSpawner.ChildProcessSpawner,
174 ChildProcessSpawner.make(() => Effect.die("unexpected spawn")),
175 ),
176 ),
177 ),
178 );
179 const onWindows = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
180 effect.pipe(
181 Effect.provideService(HostProcess.Platform, "win32"),
182 Effect.provide(
183 ConfigProvider.layer(
184 ConfigProvider.fromEnv({
185 env: { PATH: "C:\\t3-late-editors-test", PATHEXT: ".EXE" },
186 }),
187 ),
188 ),
189 );
190 return {
191 editors: {
192 resolveAvailableEditors: () => onWindows(launcher.resolveAvailableEditors()),
193 resolveFileManagerRevealKind: () => onWindows(launcher.resolveFileManagerRevealKind()),
194 },
195 probeParked: Queue.take(parkedProbes),
196 releaseProbes: Deferred.succeed(release, undefined),
197 };
198});
199
200it.effect("recovers editors after a real scan outlasts the config timeout", () =>
201 Effect.gen(function* () {
202 const { editors, probeParked, releaseProbes } = yield* makeParkedWindowsLauncher;
203
204 const snapshotFiber = yield* resolveAvailableEditorsForConfig(
205 editors.resolveAvailableEditors(),
206 ).pipe(Effect.forkChild);
207 yield* probeParked;
208 yield* TestClock.adjust(Duration.seconds(5));
209 const snapshotEditors = yield* Fiber.join(snapshotFiber);
210 assert.deepEqual(snapshotEditors, []);
211
212 const lateFiber = yield* withLateEditorConfig(
213 snapshotConfig({ availableEditors: snapshotEditors }),
214 Stream.empty,
215 editors,
216 ).pipe(Stream.runCollect, Effect.forkChild);
217 yield* releaseProbes;
218
219 const [late] = Array.from(yield* Fiber.join(lateFiber));
220 assert.equal(late?.type, "snapshot");
221 if (late?.type === "snapshot") {
222 assert.equal(late.config.availableEditors.includes("vscode"), true);
223 }
224 }).pipe(Effect.scoped),
225);
226
227it.effect("recovers a reveal kind whose real probe outlasts the config timeout", () =>
228 Effect.gen(function* () {
229 const { editors, probeParked, releaseProbes } = yield* makeParkedWindowsLauncher;
230
231 // The snapshot's bounded probe timed out: file manager, but no reveal kind.
232 const lateFiber = yield* withLateEditorConfig(
233 snapshotConfig({ availableEditors: ["file-manager"] }),
234 Stream.empty,
235 {
236 resolveAvailableEditors: () => Effect.succeed(["file-manager"]),
237 resolveFileManagerRevealKind: editors.resolveFileManagerRevealKind,
238 },
239 ).pipe(Stream.runCollect, Effect.forkChild);
240 yield* probeParked;
241 yield* TestClock.adjust(Duration.seconds(6));
242 yield* releaseProbes;
243
244 const [late] = Array.from(yield* Fiber.join(lateFiber));
245 assert.equal(late?.type, "snapshot");
246 if (late?.type === "snapshot") {
247 assert.equal(late.config.shellRevealInFileManagerKind, "file-explorer");
248 }
249 }).pipe(Effect.scoped),
250);
251
252const ReplayTestLayer = OrchestrationEventStore.layer.pipe(
253 Layer.provideMerge(SqlitePersistence.layerMemory),
254);
255
256it.live.each(["thread", "shell"] as const)(
257 "%s resume skips unknown types, synchronizes, and receives the live tail",
258 (scope) =>
259 Effect.gen(function* () {
260 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
261 const sql = yield* SqlClient.SqlClient;
262 const now = "2026-01-05T00:00:00.000Z";
263 const threadId = ThreadId.make("thread:replay");
264 const projectId = ProjectId.make("project:replay");
265 const project = {
266 id: projectId,
267 title: "Replay",
268 workspaceRoot: "/tmp/replay",
269 defaultModelSelection: null,
270 scripts: [],
271 createdAt: now,
272 updatedAt: now,
273 } satisfies OrchestrationProjectShell;
274 const insert = (version: number, unknown = false) => sql<{ sequence: number }>`
275 INSERT INTO orchestration_events (
276 event_id, aggregate_kind, stream_id, stream_version, event_type,
277 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
278 ) VALUES (
279 ${`replay:${version}`}, ${scope === "thread" ? "thread" : "project"},
280 ${scope === "thread" ? threadId : projectId}, ${version},
281 ${unknown ? "future.event" : scope === "thread" ? "provider-session.detached" : "project.meta-updated"},
282 ${now}, 'server', ${JSON.stringify(
283 scope === "thread"
284 ? { providerSessionId: "session:replay", detachedAt: now }
285 : { projectId, updatedAt: now },
286 )}, '{}', 2
287 ) RETURNING sequence
288 `;
289 const [before] = yield* insert(0);
290 yield* insert(1, true);
291 const [after] = yield* insert(2);
292 const [highWater] = yield* insert(3, true);
293 const updates = yield* PubSub.unbounded();
294 let liveSequence = 0;
295 const consume = <A extends { readonly kind: string; readonly sequence?: number }, E, R>(
296 stream: Stream.Stream<A, E, R>,
297 ) =>
298 stream.pipe(
299 Stream.tap((item) =>
300 Effect.gen(function* () {
301 if (item.kind !== "synchronized") return;
302 const [live] = yield* insert(4);
303 liveSequence = live!.sequence;
304 const events = yield* store
305 .readApplicationEvents({
306 afterSequence: highWater!.sequence,
307 throughSequence: liveSequence,
308 })
309 .pipe(Stream.runCollect);
310 yield* store.publishCommitted(events);
311 }),
312 ),
313 Stream.takeUntil((item) => item.sequence === liveSequence),
314 Stream.map((item) => ({ kind: item.kind as string, sequence: item.sequence })),
315 Stream.runCollect,
316 );
317 const subscription = Effect.gen(function* () {
318 return scope === "thread"
319 ? yield* subscribeOrchestrationV2Thread({
320 threadId,
321 afterSequence: before!.sequence,
322 requestCompletionMarker: true,
323 }).pipe(Effect.flatMap(consume))
324 : yield* subscribeOrchestrationV2Shell({
325 afterSequence: before!.sequence,
326 requestCompletionMarker: true,
327 }).pipe(Effect.flatMap(consume));
328 });
329 const items = yield* subscription.pipe(
330 Effect.provideService(ThreadManagement.ThreadManagementService, {
331 ensureLegacyTranscript: () => Effect.void,
332 streamStoredEventsFrom: (
333 input: Parameters<
334 ThreadManagement.ThreadManagementService["Service"]["streamStoredEventsFrom"]
335 >[0],
336 ) => {
337 assert.equal(input?.afterSequence, highWater!.sequence);
338 return store
339 .streamApplicationEvents({ afterSequence: input?.afterSequence ?? 0 })
340 .pipe(
341 Stream.filter((event): event is OrchestrationV2StoredEvent => "event" in event),
342 );
343 },
344 } as unknown as ThreadManagement.ThreadManagementService["Service"]),
345 Effect.provideService(ProjectStore.ProjectStoreV2, {
346 listShells: () => Effect.succeed([]),
347 } as unknown as ProjectStore.ProjectStoreV2["Service"]),
348 Effect.provideService(ProjectService.ProjectService, {
349 getShell: () => Effect.succeedSome(project),
350 } as unknown as ProjectService.ProjectService["Service"]),
351 Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, {
352 subscribeChanges: PubSub.subscribe(updates),
353 } as ProjectEnrichment.ProjectEnrichmentService["Service"]),
354 );
355 assert.deepEqual(
356 items.map((item) => item.kind),
357 scope === "thread"
358 ? ["event", "synchronized", "event"]
359 : ["project.updated", "synchronized", "project.updated"],
360 );
361 assert.deepEqual(
362 items.flatMap((item) => (item.sequence === undefined ? [] : [item.sequence])),
363 [after!.sequence, liveSequence],
364 );
365 }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)),
366);