Add worktree-backed session creation
This commit is contained in:
73
api/worktrees.py
Normal file
73
api/worktrees.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""Helpers for WebUI-managed Hermes Agent git worktrees."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import time
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def find_git_repo_root(workspace: str | Path) -> Path:
|
||||
"""Return the enclosing git repo root for *workspace*.
|
||||
|
||||
Use git itself instead of checking ``workspace/.git`` so nested workspaces
|
||||
and linked git worktrees are both handled correctly.
|
||||
"""
|
||||
ws = Path(workspace).expanduser().resolve()
|
||||
if not ws.is_dir():
|
||||
raise ValueError("Workspace path does not exist or is not a directory")
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "rev-parse", "--show-toplevel"],
|
||||
cwd=ws,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
timeout=5,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||||
raise ValueError("Workspace is not inside a git repository") from exc
|
||||
if result.returncode != 0:
|
||||
raise ValueError("Workspace is not inside a git repository")
|
||||
root = result.stdout.strip()
|
||||
if not root:
|
||||
raise ValueError("Workspace is not inside a git repository")
|
||||
return Path(root).expanduser().resolve()
|
||||
|
||||
|
||||
def _setup_agent_worktree(repo_root: str) -> dict:
|
||||
try:
|
||||
import api.config # noqa: F401 # ensure Hermes Agent dir is on sys.path
|
||||
from cli import _setup_worktree
|
||||
except Exception as exc:
|
||||
raise RuntimeError("Hermes Agent worktree helper is unavailable") from exc
|
||||
output = StringIO()
|
||||
with redirect_stdout(output), redirect_stderr(output):
|
||||
info = _setup_worktree(repo_root)
|
||||
emitted = output.getvalue().strip()
|
||||
if emitted:
|
||||
logger.debug("Hermes Agent worktree helper output: %s", emitted)
|
||||
if not info:
|
||||
raise RuntimeError("Hermes Agent failed to create a git worktree")
|
||||
return info
|
||||
|
||||
|
||||
def create_worktree_for_workspace(workspace: str | Path) -> dict:
|
||||
repo_root = find_git_repo_root(workspace)
|
||||
info = _setup_agent_worktree(str(repo_root))
|
||||
path = info.get("path")
|
||||
branch = info.get("branch")
|
||||
if not path or not branch:
|
||||
raise RuntimeError("Hermes Agent returned incomplete worktree metadata")
|
||||
return {
|
||||
"path": str(Path(path).expanduser().resolve()),
|
||||
"branch": str(branch),
|
||||
"repo_root": str(Path(info.get("repo_root") or repo_root).expanduser().resolve()),
|
||||
"created_at": time.time(),
|
||||
}
|
||||
Reference in New Issue
Block a user