1import { expect, it } from "@effect/vitest";
2import { EnvironmentId, type TerminalMetadataStreamEvent, WS_METHODS } from "@t3tools/contracts";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Fiber from "effect/Fiber";
6import * as Layer from "effect/Layer";
7import * as Stream from "effect/Stream";
8import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
9import { vi } from "vite-plus/test";
10
11import { EnvironmentRegistry } from "../connection/registry.ts";
12import { createTerminalEnvironmentAtoms } from "./terminal.ts";
13import { executeAtomQuery } from "./runtime.ts";
14
15const source = vi.hoisted(() => ({ stream: undefined as unknown }));
16vi.mock("../rpc/client.ts", async (original) => ({
17 ...(await original<typeof import("../rpc/client.ts")>()),
18 subscribe: (tag: string) => {
19 expect(tag).toBe(WS_METHODS.subscribeTerminalMetadata);
20 return source.stream;
21 },
22}));
23
24it.effect("waits for the server metadata snapshot before reporting an empty terminal list", () =>
25 Effect.scoped(
26 Effect.gen(function* () {
27 const subscribed = yield* Deferred.make<void>();
28 const snapshot = yield* Deferred.make<TerminalMetadataStreamEvent>();
29 source.stream = Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
30 Stream.flatMap(() => Stream.fromEffect(Deferred.await(snapshot))),
31 );
32 const runtime = Atom.runtime(
33 Layer.succeed(EnvironmentRegistry, {
34 followStream: (_id, stream) => Stream.concat(stream, Stream.never),
35 } as EnvironmentRegistry["Service"]),
36 );
37 const terminals = createTerminalEnvironmentAtoms(runtime);
38 const registry = AtomRegistry.make();
39 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
40 const atom = terminals.metadata({
41 environmentId: EnvironmentId.make("metadata-proof"),
42 input: null,
43 });
44 const read = yield* Effect.forkChild(AtomRegistry.getResult(registry, atom));
45 yield* Deferred.await(subscribed);
46 expect(AsyncResult.isSuccess(registry.get(atom))).toBe(false);
47 yield* Deferred.succeed(snapshot, { type: "snapshot", terminals: [] });
48 expect(yield* Fiber.join(read)).toEqual([]);
49 }),
50 ),
51);
52
53it.effect("refreshes terminal metadata while an existing subscriber keeps the stream mounted", () =>
54 Effect.scoped(
55 Effect.gen(function* () {
56 let subscriptions = 0;
57 let current: TerminalMetadataStreamEvent = { type: "snapshot", terminals: [] };
58 source.stream = Stream.suspend(() => {
59 subscriptions++;
60 return Stream.concat(Stream.succeed(current), Stream.never);
61 });
62 const runtime = Atom.runtime(
63 Layer.succeed(EnvironmentRegistry, {
64 followStream: (_id, stream) => Stream.concat(stream, Stream.never),
65 } as EnvironmentRegistry["Service"]),
66 );
67 const terminals = createTerminalEnvironmentAtoms(runtime);
68 const registry = AtomRegistry.make();
69 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
70 const target = { environmentId: EnvironmentId.make("metadata-refresh"), input: null };
71 const live = terminals.metadata(target);
72 yield* AtomRegistry.mount(registry, live);
73 expect(yield* AtomRegistry.getResult(registry, live)).toEqual([]);
74 const atom = terminals.metadataSnapshot(target);
75 const initial = yield* Effect.promise(() =>
76 executeAtomQuery(registry, atom, { refresh: true }),
77 );
78 expect(initial._tag).toBe("Success");
79 if (initial._tag === "Success") expect(initial.value).toEqual([]);
80 const subscriptionsBeforeRefresh = subscriptions;
81 current = {
82 type: "snapshot",
83 terminals: [
84 {
85 threadId: "thread",
86 terminalId: "term",
87 cwd: "/tmp",
88 worktreePath: null,
89 status: "running",
90 pid: 42,
91 exitCode: null,
92 exitSignal: null,
93 hasRunningSubprocess: true,
94 label: "sleep",
95 updatedAt: "2026-10-15T00:00:00.000Z",
96 },
97 ],
98 };
99 const result = yield* Effect.promise(() =>
100 executeAtomQuery(registry, atom, { refresh: true }),
101 );
102 expect(result._tag).toBe("Success");
103 if (result._tag === "Success") expect(result.value).toEqual(current.terminals);
104 expect(subscriptions).toBeGreaterThan(subscriptionsBeforeRefresh);
105 expect(registry.get(live)).toMatchObject({ _tag: "Success", waiting: true, value: [] });
106 }),
107 ),
108);