1import * as Context from "effect/Context";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Match from "effect/Match";
6import * as Schedule from "effect/Schedule";
7import * as Semaphore from "effect/Semaphore";
8import { ChildProcessSpawner } from "effect/process";
9
10import {
11 type VcsError,
12 VcsProcessExitError,
13 type VcsProcessExitFailureKind,
14 VcsProcessMissingExitCodeError,
15 VcsProcessOutputLimitError,
16 VcsProcessOutputReadError,
17 VcsProcessSpawnError,
18 VcsProcessStdinWriteError,
19 VcsProcessTimeoutError,
20} from "@t3tools/contracts";
21import * as ProcessRunner from "../processRunner.ts";
22
23export interface VcsProcessInput {
24 readonly operation: string;
25 readonly command: string;
26 readonly args: ReadonlyArray<string>;
27 readonly cwd: string;
28 readonly spawnCwd?: string;
29 readonly stdin?: string;
30 readonly onStdoutChunk?: (chunk: Uint8Array) => void;
31 readonly env?: NodeJS.ProcessEnv;
32 readonly allowNonZeroExit?: boolean;
33 readonly timeoutMs?: number;
34 readonly maxOutputBytes?: number;
35 readonly outputMode?: ProcessRunner.ProcessRunInput["outputMode"];
36 readonly appendTruncationMarker?: boolean;
37}
38
39export interface VcsProcessOutput {
40 readonly exitCode: ChildProcessSpawner.ExitCode;
41 readonly stdout: string;
42 readonly stderr: string;
43 readonly stdoutTruncated: boolean;
44 readonly stderrTruncated: boolean;
45 /** Present on real process output; optional so narrow test doubles remain lightweight. */
46 readonly stdoutInvalidUtf8?: boolean;
47 readonly stderrInvalidUtf8?: boolean;
48}
49
50export class VcsProcess extends Context.Service<
51 VcsProcess,
52 {
53 readonly run: (input: VcsProcessInput) => Effect.Effect<VcsProcessOutput, VcsError>;
54 }
55>()("t3/vcs/VcsProcess") {}
56
57const DEFAULT_TIMEOUT_MS = 30_000;
58const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000;
59const OUTPUT_TRUNCATED_MARKER = "\n\n[truncated]";
60const VCS_PROCESS_CONCURRENCY = 8;
61const GITHUB_PROCESS_CONCURRENCY = 4;
62
63export const CHECKPOINT_CAPTURE_OPERATION = "GitVcsDriver.checkpoints.captureCheckpoint";
64
65const classifyNonZeroExit = (command: string, stderr: string): VcsProcessExitFailureKind => {
66 const normalized = stderr.toLowerCase();
67
68 if (
69 normalized.includes("authentication failed") ||
70 normalized.includes("not logged in") ||
71 normalized.includes("gh auth login") ||
72 normalized.includes("glab auth login") ||
73 normalized.includes("az devops login") ||
74 normalized.includes("please run az login") ||
75 normalized.includes("no oauth token") ||
76 normalized.includes("unauthorized")
77 ) {
78 return "authentication";
79 }
80
81 if (
82 normalized.includes("api rate limit") ||
83 normalized.includes("rate limit exceeded") ||
84 normalized.includes("secondary rate limit") ||
85 normalized.includes("too many requests") ||
86 normalized.includes("http 429")
87 ) {
88 return "rate-limited";
89 }
90
91 if (
92 (command === "gh" &&
93 (normalized.includes("could not resolve to a pullrequest") ||
94 normalized.includes("repository.pullrequest") ||
95 normalized.includes("no pull requests found for branch") ||
96 normalized.includes("pull request not found"))) ||
97 (command === "glab" &&
98 (normalized.includes("merge request not found") ||
99 normalized.includes("not found") ||
100 normalized.includes("404"))) ||
101 (command === "az" &&
102 normalized.includes("pull request") &&
103 (normalized.includes("not found") || normalized.includes("does not exist")))
104 ) {
105 return "not-found";
106 }
107
108 return "command-failed";
109};
110
111// Classify before discarding stderr; keep paths and process output out of errors.
112const isTransientGitExit = (stderr: string) =>
113 /unable to create [^\n]*\.lock['"]?: file exists/i.test(stderr) ||
114 /(?:unable to stat|lstat\(|error: open\()[^\n]+: no such file or directory/i.test(stderr);
115
116export const make = Effect.gen(function* () {
117 const processRunner = yield* ProcessRunner.ProcessRunner;
118 const vcsProcesses = yield* Semaphore.make(VCS_PROCESS_CONCURRENCY);
119 const githubProcesses = yield* Semaphore.make(GITHUB_PROCESS_CONCURRENCY);
120
121 const runUnbounded = Effect.fn("VcsProcess.runUnbounded")(function* (input: VcsProcessInput) {
122 const baseError = {
123 operation: input.operation,
124 command: input.command,
125 cwd: input.cwd,
126 argumentCount: input.args.length,
127 };
128
129 const result = yield* processRunner
130 .run({
131 command: input.command,
132 args: input.args,
133 cwd: input.cwd,
134 ...(input.spawnCwd !== undefined ? { spawnCwd: input.spawnCwd } : {}),
135 ...(input.stdin !== undefined ? { stdin: input.stdin } : {}),
136 ...(input.onStdoutChunk !== undefined ? { onStdoutChunk: input.onStdoutChunk } : {}),
137 ...(input.env !== undefined ? { env: input.env } : {}),
138 timeout: input.timeoutMs ?? DEFAULT_TIMEOUT_MS,
139 maxOutputBytes: input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES,
140 outputMode: input.outputMode ?? "truncate",
141 truncatedMarker: input.appendTruncationMarker ? OUTPUT_TRUNCATED_MARKER : "",
142 timeoutBehavior: "error",
143 })
144 .pipe(
145 Effect.mapError(
146 Match.valueTags({
147 ProcessSpawnError: (error) =>
148 VcsProcessSpawnError.fromProcessSpawnError(baseError, error),
149 ProcessOutputLimitError: (error) =>
150 new VcsProcessOutputLimitError({
151 ...baseError,
152 stream: error.stream,
153 maxBytes: error.maxBytes,
154 observedBytes: error.observedBytes,
155 }),
156 ProcessTimeoutError: (error) =>
157 VcsProcessTimeoutError.fromProcessTimeoutError(baseError, error),
158 ProcessStdinError: (error) =>
159 new VcsProcessStdinWriteError({
160 ...baseError,
161 stdinBytes: error.stdinBytes,
162 cause: error.cause,
163 }),
164 ProcessReadError: (error) =>
165 new VcsProcessOutputReadError({
166 ...baseError,
167 stream: error.stream,
168 cause: error.cause,
169 }),
170 }),
171 ),
172 );
173
174 if (result.code === null) {
175 return yield* new VcsProcessMissingExitCodeError(baseError);
176 }
177
178 if (!input.allowNonZeroExit && result.code !== 0) {
179 const failureKind = classifyNonZeroExit(input.command, result.stderr);
180 return yield* VcsProcessExitError.fromProcessExit(
181 baseError,
182 {
183 exitCode: result.code,
184 stderr: result.stderr,
185 stderrTruncated: result.stderrTruncated,
186 },
187 failureKind,
188 input.command === "git" &&
189 failureKind === "command-failed" &&
190 isTransientGitExit(result.stderr),
191 );
192 }
193
194 return {
195 exitCode: result.code,
196 stdout: result.stdout,
197 stderr: result.stderr,
198 stdoutTruncated: result.stdoutTruncated,
199 stderrTruncated: result.stderrTruncated,
200 stdoutInvalidUtf8: result.stdoutInvalidUtf8 ?? false,
201 stderrInvalidUtf8: result.stderrInvalidUtf8 ?? false,
202 } satisfies VcsProcessOutput;
203 });
204
205 const run = Effect.fn("VcsProcess.run")(function* (input: VcsProcessInput) {
206 const bounded = vcsProcesses.withPermits(1)(runUnbounded(input));
207 if (
208 input.command === "git" &&
209 input.operation === CHECKPOINT_CAPTURE_OPERATION &&
210 input.onStdoutChunk === undefined
211 ) {
212 // Retry the failed command, retaining the private index/tree and recovery's outer deadline.
213 return yield* bounded.pipe(
214 Effect.tapError((error) =>
215 Effect.logDebug("checkpoint Git command failed", {
216 operation: input.operation,
217 errorTag: error._tag,
218 }),
219 ),
220 Effect.retry({
221 times: 2,
222 while: (error) => error._tag === "VcsProcessExitError" && error.retryable === true,
223 schedule: Schedule.spaced(Duration.millis(75)),
224 }),
225 );
226 }
227 return yield* input.command === "gh" ? githubProcesses.withPermits(1)(bounded) : bounded;
228 });
229
230 return VcsProcess.of({ run });
231});
232
233export const layer = Layer.effect(VcsProcess, make).pipe(Layer.provide(ProcessRunner.layer));