Coverage for src/keel/runtime.py: 100%
159 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-02 20:26 +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
19from .vocab import API_VENDORS, CLI_VENDORS, LOCAL_VENDORS
21KNOWN_CAPABILITIES = capabilities.KNOWN_CAPABILITIES
24@dataclass(frozen=True)
25class Capability:
26 """One detected runtime capability."""
28 name: str
29 available: bool
30 detail: str
31 source: str
33 def as_dict(self) -> dict:
34 return {
35 "name": self.name,
36 "available": self.available,
37 "detail": self.detail,
38 "source": self.source,
39 }
42@dataclass(frozen=True)
43class CapabilityReport:
44 """All capabilities detected for a run."""
46 capabilities: tuple[Capability, ...]
48 def get(self, name: str) -> Capability:
49 for cap in self.capabilities:
50 if cap.name == name:
51 return cap
52 return Capability(name, False, "unknown capability", "unknown")
54 def available(self, name: str) -> bool:
55 return self.get(name).available
57 def as_dict(self) -> dict:
58 return {"capabilities": [cap.as_dict() for cap in self.capabilities]}
60 def to_json(self) -> str:
61 return json.dumps(self.as_dict(), indent=2, sort_keys=True)
63 def render(self) -> str:
64 lines = ["keel capabilities"]
65 for cap in self.capabilities:
66 status = "yes" if cap.available else "no"
67 lines.append(f" {cap.name:<18} {status:<3} {cap.detail}")
68 return "\n".join(lines)
71@dataclass(frozen=True)
72class CapabilityRequirement:
73 """Capabilities needed by a command or extension."""
75 required: tuple[str, ...] = ()
76 optional: tuple[str, ...] = ()
78 def merged(self, other: CapabilityRequirement) -> CapabilityRequirement:
79 return CapabilityRequirement(
80 required=_unique((*self.required, *other.required)),
81 optional=_unique((*self.optional, *other.optional)),
82 )
84 def as_dict(self) -> dict:
85 return {"required": list(self.required), "optional": list(self.optional)}
88@dataclass(frozen=True)
89class CapabilityEvaluation:
90 """A requirement checked against a capability report."""
92 requirement: CapabilityRequirement
93 missing_required: tuple[str, ...]
94 missing_optional: tuple[str, ...]
96 @property
97 def ok(self) -> bool:
98 return not self.missing_required
100 def as_dict(self) -> dict:
101 return {
102 "required": list(self.requirement.required),
103 "optional": list(self.requirement.optional),
104 "missing_required": list(self.missing_required),
105 "missing_optional": list(self.missing_optional),
106 "ok": self.ok,
107 }
109 def render(self) -> str:
110 lines = [
111 "runtime capabilities:",
112 f" required: {', '.join(self.requirement.required) or '-'}",
113 f" optional: {', '.join(self.requirement.optional) or '-'}",
114 ]
115 if self.missing_required:
116 lines.append(f" missing required: {', '.join(self.missing_required)}")
117 if self.missing_optional:
118 lines.append(f" degraded optional: {', '.join(self.missing_optional)}")
119 return "\n".join(lines)
122def detect(
123 root: str | Path = ".",
124 *,
125 env: Mapping[str, str] | None = None,
126 which: Callable[[str], str | None] = shutil.which,
127 run: Callable[..., object] | None = None,
128) -> CapabilityReport:
129 """Detect capabilities for the current runtime.
131 Environment overrides intentionally use generic keel names so projects can surface
132 host-agent capabilities without hardcoding one consumer's tooling into core.
133 """
135 env = os.environ if env is None else env
136 if run is None:
137 from .runner import run_argv
139 run = run_argv
140 root_path = Path(root)
141 sh = which("sh")
142 git = which("git")
143 gh = which("gh")
144 adb = _tool_capability("adb", env_name="KEEL_ADB", env=env, which=which)
145 firebase = _tool_capability("firebase", env_name="KEEL_FIREBASE", env=env, which=which)
146 filesystem_write = _can_write(root_path)
147 gh_auth = False
148 gh_auth_detail = "gh not available"
149 if gh:
150 result = run(["gh", "auth", "status"], cwd=str(root_path), timeout=10)
151 gh_auth = bool(getattr(result, "ok", False))
152 gh_auth_detail = "authenticated" if gh_auth else "gh auth status failed"
154 caps = (
155 Capability("shell", sh is not None, sh or "sh not found", "PATH"),
156 Capability("git", git is not None, git or "git not found", "PATH"),
157 Capability("gh", gh is not None, gh or "gh not found", "PATH"),
158 Capability("gh-auth", gh_auth, gh_auth_detail, "gh auth status"),
159 Capability(
160 "github-mcp", _truthy(env.get("KEEL_GITHUB_MCP")), "KEEL_GITHUB_MCP", "environment"
161 ),
162 Capability(
163 "subagents", _truthy(env.get("KEEL_SUBAGENTS")), "KEEL_SUBAGENTS", "environment"
164 ),
165 Capability(
166 "parallel-subagents",
167 _truthy(env.get("KEEL_PARALLEL_SUBAGENTS")),
168 "KEEL_PARALLEL_SUBAGENTS",
169 "environment",
170 ),
171 Capability("browser", _truthy(env.get("KEEL_BROWSER")), "KEEL_BROWSER", "environment"),
172 adb,
173 firebase,
174 Capability(
175 "filesystem-write",
176 filesystem_write,
177 "root writable" if filesystem_write else "root not writable",
178 "filesystem",
179 ),
180 Capability(
181 "worktree",
182 git is not None and filesystem_write,
183 "requires git and writable root",
184 "derived",
185 ),
186 Capability(
187 "release-publish",
188 _truthy(env.get("KEEL_RELEASE_PUBLISH")),
189 "KEEL_RELEASE_PUBLISH",
190 "environment",
191 ),
192 Capability(
193 "secret-access",
194 _truthy(env.get("KEEL_SECRET_ACCESS")),
195 "KEEL_SECRET_ACCESS",
196 "environment",
197 ),
198 _api_token_capability(env),
199 *_provider_capabilities(env=env, which=which),
200 Capability(
201 "production-adjacent",
202 _truthy(env.get("KEEL_PRODUCTION_ADJACENT")),
203 "KEEL_PRODUCTION_ADJACENT",
204 "environment",
205 ),
206 Capability(
207 "private-setup",
208 _truthy(env.get("KEEL_PRIVATE_SETUP")),
209 "KEEL_PRIVATE_SETUP",
210 "environment",
211 ),
212 )
213 return CapabilityReport(caps)
216def evaluate(requirement: CapabilityRequirement, report: CapabilityReport) -> CapabilityEvaluation:
217 """Check required and optional capabilities against a report."""
219 missing_required = tuple(name for name in requirement.required if not report.available(name))
220 missing_optional = tuple(name for name in requirement.optional if not report.available(name))
221 return CapabilityEvaluation(requirement, missing_required, missing_optional)
224def validate_names(names: tuple[str, ...] | list[str], *, source: str) -> list[str]:
225 """Return errors for unknown capability names."""
227 return capabilities.validate_names(names, source=source)
230def _truthy(value: str | None) -> bool:
231 return (value or "").strip().lower() in {"1", "true", "yes", "on"}
234def _tool_capability(
235 name: str,
236 *,
237 env_name: str,
238 env: Mapping[str, str],
239 which: Callable[[str], str | None],
240) -> Capability:
241 if _truthy(env.get(env_name)):
242 return Capability(name, True, env_name, "environment")
243 path = which(name)
244 return Capability(name, path is not None, path or f"{name} not found", "PATH")
247def _api_token_capability(env: Mapping[str, str]) -> Capability:
248 """``api-token``: a hosted-API delegate key is present in the environment.
250 The detail names the env vars found (never their values); the per-vendor
251 dispatch check is :func:`keel.api_delegate.has_api_token`.
252 """
253 from .api_delegate import present_key_names
255 names = present_key_names(_env=env)
256 detail = (
257 ", ".join(names)
258 if names
259 else "no vendor API key (ANTHROPIC_API_KEY/OPENAI_API_KEY/GEMINI_API_KEY)"
260 )
261 return Capability("api-token", bool(names), detail, "environment")
264def _provider_capabilities(
265 *,
266 env: Mapping[str, str],
267 which: Callable[[str], str | None],
268) -> tuple[Capability, Capability]:
269 """``providers`` + ``review-vendors``: can this machine dispatch, and to how many?
271 ``providers`` is "at least one **tool-capable** implementer is available" — an
272 agent CLI that can run the git/PR steps itself. A hosted-API or local-model
273 delegate does not satisfy it: those run under keel's no-tools contract, where the
274 orchestrator performs every mutation.
276 ``review-vendors`` is "a cross-vendor review panel is possible here": at least
277 :data:`keel.providers.REVIEW_VENDOR_MINIMUM` **distinct** vendors are available
278 across all three transports. Two reviewers from one vendor is one opinion twice.
280 Deliberately the *cheap* half of the probe — ``detect`` runs on every command, so
281 this is PATH lookups and env-var names only, no subprocess and no network.
282 ``keel doctor --providers`` is the deep probe (versions, model listings, the local
283 Ollama server), and it is the only place that pays for them.
284 """
285 from . import providers
286 from .api_delegate import env_key_name
288 tool_capable = tuple(vendor for vendor in CLI_VENDORS if which(vendor))
289 local = tuple(vendor for vendor in LOCAL_VENDORS if which(vendor))
290 hosted = tuple(
291 vendor
292 for vendor in API_VENDORS
293 if (name := env_key_name(vendor)) and env.get(name, "").strip()
294 )
295 vendors = tool_capable + local + hosted
296 return (
297 Capability(
298 "providers",
299 bool(tool_capable),
300 ", ".join(tool_capable)
301 if tool_capable
302 else f"no tool-capable agent CLI on PATH ({', '.join(CLI_VENDORS)})",
303 "PATH",
304 ),
305 Capability(
306 "review-vendors",
307 len(vendors) >= providers.REVIEW_VENDOR_MINIMUM,
308 f"{len(vendors)} distinct vendor(s)" + (f": {', '.join(vendors)}" if vendors else ""),
309 "derived",
310 ),
311 )
314def _can_write(root: Path) -> bool:
315 if not root.exists():
316 return False
317 return os.access(root, os.W_OK)
320def _unique(values: tuple[str, ...]) -> tuple[str, ...]:
321 return tuple(dict.fromkeys(values))
324def build_capability_requirement(
325 command: str,
326 config,
327 loaded: dict[str, list],
328 *,
329 pr: int | None = None,
330) -> CapabilityRequirement:
331 """Build the runtime capability requirement for a given command, config,
332 and loaded extensions.
333 """
334 from . import gates, project_commands
336 del pr
337 req = CapabilityRequirement(
338 required=config.knobs.required_capabilities,
339 optional=config.knobs.optional_capabilities,
340 )
341 try:
342 specs = gates.plan_gates(config, loaded)
343 except gates.GateError:
344 return req
345 if project_command := project_commands.get_project_command(config, command):
346 req = req.merged(
347 CapabilityRequirement(
348 required=project_command.required_capabilities,
349 optional=project_command.optional_capabilities,
350 )
351 )
353 command_gate_commands = {
354 "run-gates",
355 "ship",
356 "pr-loop",
357 "wrap",
358 "work-block",
359 "overnight",
360 "implement",
361 "coverage",
362 "deps-audit",
363 "flake-audit",
364 }
365 if command in command_gate_commands and any(s.kind == "command" for s in specs):
366 req = req.merged(CapabilityRequirement(required=("shell",)))
367 worktree_commands = {"ship", "pr-loop", "wrap", "work-block", "overnight", "implement"}
368 github_read_commands = {
369 "morning",
370 "review-cycle",
371 "triage",
372 "stale-prs",
373 "regression",
374 "review-all-day",
375 "coverage",
376 "deps-audit",
377 "flake-audit",
378 "ci-check",
379 }
380 if command in worktree_commands:
381 req = req.merged(
382 CapabilityRequirement(required=("git", "worktree"), optional=("gh", "gh-auth"))
383 )
384 elif command in github_read_commands:
385 req = req.merged(CapabilityRequirement(optional=("gh", "gh-auth")))
386 for spec in specs:
387 if spec.required_capabilities or spec.optional_capabilities:
388 req = req.merged(
389 CapabilityRequirement(
390 required=spec.required_capabilities,
391 optional=spec.optional_capabilities,
392 )
393 )
394 return req
397def ci_check_capability_requirement(config) -> CapabilityRequirement:
398 """Capability requirements for ci-check command."""
399 optional = ["gh", "gh-auth"]
400 if config.knobs.ci_workflows:
401 optional.append("raw-actions-logs")
402 return CapabilityRequirement(optional=tuple(optional))
405def morning_capability_requirement(config) -> CapabilityRequirement:
406 """Capability requirements for morning command."""
407 required: list[str] = []
408 optional: list[str] = ["gh", "gh-auth"]
409 pack = config.policy_pack or {}
410 health = pack.get("health_providers") if isinstance(pack.get("health_providers"), dict) else {}
411 for provider in health.values():
412 if not isinstance(provider, dict):
413 continue
414 required.extend(provider.get("required_capabilities") or ())
415 optional.extend(provider.get("optional_capabilities") or ())
416 return CapabilityRequirement(
417 required=tuple(dict.fromkeys(required)),
418 optional=tuple(dict.fromkeys(optional)),
419 )
422def scan_capability_requirement(command: str, config) -> CapabilityRequirement:
423 """Capability requirements for scan commands (regression, review-all-day)."""
424 del config
425 if command == "regression":
426 return CapabilityRequirement(
427 required=("git", "worktree"),
428 optional=("gh", "gh-auth", "github-mcp", "parallel-subagents"),
429 )
430 return CapabilityRequirement(
431 required=("git",),
432 optional=("gh", "gh-auth", "github-mcp", "parallel-subagents"),
433 )