Coverage for core/langgraph/exec_runner.py: 89%
94 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-10-05 00:44 +0000
1"""
2exec_runner.py — engine/env command execution primitives.
4Two execution paths (mechanic A, locked):
6- Engine container: the command runs in the engine's own container via
7 `subprocess.run(["sh", "-c", command])` (see `run_engine_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 — callers decide its meaning.
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"
46# Platform-owned sidecar that runs StepFunction entrypoints in-cluster. Its
47# name is fixed (not slugged from an env name) and carries no engine secrets.
48STEP_FUNCTION_CONTAINER = "stepfunction-runner"
51class ExecError(RuntimeError):
52 """Hard execution error (ADR-0012) — never silently defaulted."""
55def slugify(name: str) -> str:
56 """ADR-0018 slug, byte-identical to the API's `git.SlugRepoName`.
58 Semantics (copied exactly from api/internal/git/service.go): lowercase;
59 each run of characters outside [a-z0-9] collapses to a single "-"; the
60 result is trimmed of leading/trailing "-" before truncation (so a
61 "-"-prefixed long name does not lose content to the hyphen) and again
62 after truncation (a cut can reintroduce a trailing hyphen); empty or
63 symbol-only input yields "-", never an error.
64 """
65 lower = name.lower()
66 parts: list[str] = []
67 last_dash = False
68 for ch in lower:
69 if ("a" <= ch <= "z") or ("0" <= ch <= "9"):
70 parts.append(ch)
71 last_dash = False
72 elif not last_dash:
73 parts.append("-")
74 last_dash = True
75 slug = "".join(parts)
76 slug = slug.strip("-")
77 if len(slug) > 64:
78 slug = slug[:64]
79 slug = slug.strip("-")
80 return slug or "-"
83def run_engine_command(
84 command: str,
85 *,
86 timeout: float | None = None,
87 cwd: str | None = None,
88 stdin: str | None = None,
89) -> dict:
90 """Run a command in the engine container via subprocess (`sh -c`).
92 Output is captured with UTF-8 (replace errors) for determinism. The
93 optional ``cwd`` sets the working directory; the optional ``stdin`` is fed
94 to the command as UTF-8 text. Both default to None — the original
95 behaviour (inherit cwd, no stdin) is unchanged.
96 """
97 try:
98 proc = subprocess.run(
99 ["sh", "-c", command],
100 capture_output=True,
101 encoding="utf-8",
102 errors="replace",
103 timeout=timeout,
104 cwd=cwd,
105 input=stdin,
106 )
107 except subprocess.TimeoutExpired as exc:
108 raise ExecError(f"command timed out after {timeout}s: {command!r}") from exc
109 return {
110 "exit_code": proc.returncode,
111 "stdout": proc.stdout,
112 "stderr": proc.stderr,
113 }
116def _read_namespace_file(path: Path) -> str:
117 """Read the pod namespace from the serviceaccount token directory."""
118 try:
119 namespace = path.read_text(encoding="utf-8").strip()
120 except OSError as exc:
121 raise ExecError(
122 f"cannot determine pod namespace ({path}): {exc} — "
123 "set the namespace explicitly when constructing KubernetesExec"
124 ) from exc
125 if not namespace:
126 raise ExecError(f"pod namespace file {path} is empty")
127 return namespace
130def _parse_exec_status(ws: Any) -> int:
131 """Map the Kubernetes exec status (error channel, v4 protocol) to an exit
132 code. Missing or uninterpretable status is a hard error (ADR-0012)."""
133 if ws.is_open():
134 raise ExecError("Kubernetes exec stream is still open — the command did not terminate")
135 raw = ws.read_channel(_EXEC_STATUS_CHANNEL)
136 if not raw:
137 raise ExecError("Kubernetes exec returned no status on the error channel — exit code unknown")
138 try:
139 status = yaml.safe_load(raw)
140 except yaml.YAMLError as exc:
141 raise ExecError(f"Kubernetes exec status is not valid YAML: {raw!r}") from exc
142 if isinstance(status, dict) and status.get("status") == "Success":
143 return 0
144 message = ""
145 if isinstance(status, dict):
146 details = status.get("details") or {}
147 for cause in details.get("causes") or []:
148 if isinstance(cause, dict) and cause.get("message"):
149 message = cause["message"]
150 m = re.search(r"(\d+)\s*$", message)
151 if m:
152 return int(m.group(1))
153 raise ExecError(f"Kubernetes exec status could not be interpreted: {raw!r}")
156class KubernetesExec:
157 """Executes a command in one of the pod's env containers (mechanic A).
159 The pod is the engine's own pod: pod name from hostname, container name
160 from slug(env name). In-cluster configuration is loaded lazily; outside a
161 cluster (no KUBERNETES_SERVICE_HOST) the run fails with a clear error.
162 Tests inject a fake api + stream function to stay hermetic.
163 """
165 def __init__(
166 self,
167 *,
168 api: Any = None,
169 stream_fn: Callable[..., Any] | None = None,
170 pod_name: str | None = None,
171 namespace: str | None = None,
172 ) -> None:
173 self._api = api
174 self._stream = stream_fn
175 self._pod_name = pod_name
176 self._namespace = namespace
178 def run(self, command: str, env_name: str) -> dict:
179 return self._exec_in_container(slugify(env_name), command)
181 def run_step_function(self, command: str) -> dict:
182 """Run a StepFunction entrypoint in the fixed platform sidecar.
184 The container name is not derived from an env name and is not checked
185 against the pod's envs; no environment or secret is forwarded. The
186 entrypoint reads its params through a file redirect the caller embeds
187 in ``command`` — no stdin channel is opened (the K8s exec v4 client has
188 no half-close and could stall a reader-to-EOF entrypoint).
189 """
190 return self._exec_in_container(STEP_FUNCTION_CONTAINER, command)
192 def _exec_in_container(self, container: str, command: str) -> dict:
193 if self._stream is None:
194 self._connect()
195 pod = self._pod_name or socket.gethostname()
196 namespace = self._namespace or _read_namespace_file(Path(_NAMESPACE_FILE))
197 ws = self._stream(
198 self._api.connect_get_namespaced_pod_exec,
199 name=pod,
200 namespace=namespace,
201 container=container,
202 command=["sh", "-c", f"cd {WORKSPACE_ROOT} && {command}"],
203 stderr=True,
204 stdin=False,
205 stdout=True,
206 tty=False,
207 _preload_content=False,
208 )
209 ws.run_forever(timeout=None)
210 return {
211 "exit_code": _parse_exec_status(ws),
212 "stdout": ws.read_stdout(),
213 "stderr": ws.read_stderr(),
214 }
216 def _connect(self) -> None:
217 """Load the in-cluster configuration and build the exec stream."""
218 if not os.getenv("KUBERNETES_SERVICE_HOST"):
219 raise ExecError(
220 "env commands require cluster execution: KUBERNETES_SERVICE_HOST "
221 "is not set — the engine is not running inside a Kubernetes pod"
222 )
223 from kubernetes import client as k8s_client, config
225 config.load_incluster_config()
226 self._api = k8s_client.CoreV1Api()
227 from kubernetes.stream import stream as k8s_stream
229 self._stream = k8s_stream