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 }),
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.gen(function* () {
239 const requests = yield* Queue.unbounded<VcsListRefsInput>();
240 const client = {
241 [WS_METHODS.vcsListRefs]: (input: VcsListRefsInput) =>
242 Effect.sync(() => {
243 Queue.offerUnsafe(requests, input);
244 return LIVE_REFS;
245 }),
246 } as unknown as WsRpcProtocolClient;
247 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
248 target: TARGET,
249 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
250 session: yield* SubscriptionRef.make(Option.some(session(client))),
251 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
252 connect: Effect.void,
253 disconnect: Effect.void,
254 retryNow: Effect.void,
255 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
256 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
257 _environmentId,
258 stream,
259 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
260 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
261 followStream,
262 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
263 const runtime = Atom.runtime(
264 Layer.merge(
265 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
266 Layer.succeed(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
267 ),
268 );
269 const atoms = createVcsEnvironmentAtoms(runtime);
270 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
271 Effect.sync(() => registry.dispose()),
272 );
273 yield* Effect.acquireRelease(
274 Effect.sync(() =>
275 registry.mount(
276 atoms.listRefs({
277 environmentId: TARGET.environmentId,
278 input: { cwd: "/repo", limit: 100 },
279 }),
280 ),
281 ),
282 (unmount) => Effect.sync(unmount),
283 );
284
285 expect(yield* Queue.take(requests)).toEqual({ cwd: "/repo", limit: 100 });
286
287 invalidateVcsRefs(registry, TARGET);
288
289 expect(yield* Queue.take(requests)).toEqual({ cwd: "/repo", limit: 100, refresh: true });
290
291 yield* Effect.acquireRelease(
292 Effect.sync(() =>
293 registry.mount(
294 atoms.listRefs({
295 environmentId: TARGET.environmentId,
296 input: { cwd: "/other-repo", limit: 100 },
297 }),
298 ),
299 ),
300 (unmount) => Effect.sync(unmount),
301 );
302 expect(yield* Queue.take(requests)).toEqual({
303 cwd: "/other-repo",
304 limit: 100,
305 refresh: true,
306 });
307
308 yield* Effect.acquireRelease(
309 Effect.sync(() =>
310 registry.mount(
311 atoms.listRefs({
312 environmentId: TARGET.environmentId,
313 input: { cwd: "/other-repo", query: "release", limit: 100 },
314 }),
315 ),
316 ),
317 (unmount) => Effect.sync(unmount),
318 );
319 expect(yield* Queue.take(requests)).toEqual({
320 cwd: "/other-repo",
321 query: "release",
322 limit: 100,
323 });
324 }),
325 );
326
327 it.effect("does not repersist a refresh superseded by ref invalidation", () =>
328 Effect.gen(function* () {
329 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
330 Effect.sync(() => registry.dispose()),
331 );
332 const saved = yield* Ref.make<ReadonlyArray<VcsListRefsResult>>([]);
333 const clears = yield* Ref.make(0);
334 const revisionsObservedDuringClear = yield* Ref.make<ReadonlyArray<number>>([]);
335 const cache = cacheWithRefs(Option.none(), {
336 saveVcsRefs: (_environmentId, _cwd, refs) =>
337 Ref.update(saved, (current) => [...current, refs]),
338 clearVcsRefs: () =>
339 Effect.all([
340 Ref.update(clears, (count) => count + 1),
341 Ref.update(revisionsObservedDuringClear, (current) => [
342 ...current,
343 registry.get(vcsRefsCacheStateAtom(TARGET)).revision,
344 ]),
345 ]).pipe(Effect.asVoid),
346 });
347
348 yield* invalidateCachedVcsRefs(registry, {
349 environmentId: TARGET.environmentId,
350 cwd: "/repo-worktree",
351 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
352
353 expect(
354 yield* commitVcsRefsRefresh(registry, cache, {
355 environmentId: TARGET.environmentId,
356 cwd: "/repo",
357 refs: CACHED_REFS,
358 expectedRevision: 0,
359 persist: true,
360 }),
361 ).toBe(false);
362
363 expect(yield* Ref.get(saved)).toEqual([]);
364 expect(yield* Ref.get(clears)).toBe(1);
365 expect(yield* Ref.get(revisionsObservedDuringClear)).toEqual([0]);
366 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual({
367 revision: 1,
368 persistedCacheReadable: true,
369 });
370
371 expect(
372 yield* commitVcsRefsRefresh(registry, cache, {
373 environmentId: TARGET.environmentId,
374 cwd: "/repo",
375 refs: LIVE_REFS,
376 expectedRevision: 1,
377 persist: true,
378 }),
379 ).toBe(true);
380 expect(yield* Ref.get(saved)).toEqual([LIVE_REFS]);
381 }),
382 );
383
384 it.effect("invalidates persisted refs when ref-affecting commands settle", () =>
385 Effect.gen(function* () {
386 const expectedError = new EnvironmentRpcUnavailableError({
387 environmentId: TARGET.environmentId,
388 message: "pull failed after fetching refs",
389 });
390 const client = {
391 [WS_METHODS.vcsPull]: () => Effect.fail(expectedError),
392 [WS_METHODS.vcsRefreshStatus]: () => Effect.void,
393 [WS_METHODS.gitPreparePullRequestThread]: () =>
394 Effect.succeed({ branch: "feature/pr", worktreePath: null }),
395 } as unknown as WsRpcProtocolClient;
396 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
397 target: TARGET,
398 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
399 session: yield* SubscriptionRef.make(Option.some(session(client))),
400 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
401 connect: Effect.void,
402 disconnect: Effect.void,
403 retryNow: Effect.void,
404 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
405 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
406 _environmentId,
407 effect,
408 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
409 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
410 run,
411 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
412 const clears = yield* Ref.make(0);
413 const runtime = Atom.runtime(
414 Layer.merge(
415 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
416 Layer.succeed(
417 Persistence.EnvironmentCacheStore,
418 cacheWithRefs(Option.none(), {
419 clearVcsRefs: () => Ref.update(clears, (count) => count + 1),
420 }),
421 ),
422 ),
423 );
424 const atoms = createVcsEnvironmentAtoms(runtime);
425 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
426 Effect.sync(() => registry.dispose()),
427 );
428
429 const result = yield* Effect.promise(() =>
430 atoms.pull.run(registry, {
431 environmentId: TARGET.environmentId,
432 input: { cwd: "/repo" },
433 }),
434 );
435
436 expect(AsyncResult.isFailure(result)).toBe(true);
437 expect(yield* Ref.get(clears)).toBe(1);
438 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(1);
439
440 const refreshResult = yield* Effect.promise(() =>
441 atoms.refreshStatus.run(registry, {
442 environmentId: TARGET.environmentId,
443 input: { cwd: "/repo" },
444 }),
445 );
446
447 expect(AsyncResult.isSuccess(refreshResult)).toBe(true);
448 expect(yield* Ref.get(clears)).toBe(2);
449 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(2);
450
451 const prepareResult = yield* Effect.promise(() =>
452 createGitEnvironmentAtoms(runtime).preparePullRequestThread.run(registry, {
453 environmentId: TARGET.environmentId,
454 input: { cwd: "/repo", reference: "42", mode: "local" },
455 }),
456 );
457
458 expect(AsyncResult.isSuccess(prepareResult)).toBe(true);
459 expect(yield* Ref.get(clears)).toBe(3);
460 expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(3);
461 }),
462 );
463
464 it.effect("a checkout does not wait behind a picker's local status refresh", () =>
465 Effect.gen(function* () {
466 const refreshStarted = yield* Deferred.make<void>();
467 const releaseRefresh = yield* Deferred.make<void>();
468 const client = {
469 [WS_METHODS.vcsRefreshStatus]: () =>
470 Deferred.succeed(refreshStarted, undefined).pipe(
471 Effect.andThen(Deferred.await(releaseRefresh)),
472 ),
473 [WS_METHODS.vcsSwitchRef]: () => Effect.succeed({ refName: "feature" }),
474 } as unknown as WsRpcProtocolClient;
475 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
476 target: TARGET,
477 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
478 session: yield* SubscriptionRef.make(Option.some(session(client))),
479 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
480 connect: Effect.void,
481 disconnect: Effect.void,
482 retryNow: Effect.void,
483 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
484 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
485 _environmentId,
486 effect,
487 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
488 const runtime = Atom.runtime(
489 Layer.merge(
490 Layer.succeed(
491 EnvironmentRegistry.EnvironmentRegistry,
492 EnvironmentRegistry.EnvironmentRegistry.of({
493 run,
494 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]),
495 ),
496 Layer.succeed(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
497 ),
498 );
499 const atoms = createVcsEnvironmentAtoms(runtime);
500 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
501 Effect.sync(() => registry.dispose()),
502 );
503 const target = { environmentId: TARGET.environmentId, input: { cwd: "/repo" } };
504
505 const refresh = atoms.refreshLocalStatusForPicker.run(registry, {
506 ...target,
507 input: { cwd: "/repo", localOnly: true },
508 });
509 yield* Deferred.await(refreshStarted);
510 const checkout = yield* Effect.promise(() =>
511 atoms.switchRef.run(registry, {
512 ...target,
513 input: { cwd: "/repo", refName: "feature" },
514 }),
515 );
516
517 expect(AsyncResult.isSuccess(checkout)).toBe(true);
518 yield* Deferred.succeed(releaseRefresh, undefined);
519 expect(AsyncResult.isSuccess(yield* Effect.promise(() => refresh))).toBe(true);
520 }),
521 );
522
523 it.effect("suppresses persisted snapshots after an environment-wide clear fails", () =>
524 Effect.gen(function* () {
525 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
526 Effect.sync(() => registry.dispose()),
527 );
528 const persisted = yield* Ref.make<Option.Option<VcsListRefsResult>>(Option.some(CACHED_REFS));
529 const loads = yield* Ref.make(0);
530 const clearAttempts = yield* Ref.make(0);
531 const cache = cacheWithRefs(Option.none(), {
532 loadVcsRefs: () =>
533 Ref.update(loads, (count) => count + 1).pipe(Effect.andThen(Ref.get(persisted))),
534 saveVcsRefs: (_environmentId, _cwd, refs) => Ref.set(persisted, Option.some(refs)),
535 clearVcsRefs: () =>
536 Ref.updateAndGet(clearAttempts, (count) => count + 1).pipe(
537 Effect.flatMap((attempt) =>
538 attempt === 1
539 ? Effect.fail(
540 new Persistence.ConnectionPersistenceError({
541 operation: "clear-vcs-refs",
542 message: "storage unavailable",
543 }),
544 )
545 : Ref.set(persisted, Option.none()),
546 ),
547 ),
548 });
549
550 yield* invalidateCachedVcsRefs(registry, {
551 environmentId: TARGET.environmentId,
552 cwd: "/repo",
553 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
554
555 const state = registry.get(vcsRefsCacheStateAtom(TARGET));
556 expect(state).toEqual({
557 revision: 1,
558 persistedCacheReadable: false,
559 });
560
561 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
562 target: TARGET,
563 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
564 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
565 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
566 connect: Effect.void,
567 disconnect: Effect.void,
568 retryNow: Effect.void,
569 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
570 const refs = yield* Stream.unwrap(
571 makeCachedVcsRefsChanges(
572 { cwd: "/repo", limit: 100 },
573 state.revision,
574 registry,
575 state.persistedCacheReadable,
576 ).pipe(
577 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
578 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
579 ),
580 ).pipe(Stream.runHead, Effect.forkChild({ startImmediately: true }));
581
582 yield* Effect.yieldNow;
583 expect(yield* Ref.get(loads)).toBe(0);
584 yield* Fiber.interrupt(refs);
585 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual(state);
586
587 expect(
588 yield* commitVcsRefsRefresh(registry, cache, {
589 environmentId: TARGET.environmentId,
590 cwd: "/repo",
591 refs: LIVE_REFS,
592 expectedRevision: state.revision,
593 persist: true,
594 }),
595 ).toBe(true);
596 const recoveredState = registry.get(vcsRefsCacheStateAtom(TARGET));
597 expect(recoveredState).toEqual({
598 revision: 1,
599 persistedCacheReadable: true,
600 });
601 expect(yield* Ref.get(clearAttempts)).toBe(2);
602
603 const recoveredRefs = yield* Stream.unwrap(
604 makeCachedVcsRefsChanges(
605 { cwd: "/repo", limit: 100 },
606 recoveredState.revision,
607 registry,
608 recoveredState.persistedCacheReadable,
609 ).pipe(
610 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
611 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
612 ),
613 ).pipe(Stream.runHead);
614
615 expect(Option.getOrThrow(recoveredRefs)).toEqual(LIVE_REFS);
616 expect(yield* Ref.get(loads)).toBe(1);
617 }),
618 );
619
620 it.effect("loads an unfiltered branch list without a connection", () =>
621 Effect.gen(function* () {
622 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
623 target: TARGET,
624 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
625 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
626 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
627 connect: Effect.void,
628 disconnect: Effect.void,
629 retryNow: Effect.void,
630 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
631 const refs = yield* Stream.unwrap(
632 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
633 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
634 Effect.provideService(
635 Persistence.EnvironmentCacheStore,
636 cacheWithRefs(Option.some(CACHED_REFS)),
637 ),
638 ),
639 ).pipe(Stream.runHead);
640
641 expect(Option.getOrThrow(refs)).toEqual(CACHED_REFS);
642 }),
643 );
644
645 it.effect("waits silently for a session while the connection phase is still connected", () =>
646 Effect.gen(function* () {
647 const logs: unknown[] = [];
648 const logger = Logger.make(({ message }) => logs.push(message));
649 const liveSession = yield* SubscriptionRef.make(Option.none<RpcSession>());
650 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
651 target: TARGET,
652 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
653 session: liveSession,
654 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
655 connect: Effect.void,
656 disconnect: Effect.void,
657 retryNow: Effect.void,
658 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
659 const refs = yield* Stream.unwrap(
660 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
661 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
662 Effect.provideService(
663 Persistence.EnvironmentCacheStore,
664 cacheWithRefs(Option.some(CACHED_REFS)),
665 ),
666 ),
667 ).pipe(
668 Stream.take(2),
669 Stream.runCollect,
670 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
671 Effect.forkChild({ startImmediately: true }),
672 );
673
674 yield* TestClock.adjust("1 minute");
675 expect(logs).toEqual([]);
676
677 const client = {
678 [WS_METHODS.vcsListRefs]: () => Effect.succeed(LIVE_REFS),
679 } as unknown as WsRpcProtocolClient;
680 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
681 expect(yield* Fiber.join(refs)).toEqual([CACHED_REFS, LIVE_REFS]);
682 expect(logs).toEqual([]);
683 }),
684 );
685
686 it.effect("cancels an in-flight refresh and restarts when the RPC session changes", () =>
687 Effect.gen(function* () {
688 const started = yield* Deferred.make<void>();
689 const interrupted = yield* Deferred.make<void>();
690 const calls = yield* Ref.make(0);
691 const interruptions = yield* Ref.make(0);
692 const client = {
693 [WS_METHODS.vcsListRefs]: () =>
694 Ref.updateAndGet(calls, (count) => count + 1).pipe(
695 Effect.flatMap((count) =>
696 count === 1
697 ? Deferred.succeed(started, undefined).pipe(
698 Effect.andThen(Effect.never),
699 Effect.onInterrupt(() =>
700 Ref.update(interruptions, (interruptions) => interruptions + 1).pipe(
701 Effect.andThen(Deferred.succeed(interrupted, undefined)),
702 ),
703 ),
704 )
705 : Effect.succeed(LIVE_REFS),
706 ),
707 ),
708 } as unknown as WsRpcProtocolClient;
709 const liveSession = yield* SubscriptionRef.make(Option.some(session(client)));
710 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
711 target: TARGET,
712 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
713 session: liveSession,
714 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
715 connect: Effect.void,
716 disconnect: Effect.void,
717 retryNow: Effect.void,
718 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
719
720 const fiber = yield* Stream.unwrap(
721 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
722 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
723 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
724 ),
725 ).pipe(Stream.runHead, Effect.forkChild);
726
727 yield* Deferred.await(started);
728 yield* SubscriptionRef.set(liveSession, Option.none());
729 yield* Deferred.await(interrupted);
730 expect(yield* Ref.get(interruptions)).toBe(1);
731
732 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
733 expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS);
734 expect(yield* Ref.get(calls)).toBe(2);
735 }),
736 );
737
738 it.effect("retries Git failures and cancels pending retries when the session is removed", () =>
739 Effect.gen(function* () {
740 const calls = yield* Ref.make(0);
741 const warnings = yield* Queue.unbounded<unknown>();
742 const logger = Logger.make(({ message }) => Queue.offerUnsafe(warnings, message));
743 const failure = new VcsProcessExitError({
744 operation: "listRefs",
745 command: "git",
746 cwd: "/repo",
747 exitCode: 128,
748 detail: "Could not read refs.",
749 });
750 const client = {
751 [WS_METHODS.vcsListRefs]: () =>
752 Ref.updateAndGet(calls, (count) => count + 1).pipe(
753 Effect.flatMap((count) =>
754 count < 3 ? Effect.fail(failure) : Effect.succeed(LIVE_REFS),
755 ),
756 ),
757 } as unknown as WsRpcProtocolClient;
758 const liveSession = yield* SubscriptionRef.make(Option.some(session(client)));
759 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
760 target: TARGET,
761 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
762 session: liveSession,
763 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
764 connect: Effect.void,
765 disconnect: Effect.void,
766 retryNow: Effect.void,
767 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
768 const refs = yield* Stream.unwrap(
769 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
770 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
771 Effect.provideService(
772 Persistence.EnvironmentCacheStore,
773 cacheWithRefs(Option.some(CACHED_REFS)),
774 ),
775 ),
776 ).pipe(
777 Stream.take(2),
778 Stream.runCollect,
779 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
780 Effect.forkChild({ startImmediately: true }),
781 );
782
783 yield* Queue.take(warnings);
784 expect(yield* Ref.get(calls)).toBe(1);
785 yield* TestClock.adjust("1 second");
786 expect(yield* Ref.get(calls)).toBe(2);
787 expect(yield* Queue.size(warnings)).toBe(1);
788
789 yield* SubscriptionRef.set(liveSession, Option.none());
790 yield* TestClock.adjust("1 minute");
791 expect(yield* Ref.get(calls)).toBe(2);
792 expect(yield* Queue.size(warnings)).toBe(1);
793
794 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
795 expect(yield* Fiber.join(refs)).toEqual([CACHED_REFS, LIVE_REFS]);
796 expect(yield* Ref.get(calls)).toBe(3);
797 }),
798 );
799
800 it.effect("emits persisted refs before a live refresh", () =>
801 Effect.gen(function* () {
802 const client = {
803 [WS_METHODS.vcsListRefs]: () => Effect.succeed(LIVE_REFS),
804 } as unknown as WsRpcProtocolClient;
805 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
806 target: TARGET,
807 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
808 session: yield* SubscriptionRef.make(Option.some(session(client))),
809 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
810 connect: Effect.void,
811 disconnect: Effect.void,
812 retryNow: Effect.void,
813 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
814
815 const refs = yield* Stream.unwrap(
816 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
817 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
818 Effect.provideService(
819 Persistence.EnvironmentCacheStore,
820 cacheWithRefs(Option.some(CACHED_REFS)),
821 ),
822 ),
823 ).pipe(Stream.take(2), Stream.runCollect);
824
825 expect(refs).toEqual([CACHED_REFS, LIVE_REFS]);
826 }),
827 );
828});
829
830vi.mock("./session.ts", () => ({
831 createEnvironmentSessionAtoms: () => ({ sessionStateAtom: grantedSessions }),
832}));
833const grantedSessions = Atom.family((_id: EnvironmentId) =>
834 Atom.make(
835 AsyncResult.success({
836 authenticated: true,
837 auth: {
838 policy: "remote-reachable",
839 bootstrapMethods: [],
840 sessionMethods: [],
841 sessionCookieName: "test",
842 },
843 scopes: [AuthSourceControlWriteScope],
844 }),
845 ),
846);
847
848function status(overrides: Partial<VcsStatusResult> = {}): VcsStatusResult {
849 return {
850 isRepo: true,
851 hasPrimaryRemote: true,
852 isDefaultRef: false,
853 refName: "feature/test",
854 hasWorkingTreeChanges: false,
855 workingTree: {
856 files: [],
857 insertions: 0,
858 deletions: 0,
859 },
860 hasUpstream: true,
861 aheadCount: 0,
862 behindCount: 0,
863 pr: null,
864 ...overrides,
865 };
866}
867
868describe("resolveQuickAction", () => {
869 it("commits local changes without pushing when the branch is behind upstream", () => {
870 const behind = status({ hasWorkingTreeChanges: true, behindCount: 2 });
871 const commit = { label: "Commit", disabled: false, kind: "run_action", action: "commit" };
872 expect(resolveQuickAction(behind, false)).toEqual(commit);
873 expect(
874 resolveQuickAction(
875 {
876 ...behind,
877 pr: {
878 number: 13,
879 title: "Open PR",
880 url: "https://example.com/pr/13",
881 baseRef: "main",
882 headRef: "feature/test",
883 state: "open",
884 },
885 },
886 false,
887 ),
888 ).toEqual(commit);
889 });
890});