Coverage for src/keel/runtime.py: 100%
150 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Runtime capability detection and requirement evaluation.
3Capabilities describe what the current execution environment can do. They are runtime
4facts, not project policy: whether local tools exist, whether GitHub access is available,
5and whether live mutation classes are possible. The detector is injectable so tests stay
6offline and deterministic.
7"""
9from __future__ import annotations
11import json
12import os
13import shutil
14from collections.abc import Callable, Mapping
15from dataclasses import dataclass
16from pathlib import Path
18from . import capabilities
20KNOWN_CAPABILITIES = capabilities.KNOWN_CAPABILITIES
22@dataclass(frozen=True)
23class Capability:
24 """One detected runtime capability."""
26 name: str
27 available: bool
28 detail: str
29 source: str
31 def as_dict(self) -> dict:
32 return {
33 "name": self.name,
34 "available": self.available,
35 "detail": self.detail,
36 "source": self.source,
37 }
40@dataclass(frozen=True)
41class CapabilityReport:
42 """All capabilities detected for a run."""
44 capabilities: tuple[Capability, ...]
46 def get(self, name: str) -> Capability:
47 for cap in self.capabilities:
48 if cap.name == name:
49 return cap
50 return Capability(name, False, "unknown capability", "unknown")
52 def available(self, name: str) -> bool:
53 return self.get(name).available
55 def as_dict(self) -> dict:
56 return {"capabilities": [cap.as_dict() for cap in self.capabilities]}
58 def to_json(self) -> str:
59 return json.dumps(self.as_dict(), indent=2, sort_keys=True)
61 def render(self) -> str:
62 lines = ["keel capabilities"]
63 for cap in self.capabilities:
64 status = "yes" if cap.available else "no"
65 lines.append(f" {cap.name:<18} {status:<3} {cap.detail}")
66 return "\n".join(lines)
69@dataclass(frozen=True)
70class CapabilityRequirement:
71 """Capabilities needed by a command or extension."""
73 required: tuple[str, ...] = ()
74 optional: tuple[str, ...] = ()
76 def merged(self, other: CapabilityRequirement) -> CapabilityRequirement:
77 return CapabilityRequirement(
78 required=_unique((*self.required, *other.required)),
79 optional=_unique((*self.optional, *other.optional)),
80 )
82 def as_dict(self) -> dict:
83 return {"required": list(self.required), "optional": list(self.optional)}
86@dataclass(frozen=True)
87class CapabilityEvaluation:
88 """A requirement checked against a capability report."""
90 requirement: CapabilityRequirement
91 missing_required: tuple[str, ...]
92 missing_optional: tuple[str, ...]
94 @property
95 def ok(self) -> bool:
96 return not self.missing_required
98 def as_dict(self) -> dict:
99 return {
100 "required": list(self.requirement.required),
101 "optional": list(self.requirement.optional),
102 "missing_required": list(self.missing_required),
103 "missing_optional": list(self.missing_optional),
104 "ok": self.ok,
105 }
107 def render(self) -> str:
108 lines = [
109 "runtime capabilities:",
110 f" required: {', '.join(self.requirement.required) or '-'}",
111 f" optional: {', '.join(self.requirement.optional) or '-'}",
112 ]
113 if self.missing_required:
114 lines.append(f" missing required: {', '.join(self.missing_required)}")
115 if self.missing_optional:
116 lines.append(f" degraded optional: {', '.join(self.missing_optional)}")
117 return "\n".join(lines)
120def detect(
121 root: str | Path = ".",
122 *,
123 env: Mapping[str, str] | None = None,
124 which: Callable[[str], str | None] = shutil.which,
125 run: Callable[..., object] | None = None,
126) -> CapabilityReport:
127 """Detect capabilities for the current runtime.
129 Environment overrides intentionally use generic keel names so projects can surface
130 host-agent capabilities without hardcoding one consumer's tooling into core.
131 """
133 env = os.environ if env is None else env
134 if run is None:
135 from .runner import run_argv
136 run = run_argv
137 root_path = Path(root)
138 sh = which("sh")
139 git = which("git")
140 gh = which("gh")
141 adb = _tool_capability("adb", env_name="KEEL_ADB", env=env, which=which)
142 firebase = _tool_capability("firebase", env_name="KEEL_FIREBASE", env=env, which=which)
143 filesystem_write = _can_write(root_path)
144 gh_auth = False
145 gh_auth_detail = "gh not available"
146 if gh:
147 result = run(["gh", "auth", "status"], cwd=str(root_path), timeout=10)
148 gh_auth = bool(getattr(result, "ok", False))
149 gh_auth_detail = "authenticated" if gh_auth else "gh auth status failed"
151 caps = (
152 Capability("shell", sh is not None, sh or "sh not found", "PATH"),
153 Capability("git", git is not None, git or "git not found", "PATH"),
154 Capability("gh", gh is not None, gh or "gh not found", "PATH"),
155 Capability("gh-auth", gh_auth, gh_auth_detail, "gh auth status"),
156 Capability("github-mcp", _truthy(env.get("KEEL_GITHUB_MCP")),
157 "KEEL_GITHUB_MCP", "environment"),
158 Capability("subagents", _truthy(env.get("KEEL_SUBAGENTS")),
159 "KEEL_SUBAGENTS", "environment"),
160 Capability("parallel-subagents", _truthy(env.get("KEEL_PARALLEL_SUBAGENTS")),
161 "KEEL_PARALLEL_SUBAGENTS", "environment"),
162 Capability("browser", _truthy(env.get("KEEL_BROWSER")),
163 "KEEL_BROWSER", "environment"),
164 adb,
165 firebase,
166 Capability("filesystem-write", filesystem_write,
167 "root writable" if filesystem_write else "root not writable", "filesystem"),
168 Capability("worktree", git is not None and filesystem_write,
169 "requires git and writable root", "derived"),
170 Capability("release-publish", _truthy(env.get("KEEL_RELEASE_PUBLISH")),
171 "KEEL_RELEASE_PUBLISH", "environment"),
172 Capability("secret-access", _truthy(env.get("KEEL_SECRET_ACCESS")),
173 "KEEL_SECRET_ACCESS", "environment"),
174 _api_token_capability(env),
175 Capability("production-adjacent", _truthy(env.get("KEEL_PRODUCTION_ADJACENT")),
176 "KEEL_PRODUCTION_ADJACENT", "environment"),
177 Capability("private-setup", _truthy(env.get("KEEL_PRIVATE_SETUP")),
178 "KEEL_PRIVATE_SETUP", "environment"),
179 )
180 return CapabilityReport(caps)
183def evaluate(requirement: CapabilityRequirement, report: CapabilityReport) -> CapabilityEvaluation:
184 """Check required and optional capabilities against a report."""
186 missing_required = tuple(name for name in requirement.required if not report.available(name))
187 missing_optional = tuple(name for name in requirement.optional if not report.available(name))
188 return CapabilityEvaluation(requirement, missing_required, missing_optional)
191def validate_names(names: tuple[str, ...] | list[str], *, source: str) -> list[str]:
192 """Return errors for unknown capability names."""
194 return capabilities.validate_names(names, source=source)
197def _truthy(value: str | None) -> bool:
198 return (value or "").strip().lower() in {"1", "true", "yes", "on"}
201def _tool_capability(
202 name: str,
203 *,
204 env_name: str,
205 env: Mapping[str, str],
206 which: Callable[[str], str | None],
207) -> Capability:
208 if _truthy(env.get(env_name)):
209 return Capability(name, True, env_name, "environment")
210 path = which(name)
211 return Capability(name, path is not None, path or f"{name} not found", "PATH")
214def _api_token_capability(env: Mapping[str, str]) -> Capability:
215 """``api-token``: a hosted-API delegate key is present in the environment.
217 The detail names the env vars found (never their values); the per-vendor
218 dispatch check is :func:`keel.api_delegate.has_api_token`.
219 """
220 from .api_delegate import present_key_names
222 names = present_key_names(_env=env)
223 detail = (
224 ", ".join(names)
225 if names
226 else "no vendor API key (ANTHROPIC_API_KEY/OPENAI_API_KEY/GEMINI_API_KEY)"
227 )
228 return Capability("api-token", bool(names), detail, "environment")
231def _can_write(root: Path) -> bool:
232 if not root.exists():
233 return False
234 return os.access(root, os.W_OK)
237def _unique(values: tuple[str, ...]) -> tuple[str, ...]:
238 return tuple(dict.fromkeys(values))
241def build_capability_requirement(
242 command: str,
243 config,
244 loaded: dict[str, list],
245 *,
246 pr: int | None = None,
247) -> CapabilityRequirement:
248 """Build the runtime capability requirement for a given command, config,
249 and loaded extensions.
250 """
251 from . import gates, project_commands
253 del pr
254 req = CapabilityRequirement(
255 required=config.knobs.required_capabilities,
256 optional=config.knobs.optional_capabilities,
257 )
258 try:
259 specs = gates.plan_gates(config, loaded)
260 except gates.GateError:
261 return req
262 if project_command := project_commands.get_project_command(config, command):
263 req = req.merged(CapabilityRequirement(
264 required=project_command.required_capabilities,
265 optional=project_command.optional_capabilities,
266 ))
268 command_gate_commands = {
269 "run-gates", "ship", "pr-loop", "wrap", "work-block", "overnight",
270 "implement", "coverage", "deps-audit", "flake-audit",
271 }
272 if command in command_gate_commands and any(s.kind == "command" for s in specs):
273 req = req.merged(CapabilityRequirement(required=("shell",)))
274 worktree_commands = {
275 "ship", "pr-loop", "wrap", "work-block", "overnight", "implement"
276 }
277 github_read_commands = {
278 "morning", "review-cycle", "triage", "stale-prs", "regression", "review-all-day",
279 "coverage", "deps-audit", "flake-audit", "ci-check",
280 }
281 if command in worktree_commands:
282 req = req.merged(CapabilityRequirement(required=("git", "worktree"),
283 optional=("gh", "gh-auth")))
284 elif command in github_read_commands:
285 req = req.merged(CapabilityRequirement(optional=("gh", "gh-auth")))
286 for spec in specs:
287 if spec.required_capabilities or spec.optional_capabilities:
288 req = req.merged(CapabilityRequirement(
289 required=spec.required_capabilities,
290 optional=spec.optional_capabilities,
291 ))
292 return req
295def ci_check_capability_requirement(config) -> CapabilityRequirement:
296 """Capability requirements for ci-check command."""
297 optional = ["gh", "gh-auth"]
298 if config.knobs.ci_workflows:
299 optional.append("raw-actions-logs")
300 return CapabilityRequirement(optional=tuple(optional))
303def morning_capability_requirement(config) -> CapabilityRequirement:
304 """Capability requirements for morning command."""
305 required: list[str] = []
306 optional: list[str] = ["gh", "gh-auth"]
307 pack = config.policy_pack or {}
308 health = pack.get("health_providers") if isinstance(pack.get("health_providers"), dict) else {}
309 for provider in health.values():
310 if not isinstance(provider, dict):
311 continue
312 required.extend(provider.get("required_capabilities") or ())
313 optional.extend(provider.get("optional_capabilities") or ())
314 return CapabilityRequirement(
315 required=tuple(dict.fromkeys(required)),
316 optional=tuple(dict.fromkeys(optional)),
317 )
320def scan_capability_requirement(command: str, config) -> CapabilityRequirement:
321 """Capability requirements for scan commands (regression, review-all-day)."""
322 del config
323 if command == "regression":
324 return CapabilityRequirement(
325 required=("git", "worktree"),
326 optional=("gh", "gh-auth", "github-mcp", "parallel-subagents"),
327 )
328 return CapabilityRequirement(
329 required=("git",),
330 optional=("gh", "gh-auth", "github-mcp", "parallel-subagents"),
331 )