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("suppresses persisted snapshots after an environment-wide clear fails", () =>
471 Effect.scoped(
472 Effect.gen(function* () {
473 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
474 Effect.sync(() => registry.dispose()),
475 );
476 const persisted = yield* Ref.make<Option.Option<VcsListRefsResult>>(
477 Option.some(CACHED_REFS),
478 );
479 const loads = yield* Ref.make(0);
480 const clearAttempts = yield* Ref.make(0);
481 const cache = cacheWithRefs(Option.none(), {
482 loadVcsRefs: () =>
483 Ref.update(loads, (count) => count + 1).pipe(Effect.andThen(Ref.get(persisted))),
484 saveVcsRefs: (_environmentId, _cwd, refs) => Ref.set(persisted, Option.some(refs)),
485 clearVcsRefs: () =>
486 Ref.updateAndGet(clearAttempts, (count) => count + 1).pipe(
487 Effect.flatMap((attempt) =>
488 attempt === 1
489 ? Effect.fail(
490 new Persistence.ConnectionPersistenceError({
491 operation: "clear-vcs-refs",
492 message: "storage unavailable",
493 }),
494 )
495 : Ref.set(persisted, Option.none()),
496 ),
497 ),
498 });
499
500 yield* invalidateCachedVcsRefs(registry, {
501 environmentId: TARGET.environmentId,
502 cwd: "/repo",
503 }).pipe(Effect.provideService(Persistence.EnvironmentCacheStore, cache));
504
505 const state = registry.get(vcsRefsCacheStateAtom(TARGET));
506 expect(state).toEqual({
507 revision: 1,
508 persistedCacheReadable: false,
509 });
510
511 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
512 target: TARGET,
513 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
514 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
515 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
516 connect: Effect.void,
517 disconnect: Effect.void,
518 retryNow: Effect.void,
519 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
520 const refs = yield* Stream.unwrap(
521 makeCachedVcsRefsChanges(
522 { cwd: "/repo", limit: 100 },
523 state.revision,
524 registry,
525 state.persistedCacheReadable,
526 ).pipe(
527 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
528 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
529 ),
530 ).pipe(Stream.runHead, Effect.forkChild({ startImmediately: true }));
531
532 yield* Effect.yieldNow;
533 expect(yield* Ref.get(loads)).toBe(0);
534 yield* Fiber.interrupt(refs);
535 expect(registry.get(vcsRefsCacheStateAtom(TARGET))).toEqual(state);
536
537 expect(
538 yield* commitVcsRefsRefresh(registry, cache, {
539 environmentId: TARGET.environmentId,
540 cwd: "/repo",
541 refs: LIVE_REFS,
542 expectedRevision: state.revision,
543 persist: true,
544 }),
545 ).toBe(true);
546 const recoveredState = registry.get(vcsRefsCacheStateAtom(TARGET));
547 expect(recoveredState).toEqual({
548 revision: 1,
549 persistedCacheReadable: true,
550 });
551 expect(yield* Ref.get(clearAttempts)).toBe(2);
552
553 const recoveredRefs = yield* Stream.unwrap(
554 makeCachedVcsRefsChanges(
555 { cwd: "/repo", limit: 100 },
556 recoveredState.revision,
557 registry,
558 recoveredState.persistedCacheReadable,
559 ).pipe(
560 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
561 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
562 ),
563 ).pipe(Stream.runHead);
564
565 expect(Option.getOrThrow(recoveredRefs)).toEqual(LIVE_REFS);
566 expect(yield* Ref.get(loads)).toBe(1);
567 }),
568 ),
569 );
570
571 it.effect("loads an unfiltered branch list without a connection", () =>
572 Effect.scoped(
573 Effect.gen(function* () {
574 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
575 target: TARGET,
576 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
577 session: yield* SubscriptionRef.make(Option.none<RpcSession>()),
578 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
579 connect: Effect.void,
580 disconnect: Effect.void,
581 retryNow: Effect.void,
582 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
583 const refs = yield* Stream.unwrap(
584 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
585 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
586 Effect.provideService(
587 Persistence.EnvironmentCacheStore,
588 cacheWithRefs(Option.some(CACHED_REFS)),
589 ),
590 ),
591 ).pipe(Stream.runHead);
592
593 expect(Option.getOrThrow(refs)).toEqual(CACHED_REFS);
594 }),
595 ),
596 );
597
598 it.effect("waits silently for a session while the connection phase is still connected", () =>
599 Effect.scoped(
600 Effect.gen(function* () {
601 const logs: unknown[] = [];
602 const logger = Logger.make(({ message }) => logs.push(message));
603 const liveSession = yield* SubscriptionRef.make(Option.none<RpcSession>());
604 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
605 target: TARGET,
606 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
607 session: liveSession,
608 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
609 connect: Effect.void,
610 disconnect: Effect.void,
611 retryNow: Effect.void,
612 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
613 const refs = yield* Stream.unwrap(
614 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
615 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
616 Effect.provideService(
617 Persistence.EnvironmentCacheStore,
618 cacheWithRefs(Option.some(CACHED_REFS)),
619 ),
620 ),
621 ).pipe(
622 Stream.take(2),
623 Stream.runCollect,
624 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
625 Effect.forkChild({ startImmediately: true }),
626 );
627
628 yield* TestClock.adjust("1 minute");
629 expect(logs).toEqual([]);
630
631 const client = {
632 [WS_METHODS.vcsListRefs]: () => Effect.succeed(LIVE_REFS),
633 } as unknown as WsRpcProtocolClient;
634 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
635 expect(yield* Fiber.join(refs)).toEqual([CACHED_REFS, LIVE_REFS]);
636 expect(logs).toEqual([]);
637 }),
638 ),
639 );
640
641 it.effect("cancels an in-flight refresh and restarts when the RPC session changes", () =>
642 Effect.scoped(
643 Effect.gen(function* () {
644 const started = yield* Deferred.make<void>();
645 const interrupted = yield* Deferred.make<void>();
646 const calls = yield* Ref.make(0);
647 const interruptions = yield* Ref.make(0);
648 const client = {
649 [WS_METHODS.vcsListRefs]: () =>
650 Ref.updateAndGet(calls, (count) => count + 1).pipe(
651 Effect.flatMap((count) =>
652 count === 1
653 ? Deferred.succeed(started, undefined).pipe(
654 Effect.andThen(Effect.never),
655 Effect.onInterrupt(() =>
656 Ref.update(interruptions, (interruptions) => interruptions + 1).pipe(
657 Effect.andThen(Deferred.succeed(interrupted, undefined)),
658 ),
659 ),
660 )
661 : Effect.succeed(LIVE_REFS),
662 ),
663 ),
664 } as unknown as WsRpcProtocolClient;
665 const liveSession = yield* SubscriptionRef.make(Option.some(session(client)));
666 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
667 target: TARGET,
668 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
669 session: liveSession,
670 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
671 connect: Effect.void,
672 disconnect: Effect.void,
673 retryNow: Effect.void,
674 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
675
676 const fiber = yield* Stream.unwrap(
677 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
678 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
679 Effect.provideService(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())),
680 ),
681 ).pipe(Stream.runHead, Effect.forkChild);
682
683 yield* Deferred.await(started);
684 yield* SubscriptionRef.set(liveSession, Option.none());
685 yield* Deferred.await(interrupted);
686 expect(yield* Ref.get(interruptions)).toBe(1);
687
688 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
689 expect(Option.getOrThrow(yield* Fiber.join(fiber))).toEqual(LIVE_REFS);
690 expect(yield* Ref.get(calls)).toBe(2);
691 }),
692 ),
693 );
694
695 it.effect("retries Git failures and cancels pending retries when the session is removed", () =>
696 Effect.scoped(
697 Effect.gen(function* () {
698 const calls = yield* Ref.make(0);
699 const warnings = yield* Queue.unbounded<unknown>();
700 const logger = Logger.make(({ message }) => Queue.offerUnsafe(warnings, message));
701 const failure = new VcsProcessExitError({
702 operation: "listRefs",
703 command: "git",
704 cwd: "/repo",
705 exitCode: 128,
706 detail: "Could not read refs.",
707 });
708 const client = {
709 [WS_METHODS.vcsListRefs]: () =>
710 Ref.updateAndGet(calls, (count) => count + 1).pipe(
711 Effect.flatMap((count) =>
712 count < 3 ? Effect.fail(failure) : Effect.succeed(LIVE_REFS),
713 ),
714 ),
715 } as unknown as WsRpcProtocolClient;
716 const liveSession = yield* SubscriptionRef.make(Option.some(session(client)));
717 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
718 target: TARGET,
719 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
720 session: liveSession,
721 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
722 connect: Effect.void,
723 disconnect: Effect.void,
724 retryNow: Effect.void,
725 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
726 const refs = yield* Stream.unwrap(
727 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
728 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
729 Effect.provideService(
730 Persistence.EnvironmentCacheStore,
731 cacheWithRefs(Option.some(CACHED_REFS)),
732 ),
733 ),
734 ).pipe(
735 Stream.take(2),
736 Stream.runCollect,
737 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
738 Effect.forkChild({ startImmediately: true }),
739 );
740
741 yield* Queue.take(warnings);
742 expect(yield* Ref.get(calls)).toBe(1);
743 yield* TestClock.adjust("1 second");
744 expect(yield* Ref.get(calls)).toBe(2);
745 expect(yield* Queue.size(warnings)).toBe(1);
746
747 yield* SubscriptionRef.set(liveSession, Option.none());
748 yield* TestClock.adjust("1 minute");
749 expect(yield* Ref.get(calls)).toBe(2);
750 expect(yield* Queue.size(warnings)).toBe(1);
751
752 yield* SubscriptionRef.set(liveSession, Option.some(session(client)));
753 expect(yield* Fiber.join(refs)).toEqual([CACHED_REFS, LIVE_REFS]);
754 expect(yield* Ref.get(calls)).toBe(3);
755 }),
756 ),
757 );
758
759 it.effect("emits persisted refs before a live refresh", () =>
760 Effect.scoped(
761 Effect.gen(function* () {
762 const client = {
763 [WS_METHODS.vcsListRefs]: () => Effect.succeed(LIVE_REFS),
764 } as unknown as WsRpcProtocolClient;
765 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
766 target: TARGET,
767 state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE),
768 session: yield* SubscriptionRef.make(Option.some(session(client))),
769 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
770 connect: Effect.void,
771 disconnect: Effect.void,
772 retryNow: Effect.void,
773 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
774
775 const refs = yield* Stream.unwrap(
776 makeCachedVcsRefsChanges({ cwd: "/repo", limit: 100 }).pipe(
777 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
778 Effect.provideService(
779 Persistence.EnvironmentCacheStore,
780 cacheWithRefs(Option.some(CACHED_REFS)),
781 ),
782 ),
783 ).pipe(Stream.take(2), Stream.runCollect);
784
785 expect(refs).toEqual([CACHED_REFS, LIVE_REFS]);
786 }),
787 ),
788 );
789});
790
791vi.mock("./session.ts", () => ({
792 createEnvironmentSessionAtoms: () => ({ sessionStateAtom: grantedSessions }),
793}));
794const grantedSessions = Atom.family((_id: EnvironmentId) =>
795 Atom.make(
796 AsyncResult.success({
797 authenticated: true,
798 auth: {
799 policy: "remote-reachable",
800 bootstrapMethods: [],
801 sessionMethods: [],
802 sessionCookieName: "test",
803 },
804 scopes: [AuthSourceControlWriteScope],
805 }),
806 ),
807);
808
809function status(overrides: Partial<VcsStatusResult> = {}): VcsStatusResult {
810 return {
811 isRepo: true,
812 hasPrimaryRemote: true,
813 isDefaultRef: false,
814 refName: "feature/test",
815 hasWorkingTreeChanges: false,
816 workingTree: {
817 files: [],
818 insertions: 0,
819 deletions: 0,
820 },
821 hasUpstream: true,
822 aheadCount: 0,
823 behindCount: 0,
824 pr: null,
825 ...overrides,
826 };
827}
828
829describe("resolveQuickAction", () => {
830 it("commits local changes without pushing when the branch is behind upstream", () => {
831 const behind = status({ hasWorkingTreeChanges: true, behindCount: 2 });
832 const commit = { label: "Commit", disabled: false, kind: "run_action", action: "commit" };
833 expect(resolveQuickAction(behind, false)).toEqual(commit);
834 expect(
835 resolveQuickAction(
836 {
837 ...behind,
838 pr: {
839 number: 13,
840 title: "Open PR",
841 url: "https://example.com/pr/13",
842 baseRef: "main",
843 headRef: "feature/test",
844 state: "open",
845 },
846 },
847 false,
848 ),
849 ).toEqual(commit);
850 });
851});