hermes-agent/tools/project_tools.py

#!/usr/bin/env python3
"""Project tools — the agent's INTENTIONAL handle on first-class Projects (per-profile
``projects.db``, the desktop sidebar's named workspaces). Creating/switching is an explicit
tool call, never a side effect of ``cd``. GUI-only: the `project` toolset stays off
``_HERMES_CORE_TOOLS``; the desktop/TUI gateway folds it in and wires
``set_project_workspace_callback`` so the live session's cwd and sidebar follow. A live
session create/switch re-anchors only that session; it must not move the profile-global
Desktop selection shared by concurrent chats."""

import json
import os
from typing import Callable, Optional

from tools.registry import registry

# Set by the GUI gateway: ``(task_id, primary_path, project_name)`` re-anchors that session's
# workspace. ``None`` in CLI/messaging — the DB write still happens, nothing to move.
_workspace_callback: Optional[Callable[[str, str, str], None]] = None


def set_project_workspace_callback(fn: Optional[Callable[[str, str, str], None]]) -> None:
    global _workspace_callback
    _workspace_callback = fn


def _primary_path(proj) -> Optional[str]:
    if getattr(proj, "primary_path", None):
        return proj.primary_path
    for folder in proj.folders:
        if folder.is_primary:
            return folder.path
    return proj.folders[0].path if proj.folders else None


def _moves_session(task_id: Optional[str], path: Optional[str]) -> bool:
    """True when a live GUI session's workspace will follow the project — ``_apply_workspace``'s gate."""
    return bool(_workspace_callback and task_id and path)


def _apply_workspace(task_id: Optional[str], path: Optional[str], name: str) -> None:
    cb = _workspace_callback
    if cb and task_id and path:
        try:
            cb(task_id, path, name)
        except Exception:
            pass


def _resolve(conn, token: str):
    from hermes_cli import projects_db as pdb
    token = (token or "").strip()
    if not token:
        return None
    projects = pdb.list_projects(conn, include_archived=True)
    # Exact id / slug / name first, then case-insensitive slug / name.
    for proj in projects:
        if token in (proj.id, proj.slug) or proj.name == token:
            return proj
    low = token.lower()
    for proj in projects:
        if proj.slug.lower() == low or proj.name.lower() == low:
            return proj
    return None


def _activated(proj, task_id: Optional[str]) -> str:
    primary = _primary_path(proj)
    _apply_workspace(task_id, primary, proj.name)
    return json.dumps({
        "success": True, "id": proj.id, "slug": proj.slug, "name": proj.name,
        "primary_path": primary})


def _calling_session_project_id(conn, task_id: Optional[str]) -> tuple[bool, Optional[str]]:
    """``(scoped, project_id)`` for the calling session. The GUI gateway registers each session's
    workspace (``cwd_source``) in the terminal override table; a caller it never registered (CLI,
    scripts) is unscoped and falls back to the profile-global pointer."""
    from hermes_cli import projects_db as pdb
    from tools.terminal_tool import resolve_task_overrides
    overrides = resolve_task_overrides(task_id) if task_id else {}
    if "cwd_source" not in overrides:
        return False, None
    cwd = overrides.get("cwd") if overrides["cwd_source"] == "session" else None
    project = pdb.project_for_path(conn, cwd) if cwd else None
    return True, project.id if project else None


def project_list(task_id: Optional[str] = None) -> str:
    from hermes_cli import projects_db as pdb
    with pdb.connect_closing() as conn:
        # Another tab's switch moves the profile-global pointer; this chat's project is its own cwd.
        scoped, active = _calling_session_project_id(conn, task_id)
        if not scoped:
            active = pdb.get_active_id(conn)
        projects = pdb.list_projects(conn)
    return json.dumps({
        "active_id": active,
        "projects": [
            {
                "id": p.id, "slug": p.slug, "name": p.name,
                "primary_path": _primary_path(p), "active": p.id == active}
            for p in projects]})


def project_create(name: str, path: Optional[str] = None, task_id: Optional[str] = None) -> str:
    name = (name or "").strip()
    if not name:
        return json.dumps({"success": False, "error": "name is required"})
    from hermes_cli import projects_db as pdb
    folder = (path or "").strip()
    if folder:
        folder = os.path.abspath(os.path.expanduser(folder))
    try:
        with pdb.connect_closing() as conn:
            existing = pdb.find_by_primary_path(conn, folder) if folder else None
            if existing is not None:
                # Idempotent create: the folder already belongs to a project. Reusing it beats minting
                # a duplicate — duplicated projects render N identical sidebar subtrees (#75820).
                proj = existing
            else:
                pid = pdb.create_project(conn, name=name, folders=[folder] if folder else [], primary_path=folder or None)
                proj = pdb.get_project(conn, pid)
            # A live session that moves owns its workspace, so it leaves the profile-global pointer
            # alone (a background chat would otherwise redirect the sidebar). Anything that can't
            # move — CLI/messaging, a pathless project — switches through the pointer instead.
            if proj is not None and not _moves_session(task_id, _primary_path(proj)):
                pdb.set_active(conn, proj.id)
    except ValueError as exc:
        return json.dumps({"success": False, "error": str(exc)})
    if proj is None:
        return json.dumps({"success": False, "error": "project vanished after create"})
    return _activated(proj, task_id)


def project_switch(project: str, task_id: Optional[str] = None) -> str:
    from hermes_cli import projects_db as pdb
    with pdb.connect_closing() as conn:
        proj = _resolve(conn, project)
        if proj is None:
            return json.dumps({"success": False, "error": f"no project matching '{project}'"})
        # Same rule as create.
        if not _moves_session(task_id, _primary_path(proj)):
            pdb.set_active(conn, proj.id)
    return _activated(proj, task_id)


_ACTIONS = {
    "list": lambda args, tid: project_list(task_id=tid),
    "create": lambda args, tid: project_create(
        name=args.get("name", ""), path=args.get("path"), task_id=tid),
    "switch": lambda args, tid: project_switch(project=args.get("name", ""), task_id=tid)}


def _handle_project(args, **kw):
    action = _ACTIONS.get((args.get("action") or "").strip())
    if action is None:
        return json.dumps({"success": False, "error": "action must be one of: create, switch, list."})
    return action(args, kw.get("task_id"))


# One action enum instead of three tools: each re-taught "desktop Projects" (244 -> ~145 tok).
# Consolidated (#95681, maintainer-directed): project_list/create/switch each re-taught "desktop Projects
# (named workspaces)"; one action enum says it once (244 -> ~145 tok).
registry.register(
    name="desktop_project",
    toolset="project",
    schema={
        "name": "desktop_project",
        "description": (
            "Create or switch desktop Projects (named workspaces). create: one and switch "
            "this chat into it — pass path to anchor it to a repo/folder (the "
            "chat's workspace moves there, the sidebar follows). switch: move "
            "this chat into an existing project by name/slug/id — the "
            "intentional way to move the session, not `cd`. list: all projects + which one this chat is in."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "action": {"type": "string", "enum": ["create", "switch", "list"]},
                "name": {"type": "string", "description": "create: human name. switch: name, slug, or id."},
                "path": {"type": "string", "description": "create: repo/folder to anchor to."},
            },
            "required": ["action"],
        },
    },
    handler=_handle_project,
)