ObsiGate/backend/tools/service.py

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

These functions are the single source of truth consumed by both the in-app
assistant and the MCP server. They delegate to the shared business-logic
services (``backend.services``) so routes and tools never diverge.
Mutating tools (create/edit/rename/move/delete) are added in later phases.
"""

from __future__ import annotations

import logging
from typing import Any

from backend.indexer import get_backlinks as _get_backlinks
from backend.indexer import get_vault_names
from backend.services.backups import diff_backup as _diff_backup
from backend.services.backups import list_backup_files as _list_backup_files
from backend.services.files import read_file_text
from backend.services.graph import get_graph as _get_graph
from backend.services.recent import list_recent as _list_recent
from backend.services.search import advanced_search_vaults, search_vaults
from backend.services.search import list_tags as _list_tags
from backend.services.search import search_paths as _search_paths
from backend.services.vaults import browse_directory, list_accessible_vaults, list_all_files
from backend.tools.context import ToolContext, ToolRisk
from backend.tools.registry import tool
from backend.tools.schemas import (
    DiffBackupInput,
    GetBacklinksInput,
    GetGraphInput,
    ListAllFilesInput,
    ListBackupsInput,
    ListDirectoryInput,
    ListRecentInput,
    ListTagsInput,
    ListVaultsInput,
    ReadFileInput,
    ReadFileRawInput,
    SearchAdvancedInput,
    SearchFulltextInput,
    SearchPathsInput,
    SuggestTagsInput,
)

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

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


# ── C1. Vaults / navigation ────────────────────────────────────────────────


@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."""
    return [
        {"name": v["name"], "file_count": v["file_count"]}
        for v in list_accessible_vaults(ctx.user)
    ]


@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 = browse_directory(params.vault, params.path)
    return [
        {"name": item["name"], "path": item["path"], "type": item["type"]}
        for item in data["items"]
    ]


@tool(
    name="list_all_files",
    description="List every file of a vault (optionally under a subdirectory), newest first.",
    input_model=ListAllFilesInput,
    risk=ToolRisk.READ,
    requires_vault=True,
)
def list_all_files_tool(ctx: ToolContext, params: ListAllFilesInput) -> dict[str, Any]:
    """Return a flat list of files with metadata."""
    return list_all_files(params.vault, dir=params.dir, limit=params.limit, recursive=params.recursive)


# ── C2. Content reading ────────────────────────────────────────────────────


@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."""
    return read_file_text(
        params.vault,
        params.path,
        redact=True,
        max_bytes=TOOL_MAX_READ_BYTES,
    )


@tool(
    name="read_file_raw",
    description="Read a file's raw text content, without the size cap (secrets redacted).",
    input_model=ReadFileRawInput,
    risk=ToolRisk.READ,
    requires_vault=True,
)
def read_file_raw(ctx: ToolContext, params: ReadFileRawInput) -> dict[str, Any]:
    """Return the full (redacted) raw text of a vault file."""
    data = read_file_text(params.vault, params.path, redact=True, max_bytes=None)
    return {"vault": data["vault"], "path": data["path"], "raw": data["content"]}


@tool(
    name="get_backlinks",
    description="List files that link to a target file via [[wikilinks]].",
    input_model=GetBacklinksInput,
    risk=ToolRisk.READ,
    requires_vault=True,
)
def get_backlinks(ctx: ToolContext, params: GetBacklinksInput) -> list[dict[str, Any]]:
    """Return backlinks, filtered to accessible vaults."""
    backlinks = _get_backlinks(params.vault, params.path)
    return [b for b in backlinks if ctx.has_vault_access(b["vault"])]


@tool(
    name="list_backups",
    description="List the available backup versions of a file (newest first).",
    input_model=ListBackupsInput,
    risk=ToolRisk.READ,
    requires_vault=True,
)
def list_backups(ctx: ToolContext, params: ListBackupsInput) -> dict[str, Any]:
    """Return the backup versions of a vault file."""
    return {
        "vault": params.vault,
        "path": params.path,
        "backups": _list_backup_files(params.vault, params.path),
    }


@tool(
    name="diff_backup",
    description="Show a unified diff between a backup version and another version or the current file.",
    input_model=DiffBackupInput,
    risk=ToolRisk.READ,
    requires_vault=True,
)
def diff_backup(ctx: ToolContext, params: DiffBackupInput) -> dict[str, Any]:
    """Return the unified diff for a backup version."""
    return _diff_backup(params.vault, params.path, params.version, params.compare_with)


@tool(
    name="get_graph",
    description="Return the graph (nodes and edges: parent/child + wikilinks) of a vault or directory.",
    input_model=GetGraphInput,
    risk=ToolRisk.READ,
    requires_vault=True,
)
def get_graph(ctx: ToolContext, params: GetGraphInput) -> dict[str, Any]:
    """Return graph data for a vault subtree or the whole vault."""
    return _get_graph(
        params.vault,
        path=params.path,
        depth=params.depth,
        scope=params.scope,
        tag=params.tag,
    )


# ── C3. Search ─────────────────────────────────────────────────────────────


@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."""
    payload = search_vaults(params.q, vault=params.vault, tag=params.tag, limit=params.limit)
    return [r for r in payload["results"] if ctx.has_vault_access(r["vault"])]


@tool(
    name="search_advanced",
    description="Advanced full-text search with operators, filters and facets.",
    input_model=SearchAdvancedInput,
    risk=ToolRisk.READ,
)
def search_advanced(ctx: ToolContext, params: SearchAdvancedInput) -> list[dict[str, Any]]:
    """Return advanced search results, filtered to accessible vaults."""
    payload = advanced_search_vaults(
        params.q,
        vault=params.vault,
        tag=params.tag,
        limit=params.limit,
        offset=params.offset,
        sort=params.sort,
        case_sensitive=params.case_sensitive,
        whole_word=params.whole_word,
        regex=params.regex,
        include_paths=params.include_paths,
        exclude_paths=params.exclude_paths,
        created=params.created,
        modified=params.modified,
        size=params.size,
    )
    return [r for r in payload["results"] if ctx.has_vault_access(r["vault"])]


@tool(
    name="search_paths",
    description="Search files and directories by path substring.",
    input_model=SearchPathsInput,
    risk=ToolRisk.READ,
)
def search_paths(ctx: ToolContext, params: SearchPathsInput) -> list[dict[str, Any]]:
    """Return path matches, filtered to accessible vaults."""
    payload = _search_paths(params.q, vault=params.vault)
    return [r for r in payload["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."""
    if params.vault and params.vault != "all":
        ctx.require_vault_access(params.vault)
        return [{"tag": tag, "count": count} for tag, count in _list_tags(params.vault).items()]

    merged: dict[str, int] = {}
    for name in get_vault_names():
        if not ctx.has_vault_access(name):
            continue
        for tag, count in _list_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])]


@tool(
    name="suggest_tags",
    description="Suggest tags matching a prefix, across accessible vaults.",
    input_model=SuggestTagsInput,
    risk=ToolRisk.READ,
)
def suggest_tags(ctx: ToolContext, params: SuggestTagsInput) -> list[dict[str, Any]]:
    """Return tag suggestions, restricted to accessible vaults."""
    from backend.search import suggest_tags as _suggest

    if params.vault and params.vault != "all":
        ctx.require_vault_access(params.vault)
        return _suggest(params.q, vault_filter=params.vault, limit=params.limit)

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


@tool(
    name="list_recent",
    description="List the current user's recently opened (or modified) files.",
    input_model=ListRecentInput,
    risk=ToolRisk.READ,
)
def list_recent(ctx: ToolContext, params: ListRecentInput) -> dict[str, Any]:
    """Return recent files for the calling user."""
    user_vaults = ctx.user.get("_token_vaults") or ctx.user.get("vaults", [])
    return _list_recent(
        ctx.username,
        user_vaults,
        vault=params.vault,
        limit=params.limit,
        mode=params.mode,
    )