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.gen(function* () {
437 const connectionState: SupervisorConnectionState = {
438 ...AVAILABLE_CONNECTION_STATE,
439 desired: true,
440 network: "online",
441 phase: "connected",
442 attempt: 1,
443 generation: 1,
444 };
445 const targetKey = { environmentId, cwd };
446 const successfulActionId = "invalidate-success";
447 const failedActionId = "invalidate-failure";
448 const interruptedActionId = "invalidate-interrupted";
449 const successfulTransportActionId = createVcsActionTransportId(targetKey, successfulActionId);
450 const failedTransportActionId = createVcsActionTransportId(targetKey, failedActionId);
451 const interruptedTransportActionId = createVcsActionTransportId(
452 targetKey,
453 interruptedActionId,
454 );
455 const rpcInputs = new Array<GitRunStackedActionInput>();
456 const client = {
457 [WS_METHODS.gitRunStackedAction]: (input: GitRunStackedActionInput) =>
458 (rpcInputs.push(input), input.actionId === successfulTransportActionId)
459 ? Stream.make(
460 progress({
461 kind: "action_finished",
462 actionId: successfulTransportActionId,
463 cwd,
464 action,
465 result,
466 }),
467 )
468 : input.actionId === interruptedTransportActionId
469 ? Stream.fromEffect(Effect.interrupt)
470 : Stream.make(
471 progress({
472 kind: "action_failed",
473 actionId: failedTransportActionId,
474 cwd,
475 action,
476 phase: "push",
477 message: "push failed after creating the branch",
478 }),
479 ),
480 } as unknown as WsRpcProtocolClient;
481 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
482 target,
483 state: yield* SubscriptionRef.make(connectionState),
484 session: yield* SubscriptionRef.make(Option.some(session(client))),
485 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
486 connect: Effect.void,
487 disconnect: Effect.void,
488 retryNow: Effect.void,
489 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
490 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
491 _environmentId,
492 effect,
493 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
494 const runStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["runStream"] = (
495 _environmentId,
496 stream,
497 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
498 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
499 run,
500 runStream,
501 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
502 const removed = new Array<string>();
503 const runtime = Atom.runtime(
504 Layer.merge(
505 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
506 Layer.succeed(
507 Persistence.EnvironmentCacheStore,
508 cacheStore((removedEnvironmentId) => {
509 removed.push(`${removedEnvironmentId}:*`);
510 }),
511 ),
512 ),
513 );
514 const manager = createVcsActionManager(runtime);
515 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
516 Effect.sync(() => registry.dispose()),
517 );
518 const state = vcsRefsCacheStateAtom({ environmentId });
519
520 expect(registry.get(state).revision).toBe(0);
521 const threadId = ThreadId.make("thread-stacked-action");
522 const projectId = ProjectId.make("project-stacked-action");
523 const successfulResult = yield* Effect.promise(() =>
524 manager.runStackedAction(targetKey).run(registry, {
525 actionId: successfulActionId,
526 action,
527 threadId,
528 projectId,
529 }),
530 );
531
532 expect(AsyncResult.isSuccess(successfulResult)).toBe(true);
533 expect(registry.get(manager.stateAtom(targetKey))).toEqual(EMPTY_VCS_ACTION_STATE);
534 expect(registry.get(state).revision).toBe(1);
535 expect(removed).toEqual([`${environmentId}:*`]);
536 expect(rpcInputs).toEqual([
537 { actionId: successfulTransportActionId, cwd, action, threadId, projectId },
538 ]);
539
540 const failedResult = yield* Effect.promise(() =>
541 manager.runStackedAction(targetKey).run(registry, {
542 actionId: failedActionId,
543 action,
544 }),
545 );
546
547 expect(AsyncResult.isFailure(failedResult)).toBe(true);
548 expect(registry.get(manager.stateAtom(targetKey))).toMatchObject({
549 isRunning: false,
550 operation: "run_change_request",
551 actionId: failedActionId,
552 error: "Source control action 'commit_push' failed during push.",
553 });
554 expect(registry.get(state).revision).toBe(2);
555 expect(removed).toEqual([`${environmentId}:*`, `${environmentId}:*`]);
556
557 const interruptedResult = yield* Effect.promise(() =>
558 manager.runStackedAction(targetKey).run(registry, {
559 actionId: interruptedActionId,
560 action,
561 }),
562 );
563
564 expect(AsyncResult.isFailure(interruptedResult)).toBe(true);
565 if (AsyncResult.isFailure(interruptedResult)) {
566 expect(Cause.hasInterruptsOnly(interruptedResult.cause)).toBe(true);
567 }
568 expect(registry.get(manager.stateAtom(targetKey))).toEqual(EMPTY_VCS_ACTION_STATE);
569 expect(registry.get(state).revision).toBe(3);
570 expect(removed).toEqual([`${environmentId}:*`, `${environmentId}:*`, `${environmentId}:*`]);
571 }),
572 );
573});
574
575vi.mock("./session.ts", () => ({
576 createEnvironmentSessionAtoms: () => ({ sessionStateAtom: grantedSessions }),
577}));
578const grantedSessions = Atom.family((_id: EnvironmentId) =>
579 Atom.make(
580 AsyncResult.success({
581 authenticated: true,
582 auth: {
583 policy: "remote-reachable",
584 bootstrapMethods: [],
585 sessionMethods: [],
586 sessionCookieName: "test",
587 },
588 scopes: [AuthOrchestrationReadScope, AuthSourceControlWriteScope],
589 }),
590 ),
591);