Agentic-OS/backend/app/services/mcp_workspace.py

"""Read/write MCP server source trees and push changes to Gitea."""
from __future__ import annotations

import logging
import os
from pathlib import Path

from app.config import settings
from app.mcp_manager.manager import get_manager

logger = logging.getLogger(__name__)

GIT_USER_NAME = "Agentic OS"
GIT_USER_EMAIL = "agentic-os@paraskeva.net"


def _local_src_dir(server: str) -> str:
    return os.path.join(os.environ.get("MCP_LOCAL_DIR", "/app/mcp-servers"), server)


def is_local_shipped(server: str) -> bool:
    return os.path.isdir(_local_src_dir(server))


def workspace_path(server: str) -> str:
    """Writable directory where the running MCP code lives."""
    manager = get_manager()
    return manager._repo_dir(server)


def _is_git_repo(path: str) -> bool:
    return os.path.isdir(os.path.join(path, ".git"))


async def _run(cmd: list[str], cwd: str) -> tuple[int, str]:
    import asyncio

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        cwd=cwd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
    )
    out, _ = await proc.communicate()
    return proc.returncode or 0, out.decode(errors="replace")


async def ensure_git_ready(server: str) -> tuple[str, bool]:
    """Return (repo_path, can_push). Clone from Gitea if local-only without git."""
    path = workspace_path(server)
    if _is_git_repo(path):
        await _ensure_remote(server, path)
        return path, True

    if not settings.gitea_token:
        return path, False

    clone_url = settings.gitea_clone_url(server)
    git_dir = os.path.join(settings.runtime_dir, "mcp-repos", server)
    if _is_git_repo(git_dir):
        await _ensure_remote(server, git_dir)
        return git_dir, True

    os.makedirs(os.path.dirname(git_dir), exist_ok=True)
    code, out = await _run(["git", "clone", "--depth", "1", clone_url, git_dir])
    if code != 0:
        logger.warning("MCP dev clone %s failed: %s", server, out[:300])
        return path, False
    return git_dir, True


async def _ensure_remote(server: str, path: str) -> None:
    from app.mcp_manager.manager import ensure_git_remote

    await ensure_git_remote(server, path)


async def list_source_files(server: str, *, max_files: int = 40) -> list[str]:
    root = Path(workspace_path(server))
    if not root.is_dir():
        return []
    out: list[str] = []
    for p in sorted(root.rglob("*")):
        if not p.is_file():
            continue
        rel = p.relative_to(root).as_posix()
        if any(part in rel for part in (".venv", "__pycache__", ".git", "node_modules")):
            continue
        if p.suffix not in {".py", ".toml", ".md", ".json", ".yaml", ".yml"}:
            continue
        out.append(rel)
        if len(out) >= max_files:
            break
    return out


async def read_source_file(server: str, rel_path: str, *, max_chars: int = 80_000) -> str:
    root = Path(workspace_path(server)).resolve()
    target = (root / rel_path).resolve()
    if not str(target).startswith(str(root)):
        raise ValueError("Path escapes MCP workspace")
    if not target.is_file():
        raise FileNotFoundError(rel_path)
    text = target.read_text(encoding="utf-8")
    return text[:max_chars]


async def write_source_file(server: str, rel_path: str, content: str) -> None:
    root = Path(workspace_path(server)).resolve()
    target = (root / rel_path).resolve()
    if not str(target).startswith(str(root)):
        raise ValueError("Path escapes MCP workspace")
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(content, encoding="utf-8")


async def apply_file_edits(server: str, edits: list[dict]) -> list[str]:
    written: list[str] = []
    for edit in edits:
        path = edit.get("path") or edit.get("file")
        content = edit.get("content")
        if not path or content is None:
            continue
        await write_source_file(server, path, content)
        written.append(path)
    return written


async def commit_and_push(
    server: str,
    message: str,
    files: list[str],
    *,
    use_git_clone: bool = True,
) -> dict:
    """Commit listed paths and push to origin. Returns {ok, commit, error}."""
    if not files:
        return {"ok": False, "error": "No files listed for git push"}

    path, can_push = await ensure_git_ready(server)
    if not can_push:
        return {
            "ok": False,
            "error": "No Gitea git repo for this MCP — changes applied at runtime only",
            "runtime_only": True,
        }

    # Sync runtime workspace files into git checkout when they differ
    runtime = workspace_path(server)
    if use_git_clone and os.path.abspath(runtime) != os.path.abspath(path):
        for rel in files:
            src = os.path.join(runtime, rel)
            dst = os.path.join(path, rel)
            if os.path.isfile(src):
                os.makedirs(os.path.dirname(dst), exist_ok=True)
                with open(src, encoding="utf-8") as fh:
                    data = fh.read()
                with open(dst, "w", encoding="utf-8") as fh:
                    fh.write(data)

    missing = [rel for rel in files if not os.path.isfile(os.path.join(path, rel))]
    if missing:
        return {
            "ok": False,
            "error": f"File(s) not found in MCP workspace: {', '.join(missing)}",
        }

    _, diff_out = await _run(["git", "diff", "HEAD", "--"] + files, cwd=path)
    _, cached_out = await _run(["git", "diff", "--cached", "HEAD", "--"] + files, cwd=path)
    if not diff_out.strip() and not cached_out.strip():
        _, status_sb = await _run(["git", "status", "-sb"], cwd=path)
        if "ahead" in status_sb:
            code, out = await _run(["git", "push"], cwd=path)
            if code != 0:
                return {"ok": False, "error": f"push failed: {out[:500]}"}
            commit_hash = (await _run(["git", "rev-parse", "--short", "HEAD"], cwd=path))[1].strip()
            return {"ok": True, "commit": commit_hash, "repo": server, "note": "pushed existing commits"}
        return {
            "ok": True,
            "commit": None,
            "note": "no changes to push — files already match the last commit",
            "already_up_to_date": True,
        }

    for rel in files:
        await _run(["git", "add", rel], cwd=path)

    code, out = await _run(
        [
            "git",
            "-c",
            f"user.email={GIT_USER_EMAIL}",
            "-c",
            f"user.name={GIT_USER_NAME}",
            "commit",
            "-m",
            message,
        ],
        cwd=path,
    )
    if code != 0:
        lowered = out.lower()
        if "nothing to commit" in lowered or "nothing added to commit" in lowered:
            return {
                "ok": True,
                "commit": None,
                "note": "no changes to commit — already up to date",
                "already_up_to_date": True,
            }
        return {"ok": False, "error": out[:500]}

    code, out = await _run(["git", "push"], cwd=path)
    if code != 0:
        return {"ok": False, "error": f"push failed: {out[:500]}"}

    commit_hash = (await _run(["git", "rev-parse", "--short", "HEAD"], cwd=path))[1].strip()
    return {"ok": True, "commit": commit_hash, "repo": server}


def list_registered_tools(server: str) -> list[str]:
    manager = get_manager()
    state = manager.servers.get(server)
    if not state:
        return []
    return [t["name"] for t in state.tools]