feat(input): support GitHub tree subdirectories

src/skillspector/input_handler.py

@@ -50,7 +50,7 @@
 from stat import S_IFMT, S_ISDIR, S_ISLNK, S_ISREG
 from time import monotonic
 from typing import BinaryIO, NoReturn, cast
-from urllib.parse import urljoin, urlparse
+from urllib.parse import unquote, urljoin, urlparse
 
 import httpx
 
@@ -739,6 +739,22 @@ def resolve(self, input_path: str) -> tuple[Path, str]:
         """
         input_path = input_path.strip()
 
+        git_target = self._github_tree_target(input_path)
+        if git_target is not None:
+            repository_url, branch, subdirectory = git_target
+            clone_dir = self._clone_git(repository_url, branch=branch)
+            try:
+                clone_root = clone_dir.resolve()
+                target = (clone_root / subdirectory).resolve()
+                target.relative_to(clone_root)
+                if not target.is_dir() or target.is_symlink():
+                    raise ValueError("Git URL subdirectory does not exist or is not a directory")
+                return target, "git"
+            except (OSError, ValueError):
+                # No caller receives the resolver after a failed selection, so it
+                # cannot clean an owned clone on our behalf.
+                self.cleanup()
+                raise
         if self._is_git_url(input_path):
             return self._clone_git(input_path), "git"
         if self._is_file_url(input_path):
@@ -1009,6 +1025,78 @@ def _is_git_url(self, path: str) -> bool:
             return True
         return False
 
+    def _github_tree_target(self, path: str) -> tuple[str, str, PurePosixPath] | None:
+        """Return a canonical clone target for a GitHub ``/tree/<ref>/<dir>`` URL.
+
+        The ref itself may contain ``/`` (for example ``feature/foo``), so the
+        split between ref and subdirectory is resolved against the remote's
+        advertised refs: the longest ``refs/heads/`` or ``refs/tags/`` name
+        that prefixes the ``/tree/`` segments wins.  Without this, a URL for
+        branch ``feature/foo`` would clone branch ``feature`` and treat
+        ``foo`` as part of the subdirectory.
+        """
+        parsed = urlparse(path)
+        if parsed.scheme != "https" or parsed.hostname != "github.com":
+            return None
+        parts = [unquote(part) for part in parsed.path.split("/") if part]
+        if len(parts) < 4 or parts[2] != "tree":
+            return None
+        owner, repository = parts[0], parts[1]
+        segments = parts[3:]
+        if any(part in {"", ".", ".."} or "/" in part or "\\" in part for part in segments):
+            raise ValueError("Git URL subdirectory must stay within the repository")
+        repository_url = f"https://github.com/{owner}/{repository}.git"
+        ref, subdirectory = self._resolve_tree_ref(repository_url, segments)
+        return (repository_url, ref, PurePosixPath(*subdirectory))
+
+    def _resolve_tree_ref(self, repository_url: str, segments: list[str]) -> tuple[str, list[str]]:
+        """Split ``/tree/`` *segments* into ``(ref, subdirectory)``.
+
+        Uses the longest remote branch/tag name that prefixes the segments, so
+        refs containing ``/`` resolve to the intended tree.  Raises ValueError
+        when no advertised ref matches the URL.
+        """
+        remote_refs = self._list_remote_refs(repository_url)
+        for end in range(len(segments), 0, -1):
+            candidate = "/".join(segments[:end])
+            if candidate in remote_refs:
+                return candidate, segments[end:]
+        raise ValueError(
+            "GitHub tree URL does not name a known branch or tag: "
+            f"{repository_url} ({'/'.join(segments)})"
+        )
+
+    def _list_remote_refs(self, repository_url: str) -> set[str]:
+        """Return the branch/tag names advertised by the remote repository.
+
+        Bounded by the ingest deadline; the host allowlist and private-IP
+        checks from URL validation apply.
+        """
+        self._validate_url_host(repository_url, ALLOWED_GIT_HOSTS)
+        deadline = self._deadline()
+        self._check_deadline(deadline, "git")
+        timeout = max(1.0, deadline - monotonic())
+        try:
+            process = subprocess.run(
+                ["git", "ls-remote", repository_url],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.DEVNULL,
+                timeout=timeout,
+                check=False,
+            )
+        except subprocess.TimeoutExpired as exc:
+            raise IngestLimitExceededError("Git ref listing exceeded its time limit") from exc
+        if process.returncode != 0:
+            raise ValueError(f"Could not list refs for GitHub tree URL: {repository_url}")
+        refs: set[str] = set()
+        for line in process.stdout.decode("utf-8", errors="replace").splitlines():
+            _, _, ref = line.partition("\t")
+            for prefix in ("refs/heads/", "refs/tags/"):
+                if ref.startswith(prefix):
+                    refs.add(ref[len(prefix) :])
+                    break
+        return refs
+
     def _is_file_url(self, path: str) -> bool:
         """Check if path is a direct file URL."""
         if not path.startswith("https://"):
@@ -1044,7 +1132,7 @@ def _validate_url_host(self, url: str, allowed_hosts: frozenset[str]) -> str:
             )
         return host
 
-    def _clone_git(self, url: str) -> Path:
+    def _clone_git(self, url: str, *, branch: str | None = None) -> Path:
         """Clone a Git repository to a temporary directory, bounded by ``INGEST_MAX_BYTES``."""
         remaining_seconds = self._remaining_seconds()
         remaining_bytes = self._remaining_bytes()
@@ -1070,6 +1158,8 @@ def _clone_git(self, url: str) -> Path:
             url,
             str(clone_dir),
         ]
+        if branch is not None:
+            clone_command[6:6] = ["--branch", branch]
         if remaining_bytes is not None:
             clone_command.insert(6, f"--filter=blob:limit={remaining_bytes}")
         process: subprocess.Popen[bytes] | None = None

tests/unit/test_input_handler.py

@@ -19,7 +19,7 @@
 import os
 import sys
 from errno import ENOENT
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
@@ -455,6 +455,100 @@ def test_scp_url_is_git_url() -> None:
     assert InputHandler()._is_git_url("git@github.com:org/repo.git") is True
 
 
+def test_github_tree_url_resolves_a_checked_out_subdirectory(tmp_path: Path) -> None:
+    handler = InputHandler()
+    clone = tmp_path / "repo"
+    (clone / "skills" / "biome-gritql").mkdir(parents=True)
+    with (
+        patch.object(handler, "_clone_git", return_value=clone) as clone_git,
+        patch.object(handler, "_list_remote_refs", return_value={"main"}),
+    ):
+        resolved, source_type = handler.resolve(
+            "https://github.com/somtougeh/somto-dev-toolkit/tree/main/skills/biome-gritql"
+        )
+    assert resolved == clone / "skills" / "biome-gritql"
+    assert source_type == "git"
+    clone_git.assert_called_once_with(
+        "https://github.com/somtougeh/somto-dev-toolkit.git", branch="main"
+    )
+
+
+def test_github_tree_url_resolves_slash_containing_ref(tmp_path: Path) -> None:
+    """A branch name containing / must not be split into ref + subdirectory."""
+    handler = InputHandler()
+    clone = tmp_path / "repo"
+    (clone / "skills" / "demo").mkdir(parents=True)
+    with (
+        patch.object(handler, "_clone_git", return_value=clone) as clone_git,
+        patch.object(handler, "_list_remote_refs", return_value={"main", "feature", "feature/foo"}),
+    ):
+        resolved, source_type = handler.resolve(
+            "https://github.com/example/repo/tree/feature/foo/skills/demo"
+        )
+    assert resolved == clone / "skills" / "demo"
+    assert source_type == "git"
+    clone_git.assert_called_once_with("https://github.com/example/repo.git", branch="feature/foo")
+
+
+def test_github_tree_url_prefers_shorter_ref_when_longest_absent() -> None:
+    """The longest *advertised* ref wins, not the longest URL prefix."""
+    handler = InputHandler()
+    with patch.object(handler, "_list_remote_refs", return_value={"feature"}):
+        repository_url, ref, subdirectory = handler._github_tree_target(
+            "https://github.com/example/repo/tree/feature/sub"
+        )
+    assert repository_url == "https://github.com/example/repo.git"
+    assert ref == "feature"
+    assert subdirectory == PurePosixPath("sub")
+
+
+def test_github_tree_url_rejects_unknown_ref() -> None:
+    handler = InputHandler()
+    with (
+        patch.object(handler, "_list_remote_refs", return_value={"main"}),
+        pytest.raises(ValueError, match="does not name a known branch or tag"),
+    ):
+        handler._github_tree_target("https://github.com/example/repo/tree/nope/sub")
+
+
+def test_github_tree_url_supports_ref_without_subdirectory() -> None:
+    handler = InputHandler()
+    with patch.object(handler, "_list_remote_refs", return_value={"main"}):
+        repository_url, ref, subdirectory = handler._github_tree_target(
+            "https://github.com/example/repo/tree/main"
+        )
+    assert repository_url == "https://github.com/example/repo.git"
+    assert ref == "main"
+    assert subdirectory == PurePosixPath(".")
+
+
+@pytest.mark.parametrize("segment", ["%2Fetc", "%2E%2E%2Frepo", "%5Coutside"])
+def test_github_tree_url_rejects_encoded_path_escapes(segment: str) -> None:
+    with pytest.raises(ValueError, match="stay within the repository"):
+        InputHandler()._github_tree_target(
+            f"https://github.com/example/repo/tree/main/skills/{segment}"
+        )
+
+
+@pytest.mark.parametrize("target", ["missing", "SKILL.md"])
+def test_github_tree_url_selection_failure_cleans_owned_clone(tmp_path: Path, target: str) -> None:
+    """A post-clone tree selection error must not strand the owned checkout."""
+    handler = InputHandler()
+    clone = tmp_path / "repo"
+    clone.mkdir()
+    if target == "SKILL.md":
+        (clone / target).write_text("# skill\n")
+    handler._temp_dir = tmp_path
+    with (
+        patch.object(handler, "_clone_git", return_value=clone),
+        patch.object(handler, "_list_remote_refs", return_value={"main"}),
+    ):
+        with pytest.raises(ValueError):
+            handler.resolve(f"https://github.com/example/repo/tree/main/{target}")
+    assert not tmp_path.exists()
+    assert handler.temp_dir_for_cleanup() is None
+
+
 def test_http_urls_are_not_accepted_as_remote_inputs() -> None:
     """Network inputs require HTTPS unless they use SSH's scp-style syntax."""
     handler = InputHandler()