ObsiGate/backend/tools/service.py

"""Built-in tool services (Phase 0 — read/search).

These functions are the single source of truth consumed by both the in-app
assistant and the MCP server. They delegate to the existing core modules
(``backend.indexer``, ``backend.search``) rather than duplicating route logic.
Mutating tools (create/edit/rename/move/delete) are added in later phases.
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any

from backend.tools.context import (
    ToolContext,
    ToolError,
    ToolNotFoundError,
    ToolRisk,
    resolve_safe_path,
)
from backend.tools.registry import tool
from backend.tools.schemas import (
    ListDirectoryInput,
    ListTagsInput,
    ListVaultsInput,
    ReadFileInput,
    SearchFulltextInput,
)

logger = logging.getLogger("obsigate.tools.service")

# Maximum file size returned by ``read_file`` (bytes).
TOOL_MAX_READ_BYTES = 200_000


def _vault_data_or_raise(vault: str) -> dict[str, Any]:
    from backend.indexer import get_vault_data

    data = get_vault_data(vault)
    if not data:
        raise ToolNotFoundError(f"Vault '{vault}' not found", details={"vault": vault})
    return data


@tool(
    name="list_vaults",
    description="List the vaults the current user is allowed to access.",
    input_model=ListVaultsInput,
    risk=ToolRisk.READ,
)
def list_vaults(ctx: ToolContext, _params: ListVaultsInput) -> list[dict[str, Any]]:
    """Return accessible vaults with a file count."""
    from backend.indexer import get_vault_data, get_vault_names

    vaults: list[dict[str, Any]] = []
    for name in get_vault_names():
        if not ctx.has_vault_access(name):
            continue
        data = get_vault_data(name) or {}
        vaults.append({"name": name, "file_count": len(data.get("files", []))})
    return vaults


@tool(
    name="list_directory",
    description="List files and subdirectories of a directory inside a vault.",
    input_model=ListDirectoryInput,
    risk=ToolRisk.READ,
    requires_vault=True,
)
def list_directory(ctx: ToolContext, params: ListDirectoryInput) -> list[dict[str, Any]]:
    """Return the entries of a vault directory (direct children only)."""
    data = _vault_data_or_raise(params.vault)
    root = Path(data["path"])
    target = resolve_safe_path(root, params.path) if params.path else root.resolve()

    if not target.exists() or not target.is_dir():
        raise ToolNotFoundError(f"Directory not found: {params.path}", details={"vault": params.vault, "path": params.path})

    from backend.vault_settings import get_vault_setting

    hide_hidden = (get_vault_setting(params.vault) or {}).get("hideHiddenFiles", False)

    items: list[dict[str, Any]] = []
    for entry in sorted(target.iterdir(), key=lambda e: (not e.is_dir(), e.name.lower())):
        if hide_hidden and entry.name.startswith("."):
            continue
        rel = str(entry.relative_to(root)).replace("\\", "/")
        items.append({
            "name": entry.name,
            "path": rel,
            "type": "directory" if entry.is_dir() else "file",
        })
    return items


@tool(
    name="read_file",
    description="Read the text content of a file inside a vault (secrets redacted).",
    input_model=ReadFileInput,
    risk=ToolRisk.READ,
    requires_vault=True,
)
def read_file(ctx: ToolContext, params: ReadFileInput) -> dict[str, Any]:
    """Return the (redacted) content of a vault file."""
    data = _vault_data_or_raise(params.vault)
    root = Path(data["path"])
    target = resolve_safe_path(root, params.path)

    if not target.exists() or not target.is_file():
        raise ToolNotFoundError(f"File not found: {params.path}", details={"vault": params.vault, "path": params.path})

    size = target.stat().st_size
    if size > TOOL_MAX_READ_BYTES:
        raise ToolError(
            f"File too large ({size} bytes > {TOOL_MAX_READ_BYTES})",
            code="file_too_large",
            details={"vault": params.vault, "path": params.path, "size": size},
        )

    content = target.read_text(encoding="utf-8", errors="replace")

    from backend.secret_redactor import redact_file_content

    content = redact_file_content(content, params.path)
    return {"vault": params.vault, "path": params.path, "size": size, "content": content}


@tool(
    name="search_fulltext",
    description="Full-text search across one vault or all accessible vaults.",
    input_model=SearchFulltextInput,
    risk=ToolRisk.READ,
)
def search_fulltext(ctx: ToolContext, params: SearchFulltextInput) -> list[dict[str, Any]]:
    """Return ranked search results, filtered to accessible vaults."""
    from backend.search import search

    results = search(params.q, vault_filter=params.vault, tag_filter=params.tag, limit=params.limit)
    return [r for r in results if ctx.has_vault_access(r["vault"])]


@tool(
    name="list_tags",
    description="List tags with their occurrence counts for a vault or all accessible vaults.",
    input_model=ListTagsInput,
    risk=ToolRisk.READ,
)
def list_tags(ctx: ToolContext, params: ListTagsInput) -> list[dict[str, Any]]:
    """Return tags sorted by descending count."""
    from backend.search import get_all_tags

    if params.vault and params.vault != "all":
        ctx.require_vault_access(params.vault)
        return [{"tag": tag, "count": count} for tag, count in get_all_tags(params.vault).items()]

    from backend.indexer import get_vault_names

    merged: dict[str, int] = {}
    for name in get_vault_names():
        if not ctx.has_vault_access(name):
            continue
        for tag, count in get_all_tags(name).items():
            merged[tag] = merged.get(tag, 0) + count
    return [{"tag": tag, "count": count} for tag, count in sorted(merged.items(), key=lambda x: -x[1])]