Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 73 additions & 6 deletions src/specify_cli/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,9 @@
import shutil
import stat
import subprocess
import sys
import tempfile
import warnings
import yaml
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Any
Expand All @@ -19,19 +21,84 @@
DOCKER_AGENT_CHECK_TIMEOUT = 5


def docker_agent_command(executable: str | None = None) -> list[str] | None:
def _docker_path_is_excluded(path: Path, roots: tuple[Path, ...]) -> bool:
if any(path.is_relative_to(root) for root in roots):
return True
# resolve() can retain a different spelling on case-insensitive filesystems.
for ancestor in (path, *path.parents):
for root in roots:
try:
if ancestor.samefile(root):
return True
except (FileNotFoundError, NotADirectoryError):
continue
return False


def _find_docker_executable(
name: str, *, project_root: Path | None = None
) -> str | None:
"""Search absolute PATH entries outside launch and runtime directories."""
roots = (Path.cwd().resolve(),)
if project_root is not None:
roots += (project_root.resolve(),)
names = [name]
if sys.platform == "win32":
# Python 3.11's which() does not expand PATHEXT for qualified names.
pathext = os.environ.get("PATHEXT") or ".COM;.EXE;.BAT;.CMD"
names = [
name + extension
for extension in pathext.split(";")
if extension
]
for entry in os.environ.get("PATH", os.defpath).split(os.pathsep):
directory = Path(entry)
if not directory.is_absolute():
continue
try:
directory = directory.resolve()
if _docker_path_is_excluded(directory, roots):
continue
except (OSError, RuntimeError) as exc:
warnings.warn(
f"Skipping Docker executable search candidate {directory}: {exc}",
stacklevel=2,
)
continue
# A qualified name bypasses Python 3.11's implicit Windows CWD search.
for filename in names:
candidate = directory / filename
try:
resolved = shutil.which(str(candidate))
if resolved is not None:
executable = Path(resolved)
if executable.is_absolute() and not _docker_path_is_excluded(
executable.resolve(), roots
):
return str(executable)
except (OSError, RuntimeError) as exc:
warnings.warn(
f"Skipping Docker executable search candidate {candidate}: {exc}",
stacklevel=2,
)
return None


def docker_agent_command(
executable: str | None = None, *, project_root: Path | None = None
) -> list[str] | None:
"""Return a runnable Docker Agent command, or ``None`` if unavailable.

Docker Agent is distributed either as the standalone ``docker-agent``
executable or as the ``docker agent`` Docker CLI plugin. The plugin form
is verified with a bounded, read-only version probe so a plain Docker CLI
is not mistaken for an installed Docker Agent.
"""
resolved_from_path = executable is None
if executable is None:
if shutil.which("docker-agent"):
return ["docker-agent", "run"]
executable = shutil.which("docker")
standalone = _find_docker_executable("docker-agent", project_root=project_root)
if standalone is not None:
return [standalone, "run"]
executable = _find_docker_executable("docker", project_root=project_root)
if executable is None:
return None

Expand All @@ -54,7 +121,7 @@ def docker_agent_command(executable: str | None = None) -> list[str] | None:
return None
if result.returncode != 0:
return None
return ["docker", "agent", "run"] if resolved_from_path else run_command
return run_command


def relative_extension_path_violation(value: Any) -> str | None:
Expand Down
21 changes: 12 additions & 9 deletions src/specify_cli/integrations/docker_agent/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,20 +50,20 @@ class DockerAgentIntegration(SkillsIntegration):
# ``agents.<name>.hooks``. Spec Kit does not edit that user-owned file, so
# hooks are intentionally not exposed through the integration event system.

def _agent_command(self) -> list[str]:
def _agent_command(self, project_root: Path | None = None) -> list[str] | None:
"""Return the available Docker Agent command form."""

# The shared executable override supports both a standalone
# ``docker-agent`` binary and the Docker CLI plugin form.
executable = self._resolve_executable()
command = docker_agent_command(
None if executable == self.key else executable
# An explicit override can have the same spelling as the default key.
has_override = bool(
os.environ.get("SPECKIT_INTEGRATION_DOCKER_AGENT_EXECUTABLE", "").strip()
)
return docker_agent_command(
None if executable == self.key and not has_override else executable,
project_root=project_root,
)
if command is None:
# Preserve the normal executable-shaped argv for dispatch callers;
# preflight and the subprocess runner report the unavailable CLI.
return [executable, "run"]
return command

@classmethod
def options(cls) -> list[IntegrationOption]:
Expand Down Expand Up @@ -128,7 +128,10 @@ def build_exec_args(
"reference, for example ./agent.yaml"
)

args = [*self._agent_command(), "--exec"]
command = self._agent_command(project_root)
if command is None:
return None
args = [*command, "--exec"]

args.extend(runtime_args or legacy_args)

Expand Down
Loading
Loading