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.
This commit is contained in:
2026-07-15 10:22:00 +08:00
parent c402555778
commit 5b4d3985b7
3 changed files with 207 additions and 0 deletions
+11
View File
@@ -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
+86
View File
@@ -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,
]
+110
View File
@@ -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()