1import { vi } from "vite-plus/test";
2import {
3 EnvironmentId,
4 AuthSourceControlWriteScope,
5 WS_METHODS,
6 VcsProcessExitError,
7 type VcsListRefsInput,
8 type VcsListRefsResult,
9 type VcsStatusResult,
10 type ServerConfig,
11} from "@t3tools/contracts";
12import { describe, expect, it } from "@effect/vitest";
13import * as Deferred from "effect/Deferred";
14import * as Effect from "effect/Effect";
15import * as Fiber from "effect/Fiber";
16import * as Layer from "effect/Layer";
17import * as Logger from "effect/Logger";
18import * as Option from "effect/Option";
19import * as Queue from "effect/Queue";
20import * as Ref from "effect/Ref";
21import * as Stream from "effect/Stream";
22import * as SubscriptionRef from "effect/SubscriptionRef";
23import * as TestClock from "effect/testing/TestClock";
24import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
25
26import {
27 AVAILABLE_CONNECTION_STATE,
28 PrimaryConnectionTarget,
29 type PreparedConnection,
30 type SupervisorConnectionState,
31} from "../connection/model.ts";
32import * as EnvironmentRegistry from "../connection/registry.ts";
33import * as EnvironmentSupervisor from "../connection/supervisor.ts";
34import * as Persistence from "../platform/persistence.ts";
35import { EnvironmentRpcUnavailableError } from "../rpc/client.ts";
36import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
37import type { RpcSession } from "../rpc/session.ts";
38import { createGitEnvironmentAtoms } from "./git.ts";
39
40import {
41 commitVcsRefsRefresh,
42 createVcsEnvironmentAtoms,
43 makeCachedVcsRefsChanges,
44 resolveQuickAction,
45} from "./vcs.ts";
46import {
47 invalidateCachedVcsRefs,
48 invalidateVcsRefs,
49 vcsRefsCacheStateAtom,
50} from "./vcsRefInvalidation.ts";
51
52const TARGET = new PrimaryConnectionTarget({
53 environmentId: EnvironmentId.make("environment-1"),
54 label: "Test environment",
55 httpBaseUrl: "https://environment.example.test",
56 wsBaseUrl: "wss://environment.example.test",
57});
58
59const CONNECTED_CONNECTION_STATE: SupervisorConnectionState = {
60 ...AVAILABLE_CONNECTION_STATE,
61 desired: true,
62 network: "online",
63 phase: "connected",
64 attempt: 1,
65 generation: 1,
66};
67
68const CACHED_REFS: VcsListRefsResult = {
69 refs: [
70 {
71 name: "main",
72 current: true,
73 isDefault: true,
74 worktreePath: "/repo",
75 },
76 ],
77 isRepo: true,
78 hasPrimaryRemote: true,
79 nextCursor: null,
80 totalCount: 1,
81};
82
83const LIVE_REFS: VcsListRefsResult = {
84 ...CACHED_REFS,
85 refs: [
86 {
87 name: "release",
88 current: true,
89 isDefault: true,
90 worktreePath: "/repo",
91 },
92 ],
93};
94
95function session(client: WsRpcProtocolClient): RpcSession {
96 return {
97 client,
98 initialConfig: Effect.never,
99 subscribeServerConfig: (input = {}) => client.subscribeServerConfig(input),
100 ready: Effect.void,
101 probe: Effect.void,
102 inboundFrames: Effect.succeed(0),
103 closed: Effect.never,
104 };
105}
106
107function cacheWithRefs(
108 refs: Option.Option<VcsListRefsResult>,
109 overrides: Partial<Persistence.EnvironmentCacheStore["Service"]> = {},
110) {
111 return Persistence.EnvironmentCacheStore.of({
112 loadShell: () => Effect.succeedNone,
113 saveShell: () => Effect.void,
114 loadThread: () => Effect.succeedNone,
115 saveThread: () => Effect.void,
116 removeThread: () => Effect.void,
117 loadServerConfig: () => Effect.succeedNone,
118 saveServerConfig: () => Effect.void,
119 loadVcsRefs: () => Effect.succeed(refs),
120 saveVcsRefs: () => Effect.void,
121 removeVcsRefs: () => Effect.void,
122 clearVcsRefs: () => Effect.void,
123 clear: () => Effect.void,
124 ...overrides,
125 });
126}
127
128describe("branch subscriptions", () => {
129 it.effect.each([true, false])(
130 "shares and releases branch streams across session replacement (capability %s)",
131 (supported) =>
132 Effect.gen(function* () {
133 const released = yield* Queue.unbounded<string>();
134 let subscriptions = 0;
135 const clientFor = (refName: string, branchSupported: boolean) =>
136 ({
137 [branchSupported ? WS_METHODS.subscribeVcsBranch : WS_METHODS.subscribeVcsStatus]:
138 () => {
139 subscriptions += 1;
140 return Stream.concat(
141 Stream.make(
142 branchSupported
143 ? { refName, isRepo: true }
144 : { _tag: "snapshot", local: status({ refName }), remote: null },
145 ),
146 Stream.never,
147 ).pipe(Stream.ensuring(Queue.offer(released, refName)));
148 },
149 }) as unknown as WsRpcProtocolClient;
150 const branchSession = (refName: string, branchSupported: boolean): RpcSession => ({
151 ...session(clientFor(refName, branchSupported)),
152 initialConfig: Effect.succeed({
153 environment: { capabilities: branchSupported ? { vcsBranchStatus: true } : {} },
154 } as ServerConfig),
155 });
156 const activeSession = yield* SubscriptionRef.make(
157 Option.some(branchSession("main", supported)),
158 );
159 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
160 target: TARGET,
161 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
162 session: activeSession,
163 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
164 connect: Effect.void,
165 disconnect: Effect.void,
166 retryNow: Effect.void,
167 });
168 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
169 followStream: (_environmentId, stream) =>
170 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
171 } as EnvironmentRegistry.EnvironmentRegistry["Service"]);
172 const runtime = Atom.runtime(
173 Layer.merge(
174 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
175 Layer.succeed(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
176 ),
177 );
178 const atoms = createVcsEnvironmentAtoms(runtime);
179 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (value) =>
180 Effect.sync(() => value.dispose()),
181 );
182 const target = { environmentId: TARGET.environmentId, input: { cwd: "/repo" } };
183 const first = yield* Stream.toPull(
184 AtomRegistry.toStreamResult(registry, atoms.branch(target)),
185 );
186 expect(yield* first).toEqual([{ refName: "main", isRepo: true }]);
187 const second = yield* Stream.toPull(
188 AtomRegistry.toStreamResult(registry, atoms.branch(target)),
189 );
190 expect(yield* second).toEqual([{ refName: "main", isRepo: true }]);
191 expect(subscriptions).toBe(1);
192 yield* SubscriptionRef.set(
193 activeSession,
194 Option.some(branchSession("external", !supported)),
195 );
196 expect(yield* Queue.take(released)).toBe("main");
197 expect(yield* first).toEqual([{ refName: "external", isRepo: true }]);
198 expect(yield* second).toEqual([{ refName: "external", isRepo: true }]);
199 registry.dispose();
200 expect(yield* Queue.take(released)).toBe("external");
201 }).pipe(Effect.scoped),
202 );
203});
204
205describe("cached VCS refs", () => {
206 it("invalidates all ref streams in the mutated environment", () => {
207 const registry = AtomRegistry.make();
208 const environment = {
209 environmentId: TARGET.environmentId,
210 };
211 const otherEnvironment = {
212 environmentId: EnvironmentId.make("environment-2"),
213 };
214
215 expect(registry.get(vcsRefsCacheStateAtom(environment))).toEqual({
216 revision: 0,
217 persistedCacheReadable: true,
218 });
219 expect(registry.get(vcsRefsCacheStateAtom(otherEnvironment))).toEqual({
220 revision: 0,
221 persistedCacheReadable: true,
222 });
223
224 invalidateVcsRefs(registry, environment);
225
226 expect(registry.get(vcsRefsCacheStateAtom(environment))).toEqual({
227 revision: 1,
228 persistedCacheReadable: true,
229 });
230 expect(registry.get(vcsRefsCacheStateAtom(otherEnvironment))).toEqual({
231 revision: 0,
232 persistedCacheReadable: true,
233 });
234 registry.dispose();
235 });
236
237 it.effect("bypasses the server ref snapshot when a mounted list is invalidated", () =>
238 Effect.scoped(
239 Effect.gen(function* () {
240 const requests = yield* Queue.unbounded<VcsListRefsInput>();
241 const client = {
242 [WS_METHODS.vcsListRefs]: (input: VcsListRefsInput) =>
243 Effect.sync(() => {
244 Queue.offerUnsafe(requests, input);
245 return LIVE_REFS;
246 }),
247 } as unknown as WsRpcProtocolClient;
248 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
249 target: TARGET,
250 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
251 session: yield* SubscriptionRef.make(Option.some(session(client))),
252 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
253 connect: Effect.void,
254 disconnect: Effect.void,
255 retryNow: Effect.void,
256 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
257 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
258 _environmentId,
259 stream,
260 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
261 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
262 followStream,
263 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
264 const runtime = Atom.runtime(
265 Layer.merge(
266 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
267 Layer.succeed(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
268 ),
269 );
270 const atoms = createVcsEnvironmentAtoms(runtime);
271 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
272 Effect.sync(() => registry.dispose()),
273 );
274 yield* Effect.acquireRelease(
275 Effect.sync(() =>
276 registry.mount(
277 atoms.listRefs({
278 environmentId: TARGET.environmentId,
279 input: { cwd: "/repo", limit: 100 },
280 }),
281 ),
282 ),
283 (unmount) => Effect.sync(unmount),
284 );
285
286 expect(yield* Queue.take(requests)).toEqual({ cwd: "/repo", limit: 100 });
287
288 invalidateVcsRefs(registry, TARGET);
289
290 expect(yield* Queue.take(requests)).toEqual({ cwd: "/repo", limit: 100, refresh: true });
291
292 yield* Effect.acquireRelease(
293 Effect.sync(() =>
294 registry.mount(
295 atoms.listRefs({
296 environmentId: TARGET.environmentId,
297 input: { cwd: "/other-repo", limit: 100 },
298 }),
299 ),
300 ),
301 (unmount) => Effect.sync(unmount),
302 );
303 expect(yield* Queue.take(requests)).toEqual({
304 cwd: "/other-repo",
305 limit: 100,
306 refresh: true,
307 });
308
309 yield* Effect.acquireRelease(
310 Effect.sync(() =>
311 registry.mount(
312 atoms.listRefs({
313 environmentId: TARGET.environmentId,
314 input: { cwd: "/other-repo", query: "release", limit: 100 },
315 }),
316 ),
317 ),
318 (unmount) => Effect.sync(unmount),
319 );
320 expect(yield* Queue.take(requests)).toEqual({
321 cwd: "/other-repo",
322 query: "release",
323 limit: 100,
324 });
325 }),
326 ),
327 );
328
329 it.effect("does not repersist a refresh superseded by ref invalidation", () =>
330 Effect.scoped(
331 Effect.gen(function* () {
332 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
333 Effect.sync(() => registry.dispose()),
334 );
335 const saved = yield* Ref.make<ReadonlyArray<VcsListRefsResult>>([]);
336 const clears = yield* Ref.make(0);
337 const revisionsObservedDuringClear = yield* Ref.make<ReadonlyArray<number>>([]);
338 const cache = cacheWithRefs(Option.none(), {
339 saveVcsRefs: (_environmentId, _cwd, refs) =>
340 Ref.update(saved, (current) => [...current, refs]),
341 clearVcsRefs: () =>
342 Effect.all([
343 Ref.update(clears, (count) => count + 1),
344 Ref.update(revisionsObservedDuringClear, (current) => [
345 ...current,
346 registry.get(vcsRefsCacheStateAtom(TARGET)).revision,
347 ]),
348 ]).pipe(Effect.asVoid),
349 });
350
351 yield* invalidateCachedVcsRefs(registry, {
352 environmentId: TARGET.environmentId,
353 cwd: "/repo-worktree",
354 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
355
356 expect(
357 yield* commitVcsRefsRefresh(registry, cache, {
358 environmentId: TARGET.environmentId,
359 cwd: "/repo",
360 refs: CACHED_REFS,
361 expectedRevision: 0,
362 persist: true,
363 }),
364 ).toBe(false);
365
366 expect(yield* Ref.get(saved)).toEqual([]);
367 expect(yield* Ref.get(clears)).toBe(1);
368 expect(yield* Ref.get(revisionsObservedDuringClear)).toEqual([0]);
369 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual({
370 revision: 1,
371 persistedCacheReadable: true,
372 });
373
374 expect(
375 yield* commitVcsRefsRefresh(registry, cache, {
376 environmentId: TARGET.environmentId,
377 cwd: "/repo",
378 refs: LIVE_REFS,
379 expectedRevision: 1,
380 persist: true,
381 }),
382 ).toBe(true);
383 expect(yield* Ref.get(saved)).toEqual([LIVE_REFS]);
384 }),
385 ),
386 );
387
388 it.effect("invalidates persisted refs when ref-affecting commands settle", () =>
389 Effect.scoped(
390 Effect.gen(function* () {
391 const expectedError = new EnvironmentRpcUnavailableError({
392 environmentId: TARGET.environmentId,
393 message: "pull failed after fetching refs",
394 });
395 const client = {
396 [WS_METHODS.vcsPull]: () => Effect.fail(expectedError),
397 [WS_METHODS.vcsRefreshStatus]: () => Effect.void,
398 [WS_METHODS.gitPreparePullRequestThread]: () =>
399 Effect.succeed({ branch: "feature/pr", worktreePath: null }),
400 } as unknown as WsRpcProtocolClient;
401 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
402 target: TARGET,
403 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
404 session: yield* SubscriptionRef.make(Option.some(session(client))),
405 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
406 connect: Effect.void,
407 disconnect: Effect.void,
408 retryNow: Effect.void,
409 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
410 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
411 _environmentId,
412 effect,
413 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
414 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
415 run,
416 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
417 const clears = yield* Ref.make(0);
418 const runtime = Atom.runtime(
419 Layer.merge(
420 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
421 Layer.succeed(
422 Persistence.EnvironmentCacheStore,
423 cacheWithRefs(Option.none(), {
424 clearVcsRefs: () => Ref.update(clears, (count) => count + 1),
425 }),
426 ),
427 ),
428 );
429 const atoms = createVcsEnvironmentAtoms(runtime);
430 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
431 Effect.sync(() => registry.dispose()),
432 );
433
434 const result = yield* Effect.promise(() =>
435 atoms.pull.run(registry, {
436 environmentId: TARGET.environmentId,
437 input: { cwd: "/repo" },
438 }),
439 );
440
441 expect(AsyncResult.isFailure(result)).toBe(true);
442 expect(yield* Ref.get(clears)).toBe(1);
443 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(1);
444
445 const refreshResult = yield* Effect.promise(() =>
446 atoms.refreshStatus.run(registry, {
447 environmentId: TARGET.environmentId,
448 input: { cwd: "/repo" },
449 }),
450 );
451
452 expect(AsyncResult.isSuccess(refreshResult)).toBe(true);
453 expect(yield* Ref.get(clears)).toBe(2);
454 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(2);
455
456 const prepareResult = yield* Effect.promise(() =>
457 createGitEnvironmentAtoms(runtime).preparePullRequestThread.run(registry, {
458 environmentId: TARGET.environmentId,
459 input: { cwd: "/repo", reference: "42", mode: "local" },
460 }),
461 );
462
463 expect(AsyncResult.isSuccess(prepareResult)).toBe(true);
464 expect(yield* Ref.get(clears)).toBe(3);
465 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(3);
466 }),
467 ),
468 );
469
470 it.effect("a checkout does not wait behind a picker's local status refresh", () =>
471 Effect.scoped(
472 Effect.gen(function* () {
473 const refreshStarted = yield* Deferred.make<void>();
474 const releaseRefresh = yield* Deferred.make<void>();
475 const client = {
476 [WS_METHODS.vcsRefreshStatus]: () =>
477 Deferred.succeed(refreshStarted, undefined).pipe(
478 Effect.andThen(Deferred.await(releaseRefresh)),
479 ),
480 [WS_METHODS.vcsSwitchRef]: () => Effect.succeed({ refName: "feature" }),
481 } as unknown as WsRpcProtocolClient;
482 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
483 target: TARGET,
484 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
485 session: yield* SubscriptionRef.make(Option.some(session(client))),
486 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
487 connect: Effect.void,
488 disconnect: Effect.void,
489 retryNow: Effect.void,
490 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
491 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
492 _environmentId,
493 effect,
494 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
495 const runtime = Atom.runtime(
496 Layer.merge(
497 Layer.succeed(
498 EnvironmentRegistry.EnvironmentRegistry,
499 EnvironmentRegistry.EnvironmentRegistry.of({
500 run,
501 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]),
502 ),
503 Layer.succeed(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
504 ),
505 );
506 const atoms = createVcsEnvironmentAtoms(runtime);
507 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
508 Effect.sync(() => registry.dispose()),
509 );
510 const target = { environmentId: TARGET.environmentId, input: { cwd: "/repo" } };
511
512 const refresh = atoms.refreshLocalStatusForPicker.run(registry, {
513 ...target,
514 input: { cwd: "/repo", localOnly: true },
515 });
516 yield* Deferred.await(refreshStarted);
517 const checkout = yield* Effect.promise(() =>
518 atoms.switchRef.run(registry, {
519 ...target,
520 input: { cwd: "/repo", refName: "feature" },
521 }),
522 );
523
524 expect(AsyncResult.isSuccess(checkout)).toBe(true);
525 yield* Deferred.succeed(releaseRefresh, undefined);
526 expect(AsyncResult.isSuccess(yield* Effect.promise(() => refresh))).toBe(true);
527 }),
528 ),
529 );
530
531 it.effect("suppresses persisted snapshots after an environment-wide clear fails", () =>
532 Effect.scoped(
533 Effect.gen(function* () {
534 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
535 Effect.sync(() => registry.dispose()),
536 );
537 const persisted = yield* Ref.make<Option.Option<VcsListRefsResult>>(
538 Option.some(CACHED_REFS),
539 );
540 const loads = yield* Ref.make(0);
541 const clearAttempts = yield* Ref.make(0);
542 const cache = cacheWithRefs(Option.none(), {
543 loadVcsRefs: () =>
544 Ref.update(loads, (count) => count + 1).pipe(Effect.andThen(Ref.get(persisted))),
545 saveVcsRefs: (_environmentId, _cwd, refs) => Ref.set(persisted, Option.some(refs)),
546 clearVcsRefs: () =>
547 Ref.updateAndGet(clearAttempts, (count) => count + 1).pipe(
548 Effect.flatMap((attempt) =>
549 attempt === 1
550 ? Effect.fail(
551 new Persistence.ConnectionPersistenceError({
552 operation: "clear-vcs-refs",
553 message: "storage unavailable",
554 }),
555 )
556 : Ref.set(persisted, Option.none()),
557 ),
558 ),
559 });
560
561 yield* invalidateCachedVcsRefs(registry, {
562 environmentId: TARGET.environmentId,
563 cwd: "/repo",
564 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
565
566 const state = registry.get(vcsRefsCacheStateAtom(TARGET));
567 expect(state).toEqual({
568 revision: 1,
569 persistedCacheReadable: false,
570 });
571
572 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
573 target: TARGET,
574 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
575 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
576 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
577 connect: Effect.void,
578 disconnect: Effect.void,
579 retryNow: Effect.void,
580 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
581 const refs = yield* Stream.unwrap(
582 makeCachedVcsRefsChanges(
583 { cwd: "/repo", limit: 100 },
584 state.revision,
585 registry,
586 state.persistedCacheReadable,
587 ).pipe(
588 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
589 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
590 ),
591 ).pipe(Stream.runHead, Effect.forkChild({ startImmediately: true }));
592
593 yield* Effect.yieldNow;
594 expect(yield* Ref.get(loads)).toBe(0);
595 yield* Fiber.interrupt(refs);
596 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual(state);
597
598 expect(
599 yield* commitVcsRefsRefresh(registry, cache, {
600 environmentId: TARGET.environmentId,
601 cwd: "/repo",
602 refs: LIVE_REFS,
603 expectedRevision: state.revision,
604 persist: true,
605 }),
606 ).toBe(true);
607 const recoveredState = registry.get(vcsRefsCacheStateAtom(TARGET));
608 expect(recoveredState).toEqual({
609 revision: 1,
610 persistedCacheReadable: true,
611 });
612 expect(yield* Ref.get(clearAttempts)).toBe(2);
613
614 const recoveredRefs = yield* Stream.unwrap(
615 makeCachedVcsRefsChanges(
616 { cwd: "/repo", limit: 100 },
617 recoveredState.revision,
618 registry,
619 recoveredState.persistedCacheReadable,
620 ).pipe(
621 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
622 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
623 ),
624 ).pipe(Stream.runHead);
625
626 expect(Option.getOrThrow(recoveredRefs)).toEqual(LIVE_REFS);
627 expect(yield* Ref.get(loads)).toBe(1);
628 }),
629 ),
630 );
631
632 it.effect("loads an unfiltered branch list without a connection", () =>
633 Effect.scoped(
634 Effect.gen(function* () {
635 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
636 target: TARGET,
637 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
638 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
639 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
640 connect: Effect.void,
641 disconnect: Effect.void,
642 retryNow: Effect.void,
643 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
644 const refs = yield* Stream.unwrap(
645 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
646 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
647 Effect.provideService(
648 Persistence.EnvironmentCacheStore,
649 cacheWithRefs(Option.some(CACHED_REFS)),
650 ),
651 ),
652 ).pipe(Stream.runHead);
653
654 expect(Option.getOrThrow(refs)).toEqual(CACHED_REFS);
655 }),
656 ),
657 );
658
659 it.effect("waits silently for a session while the connection phase is still connected", () =>
660 Effect.scoped(
661 Effect.gen(function* () {
662 const logs: unknown[] = [];
663 const logger = Logger.make(({ message }) => logs.push(message));
664 const liveSession = yield* SubscriptionRef.make(Option.none<RpcSession>());
665 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
666 target: TARGET,
667 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
668 session: liveSession,
669 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
670 connect: Effect.void,
671 disconnect: Effect.void,
672 retryNow: Effect.void,
673 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
674 const refs = yield* Stream.unwrap(
675 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
676 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
677 Effect.provideService(
678 Persistence.EnvironmentCacheStore,
679 cacheWithRefs(Option.some(CACHED_REFS)),
680 ),
681 ),
682 ).pipe(
683 Stream.take(2),
684 Stream.runCollect,
685 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
686 Effect.forkChild({ startImmediately: true }),
687 );
688
689 yield* TestClock.adjust("1 minute");
690 expect(logs).toEqual([]);
691
692 const client = {
693 [WS_METHODS.vcsListRefs]: () => Effect.succeed(LIVE_REFS),
694 } as unknown as WsRpcProtocolClient;
695 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
696 expect(yield* Fiber.join(refs)).toEqual([CACHED_REFS, LIVE_REFS]);
697 expect(logs).toEqual([]);
698 }),
699 ),
700 );
701
702 it.effect("cancels an in-flight refresh and restarts when the RPC session changes", () =>
703 Effect.scoped(
704 Effect.gen(function* () {
705 const started = yield* Deferred.make<void>();
706 const interrupted = yield* Deferred.make<void>();
707 const calls = yield* Ref.make(0);
708 const interruptions = yield* Ref.make(0);
709 const client = {
710 [WS_METHODS.vcsListRefs]: () =>
711 Ref.updateAndGet(calls, (count) => count + 1).pipe(
712 Effect.flatMap((count) =>
713 count === 1
714 ? Deferred.succeed(started, undefined).pipe(
715 Effect.andThen(Effect.never),
716 Effect.onInterrupt(() =>
717 Ref.update(interruptions, (interruptions) => interruptions + 1).pipe(
718 Effect.andThen(Deferred.succeed(interrupted, undefined)),
719 ),
720 ),
721 )
722 : Effect.succeed(LIVE_REFS),
723 ),
724 ),
725 } as unknown as WsRpcProtocolClient;
726 const liveSession = yield* SubscriptionRef.make(Option.some(session(client)));
727 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
728 target: TARGET,
729 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
730 session: liveSession,
731 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
732 connect: Effect.void,
733 disconnect: Effect.void,
734 retryNow: Effect.void,
735 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
736
737 const fiber = yield* Stream.unwrap(
738 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
739 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
740 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
741 ),
742 ).pipe(Stream.runHead, Effect.forkChild);
743
744 yield* Deferred.await(started);
745 yield* SubscriptionRef.set(liveSession, Option.none());
746 yield* Deferred.await(interrupted);
747 expect(yield* Ref.get(interruptions)).toBe(1);
748
749 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
750 expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS);
751 expect(yield* Ref.get(calls)).toBe(2);
752 }),
753 ),
754 );
755
756 it.effect("retries Git failures and cancels pending retries when the session is removed", () =>
757 Effect.scoped(
758 Effect.gen(function* () {
759 const calls = yield* Ref.make(0);
760 const warnings = yield* Queue.unbounded<unknown>();
761 const logger = Logger.make(({ message }) => Queue.offerUnsafe(warnings, message));
762 const failure = new VcsProcessExitError({
763 operation: "listRefs",
764 command: "git",
765 cwd: "/repo",
766 exitCode: 128,
767 detail: "Could not read refs.",
768 });
769 const client = {
770 [WS_METHODS.vcsListRefs]: () =>
771 Ref.updateAndGet(calls, (count) => count + 1).pipe(
772 Effect.flatMap((count) =>
773 count < 3 ? Effect.fail(failure) : Effect.succeed(LIVE_REFS),
774 ),
775 ),
776 } as unknown as WsRpcProtocolClient;
777 const liveSession = yield* SubscriptionRef.make(Option.some(session(client)));
778 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
779 target: TARGET,
780 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
781 session: liveSession,
782 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
783 connect: Effect.void,
784 disconnect: Effect.void,
785 retryNow: Effect.void,
786 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
787 const refs = yield* Stream.unwrap(
788 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
789 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
790 Effect.provideService(
791 Persistence.EnvironmentCacheStore,
792 cacheWithRefs(Option.some(CACHED_REFS)),
793 ),
794 ),
795 ).pipe(
796 Stream.take(2),
797 Stream.runCollect,
798 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
799 Effect.forkChild({ startImmediately: true }),
800 );
801
802 yield* Queue.take(warnings);
803 expect(yield* Ref.get(calls)).toBe(1);
804 yield* TestClock.adjust("1 second");
805 expect(yield* Ref.get(calls)).toBe(2);
806 expect(yield* Queue.size(warnings)).toBe(1);
807
808 yield* SubscriptionRef.set(liveSession, Option.none());
809 yield* TestClock.adjust("1 minute");
810 expect(yield* Ref.get(calls)).toBe(2);
811 expect(yield* Queue.size(warnings)).toBe(1);
812
813 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
814 expect(yield* Fiber.join(refs)).toEqual([CACHED_REFS, LIVE_REFS]);
815 expect(yield* Ref.get(calls)).toBe(3);
816 }),
817 ),
818 );
819
820 it.effect("emits persisted refs before a live refresh", () =>
821 Effect.scoped(
822 Effect.gen(function* () {
823 const client = {
824 [WS_METHODS.vcsListRefs]: () => Effect.succeed(LIVE_REFS),
825 } as unknown as WsRpcProtocolClient;
826 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
827 target: TARGET,
828 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
829 session: yield* SubscriptionRef.make(Option.some(session(client))),
830 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
831 connect: Effect.void,
832 disconnect: Effect.void,
833 retryNow: Effect.void,
834 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
835
836 const refs = yield* Stream.unwrap(
837 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
838 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
839 Effect.provideService(
840 Persistence.EnvironmentCacheStore,
841 cacheWithRefs(Option.some(CACHED_REFS)),
842 ),
843 ),
844 ).pipe(Stream.take(2), Stream.runCollect);
845
846 expect(refs).toEqual([CACHED_REFS, LIVE_REFS]);
847 }),
848 ),
849 );
850});
851
852vi.mock("./session.ts", () => ({
853 createEnvironmentSessionAtoms: () => ({ sessionStateAtom: grantedSessions }),
854}));
855const grantedSessions = Atom.family((_id: EnvironmentId) =>
856 Atom.make(
857 AsyncResult.success({
858 authenticated: true,
859 auth: {
860 policy: "remote-reachable",
861 bootstrapMethods: [],
862 sessionMethods: [],
863 sessionCookieName: "test",
864 },
865 scopes: [AuthSourceControlWriteScope],
866 }),
867 ),
868);
869
870function status(overrides: Partial<VcsStatusResult> = {}): VcsStatusResult {
871 return {
872 isRepo: true,
873 hasPrimaryRemote: true,
874 isDefaultRef: false,
875 refName: "feature/test",
876 hasWorkingTreeChanges: false,
877 workingTree: {
878 files: [],
879 insertions: 0,
880 deletions: 0,
881 },
882 hasUpstream: true,
883 aheadCount: 0,
884 behindCount: 0,
885 pr: null,
886 ...overrides,
887 };
888}
889
890describe("resolveQuickAction", () => {
891 it("commits local changes without pushing when the branch is behind upstream", () => {
892 const behind = status({ hasWorkingTreeChanges: true, behindCount: 2 });
893 const commit = { label: "Commit", disabled: false, kind: "run_action", action: "commit" };
894 expect(resolveQuickAction(behind, false)).toEqual(commit);
895 expect(
896 resolveQuickAction(
897 {
898 ...behind,
899 pr: {
900 number: 13,
901 title: "Open PR",
902 url: "https://example.com/pr/13",
903 baseRef: "main",
904 headRef: "feature/test",
905 state: "open",
906 },
907 },
908 false,
909 ),
910 ).toEqual(commit);
911 });
912});