From 5b4d3985b77d6ea20ee6855b69eebbf046cd9d5b Mon Sep 17 00:00:00 2001 From: worldmozara Date: Wed, 15 Jul 2026 10:22:00 +0800 Subject: [PATCH] core/tools/vcs: add read-only vcs_* agent tools Expose vcs_kind/status/diff/log/branch over detect_vcs; cover git and custom detector registration. --- src/plyngent/tools/vcs/__init__.py | 11 +++ src/plyngent/tools/vcs/tools.py | 86 ++++++++++++++++++++++ tests/test_tools/test_vcs.py | 110 +++++++++++++++++++++++++++++ 3 files changed, 207 insertions(+) create mode 100644 src/plyngent/tools/vcs/__init__.py create mode 100644 src/plyngent/tools/vcs/tools.py create mode 100644 tests/test_tools/test_vcs.py diff --git a/src/plyngent/tools/vcs/__init__.py b/src/plyngent/tools/vcs/__init__.py new file mode 100644 index 0000000..c788c1b --- /dev/null +++ b/src/plyngent/tools/vcs/__init__.py @@ -0,0 +1,11 @@ +from .backend import VcsBackend as VcsBackend +from .detect import clear_extra_detectors as clear_extra_detectors +from .detect import detect_vcs as detect_vcs +from .detect import register_detector as register_detector +from .git_backend import GitBackend as GitBackend +from .tools import VCS_TOOLS as VCS_TOOLS +from .tools import vcs_branch as vcs_branch +from .tools import vcs_diff as vcs_diff +from .tools import vcs_kind as vcs_kind +from .tools import vcs_log as vcs_log +from .tools import vcs_status as vcs_status diff --git a/src/plyngent/tools/vcs/tools.py b/src/plyngent/tools/vcs/tools.py new file mode 100644 index 0000000..5f48619 --- /dev/null +++ b/src/plyngent/tools/vcs/tools.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from plyngent.agent import tool +from plyngent.tools.workspace import WorkspaceError, get_workspace_root + +from .detect import detect_vcs + +if TYPE_CHECKING: + from .backend import VcsBackend + + +def _backend_or_error() -> VcsBackend | str: + try: + root = get_workspace_root() + except WorkspaceError as exc: + return f"error: {exc}" + backend = detect_vcs(root) + if backend is None: + return ( + "error: no supported VCS detected under workspace " + "(currently: git; other systems can register detectors)" + ) + return backend + + +@tool +def vcs_kind() -> str: + """Return the detected VCS kind under the workspace (e.g. ``git``), or an error.""" + backend = _backend_or_error() + if isinstance(backend, str): + return backend + return backend.kind + + +@tool +def vcs_status() -> str: + """Show working-tree status for the detected VCS (read-only).""" + backend = _backend_or_error() + if isinstance(backend, str): + return backend + return backend.status() + + +@tool +def vcs_diff(path: str = "", *, staged: bool = False) -> str: + """Show a unified diff for the detected VCS (read-only). + + ``path`` is optional and relative to the workspace. ``staged=true`` is + honored by git (index vs HEAD); other backends may ignore it. + """ + backend = _backend_or_error() + if isinstance(backend, str): + return backend + rel = path.strip() or None + return backend.diff(staged=staged, path=rel) + + +@tool +def vcs_log(limit: int = 10) -> str: + """Show recent commits for the detected VCS (read-only).""" + if limit < 1: + return "error: limit must be >= 1" + backend = _backend_or_error() + if isinstance(backend, str): + return backend + return backend.log(limit=limit) + + +@tool +def vcs_branch() -> str: + """Show the current branch / named head for the detected VCS (read-only).""" + backend = _backend_or_error() + if isinstance(backend, str): + return backend + return backend.branch() + + +VCS_TOOLS = [ + vcs_kind, + vcs_status, + vcs_diff, + vcs_log, + vcs_branch, +] diff --git a/tests/test_tools/test_vcs.py b/tests/test_tools/test_vcs.py new file mode 100644 index 0000000..ae0bef1 --- /dev/null +++ b/tests/test_tools/test_vcs.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import shutil +import subprocess +from pathlib import Path +from typing import TYPE_CHECKING + +import pytest + +from plyngent.tools.vcs import ( + detect_vcs, + vcs_branch, + vcs_diff, + vcs_kind, + vcs_log, + vcs_status, +) +from plyngent.tools.vcs.detect import clear_extra_detectors, register_detector +from tests.test_tools.helpers import call_sync + +if TYPE_CHECKING: + from plyngent.tools.vcs.backend import VcsBackend + +pytestmark = pytest.mark.skipif(shutil.which("git") is None, reason="git not installed") + + +def _git(root: Path, *args: str) -> None: + completed = subprocess.run( + ["git", *args], + cwd=root, + check=True, + capture_output=True, + text=True, + ) + del completed + + +def _init_repo(root: Path) -> None: + _git(root, "init") + _git(root, "config", "user.email", "test@example.com") + _git(root, "config", "user.name", "Test") + _ = (root / "readme.txt").write_text("hello\n", encoding="utf-8") + _git(root, "add", "readme.txt") + _git(root, "commit", "-m", "initial") + + +def test_detect_none(workspace: object) -> None: + assert isinstance(workspace, Path) + assert detect_vcs(workspace) is None + assert "no supported VCS" in call_sync(vcs_status) + + +def test_git_status_and_kind(workspace: object) -> None: + assert isinstance(workspace, Path) + _init_repo(workspace) + assert call_sync(vcs_kind) == "git" + status = call_sync(vcs_status) + assert "readme" in status or "main" in status or "master" in status or status == "(clean)" + + +def test_git_log_and_branch(workspace: object) -> None: + assert isinstance(workspace, Path) + _init_repo(workspace) + log = call_sync(vcs_log, limit=5) + assert "initial" in log + branch = call_sync(vcs_branch) + assert branch # main/master/detached — non-empty + + +def test_git_diff(workspace: object) -> None: + assert isinstance(workspace, Path) + _init_repo(workspace) + _ = (workspace / "readme.txt").write_text("hello\nworld\n", encoding="utf-8") + diff = call_sync(vcs_diff) + assert "world" in diff or "readme" in diff + + +def test_register_custom_detector(workspace: object) -> None: + assert isinstance(workspace, Path) + + class FakeBackend: + @property + def kind(self) -> str: + return "fake" + + def status(self) -> str: + return "fake-status" + + def diff(self, *, staged: bool = False, path: str | None = None) -> str: + del staged, path + return "fake-diff" + + def log(self, *, limit: int = 10) -> str: + del limit + return "fake-log" + + def branch(self) -> str: + return "fake-branch" + + def detect_fake(root: Path) -> VcsBackend | None: + del root + return FakeBackend() + + register_detector(detect_fake, prepend=True) + try: + assert call_sync(vcs_kind) == "fake" + assert call_sync(vcs_status) == "fake-status" + assert call_sync(vcs_branch) == "fake-branch" + finally: + clear_extra_detectors()