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