Coverage for core/langgraph/tools/git_checkout.py: 97%
144 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 03:42 +0000
1"""Per-tool module for the `git-checkout` deterministic builtin.
3Owns the tool's model-facing definition, its instruction surface (none), the
4git-checkout-only helpers (env requirements, project reset, the streaming and
5capturing git runners, branch derivation/resolution) and the checkout
6execution itself (ADR-0012/ADR-0018). The definition and the execution here
7are the single source; the legacy builtin site delegates to the pure
8`_git_checkout` handler.
9"""
10from __future__ import annotations
12import json
13import os
14import shutil
15import subprocess
16from typing import Any
17from urllib.parse import urlparse, urlunparse
19from .contract import ToolContext, ToolResult
21try:
22 from exec_runner import slugify
23except ImportError: # pragma: no cover — main.py run (qualified imports)
24 from langgraph.exec_runner import slugify
26GIT_CHECKOUT_TOOL_NAME = "git-checkout"
28GIT_CHECKOUT_DEFINITION: dict[str, Any] = {
29 "type": "function",
30 "function": {
31 "name": GIT_CHECKOUT_TOOL_NAME,
32 "description": (
33 "Prepare the working tree with `git-checkout` before any step that "
34 "reads or writes repository files. It clones the target repository "
35 "for the active run profile and exposes the checkout location and "
36 "branch facts in state; do not clone the repository yourself."
37 ),
38 "parameters": {
39 "type": "object",
40 "properties": {},
41 "additionalProperties": False,
42 },
43 },
44}
46_PATCH_KEYS = (
47 "checkout_done",
48 "project_dir",
49 "ticket_branch",
50 "ticket_base_branch",
51 "pr_diff",
52 "changed_files",
53)
55_PR_DIFF_MAX_CHARS = 20_000
56_PR_DIFF_TRUNCATION_SUFFIX = "\n… (truncated)"
59def _hard_failure_type() -> type[Exception]:
60 """The runner's hard-failure type, resolved without an import cycle."""
61 try:
62 from runner import RunHardFailure
63 except ImportError: # pragma: no cover — main.py run (qualified imports)
64 from langgraph.runner import RunHardFailure
65 return RunHardFailure
68def _hard_failure(code: str, message: str, details: dict | None = None) -> Exception:
69 return _hard_failure_type()(code, message, details)
72def _git_argv(*args: str) -> list[str]:
73 return ["git", "-c", "safe.directory=*", *args]
76def _git_failure(args: tuple[str, ...], exc: subprocess.CalledProcessError) -> Exception:
77 detail = (exc.stderr or exc.stdout or "").strip()
78 message = f"git-checkout: 'git {' '.join(args)}' failed (exit {exc.returncode})"
79 return _hard_failure("EXECUTION_ERROR", f"{message}: {detail}" if detail else message)
82def _git_spawn_failure(args: tuple[str, ...], exc: OSError) -> Exception:
83 return _hard_failure(
84 "EXECUTION_ERROR",
85 f"git-checkout: 'git {' '.join(args)}' could not be started: {exc}",
86 )
89def _run_git(args: tuple[str, ...], cwd: str | None = None) -> None:
90 """Run a git command with its output streaming to the run log.
92 Clone/checkout/fetch output must stay observable, so git inherits the
93 parent's stdout/stderr instead of being captured. A subprocess failure
94 (nonzero exit or spawn failure, e.g. a missing git binary) becomes a
95 typed hard failure — never a raw exception.
96 """
97 try:
98 subprocess.run(_git_argv(*args), cwd=cwd, check=True)
99 except subprocess.CalledProcessError as exc:
100 raise _git_failure(args, exc) from exc
101 except OSError as exc:
102 raise _git_spawn_failure(args, exc) from exc
105def _capture_git(
106 args: tuple[str, ...], cwd: str | None = None
107) -> subprocess.CompletedProcess:
108 """Run a git command whose stdout is parsed by the caller.
110 Capture is genuinely required for ls-remote/diff (the output is
111 consumed), unlike the streaming clone/checkout path in `_run_git`.
112 """
113 try:
114 return subprocess.run(
115 _git_argv(*args),
116 cwd=cwd,
117 capture_output=True,
118 text=True,
119 check=True,
120 )
121 except subprocess.CalledProcessError as exc:
122 raise _git_failure(args, exc) from exc
123 except OSError as exc:
124 raise _git_spawn_failure(args, exc) from exc
127def _reset_project_dir(project_dir: str) -> None:
128 if not os.path.exists(project_dir):
129 return
130 try:
131 with os.scandir(project_dir) as entries:
132 for entry in entries:
133 if entry.is_symlink() or not entry.is_dir():
134 os.unlink(entry.path)
135 else:
136 shutil.rmtree(entry.path)
137 except OSError as exc:
138 raise _hard_failure(
139 "EXECUTION_ERROR",
140 f"git-checkout: failed to reset {project_dir!r}: {exc}",
141 ) from exc
144def _require_env(name: str) -> str:
145 value = os.environ.get(name)
146 if not value:
147 raise _hard_failure(
148 "VALIDATION_ERROR",
149 f"git-checkout: required env var {name!r} is not set",
150 )
151 return value
154def _resolve_default_branch(project_dir: str) -> str:
155 """Resolve the repo's remote default branch (repo fact, not a fallback).
157 TICKET profile with an empty BASE_BRANCH: the base branch is the remote
158 default reported by `git ls-remote --symref origin HEAD` — never a
159 hardcoded "main" assumption (ADR-0012, contract BASE_BRANCH row).
160 """
161 proc = _capture_git(("ls-remote", "--symref", "origin", "HEAD"), cwd=project_dir)
162 for line in proc.stdout.splitlines():
163 if line.startswith("ref: "):
164 # Format: "ref: refs/heads/main\tHEAD"
165 parts = line.split("\t")
166 ref = parts[0][len("ref: "):].strip()
167 if ref.startswith("refs/heads/"):
168 return ref[len("refs/heads/"):]
169 raise _hard_failure(
170 "EXECUTION_ERROR",
171 "git-checkout: could not determine the remote default branch from "
172 "'git ls-remote --symref origin HEAD' (no ref: refs/heads/... HEAD line)",
173 )
176def _ticket_branch(issue_key: str, title: str) -> str:
177 """Deterministic per-run feature branch for the TICKET profile.
179 Format (locked): ticket/<issue_key>-<slug(title)>; slugify is the
180 ADR-0018 slug. Retries and continuation runs reuse the same branch:
181 git-checkout resumes an existing remote branch instead of creating a
182 new one per run, and the push is plain — never a forced push.
183 """
184 return f"ticket/{issue_key}-{slugify(title)}"
187def _remote_branch_exists(remote_url: str, branch: str) -> bool:
188 """True when the remote already has the branch (TICKET resume path).
190 `git ls-remote --heads <url> refs/heads/<branch>` runs BEFORE the
191 clone — a non-empty stdout line means the branch exists and the
192 checkout resumes it instead of creating a new one (continuation runs
193 continue the SAME branch and PR). Failure → hard error (fail-fast,
194 ADR-0012).
195 """
196 proc = _capture_git(
197 ("ls-remote", "--heads", remote_url, f"refs/heads/{branch}"),
198 )
199 return bool(proc.stdout.strip())
202def _join_pr_diff(patches: list[str]) -> str:
203 joined = "\n".join(p for p in patches if p)
204 if len(joined) > _PR_DIFF_MAX_CHARS:
205 joined = joined[:_PR_DIFF_MAX_CHARS].rstrip() + _PR_DIFF_TRUNCATION_SUFFIX
206 return joined
209def _git_checkout(step: dict, state: dict) -> dict:
210 url = _require_env("GIT_REPO_URL")
211 token = _require_env("GIT_TOKEN")
212 run_profile = _require_env("RUN_PROFILE")
214 if run_profile == "PR":
215 branch = _require_env("PR_HEAD_BRANCH")
216 elif run_profile == "BRANCH":
217 branch = _require_env("BASE_BRANCH")
218 elif run_profile != "TICKET":
219 raise _hard_failure(
220 "VALIDATION_ERROR",
221 f"git-checkout: unsupported RUN_PROFILE {run_profile!r} "
222 "(expected 'PR', 'BRANCH', or 'TICKET')",
223 )
225 parsed = urlparse(url)
226 host = parsed.hostname or ""
227 new_netloc = f"x-access-token:{token}@{host}"
228 url = urlunparse(parsed._replace(netloc=new_netloc))
230 project_dir = "/project"
231 _reset_project_dir(project_dir)
233 if run_profile == "TICKET":
234 base_branch = os.environ.get("BASE_BRANCH") or ""
235 issue_key = _require_env("ISSUE_KEY")
236 title = _require_env("PR_TITLE")
237 feature_branch = _ticket_branch(issue_key, title)
239 if _remote_branch_exists(url, feature_branch):
240 _run_git(
241 ("clone", "--depth", "1", "--branch", feature_branch, url, project_dir),
242 )
243 if not base_branch:
244 # Resume + empty BASE_BRANCH: the PR base is the remote default.
245 base_branch = _resolve_default_branch(project_dir)
246 else:
247 if base_branch:
248 _run_git(
249 ("clone", "--depth", "1", "--branch", base_branch, url, project_dir),
250 )
251 else:
252 # Empty BASE_BRANCH → clone the default branch, then resolve the
253 # default branch name as a repo fact (PR base field + state).
254 _run_git(
255 ("clone", "--depth", "1", url, project_dir),
256 )
257 base_branch = _resolve_default_branch(project_dir)
258 _run_git(("checkout", "-b", feature_branch), cwd=project_dir)
259 result = {**state, "checkout_done": True, "project_dir": project_dir,
260 "ticket_branch": feature_branch, "ticket_base_branch": base_branch}
261 return result
263 _run_git(("clone", "--depth", "1", "--branch", branch, url, project_dir))
265 result = {**state, "checkout_done": True, "project_dir": project_dir}
267 if run_profile == "PR":
268 base_branch = _require_env("PR_BASE_BRANCH")
269 diff_base = f"origin/{base_branch}"
271 _run_git(
272 ("fetch", "--depth", "1", "origin", f"{base_branch}:refs/remotes/origin/{base_branch}"),
273 cwd=project_dir,
274 )
275 changed = _capture_git(("diff", "--name-only", f"{diff_base}..HEAD"), cwd=project_dir)
276 diff_out = _capture_git(("diff", f"{diff_base}..HEAD"), cwd=project_dir)
277 changed_files = [line for line in changed.stdout.splitlines() if line.strip()]
278 pr_diff = _join_pr_diff([diff_out.stdout])
279 result = {**result, "pr_diff": pr_diff, "changed_files": changed_files}
281 return result
284def _state_patch(new_state: dict) -> dict[str, Any]:
285 return {key: new_state[key] for key in _PATCH_KEYS if key in new_state}
288def _success_message(patch: dict[str, Any]) -> str:
289 return json.dumps(
290 {"name": GIT_CHECKOUT_TOOL_NAME, "ok": True, "output": patch},
291 ensure_ascii=False,
292 )
295def _failure_message(error: str) -> str:
296 return json.dumps(
297 {"name": GIT_CHECKOUT_TOOL_NAME, "ok": False, "error": error},
298 ensure_ascii=False,
299 )
302class _GitCheckoutTool:
303 name = GIT_CHECKOUT_TOOL_NAME
305 def definition(self) -> dict[str, Any]:
306 return GIT_CHECKOUT_DEFINITION
308 def instruction(self) -> str | None:
309 return None
311 def params(self) -> list[dict[str, Any]] | None:
312 return None
314 def execute(self, context: ToolContext) -> ToolResult:
315 try:
316 new_state = _git_checkout(context.step or {}, context.state or {})
317 except _hard_failure_type() as exc:
318 return ToolResult(
319 ok=False,
320 content=_failure_message(exc.message),
321 error=exc.message,
322 )
323 patch = _state_patch(new_state)
324 return ToolResult(
325 ok=True,
326 content=_success_message(patch),
327 state_patch=patch,
328 )
331TOOL = _GitCheckoutTool()