1import { vi } from "vite-plus/test";
2import {
3 EnvironmentId,
4 AuthSourceControlWriteScope,
5 ThreadId,
6 ProjectId,
7 WS_METHODS,
8 type GitActionProgressEvent,
9 type GitRunStackedActionInput,
10 type GitRunStackedActionResult,
11} from "@t3tools/contracts";
12import { describe, expect, it } from "@effect/vitest";
13import * as Cause from "effect/Cause";
14import * as Effect from "effect/Effect";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as Schema from "effect/Schema";
18import * as Stream from "effect/Stream";
19import * as SubscriptionRef from "effect/SubscriptionRef";
20import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
21
22import {
23 AVAILABLE_CONNECTION_STATE,
24 PrimaryConnectionTarget,
25 type PreparedConnection,
26 type SupervisorConnectionState,
27} from "../connection/model.ts";
28import * as EnvironmentRegistry from "../connection/registry.ts";
29import * as EnvironmentSupervisor from "../connection/supervisor.ts";
30import * as Persistence from "../platform/persistence.ts";
31import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
32import type { RpcSession } from "../rpc/session.ts";
33import type { AtomCommandResult } from "./runtime.ts";
34import {
35 applyVcsActionProgressEvent,
36 beginVcsActionState,
37 consumeVcsActionProgress,
38 createVcsActionManager,
39 createVcsActionTransportId,
40 EMPTY_VCS_ACTION_STATE,
41 getVcsActionTargetKey,
42 normalizeVcsActionProgressEvent,
43 parseVcsActionTargetKey,
44 VcsActionMissingTerminalEventError,
45 VcsActionRemoteFailureError,
46 VcsActionTargetKeyParseError,
47 VcsActionUnavailableError,
48} from "./vcsAction.ts";
49import { vcsRefsCacheStateAtom } from "./vcsRefInvalidation.ts";
50
51const actionId = "action-123";
52const action = "commit_push" as const;
53const cwd = "/repo";
54const environmentId = EnvironmentId.make("environment-1");
55const isVcsActionUnavailableError = Schema.is(VcsActionUnavailableError);
56const result: GitRunStackedActionResult = {
57 action,
58 branch: {
59 status: "skipped_not_requested",
60 },
61 commit: {
62 status: "created",
63 commitSha: "abc123",
64 subject: "Test commit",
65 },
66 push: {
67 status: "pushed",
68 branch: "feature",
69 },
70 pr: {
71 status: "skipped_not_requested",
72 },
73 toast: {
74 title: "Changes pushed",
75 cta: {
76 kind: "none",
77 },
78 },
79};
80
81const target = new PrimaryConnectionTarget({
82 environmentId,
83 label: "Test environment",
84 httpBaseUrl: "https://environment.example.test",
85 wsBaseUrl: "wss://environment.example.test",
86});
87
88function session(client: WsRpcProtocolClient): RpcSession {
89 return {
90 client,
91 initialConfig: Effect.never,
92 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
93 ready: Effect.void,
94 probe: Effect.void,
95 closed: Effect.never,
96 };
97}
98
99function progress<T extends GitActionProgressEvent>(event: T): T {
100 return event;
101}
102
103function cacheStore(onClearVcsRefs: (environmentId: EnvironmentId) => void) {
104 return Persistence.EnvironmentCacheStore.of({
105 loadShell: () => Effect.succeedNone,
106 saveShell: () => Effect.void,
107 loadThread: () => Effect.succeedNone,
108 saveThread: () => Effect.void,
109 removeThread: () => Effect.void,
110 loadServerConfig: () => Effect.succeedNone,
111 saveServerConfig: () => Effect.void,
112 loadVcsRefs: () => Effect.succeedNone,
113 saveVcsRefs: () => Effect.void,
114 removeVcsRefs: () => Effect.void,
115 clearVcsRefs: (environmentId) => Effect.sync(() => onClearVcsRefs(environmentId)),
116 clear: () => Effect.void,
117 });
118}
119
120describe("vcsActionState", () => {
121 it("preserves malformed target key diagnostics and the native cause without copying the key", () => {
122 const key = "not-json-with-credential=do-not-log";
123 let error: unknown;
124
125 try {
126 parseVcsActionTargetKey(key);
127 } catch (cause) {
128 error = cause;
129 }
130
131 expect(error).toBeInstanceOf(VcsActionTargetKeyParseError);
132 expect(error).toMatchObject({ keyLength: key.length, cause: expect.any(SyntaxError) });
133 expect(error).not.toHaveProperty("key");
134 expect((error as Error).message).not.toContain(key);
135 });
136
137 it("rejects invalid target key shapes", () => {
138 const key = JSON.stringify([environmentId]);
139
140 expect(() => parseVcsActionTargetKey(key)).toThrowError(VcsActionTargetKeyParseError);
141 });
142
143 it("projects phase and hook progress without owning the async operation", () => {
144 const initial = beginVcsActionState({
145 operation: "run_change_request",
146 label: "Running source control action",
147 actionId,
148 });
149 const phase = applyVcsActionProgressEvent(
150 initial,
151 progress({
152 actionId,
153 action,
154 cwd,
155 kind: "phase_started",
156 phase: "commit",
157 label: "Committing...",
158 }),
159 );
160 const hook = applyVcsActionProgressEvent(
161 phase,
162 progress({
163 actionId,
164 action,
165 cwd,
166 kind: "hook_started",
167 hookName: "post-commit",
168 }),
169 );
170 const output = applyVcsActionProgressEvent(
171 hook,
172 progress({
173 actionId,
174 action,
175 cwd,
176 kind: "hook_output",
177 hookName: "post-commit",
178 stream: "stdout",
179 text: "hook output",
180 }),
181 );
182 const finished = applyVcsActionProgressEvent(
183 output,
184 progress({
185 actionId,
186 action,
187 cwd,
188 kind: "hook_finished",
189 hookName: "post-commit",
190 exitCode: 0,
191 durationMs: 12,
192 }),
193 );
194
195 expect(phase).toMatchObject({
196 isRunning: true,
197 currentLabel: "Committing...",
198 currentPhaseLabel: "Committing...",
199 });
200 expect(output).toMatchObject({
201 currentLabel: "Running post-commit...",
202 hookName: "post-commit",
203 lastOutputLine: "hook output",
204 });
205 expect(finished).toMatchObject({
206 currentLabel: "Committing...",
207 hookName: null,
208 lastOutputLine: null,
209 });
210 });
211
212 it("keeps running presentation state until the finished action settles", () => {
213 const initial = beginVcsActionState({
214 operation: "run_change_request",
215 label: "Running source control action",
216 actionId,
217 });
218 const pushing = applyVcsActionProgressEvent(
219 initial,
220 progress({
221 actionId,
222 action,
223 cwd,
224 kind: "phase_started",
225 phase: "push",
226 label: "Pushing...",
227 }),
228 );
229 const finished = applyVcsActionProgressEvent(
230 pushing,
231 progress({
232 actionId,
233 action,
234 cwd,
235 kind: "action_finished",
236 result,
237 }),
238 );
239
240 expect(finished).toMatchObject({
241 isRunning: true,
242 operation: "run_change_request",
243 actionId,
244 action,
245 currentLabel: "Pushing...",
246 currentPhaseLabel: "Pushing...",
247 lastOutputLine: null,
248 error: null,
249 });
250 });
251
252 it("keeps running presentation state and retains a terminal action error until settle", () => {
253 const initial = beginVcsActionState({
254 operation: "run_change_request",
255 label: "Running source control action",
256 actionId,
257 });
258 const failed = applyVcsActionProgressEvent(
259 initial,
260 progress({
261 actionId,
262 action,
263 cwd,
264 kind: "action_failed",
265 phase: null,
266 message: "Push failed.",
267 }),
268 );
269
270 expect(failed).toMatchObject({
271 isRunning: true,
272 operation: "run_change_request",
273 actionId,
274 action,
275 error: "Push failed.",
276 });
277 });
278
279 it("ignores progress after a newer action owns the target", () => {
280 const current = beginVcsActionState({
281 operation: "pull",
282 label: "Pulling latest changes",
283 actionId: "newer-action",
284 });
285
286 expect(
287 applyVcsActionProgressEvent(
288 current,
289 progress({
290 actionId,
291 action,
292 cwd,
293 kind: "phase_started",
294 phase: "push",
295 label: "Pushing...",
296 }),
297 ),
298 ).toBe(current);
299 });
300
301 it("keys presentation state only when the environment and repository are known", () => {
302 expect(
303 getVcsActionTargetKey({
304 environmentId,
305 cwd,
306 }),
307 ).toBe(JSON.stringify([environmentId, cwd]));
308 expect(getVcsActionTargetKey({ environmentId: null, cwd })).toBeNull();
309 expect(
310 getVcsActionTargetKey({
311 environmentId,
312 cwd: null,
313 }),
314 ).toBeNull();
315 });
316
317 it("normalizes progress only for the matching environment-scoped action", () => {
318 const target = { environmentId, cwd };
319 const otherTarget = {
320 environmentId: EnvironmentId.make("environment-2"),
321 cwd,
322 };
323 const transportActionId = createVcsActionTransportId(target, actionId);
324 const event = progress({
325 actionId: createVcsActionTransportId(otherTarget, actionId),
326 action,
327 cwd,
328 kind: "phase_started",
329 phase: "push",
330 label: "Pushing...",
331 });
332
333 expect(normalizeVcsActionProgressEvent(target, transportActionId, actionId, event)).toBeNull();
334 expect(
335 normalizeVcsActionProgressEvent(target, transportActionId, actionId, {
336 ...event,
337 actionId: transportActionId,
338 }),
339 ).toEqual({
340 ...event,
341 actionId,
342 });
343 });
344
345 it.effect("consumes progress through the terminal event and returns its result", () =>
346 Effect.gen(function* () {
347 const target = { environmentId, cwd };
348 const transportActionId = createVcsActionTransportId(target, actionId);
349 const observed: GitActionProgressEvent[] = [];
350 const events: GitActionProgressEvent[] = [
351 {
352 actionId: "unrelated-action",
353 action,
354 cwd,
355 kind: "phase_started",
356 phase: "commit",
357 label: "Ignored",
358 },
359 {
360 actionId: transportActionId,
361 action,
362 cwd,
363 kind: "phase_started",
364 phase: "push",
365 label: "Pushing...",
366 },
367 {
368 actionId: transportActionId,
369 action,
370 cwd,
371 kind: "action_finished",
372 result,
373 },
374 ];
375
376 const actual = yield* consumeVcsActionProgress(Stream.fromIterable(events), {
377 target,
378 transportActionId,
379 actionId,
380 action,
381 onProgress: (event) =>
382 Effect.sync(() => {
383 observed.push(event);
384 }),
385 });
386
387 expect(actual).toEqual(result);
388 expect(observed.map((event) => event.actionId)).toEqual([actionId, actionId]);
389 expect(observed.map((event) => event.kind)).toEqual(["phase_started", "action_finished"]);
390 }),
391 );
392
393 it.effect("retains structural remote failure context without copying the remote payload", () =>
394 Effect.gen(function* () {
395 const target = { environmentId, cwd };
396 const transportActionId = createVcsActionTransportId(target, actionId);
397 const remoteMessage = "The remote rejected the push with credential=do-not-log.";
398 const error = yield* consumeVcsActionProgress(
399 Stream.fromIterable<GitActionProgressEvent>([
400 {
401 actionId: transportActionId,
402 action,
403 cwd,
404 kind: "action_failed",
405 phase: "push",
406 message: remoteMessage,
407 },
408 ]),
409 {
410 target,
411 transportActionId,
412 actionId,
413 action,
414 onProgress: () => Effect.void,
415 },
416 ).pipe(Effect.flip);
417
418 expect(error).toBeInstanceOf(VcsActionRemoteFailureError);
419 expect(error).toMatchObject({
420 actionId,
421 transportActionId,
422 action,
423 environmentId,
424 cwd,
425 phase: "push",
426 remoteMessageLength: remoteMessage.length,
427 });
428 expect(error).not.toHaveProperty("detail");
429 expect(error.message).toBe("Source control action 'commit_push' failed during push.");
430 expect(error.message).not.toContain(remoteMessage);
431 }),
432 );
433
434 it.effect("reports a missing terminal event as a protocol failure", () =>
435 Effect.gen(function* () {
436 const target = { environmentId, cwd };
437 const transportActionId = createVcsActionTransportId(target, actionId);
438 const error = yield* consumeVcsActionProgress(
439 Stream.fromIterable<GitActionProgressEvent>([
440 {
441 actionId: transportActionId,
442 action,
443 cwd,
444 kind: "phase_started",
445 phase: "commit",
446 label: "Committing...",
447 },
448 ]),
449 {
450 target,
451 transportActionId,
452 actionId,
453 action,
454 onProgress: () => Effect.void,
455 },
456 ).pipe(Effect.flip);
457
458 expect(error).toBeInstanceOf(VcsActionMissingTerminalEventError);
459 expect(error).toMatchObject({
460 actionId,
461 transportActionId,
462 action,
463 environmentId,
464 cwd,
465 });
466 expect(error.message).toBe(
467 "Source control action 'commit_push' ended without a terminal result.",
468 );
469 }),
470 );
471
472 it("keys mutation ownership by environment and cwd", () => {
473 const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
474 EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore,
475 never
476 >;
477 const manager = createVcsActionManager(runtime);
478 const registry = AtomRegistry.make();
479 const target = { environmentId, cwd };
480 const otherTarget = {
481 environmentId: EnvironmentId.make("environment-2"),
482 cwd,
483 };
484
485 expect(manager.runStackedAction(target)).toBe(manager.runStackedAction({ ...target }));
486 expect(manager.runStackedAction(target)).not.toBe(manager.runStackedAction(otherTarget));
487 expect(registry.get(manager.stateAtom(target))).toEqual(EMPTY_VCS_ACTION_STATE);
488
489 registry.dispose();
490 });
491
492 it("retains the incomplete target and operation when tracking is unavailable", async () => {
493 const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
494 EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore,
495 never
496 >;
497 const manager = createVcsActionManager(runtime);
498 const registry = AtomRegistry.make();
499 const result = await manager.track(
500 registry,
501 { environmentId, cwd: null },
502 { operation: "pull", label: "Pulling latest changes" },
503 async () => AsyncResult.success(undefined),
504 );
505
506 expect(AsyncResult.isFailure(result)).toBe(true);
507 if (AsyncResult.isFailure(result)) {
508 const error = Cause.squash(result.cause);
509 expect(error).toBeInstanceOf(VcsActionUnavailableError);
510 if (!isVcsActionUnavailableError(error)) {
511 throw error;
512 }
513 expect(error).toMatchObject({
514 operation: "pull",
515 environmentId,
516 cwd: null,
517 });
518 expect(error.message).toBe("Source control operation 'pull' is unavailable.");
519 }
520
521 registry.dispose();
522 });
523
524 it("tracks finite mutations without letting an older completion clear newer state", async () => {
525 const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
526 EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore,
527 never
528 >;
529 const manager = createVcsActionManager(runtime);
530 const registry = AtomRegistry.make();
531 const target = { environmentId, cwd };
532 let finishFirst!: () => void;
533 let failSecond!: (error: Error) => void;
534 const firstAction = new Promise<AtomCommandResult<void, never>>((resolve) => {
535 finishFirst = () => resolve(AsyncResult.success(undefined));
536 });
537 const secondAction = new Promise<AtomCommandResult<void, Error>>((resolve) => {
538 failSecond = (error) => resolve(AsyncResult.failure(Cause.fail(error)));
539 });
540
541 const first = manager.track(
542 registry,
543 target,
544 { operation: "pull", label: "Pulling latest changes" },
545 () => firstAction,
546 );
547 const firstActionId = registry.get(manager.stateAtom(target)).actionId;
548 const second = manager.track(
549 registry,
550 target,
551 { operation: "switch_ref", label: "Switching branch" },
552 () => secondAction,
553 );
554 const secondActionId = registry.get(manager.stateAtom(target)).actionId;
555
556 finishFirst();
557 await first;
558 expect(registry.get(manager.stateAtom(target))).toMatchObject({
559 actionId: secondActionId,
560 isRunning: true,
561 operation: "switch_ref",
562 });
563 expect(secondActionId).not.toBe(firstActionId);
564
565 failSecond(new Error("switch failed"));
566 const secondFailure = await second;
567 expect(AsyncResult.isFailure(secondFailure)).toBe(true);
568 expect(registry.get(manager.stateAtom(target))).toMatchObject({
569 actionId: secondActionId,
570 error: "switch failed",
571 isRunning: false,
572 operation: "switch_ref",
573 });
574
575 registry.dispose();
576 });
577
578 it.effect("invalidates persisted refs after successful and failed stacked actions", () =>
579 Effect.scoped(
580 Effect.gen(function* () {
581 const connectionState: SupervisorConnectionState = {
582 ...AVAILABLE_CONNECTION_STATE,
583 desired: true,
584 network: "online",
585 phase: "connected",
586 attempt: 1,
587 generation: 1,
588 };
589 const targetKey = { environmentId, cwd };
590 const successfulActionId = "invalidate-success";
591 const failedActionId = "invalidate-failure";
592 const interruptedActionId = "invalidate-interrupted";
593 const successfulTransportActionId = createVcsActionTransportId(
594 targetKey,
595 successfulActionId,
596 );
597 const failedTransportActionId = createVcsActionTransportId(targetKey, failedActionId);
598 const interruptedTransportActionId = createVcsActionTransportId(
599 targetKey,
600 interruptedActionId,
601 );
602 const rpcInputs = new Array<GitRunStackedActionInput>();
603 const client = {
604 [WS_METHODS.gitRunStackedAction]: (input: GitRunStackedActionInput) =>
605 (rpcInputs.push(input), input.actionId === successfulTransportActionId)
606 ? Stream.make(
607 progress({
608 kind: "action_finished",
609 actionId: successfulTransportActionId,
610 cwd,
611 action,
612 result,
613 }),
614 )
615 : input.actionId === interruptedTransportActionId
616 ? Stream.fromEffect(Effect.interrupt)
617 : Stream.make(
618 progress({
619 kind: "action_failed",
620 actionId: failedTransportActionId,
621 cwd,
622 action,
623 phase: "push",
624 message: "push failed after creating the branch",
625 }),
626 ),
627 } as unknown as WsRpcProtocolClient;
628 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
629 target,
630 state: yield* SubscriptionRef.make(connectionState),
631 session: yield* SubscriptionRef.make(Option.some(session(client))),
632 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
633 connect: Effect.void,
634 disconnect: Effect.void,
635 retryNow: Effect.void,
636 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
637 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
638 _environmentId,
639 effect,
640 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
641 const runStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["runStream"] = (
642 _environmentId,
643 stream,
644 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
645 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
646 run,
647 runStream,
648 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
649 const removed = new Array<string>();
650 const runtime = Atom.runtime(
651 Layer.merge(
652 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
653 Layer.succeed(
654 Persistence.EnvironmentCacheStore,
655 cacheStore((removedEnvironmentId) => {
656 removed.push(`${removedEnvironmentId}:*`);
657 }),
658 ),
659 ),
660 );
661 const manager = createVcsActionManager(runtime);
662 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
663 Effect.sync(() => registry.dispose()),
664 );
665 const state = vcsRefsCacheStateAtom({ environmentId });
666
667 expect(registry.get(state).revision).toBe(0);
668 const threadId = ThreadId.make("thread-stacked-action");
669 const projectId = ProjectId.make("project-stacked-action");
670 const successfulResult = yield* Effect.promise(() =>
671 manager.runStackedAction(targetKey).run(registry, {
672 actionId: successfulActionId,
673 action,
674 threadId,
675 projectId,
676 }),
677 );
678
679 expect(AsyncResult.isSuccess(successfulResult)).toBe(true);
680 expect(registry.get(manager.stateAtom(targetKey))).toEqual(EMPTY_VCS_ACTION_STATE);
681 expect(registry.get(state).revision).toBe(1);
682 expect(removed).toEqual([`${environmentId}:*`]);
683 // The server links a created pull request to this thread, so the id must ride along.
684 expect(rpcInputs).toEqual([
685 { actionId: successfulTransportActionId, cwd, action, threadId, projectId },
686 ]);
687
688 const failedResult = yield* Effect.promise(() =>
689 manager.runStackedAction(targetKey).run(registry, {
690 actionId: failedActionId,
691 action,
692 }),
693 );
694
695 expect(AsyncResult.isFailure(failedResult)).toBe(true);
696 expect(registry.get(manager.stateAtom(targetKey))).toMatchObject({
697 isRunning: false,
698 operation: "run_change_request",
699 actionId: failedActionId,
700 error: "Source control action 'commit_push' failed during push.",
701 });
702 expect(registry.get(state).revision).toBe(2);
703 expect(removed).toEqual([`${environmentId}:*`, `${environmentId}:*`]);
704
705 const interruptedResult = yield* Effect.promise(() =>
706 manager.runStackedAction(targetKey).run(registry, {
707 actionId: interruptedActionId,
708 action,
709 }),
710 );
711
712 expect(AsyncResult.isFailure(interruptedResult)).toBe(true);
713 if (AsyncResult.isFailure(interruptedResult)) {
714 expect(Cause.hasInterruptsOnly(interruptedResult.cause)).toBe(true);
715 }
716 expect(registry.get(manager.stateAtom(targetKey))).toEqual(EMPTY_VCS_ACTION_STATE);
717 expect(registry.get(state).revision).toBe(3);
718 expect(removed).toEqual([`${environmentId}:*`, `${environmentId}:*`, `${environmentId}:*`]);
719 }),
720 ),
721 );
722});
723
724// Transport fixtures have a source-control-only session; authorization edge cases
725// are exercised by commandPermissions.test.ts.
726vi.mock("./session.ts", () => ({
727 createEnvironmentSessionAtoms: () => ({ sessionStateAtom: grantedSessions }),
728}));
729const grantedSessions = Atom.family((_id: EnvironmentId) =>
730 Atom.make(
731 AsyncResult.success({
732 authenticated: true,
733 auth: {
734 policy: "remote-reachable",
735 bootstrapMethods: [],
736 sessionMethods: [],
737 sessionCookieName: "test",
738 },
739 scopes: [AuthSourceControlWriteScope],
740 }),
741 ),
742);