1import { vi } from "vite-plus/test";
2import {
3 EnvironmentId,
4 AuthSourceControlWriteScope,
5 WS_METHODS,
6 type VcsListRefsInput,
7 type VcsListRefsResult,
8} from "@t3tools/contracts";
9import { describe, expect, it } from "@effect/vitest";
10import * as Effect from "effect/Effect";
11import * as Fiber from "effect/Fiber";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Ref from "effect/Ref";
15import * as Stream from "effect/Stream";
16import * as SubscriptionRef from "effect/SubscriptionRef";
17import * as TestClock from "effect/testing/TestClock";
18import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
19
20import {
21 AVAILABLE_CONNECTION_STATE,
22 PrimaryConnectionTarget,
23 type PreparedConnection,
24 type SupervisorConnectionState,
25} from "../connection/model.ts";
26import * as EnvironmentRegistry from "../connection/registry.ts";
27import * as EnvironmentSupervisor from "../connection/supervisor.ts";
28import * as Persistence from "../platform/persistence.ts";
29import { EnvironmentRpcUnavailableError } from "../rpc/client.ts";
30import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
31import type { RpcSession } from "../rpc/session.ts";
32
33import {
34 commitVcsRefsRefresh,
35 createVcsEnvironmentAtoms,
36 makeCachedVcsRefsChanges,
37} from "./vcs.ts";
38import {
39 invalidateCachedVcsRefs,
40 invalidateVcsRefs,
41 vcsRefsCacheStateAtom,
42} from "./vcsRefInvalidation.ts";
43
44const TARGET = new PrimaryConnectionTarget({
45 environmentId: EnvironmentId.make("environment-1"),
46 label: "Test environment",
47 httpBaseUrl: "https://environment.example.test",
48 wsBaseUrl: "wss://environment.example.test",
49});
50
51const CONNECTED_CONNECTION_STATE: SupervisorConnectionState = {
52 ...AVAILABLE_CONNECTION_STATE,
53 desired: true,
54 network: "online",
55 phase: "connected",
56 attempt: 1,
57 generation: 1,
58};
59
60const CACHED_REFS: VcsListRefsResult = {
61 refs: [
62 {
63 name: "main",
64 current: true,
65 isDefault: true,
66 worktreePath: "/repo",
67 },
68 ],
69 isRepo: true,
70 hasPrimaryRemote: true,
71 nextCursor: null,
72 totalCount: 1,
73};
74
75const LIVE_REFS: VcsListRefsResult = {
76 ...CACHED_REFS,
77 refs: [
78 {
79 name: "release",
80 current: true,
81 isDefault: true,
82 worktreePath: "/repo",
83 },
84 ],
85};
86
87function session(client: WsRpcProtocolClient): RpcSession {
88 return {
89 client,
90 initialConfig: Effect.never,
91 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
92 ready: Effect.void,
93 probe: Effect.void,
94 closed: Effect.never,
95 };
96}
97
98function cacheWithRefs(
99 refs: Option.Option<VcsListRefsResult>,
100 overrides: Partial<Persistence.EnvironmentCacheStore["Service"]> = {},
101) {
102 return Persistence.EnvironmentCacheStore.of({
103 loadShell: () => Effect.succeedNone,
104 saveShell: () => Effect.void,
105 loadThread: () => Effect.succeedNone,
106 saveThread: () => Effect.void,
107 removeThread: () => Effect.void,
108 loadServerConfig: () => Effect.succeedNone,
109 saveServerConfig: () => Effect.void,
110 loadVcsRefs: () => Effect.succeed(refs),
111 saveVcsRefs: () => Effect.void,
112 removeVcsRefs: () => Effect.void,
113 clearVcsRefs: () => Effect.void,
114 clear: () => Effect.void,
115 ...overrides,
116 });
117}
118
119describe("cached VCS refs", () => {
120 it("invalidates all ref streams in the mutated environment", () => {
121 const registry = AtomRegistry.make();
122 const environment = {
123 environmentId: TARGET.environmentId,
124 };
125 const otherEnvironment = {
126 environmentId: EnvironmentId.make("environment-2"),
127 };
128
129 expect(registry.get(vcsRefsCacheStateAtom(environment))).toEqual({
130 revision: 0,
131 persistedCacheReadable: true,
132 });
133 expect(registry.get(vcsRefsCacheStateAtom(otherEnvironment))).toEqual({
134 revision: 0,
135 persistedCacheReadable: true,
136 });
137
138 invalidateVcsRefs(registry, environment);
139
140 expect(registry.get(vcsRefsCacheStateAtom(environment))).toEqual({
141 revision: 1,
142 persistedCacheReadable: true,
143 });
144 expect(registry.get(vcsRefsCacheStateAtom(otherEnvironment))).toEqual({
145 revision: 0,
146 persistedCacheReadable: true,
147 });
148 registry.dispose();
149 });
150
151 it.effect("preserves the caller's repository snapshot refresh policy", () =>
152 Effect.scoped(
153 Effect.gen(function* () {
154 const requests = yield* Ref.make<ReadonlyArray<VcsListRefsInput>>([]);
155 const client = {
156 [WS_METHODS.vcsListRefs]: (input: VcsListRefsInput) =>
157 Ref.update(requests, (current) => [...current, input]).pipe(Effect.as(LIVE_REFS)),
158 } as unknown as WsRpcProtocolClient;
159 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
160 target: TARGET,
161 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
162 session: yield* SubscriptionRef.make(Option.some(session(client))),
163 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
164 connect: Effect.void,
165 disconnect: Effect.void,
166 retryNow: Effect.void,
167 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
168
169 yield* Stream.unwrap(
170 makeCachedVcsRefsChanges({
171 cwd: "/repo",
172 limit: 20,
173 query: "release",
174 refKind: "remote",
175 }).pipe(
176 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
177 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
178 ),
179 ).pipe(Stream.runHead);
180
181 expect(yield* Ref.get(requests)).toEqual([
182 {
183 cwd: "/repo",
184 limit: 20,
185 query: "release",
186 refKind: "remote",
187 },
188 ]);
189
190 yield* Stream.unwrap(
191 makeCachedVcsRefsChanges({
192 cwd: "/repo",
193 limit: 20,
194 query: "release",
195 refKind: "remote",
196 refresh: true,
197 }).pipe(
198 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
199 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
200 ),
201 ).pipe(Stream.runHead);
202
203 expect(yield* Ref.get(requests)).toEqual([
204 {
205 cwd: "/repo",
206 limit: 20,
207 query: "release",
208 refKind: "remote",
209 },
210 {
211 cwd: "/repo",
212 limit: 20,
213 query: "release",
214 refKind: "remote",
215 refresh: true,
216 },
217 ]);
218 }),
219 ),
220 );
221
222 it.effect("does not repersist a refresh superseded by ref invalidation", () =>
223 Effect.scoped(
224 Effect.gen(function* () {
225 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
226 Effect.sync(() => registry.dispose()),
227 );
228 const saved = yield* Ref.make<ReadonlyArray<VcsListRefsResult>>([]);
229 const clears = yield* Ref.make(0);
230 const revisionsObservedDuringClear = yield* Ref.make<ReadonlyArray<number>>([]);
231 const cache = cacheWithRefs(Option.none(), {
232 saveVcsRefs: (_environmentId, _cwd, refs) =>
233 Ref.update(saved, (current) => [...current, refs]),
234 clearVcsRefs: () =>
235 Effect.all([
236 Ref.update(clears, (count) => count + 1),
237 Ref.update(revisionsObservedDuringClear, (current) => [
238 ...current,
239 registry.get(vcsRefsCacheStateAtom(TARGET)).revision,
240 ]),
241 ]).pipe(Effect.asVoid),
242 });
243
244 yield* invalidateCachedVcsRefs(registry, {
245 environmentId: TARGET.environmentId,
246 cwd: "/repo-worktree",
247 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
248
249 expect(
250 yield* commitVcsRefsRefresh(registry, cache, {
251 environmentId: TARGET.environmentId,
252 cwd: "/repo",
253 refs: CACHED_REFS,
254 expectedRevision: 0,
255 persist: true,
256 }),
257 ).toBe(false);
258
259 expect(yield* Ref.get(saved)).toEqual([]);
260 expect(yield* Ref.get(clears)).toBe(1);
261 expect(yield* Ref.get(revisionsObservedDuringClear)).toEqual([0]);
262 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual({
263 revision: 1,
264 persistedCacheReadable: true,
265 });
266
267 expect(
268 yield* commitVcsRefsRefresh(registry, cache, {
269 environmentId: TARGET.environmentId,
270 cwd: "/repo",
271 refs: LIVE_REFS,
272 expectedRevision: 1,
273 persist: true,
274 }),
275 ).toBe(true);
276 expect(yield* Ref.get(saved)).toEqual([LIVE_REFS]);
277 }),
278 ),
279 );
280
281 it.effect("invalidates persisted refs when ref-affecting commands settle", () =>
282 Effect.scoped(
283 Effect.gen(function* () {
284 const expectedError = new EnvironmentRpcUnavailableError({
285 environmentId: TARGET.environmentId,
286 message: "pull failed after fetching refs",
287 });
288 const client = {
289 [WS_METHODS.vcsPull]: () => Effect.fail(expectedError),
290 [WS_METHODS.vcsRefreshStatus]: () => Effect.void,
291 } as unknown as WsRpcProtocolClient;
292 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
293 target: TARGET,
294 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
295 session: yield* SubscriptionRef.make(Option.some(session(client))),
296 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
297 connect: Effect.void,
298 disconnect: Effect.void,
299 retryNow: Effect.void,
300 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
301 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
302 _environmentId,
303 effect,
304 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
305 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
306 run,
307 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
308 const clears = yield* Ref.make(0);
309 const runtime = Atom.runtime(
310 Layer.merge(
311 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
312 Layer.succeed(
313 Persistence.EnvironmentCacheStore,
314 cacheWithRefs(Option.none(), {
315 clearVcsRefs: () => Ref.update(clears, (count) => count + 1),
316 }),
317 ),
318 ),
319 );
320 const atoms = createVcsEnvironmentAtoms(runtime);
321 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
322 Effect.sync(() => registry.dispose()),
323 );
324
325 const result = yield* Effect.promise(() =>
326 atoms.pull.run(registry, {
327 environmentId: TARGET.environmentId,
328 input: { cwd: "/repo" },
329 }),
330 );
331
332 expect(AsyncResult.isFailure(result)).toBe(true);
333 expect(yield* Ref.get(clears)).toBe(1);
334 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(1);
335
336 const refreshResult = yield* Effect.promise(() =>
337 atoms.refreshStatus.run(registry, {
338 environmentId: TARGET.environmentId,
339 input: { cwd: "/repo" },
340 }),
341 );
342
343 expect(AsyncResult.isSuccess(refreshResult)).toBe(true);
344 expect(yield* Ref.get(clears)).toBe(2);
345 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(2);
346 }),
347 ),
348 );
349
350 it.effect("suppresses persisted snapshots after an environment-wide clear fails", () =>
351 Effect.scoped(
352 Effect.gen(function* () {
353 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
354 Effect.sync(() => registry.dispose()),
355 );
356 const persisted = yield* Ref.make<Option.Option<VcsListRefsResult>>(
357 Option.some(CACHED_REFS),
358 );
359 const loads = yield* Ref.make(0);
360 const clearAttempts = yield* Ref.make(0);
361 const cache = cacheWithRefs(Option.none(), {
362 loadVcsRefs: () =>
363 Ref.update(loads, (count) => count + 1).pipe(Effect.andThen(Ref.get(persisted))),
364 saveVcsRefs: (_environmentId, _cwd, refs) => Ref.set(persisted, Option.some(refs)),
365 clearVcsRefs: () =>
366 Ref.updateAndGet(clearAttempts, (count) => count + 1).pipe(
367 Effect.flatMap((attempt) =>
368 attempt === 1
369 ? Effect.fail(
370 new Persistence.ConnectionPersistenceError({
371 operation: "clear-vcs-refs",
372 message: "storage unavailable",
373 }),
374 )
375 : Ref.set(persisted, Option.none()),
376 ),
377 ),
378 });
379
380 yield* invalidateCachedVcsRefs(registry, {
381 environmentId: TARGET.environmentId,
382 cwd: "/repo",
383 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
384
385 const state = registry.get(vcsRefsCacheStateAtom(TARGET));
386 expect(state).toEqual({
387 revision: 1,
388 persistedCacheReadable: false,
389 });
390
391 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
392 target: TARGET,
393 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
394 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
395 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
396 connect: Effect.void,
397 disconnect: Effect.void,
398 retryNow: Effect.void,
399 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
400 const refs = yield* Stream.unwrap(
401 makeCachedVcsRefsChanges(
402 { cwd: "/repo", limit: 100 },
403 state.revision,
404 registry,
405 state.persistedCacheReadable,
406 ).pipe(
407 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
408 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
409 ),
410 ).pipe(Stream.runHead, Effect.forkChild({ startImmediately: true }));
411
412 yield* Effect.yieldNow;
413 expect(yield* Ref.get(loads)).toBe(0);
414 yield* Fiber.interrupt(refs);
415 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual(state);
416
417 expect(
418 yield* commitVcsRefsRefresh(registry, cache, {
419 environmentId: TARGET.environmentId,
420 cwd: "/repo",
421 refs: LIVE_REFS,
422 expectedRevision: state.revision,
423 persist: true,
424 }),
425 ).toBe(true);
426 const recoveredState = registry.get(vcsRefsCacheStateAtom(TARGET));
427 expect(recoveredState).toEqual({
428 revision: 1,
429 persistedCacheReadable: true,
430 });
431 expect(yield* Ref.get(clearAttempts)).toBe(2);
432
433 const recoveredRefs = yield* Stream.unwrap(
434 makeCachedVcsRefsChanges(
435 { cwd: "/repo", limit: 100 },
436 recoveredState.revision,
437 registry,
438 recoveredState.persistedCacheReadable,
439 ).pipe(
440 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
441 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
442 ),
443 ).pipe(Stream.runHead);
444
445 expect(Option.getOrThrow(recoveredRefs)).toEqual(LIVE_REFS);
446 expect(yield* Ref.get(loads)).toBe(1);
447 }),
448 ),
449 );
450
451 it.effect("loads an unfiltered branch list without a connection", () =>
452 Effect.scoped(
453 Effect.gen(function* () {
454 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
455 target: TARGET,
456 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
457 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
458 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
459 connect: Effect.void,
460 disconnect: Effect.void,
461 retryNow: Effect.void,
462 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
463 const refs = yield* Stream.unwrap(
464 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
465 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
466 Effect.provideService(
467 Persistence.EnvironmentCacheStore,
468 cacheWithRefs(Option.some(CACHED_REFS)),
469 ),
470 ),
471 ).pipe(Stream.runHead);
472
473 expect(Option.getOrThrow(refs)).toEqual(CACHED_REFS);
474 }),
475 ),
476 );
477
478 it.effect("retries a transient live failure while connected", () =>
479 Effect.scoped(
480 Effect.gen(function* () {
481 const expectedError = new Error("Could not list Git refs.");
482 const calls = yield* Ref.make(0);
483 const connectionState = yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE);
484 const client = {
485 [WS_METHODS.vcsListRefs]: () =>
486 Ref.updateAndGet(calls, (count) => count + 1).pipe(
487 Effect.flatMap((count) =>
488 count === 1 ? Effect.fail(expectedError) : Effect.succeed(LIVE_REFS),
489 ),
490 ),
491 } as unknown as WsRpcProtocolClient;
492 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
493 target: TARGET,
494 state: connectionState,
495 session: yield* SubscriptionRef.make(Option.some(session(client))),
496 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
497 connect: Effect.void,
498 disconnect: Effect.void,
499 retryNow: Effect.void,
500 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
501
502 const result = Stream.unwrap(
503 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
504 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
505 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
506 ),
507 ).pipe(Stream.runHead);
508 const fiber = yield* Effect.forkChild(result);
509
510 for (let attempt = 0; attempt < 100 && (yield* Ref.get(calls)) < 1; attempt += 1) {
511 yield* Effect.yieldNow;
512 }
513 expect(yield* Ref.get(calls)).toBe(1);
514
515 yield* TestClock.adjust("1 second");
516 expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS);
517 expect(yield* Ref.get(calls)).toBe(2);
518 }).pipe(Effect.provide(TestClock.layer())),
519 ),
520 );
521
522 it.effect("cancels an in-flight refresh when the connection generation changes", () =>
523 Effect.scoped(
524 Effect.gen(function* () {
525 const calls = yield* Ref.make(0);
526 const interruptions = yield* Ref.make(0);
527 const connectionState = yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE);
528 const client = {
529 [WS_METHODS.vcsListRefs]: () =>
530 Ref.updateAndGet(calls, (count) => count + 1).pipe(
531 Effect.flatMap((count) =>
532 count === 1
533 ? Effect.never.pipe(
534 Effect.onInterrupt(() =>
535 Ref.update(interruptions, (interruptions) => interruptions + 1),
536 ),
537 )
538 : Effect.succeed(LIVE_REFS),
539 ),
540 ),
541 } as unknown as WsRpcProtocolClient;
542 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
543 target: TARGET,
544 state: connectionState,
545 session: yield* SubscriptionRef.make(Option.some(session(client))),
546 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
547 connect: Effect.void,
548 disconnect: Effect.void,
549 retryNow: Effect.void,
550 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
551
552 const result = Stream.unwrap(
553 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
554 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
555 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
556 ),
557 ).pipe(Stream.runHead);
558 const fiber = yield* Effect.forkChild(result);
559
560 for (let attempt = 0; attempt < 100 && (yield* Ref.get(calls)) < 1; attempt += 1) {
561 yield* Effect.yieldNow;
562 }
563 yield* SubscriptionRef.set(connectionState, AVAILABLE_CONNECTION_STATE);
564 for (let attempt = 0; attempt < 100 && (yield* Ref.get(interruptions)) < 1; attempt += 1) {
565 yield* Effect.yieldNow;
566 }
567 expect(yield* Ref.get(interruptions)).toBe(1);
568
569 yield* SubscriptionRef.set(connectionState, {
570 ...CONNECTED_CONNECTION_STATE,
571 generation: 2,
572 });
573 expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS);
574 expect(yield* Ref.get(calls)).toBe(2);
575 }),
576 ),
577 );
578
579 it.effect("does not poll refs while the connection remains stable", () =>
580 Effect.scoped(
581 Effect.gen(function* () {
582 const calls = yield* Ref.make(0);
583 const client = {
584 [WS_METHODS.vcsListRefs]: () =>
585 Ref.update(calls, (count) => count + 1).pipe(Effect.as(CACHED_REFS)),
586 } as unknown as WsRpcProtocolClient;
587 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
588 target: TARGET,
589 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
590 session: yield* SubscriptionRef.make(Option.some(session(client))),
591 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
592 connect: Effect.void,
593 disconnect: Effect.void,
594 retryNow: Effect.void,
595 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
596 const stream = Stream.unwrap(
597 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
598 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
599 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
600 ),
601 ).pipe(Stream.runDrain);
602 const fiber = yield* Effect.forkChild(stream);
603
604 for (let attempt = 0; attempt < 100 && (yield* Ref.get(calls)) < 1; attempt += 1) {
605 yield* Effect.yieldNow;
606 }
607 expect(yield* Ref.get(calls)).toBe(1);
608
609 yield* TestClock.adjust("1 minute");
610 yield* Effect.yieldNow;
611 expect(yield* Ref.get(calls)).toBe(1);
612 yield* Fiber.interrupt(fiber);
613 }).pipe(Effect.provide(TestClock.layer())),
614 ),
615 );
616
617 it.effect("emits persisted refs before a live refresh", () =>
618 Effect.scoped(
619 Effect.gen(function* () {
620 const client = {
621 [WS_METHODS.vcsListRefs]: () => Effect.succeed(LIVE_REFS),
622 } as unknown as WsRpcProtocolClient;
623 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
624 target: TARGET,
625 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
626 session: yield* SubscriptionRef.make(Option.some(session(client))),
627 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
628 connect: Effect.void,
629 disconnect: Effect.void,
630 retryNow: Effect.void,
631 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
632
633 const refs = yield* Stream.unwrap(
634 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
635 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
636 Effect.provideService(
637 Persistence.EnvironmentCacheStore,
638 cacheWithRefs(Option.some(CACHED_REFS)),
639 ),
640 ),
641 ).pipe(Stream.take(2), Stream.runCollect);
642
643 expect(refs).toEqual([CACHED_REFS, LIVE_REFS]);
644 }),
645 ),
646 );
647});
648
649// Transport fixtures have a source-control-only session; authorization edge cases
650// are exercised by commandPermissions.test.ts.
651vi.mock("./session.ts", () => ({
652 createEnvironmentSessionAtoms: () => ({ sessionStateAtom: grantedSessions }),
653}));
654const grantedSessions = Atom.family((_id: EnvironmentId) =>
655 Atom.make(
656 AsyncResult.success({
657 authenticated: true,
658 auth: {
659 policy: "remote-reachable",
660 bootstrapMethods: [],
661 sessionMethods: [],
662 sessionCookieName: "test",
663 },
664 scopes: [AuthSourceControlWriteScope],
665 }),
666 ),
667);