Coverage for core/langgraph/tools/git_checkout.py: 97%
139 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-10-05 00:44 +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)
56def _hard_failure_type() -> type[Exception]:
57 """The runner's hard-failure type, resolved without an import cycle."""
58 try:
59 from runner import RunHardFailure
60 except ImportError: # pragma: no cover — main.py run (qualified imports)
61 from langgraph.runner import RunHardFailure
62 return RunHardFailure
65def _hard_failure(code: str, message: str, details: dict | None = None) -> Exception:
66 return _hard_failure_type()(code, message, details)
69def _git_argv(*args: str) -> list[str]:
70 return ["git", "-c", "safe.directory=*", *args]
73def _git_failure(args: tuple[str, ...], exc: subprocess.CalledProcessError) -> Exception:
74 detail = (exc.stderr or exc.stdout or "").strip()
75 message = f"git-checkout: 'git {' '.join(args)}' failed (exit {exc.returncode})"
76 return _hard_failure("EXECUTION_ERROR", f"{message}: {detail}" if detail else message)
79def _git_spawn_failure(args: tuple[str, ...], exc: OSError) -> Exception:
80 return _hard_failure(
81 "EXECUTION_ERROR",
82 f"git-checkout: 'git {' '.join(args)}' could not be started: {exc}",
83 )
86def _run_git(args: tuple[str, ...], cwd: str | None = None) -> None:
87 """Run a git command with its output streaming to the run log.
89 Clone/checkout/fetch output must stay observable, so git inherits the
90 parent's stdout/stderr instead of being captured. A subprocess failure
91 (nonzero exit or spawn failure, e.g. a missing git binary) becomes a
92 typed hard failure — never a raw exception.
93 """
94 try:
95 subprocess.run(_git_argv(*args), cwd=cwd, check=True)
96 except subprocess.CalledProcessError as exc:
97 raise _git_failure(args, exc) from exc
98 except OSError as exc:
99 raise _git_spawn_failure(args, exc) from exc
102def _capture_git(
103 args: tuple[str, ...], cwd: str | None = None
104) -> subprocess.CompletedProcess:
105 """Run a git command whose stdout is parsed by the caller.
107 Capture is genuinely required for ls-remote/diff (the output is
108 consumed), unlike the streaming clone/checkout path in `_run_git`.
109 """
110 try:
111 return subprocess.run(
112 _git_argv(*args),
113 cwd=cwd,
114 capture_output=True,
115 text=True,
116 check=True,
117 )
118 except subprocess.CalledProcessError as exc:
119 raise _git_failure(args, exc) from exc
120 except OSError as exc:
121 raise _git_spawn_failure(args, exc) from exc
124def _reset_project_dir(project_dir: str) -> None:
125 if not os.path.exists(project_dir):
126 return
127 try:
128 with os.scandir(project_dir) as entries:
129 for entry in entries:
130 if entry.is_symlink() or not entry.is_dir():
131 os.unlink(entry.path)
132 else:
133 shutil.rmtree(entry.path)
134 except OSError as exc:
135 raise _hard_failure(
136 "EXECUTION_ERROR",
137 f"git-checkout: failed to reset {project_dir!r}: {exc}",
138 ) from exc
141def _require_env(name: str) -> str:
142 value = os.environ.get(name)
143 if not value:
144 raise _hard_failure(
145 "VALIDATION_ERROR",
146 f"git-checkout: required env var {name!r} is not set",
147 )
148 return value
151def _resolve_default_branch(project_dir: str) -> str:
152 """Resolve the repo's remote default branch (repo fact, not a fallback).
154 TICKET profile with an empty BASE_BRANCH: the base branch is the remote
155 default reported by `git ls-remote --symref origin HEAD` — never a
156 hardcoded "main" assumption (ADR-0012, contract BASE_BRANCH row).
157 """
158 proc = _capture_git(("ls-remote", "--symref", "origin", "HEAD"), cwd=project_dir)
159 for line in proc.stdout.splitlines():
160 if line.startswith("ref: "):
161 # Format: "ref: refs/heads/main\tHEAD"
162 parts = line.split("\t")
163 ref = parts[0][len("ref: "):].strip()
164 if ref.startswith("refs/heads/"):
165 return ref[len("refs/heads/"):]
166 raise _hard_failure(
167 "EXECUTION_ERROR",
168 "git-checkout: could not determine the remote default branch from "
169 "'git ls-remote --symref origin HEAD' (no ref: refs/heads/... HEAD line)",
170 )
173def _ticket_branch(issue_key: str, title: str) -> str:
174 """Deterministic per-run feature branch for the TICKET profile.
176 Format (locked): ticket/<issue_key>-<slug(title)>; slugify is the
177 ADR-0018 slug. Retries and continuation runs reuse the same branch:
178 git-checkout resumes an existing remote branch instead of creating a
179 new one per run, and the push is plain — never a forced push.
180 """
181 return f"ticket/{issue_key}-{slugify(title)}"
184def _remote_branch_exists(remote_url: str, branch: str) -> bool:
185 """True when the remote already has the branch (TICKET resume path).
187 `git ls-remote --heads <url> refs/heads/<branch>` runs BEFORE the
188 clone — a non-empty stdout line means the branch exists and the
189 checkout resumes it instead of creating a new one (continuation runs
190 continue the SAME branch and PR). Failure → hard error (fail-fast,
191 ADR-0012).
192 """
193 proc = _capture_git(
194 ("ls-remote", "--heads", remote_url, f"refs/heads/{branch}"),
195 )
196 return bool(proc.stdout.strip())
199def _join_pr_diff(patches: list[str]) -> str:
200 return "\n".join(p for p in patches if p)
203def _git_checkout(step: dict, state: dict) -> dict:
204 url = _require_env("GIT_REPO_URL")
205 token = _require_env("GIT_TOKEN")
206 run_profile = _require_env("RUN_PROFILE")
208 if run_profile == "PR":
209 branch = _require_env("PR_HEAD_BRANCH")
210 elif run_profile == "BRANCH":
211 branch = _require_env("BASE_BRANCH")
212 elif run_profile != "TICKET":
213 raise _hard_failure(
214 "VALIDATION_ERROR",
215 f"git-checkout: unsupported RUN_PROFILE {run_profile!r} "
216 "(expected 'PR', 'BRANCH', or 'TICKET')",
217 )
219 parsed = urlparse(url)
220 host = parsed.hostname or ""
221 new_netloc = f"x-access-token:{token}@{host}"
222 url = urlunparse(parsed._replace(netloc=new_netloc))
224 project_dir = "/project"
225 _reset_project_dir(project_dir)
227 if run_profile == "TICKET":
228 base_branch = os.environ.get("BASE_BRANCH") or ""
229 issue_key = _require_env("ISSUE_KEY")
230 title = _require_env("PR_TITLE")
231 feature_branch = _ticket_branch(issue_key, title)
233 if _remote_branch_exists(url, feature_branch):
234 _run_git(
235 ("clone", "--depth", "1", "--branch", feature_branch, url, project_dir),
236 )
237 if not base_branch:
238 # Resume + empty BASE_BRANCH: the PR base is the remote default.
239 base_branch = _resolve_default_branch(project_dir)
240 else:
241 if base_branch:
242 _run_git(
243 ("clone", "--depth", "1", "--branch", base_branch, url, project_dir),
244 )
245 else:
246 # Empty BASE_BRANCH → clone the default branch, then resolve the
247 # default branch name as a repo fact (PR base field + state).
248 _run_git(
249 ("clone", "--depth", "1", url, project_dir),
250 )
251 base_branch = _resolve_default_branch(project_dir)
252 _run_git(("checkout", "-b", feature_branch), cwd=project_dir)
253 result = {**state, "checkout_done": True, "project_dir": project_dir,
254 "ticket_branch": feature_branch, "ticket_base_branch": base_branch}
255 return result
257 _run_git(("clone", "--depth", "1", "--branch", branch, url, project_dir))
259 result = {**state, "checkout_done": True, "project_dir": project_dir}
261 if run_profile == "PR":
262 base_branch = _require_env("PR_BASE_BRANCH")
263 diff_base = f"origin/{base_branch}"
265 _run_git(
266 ("fetch", "--depth", "1", "origin", f"{base_branch}:refs/remotes/origin/{base_branch}"),
267 cwd=project_dir,
268 )
269 changed = _capture_git(("diff", "--name-only", f"{diff_base}..HEAD"), cwd=project_dir)
270 diff_out = _capture_git(("diff", f"{diff_base}..HEAD"), cwd=project_dir)
271 changed_files = [line for line in changed.stdout.splitlines() if line.strip()]
272 pr_diff = _join_pr_diff([diff_out.stdout])
273 result = {**result, "pr_diff": pr_diff, "changed_files": changed_files}
275 return result
278def _state_patch(new_state: dict) -> dict[str, Any]:
279 return {key: new_state[key] for key in _PATCH_KEYS if key in new_state}
282def _success_message(patch: dict[str, Any]) -> str:
283 return json.dumps(
284 {"name": GIT_CHECKOUT_TOOL_NAME, "ok": True, "output": patch},
285 ensure_ascii=False,
286 )
289def _failure_message(error: str) -> str:
290 return json.dumps(
291 {"name": GIT_CHECKOUT_TOOL_NAME, "ok": False, "error": error},
292 ensure_ascii=False,
293 )
296class _GitCheckoutTool:
297 name = GIT_CHECKOUT_TOOL_NAME
299 def definition(self) -> dict[str, Any]:
300 return GIT_CHECKOUT_DEFINITION
302 def instruction(self) -> str | None:
303 return None
305 def params(self) -> list[dict[str, Any]] | None:
306 return None
308 def execute(self, context: ToolContext) -> ToolResult:
309 try:
310 new_state = _git_checkout(context.step or {}, context.state or {})
311 except _hard_failure_type() as exc:
312 return ToolResult(
313 ok=False,
314 content=_failure_message(exc.message),
315 error=exc.message,
316 )
317 patch = _state_patch(new_state)
318 return ToolResult(
319 ok=True,
320 content=_success_message(patch),
321 state_patch=patch,
322 )
325TOOL = _GitCheckoutTool()