Coverage for core/langgraph/exec_runner.py: 90%
98 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 02:20 +0000
1"""
2exec_runner.py — command execution for manifest `command` steps.
4Two execution paths (mechanic A, locked):
6- Engine container (default): the command runs in the engine's own
7 container via `subprocess.run(["sh", "-c", command])`.
8- Env container: the command runs inside the pod's env container through
9 the Kubernetes exec API — pod name from hostname (pod identity, M4),
10 container = slug(env name) (ADR-0018, identical to the API's
11 `git.SlugRepoName`), command `sh -c <command>`. In-cluster config is used
12 when KUBERNETES_SERVICE_HOST is set; outside a cluster (local Docker
13 Compose mode) env execution fails with a clear error — no silent fallback
14 (ADR-0012).
16Result shape for both paths:
17 {"exit_code": int, "stdout": str, "stderr": str}
19A non-zero exit code is a result, not a failure — LLM steps consume it.
20"""
21from __future__ import annotations
23import logging
24import os
25import re
26import socket
27import subprocess
28from pathlib import Path
29from typing import Any, Callable
31import yaml
33log = logging.getLogger(__name__)
35# Kubernetes exec v4 protocol channels (kubernetes.stream.ws_client):
36# 0 stdin, 1 stdout, 2 stderr, 3 error/status channel.
37_EXEC_STATUS_CHANNEL = 3
39_NAMESPACE_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
41# ADR-0024 D2: env-container commands run in the workspace root as the
42# default working directory — the sh -c command carries a cd prefix. The
43# path must match the API's workspace mount point (/project, ADR-0024 D1).
44WORKSPACE_ROOT = "/project"
47class ExecError(RuntimeError):
48 """Hard execution error (ADR-0012) — never silently defaulted."""
51def slugify(name: str) -> str:
52 """ADR-0018 slug, byte-identical to the API's `git.SlugRepoName`.
54 Semantics (copied exactly from api/internal/git/service.go): lowercase;
55 each run of characters outside [a-z0-9] collapses to a single "-"; the
56 result is trimmed of leading/trailing "-" before truncation (so a
57 "-"-prefixed long name does not lose content to the hyphen) and again
58 after truncation (a cut can reintroduce a trailing hyphen); empty or
59 symbol-only input yields "-", never an error.
60 """
61 lower = name.lower()
62 parts: list[str] = []
63 last_dash = False
64 for ch in lower:
65 if ("a" <= ch <= "z") or ("0" <= ch <= "9"):
66 parts.append(ch)
67 last_dash = False
68 elif not last_dash:
69 parts.append("-")
70 last_dash = True
71 slug = "".join(parts)
72 slug = slug.strip("-")
73 if len(slug) > 64:
74 slug = slug[:64]
75 slug = slug.strip("-")
76 return slug or "-"
79def run_engine_command(command: str, *, timeout: float | None = None) -> dict:
80 """Run a command in the engine container via subprocess (`sh -c`).
82 Output is captured with UTF-8 (replace errors) for determinism.
83 """
84 try:
85 proc = subprocess.run(
86 ["sh", "-c", command],
87 capture_output=True,
88 encoding="utf-8",
89 errors="replace",
90 timeout=timeout,
91 )
92 except subprocess.TimeoutExpired as exc:
93 raise ExecError(f"command timed out after {timeout}s: {command!r}") from exc
94 return {
95 "exit_code": proc.returncode,
96 "stdout": proc.stdout,
97 "stderr": proc.stderr,
98 }
101def _read_namespace_file(path: Path) -> str:
102 """Read the pod namespace from the serviceaccount token directory."""
103 try:
104 namespace = path.read_text(encoding="utf-8").strip()
105 except OSError as exc:
106 raise ExecError(
107 f"cannot determine pod namespace ({path}): {exc} — "
108 "set the namespace explicitly when constructing KubernetesExec"
109 ) from exc
110 if not namespace:
111 raise ExecError(f"pod namespace file {path} is empty")
112 return namespace
115def _parse_exec_status(ws: Any) -> int:
116 """Map the Kubernetes exec status (error channel, v4 protocol) to an exit
117 code. Missing or uninterpretable status is a hard error (ADR-0012)."""
118 if ws.is_open():
119 raise ExecError("Kubernetes exec stream is still open — the command did not terminate")
120 raw = ws.read_channel(_EXEC_STATUS_CHANNEL)
121 if not raw:
122 raise ExecError("Kubernetes exec returned no status on the error channel — exit code unknown")
123 try:
124 status = yaml.safe_load(raw)
125 except yaml.YAMLError as exc:
126 raise ExecError(f"Kubernetes exec status is not valid YAML: {raw!r}") from exc
127 if isinstance(status, dict) and status.get("status") == "Success":
128 return 0
129 message = ""
130 if isinstance(status, dict):
131 details = status.get("details") or {}
132 for cause in details.get("causes") or []:
133 if isinstance(cause, dict) and cause.get("message"):
134 message = cause["message"]
135 m = re.search(r"(\d+)\s*$", message)
136 if m:
137 return int(m.group(1))
138 raise ExecError(f"Kubernetes exec status could not be interpreted: {raw!r}")
141class KubernetesExec:
142 """Executes a command in one of the pod's env containers (mechanic A).
144 The pod is the engine's own pod: pod name from hostname, container name
145 from slug(env name). In-cluster configuration is loaded lazily; outside a
146 cluster (no KUBERNETES_SERVICE_HOST) the run fails with a clear error.
147 Tests inject a fake api + stream function to stay hermetic.
148 """
150 def __init__(
151 self,
152 *,
153 api: Any = None,
154 stream_fn: Callable[..., Any] | None = None,
155 pod_name: str | None = None,
156 namespace: str | None = None,
157 ) -> None:
158 self._api = api
159 self._stream = stream_fn
160 self._pod_name = pod_name
161 self._namespace = namespace
163 def run(self, command: str, env_name: str) -> dict:
164 if self._stream is None:
165 self._connect()
166 pod = self._pod_name or socket.gethostname()
167 namespace = self._namespace or _read_namespace_file(Path(_NAMESPACE_FILE))
168 container = slugify(env_name)
169 ws = self._stream(
170 self._api.connect_get_namespaced_pod_exec,
171 name=pod,
172 namespace=namespace,
173 container=container,
174 command=["sh", "-c", f"cd {WORKSPACE_ROOT} && {command}"],
175 stderr=True,
176 stdin=False,
177 stdout=True,
178 tty=False,
179 _preload_content=False,
180 )
181 ws.run_forever(timeout=None)
182 return {
183 "exit_code": _parse_exec_status(ws),
184 "stdout": ws.read_stdout(),
185 "stderr": ws.read_stderr(),
186 }
188 def _connect(self) -> None:
189 """Load the in-cluster configuration and build the exec stream."""
190 if not os.getenv("KUBERNETES_SERVICE_HOST"):
191 raise ExecError(
192 "env commands require cluster execution: KUBERNETES_SERVICE_HOST "
193 "is not set — the engine is not running inside a Kubernetes pod"
194 )
195 from kubernetes import client as k8s_client, config
197 config.load_incluster_config()
198 self._api = k8s_client.CoreV1Api()
199 from kubernetes.stream import stream as k8s_stream
201 self._stream = k8s_stream
204class CommandExecutor:
205 """Runs manifest commands: engine container by default (subprocess) or an
206 env container via Kubernetes exec (mechanic A)."""
208 def __init__(self, kubernetes_exec: KubernetesExec | None = None) -> None:
209 self._kubernetes_exec = kubernetes_exec
211 def run_in_engine(self, command: str, *, timeout: float | None = None) -> dict:
212 return run_engine_command(command, timeout=timeout)
214 def run_in_env(self, command: str, env_name: str) -> dict:
215 k8s = self._kubernetes_exec or KubernetesExec()
216 return k8s.run(command, env_name)