hermes-agent/pm/workspace.py

"""Resolve core plus plugin requirements in a writable build snapshot.

Shipped source and locks are inputs, never mutation targets. Candidate
failure propagates without changing plugin configuration or the live venv.
"""

from __future__ import annotations

import hashlib
import os
import shutil
from collections.abc import Mapping
from pathlib import Path
from typing import TYPE_CHECKING, Optional

if TYPE_CHECKING:
    from pm.environment import PythonEnvironment

from pm import paths
from pm.package import InstallError
from pm.plugin_declarations import read_python_declaration, manifest_version_error

_MEMBER_EXCLUDE = frozenset({".git", ".venv", "venv", "node_modules", "__pycache__"})


def _member_ignored(directory, names):
    return [name for name in names if name in _MEMBER_EXCLUDE or name.endswith(".egg-info")]


# The uv failure classifier lives beside the uv runner (stdlib-only imports): the bootstrap
# runner streams uv output from a pre-3.11 system python where this module's tomllib import
# cannot load. Workspace callers keep reaching it from here.
from pm.environment import ResolutionConflict, classify_uv_failure  # noqa: E402,F401


def member_sources(plugin_dirs) -> dict[Path, Path]:
    """Map installed identities to build inputs, including staged plugin updates."""
    rows = plugin_dirs.items() if isinstance(plugin_dirs, Mapping) else ((path, path) for path in plugin_dirs)
    return {Path(identity).resolve(): Path(source).resolve() for identity, source in rows}


def members_stamp(plugin_dirs) -> str:
    """Hash the member inputs copied into a generation, independent of staging paths."""
    h = hashlib.sha256()
    for identity, entry in sorted(member_sources(plugin_dirs).items()):
        h.update(str(identity).encode("utf-8"))
        h.update(b"\0")
        declaration = read_python_declaration(entry)
        for source in declaration.files:
            h.update(source.name.encode("utf-8"))
            h.update(source.read_bytes())
            h.update(b"\0")
        if (entry / "pyproject.toml").is_file():
            for directory, dirs, files in os.walk(entry):
                dirs[:] = sorted(set(dirs) - set(_member_ignored(directory, dirs)))
                for name in sorted(set(files) - set(_member_ignored(directory, files))):
                    path = Path(directory) / name
                    h.update(path.relative_to(entry).as_posix().encode())
                    h.update(b"\0")
                    h.update(os.readlink(path).encode() if path.is_symlink() else path.read_bytes())
                    h.update(b"\0")
    return h.hexdigest()


def _copy_core_inputs(source: Path, destination: Path) -> None:
    """Build from a writable snapshot, never from signed/read-only source."""
    import fnmatch
    import tomllib

    metadata = tomllib.loads((source / "pyproject.toml").read_text(encoding="utf-8-sig"))
    project = metadata.get("project", {})
    setuptools = metadata.get("tool", {}).get("setuptools", {})
    patterns = setuptools.get("packages", {}).get("find", {}).get("include", ["*"])
    package_roots = {pattern.split(".", 1)[0] for pattern in patterns}
    files = {"pyproject.toml", "setup.py", "setup.cfg"}
    readme = project.get("readme")
    if isinstance(readme, str):
        files.add(readme)
    elif isinstance(readme, dict) and "file" in readme:
        files.add(readme["file"])
    for pattern in project.get("license-files", []):
        files.update(str(p.relative_to(source)) for p in source.glob(pattern))
    files.update(p.name for p in source.glob("*.py"))

    excluded = {".git", ".venv", "venv", "node_modules", "__pycache__", "build", "dist", "release", "uv.lock"}
    def ignore(directory, names):
        return [name for name in names if name in excluded or name.startswith(".")
                or name.endswith(".egg-info") or (Path(directory) / name).is_symlink()]

    for entry in source.iterdir():
        if (entry.is_dir() and not entry.is_symlink() and entry.name not in excluded
                and not entry.name.startswith(".") and entry.resolve() != destination.resolve()
                and any(fnmatch.fnmatchcase(entry.name, pattern) for pattern in package_roots)):
            target = destination / entry.name
            shutil.copytree(entry, target, ignore=ignore)
    for name in files:
        entry = source / name
        if not entry.is_file() or entry.is_symlink():
            continue
        if not entry.resolve().is_relative_to(source.resolve()):
            raise InstallError("venv", f"build input escapes the core project: {name}")
        target = destination / name
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(entry, target)


def _generate_pyproject(plugin_dirs: list[Path] | Mapping[Path, Path], root: Path, *, source: Path) -> None:
    """Snapshot core and plugin build inputs into a fresh generation."""
    source = source.resolve()
    if root.resolve() == source or source.is_relative_to(root.resolve()):
        raise InstallError("venv", "workspace must not replace the core source")
    root.mkdir(parents=True)

    core_pyproject = source / "pyproject.toml"
    core_text = core_pyproject.read_text(encoding="utf-8-sig")

    members = [_workspace_member(source, root, identity=identity).relative_to(root).as_posix()
               for identity, source in member_sources(plugin_dirs).items()
               if _is_member_candidate(source)]

    lines = [core_text.rstrip("\n")]
    if members:
        lines.append("")
        lines.append("[tool.uv.workspace]")
        lines.append("members = [" + ", ".join(f'"{m}"' for m in sorted(members)) + "]")

    text = "\n".join(lines) + "\n"
    target = root / "pyproject.toml"
    _copy_core_inputs(source, root)
    target.write_text(text, encoding="utf-8")


def _is_member_candidate(plugin_dir: Path) -> bool:
    return read_python_declaration(plugin_dir).is_member


def enabled_plugin_dirs(*, proposed_home=None, enabled=None, disabled=None, installing: Path | None = None) -> list[Path]:
    """Resolve the effective plugin selection without filtering dependency declarations."""
    from pm.plugins_state import _is_directory, enabled_plugins_ordered

    selection = enabled_plugins_ordered(
        proposed_home=proposed_home, enabled=enabled, disabled=disabled, installing=installing,
    )
    members = []
    for plugins_dir, names in selection.items():
        for name in names:
            relative = Path(name)
            if relative.is_absolute() or ".." in relative.parts:
                raise InstallError("venv", f"invalid plugin key: {name}")
            plugin_dir = plugins_dir / relative
            proposed = installing is not None and plugin_dir.resolve() == installing.resolve()
            if not proposed and not _is_directory(plugin_dir):
                plugin_dir = paths.repo_root() / "plugins" / relative
            if proposed or _is_directory(plugin_dir):
                members.append(plugin_dir)
    return list(dict.fromkeys(members))


def enabled_member_dirs(*, proposed_home=None, enabled=None, disabled=None) -> list[Path]:
    """Keep every selected member or refuse an incompatible selection."""
    selected = enabled_plugin_dirs(proposed_home=proposed_home, enabled=enabled, disabled=disabled)
    members = []
    for path in selected:
        declaration = read_python_declaration(path)
        reason = manifest_version_error(declaration.manifest, path.name)
        if reason:
            raise InstallError("venv", reason)
        if declaration.is_member:
            members.append(path)
    return members


def _workspace_member(plugin_dir: Path, root: Path, *, identity: Path) -> Path:
    """Keep workspace members with their generation, not a temporary install clone."""
    import json
    import tomllib

    key = hashlib.sha256(str(identity.resolve()).encode()).hexdigest()[:16]
    declaration = read_python_declaration(plugin_dir)
    pyproject = declaration.pyproject
    if pyproject is not None:
        member = root / "plugin-sources" / key
        shutil.copytree(plugin_dir, member, symlinks=True,
                        ignore=_member_ignored)
        document = tomllib.loads(pyproject.read_text(encoding="utf-8-sig"))
        changed = declaration.install_requirements != declaration.requirements
        if changed:
            document["project"]["dependencies"] = list(declaration.install_requirements)
        for sources in document.get("tool", {}).get("uv", {}).get("sources", {}).values():
            for spec in sources if isinstance(sources, list) else [sources]:
                if not isinstance(spec, dict) or "path" not in spec:
                    continue
                relative = Path(spec["path"])
                if relative.is_absolute():
                    continue
                resolved = (plugin_dir / relative).resolve()
                if resolved.is_relative_to(plugin_dir.resolve()):
                    continue  # The referenced tree was copied with this member.
                spec["path"] = (identity / relative).resolve().as_posix()
                changed = True
        if changed:
            import tomli_w

            (member / "pyproject.toml").write_text(tomli_w.dumps(document), encoding="utf-8")
        return member
    specs = declaration.install_requirements
    member = root / "plugin-deps" / key
    member.mkdir(parents=True)
    (member / "pyproject.toml").write_text(
        f'[project]\nname = "hermes-plugin-{key}"\nversion = "0.0.0"\n'
        'requires-python = ">=3.11"\n'
        f'dependencies = {json.dumps(specs)}\n[tool.uv]\npackage = false\n',
        encoding="utf-8",
    )
    return member


def install_node_sidecar(
    plugin_dir: Path,
    *,
    explicit: bool = False,
) -> Optional[str]:
    """Install plugin-local dependencies using PM's paired npm/Node context.

    Explicit user consent permits acquisition even when on-demand installs
    are disabled. Returns None on success, otherwise a diagnostic.
    """
    package_json = plugin_dir / "package.json"
    if not package_json.is_file():
        return None  # nothing to install

    import pm
    from pm.install import lazy_installs_allowed

    # Tool availability does not authorize mutation of the sidecar itself.
    if not explicit and not lazy_installs_allowed():
        return "lazy installs are disabled — run `hermes plugins install` and approve Node dependencies"

    # a lockfile means reproducible `npm ci`; plain `npm install` otherwise
    install_cmd = ["ci"] if (plugin_dir / "package-lock.json").is_file() else ["install"]
    try:
        runner = pm.ensure("npm", explicit=explicit)
        # Resolve inside the composed context, including npm.cmd on Windows;
        # CreateProcess does not search a child's replacement PATH itself.
        npm = shutil.which("npm", path=runner.env.get("PATH", ""))
        if npm is None:
            return "npm is missing from the prepared PM environment"
        proc = runner.run(
            [npm, *install_cmd, "--no-audit", "--no-fund"],
            cwd=str(plugin_dir),
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=900,
        )
    except Exception as exc:
        return f"npm {install_cmd[0]} failed to run: {exc}"
    if proc.returncode != 0:
        tail = (proc.stderr or proc.stdout or "").strip()[-300:]
        return f"npm {install_cmd[0]} exited {proc.returncode}: {tail}"
    return None


def lock_and_sync(
    plugin_dirs: list[Path] | Mapping[Path, Path],
    extras: list[str],
    *,
    root: Path,
    source: Path,
    seed_lock: Path | None,
    environment: PythonEnvironment,
    frozen: bool = False,
    replay: Path | None = None,
) -> None:
    """Prepare a fresh generation using explicit inputs and a prepared engine.

    The caller selects the seed; uv retains its compatible versions. Repair
    copies the recorded build inputs and never reads current manifests.
    Resolver conflicts remain distinct from download/build failures.
    """
    if root.exists() or root.is_symlink():
        raise InstallError("venv", f"workspace must be fresh: {root}")
    if replay is None:
        _generate_pyproject(plugin_dirs, root, source=source)
        if seed_lock is not None:
            (root / "uv.lock").write_bytes(seed_lock.read_bytes())
    else:
        if not (replay / "pyproject.toml").is_file() or not (replay / "uv.lock").is_file():
            raise InstallError("venv", f"recorded workspace is missing: {replay}")
        # Sibling generations keep external relative paths at the same depth;
        # snapshotted members and their exact lock travel with the workspace.
        # Use the snapshot's exclusions: build/ may hold an in-tree backend.
        shutil.copytree(replay, root, symlinks=True, ignore=_member_ignored)
        frozen = True

    environment.sync(root, extras=extras, frozen=frozen)