ObsiGate/backend/tools/service.py

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

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/restore) are registered with
``ToolRisk.WRITE`` or ``ToolRisk.DANGEROUS`` and gated by the registry's
confirmation mechanism (two-step propose/apply).
"""

from __future__ import annotations

import logging
import os
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.mutations import (
    append_to_file as _append_to_file,
)
from backend.services.mutations import (
    create_directory as _create_directory,
)
from backend.services.mutations import (
    create_file as _create_file,
)
from backend.services.mutations import (
    delete_directory as _delete_directory,
)
from backend.services.mutations import (
    delete_file as _delete_file,
)
from backend.services.mutations import (
    edit_file as _edit_file,
)
from backend.services.mutations import (
    move_path as _move_path,
)
from backend.services.mutations import (
    rename_directory as _rename_directory,
)
from backend.services.mutations import (
    rename_file as _rename_file,
)
from backend.services.mutations import (
    replace_in_files as _replace_in_files,
)
from backend.services.mutations import (
    restore_backup as _restore_backup,
)
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 (
    AppendToFileInput,
    CreateDirectoryInput,
    CreateFileInput,
    DeleteDirectoryInput,
    DeleteFileInput,
    DiffBackupInput,
    EditFileInput,
    GetBacklinksInput,
    GetGraphInput,
    ListAllFilesInput,
    ListBackupsInput,
    ListDirectoryInput,
    ListRecentInput,
    ListTagsInput,
    ListVaultsInput,
    MovePathInput,
    ReadFileInput,
    ReadFileRawInput,
    RenameDirectoryInput,
    RenameFileInput,
    ReplaceInFilesInput,
    RestoreBackupInput,
    SearchAdvancedInput,
    SearchFulltextInput,
    SearchPathsInput,
    SuggestTagsInput,
)

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

# Maximum file size returned by ``read_file`` (bytes). Quota configurable via
# ``BOOKSLM_MAX_TOOL_READ_BYTES``.
TOOL_MAX_READ_BYTES = int(os.environ.get("BOOKSLM_MAX_TOOL_READ_BYTES", "200000"))


# ── 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,
    )


# ── D. Mutations ───────────────────────────────────────────────────────────


@tool(
    name="create_file",
    description=(
        "Create a new text file in a vault with optional initial content. "
        "Parent directories are created automatically, so a single call with a "
        "nested path (e.g. 'Folder/note.md') is enough to create a file inside "
        "a new folder."
    ),
    input_model=CreateFileInput,
    risk=ToolRisk.WRITE,
    requires_vault=True,
)
def create_file(ctx: ToolContext, params: CreateFileInput) -> dict[str, Any]:
    """Create a vault file (fails if it already exists)."""
    return _create_file(params.vault, params.path, params.content)


@tool(
    name="create_directory",
    description=(
        "Create a new directory (and parents) in a vault. Succeeds if it "
        "already exists. Optional when creating a file: create_file already "
        "creates parent directories."
    ),
    input_model=CreateDirectoryInput,
    risk=ToolRisk.WRITE,
    requires_vault=True,
)
def create_directory(ctx: ToolContext, params: CreateDirectoryInput) -> dict[str, Any]:
    """Create a vault directory (idempotent)."""
    return _create_directory(params.vault, params.path, exist_ok=True)


@tool(
    name="edit_file",
    description="Overwrite the full content of an existing file (automatic backup).",
    input_model=EditFileInput,
    risk=ToolRisk.WRITE,
    requires_vault=True,
)
def edit_file(ctx: ToolContext, params: EditFileInput) -> dict[str, Any]:
    """Replace a vault file's content."""
    return _edit_file(params.vault, params.path, params.content)


@tool(
    name="append_to_file",
    description="Append text to the end of an existing file (automatic backup).",
    input_model=AppendToFileInput,
    risk=ToolRisk.WRITE,
    requires_vault=True,
)
def append_to_file(ctx: ToolContext, params: AppendToFileInput) -> dict[str, Any]:
    """Append content to a vault file."""
    return _append_to_file(params.vault, params.path, params.content)


@tool(
    name="rename_file",
    description="Rename a file in place (same parent directory). Destructive: requires confirmation.",
    input_model=RenameFileInput,
    risk=ToolRisk.DANGEROUS,
    requires_vault=True,
)
def rename_file(ctx: ToolContext, params: RenameFileInput) -> dict[str, Any]:
    """Rename a vault file."""
    return _rename_file(params.vault, params.path, params.new_name)


@tool(
    name="rename_directory",
    description="Rename a directory in place (same parent directory). Destructive: requires confirmation.",
    input_model=RenameDirectoryInput,
    risk=ToolRisk.DANGEROUS,
    requires_vault=True,
)
def rename_directory(ctx: ToolContext, params: RenameDirectoryInput) -> dict[str, Any]:
    """Rename a vault directory."""
    return _rename_directory(params.vault, params.path, params.new_name)


@tool(
    name="move_path",
    description="Move a file or directory to another directory in the same vault. Destructive: requires confirmation.",
    input_model=MovePathInput,
    risk=ToolRisk.DANGEROUS,
    requires_vault=True,
)
def move_path(ctx: ToolContext, params: MovePathInput) -> dict[str, Any]:
    """Move a vault file or directory."""
    return _move_path(params.vault, params.source_path, params.destination_dir)


@tool(
    name="replace_in_files",
    description=(
        "Find and replace text across vault files. Previews by default "
        "(dry_run); set replace_all=true to apply. Destructive: requires confirmation."
    ),
    input_model=ReplaceInFilesInput,
    risk=ToolRisk.DANGEROUS,
)
def replace_in_files(ctx: ToolContext, params: ReplaceInFilesInput) -> dict[str, Any]:
    """Preview or apply a find/replace, filtered to permitted vaults."""
    if params.vault and params.vault != "all":
        ctx.require_vault_access(params.vault)
        ctx.require_destructive_allowed(params.vault)

    dry_run = params.dry_run if params.dry_run is not None else not params.replace_all

    def _allowed(vault: str) -> bool:
        return ctx.has_vault_access(vault) and ctx.destructive_tools_enabled(vault)

    return _replace_in_files(
        params.find,
        params.replace,
        vault=params.vault,
        case_sensitive=params.case_sensitive,
        whole_word=params.whole_word,
        regex=params.regex,
        include_paths=params.include_paths,
        exclude_paths=params.exclude_paths,
        replace_all=params.replace_all,
        dry_run=dry_run,
        is_vault_allowed=_allowed,
    )


@tool(
    name="delete_file",
    description="Delete a file from a vault (automatic backup). Destructive: requires confirmation.",
    input_model=DeleteFileInput,
    risk=ToolRisk.DANGEROUS,
    requires_vault=True,
)
def delete_file(ctx: ToolContext, params: DeleteFileInput) -> dict[str, Any]:
    """Delete a vault file."""
    return _delete_file(params.vault, params.path)


@tool(
    name="delete_directory",
    description="Delete a directory (recursive by default) from a vault. Destructive: requires confirmation.",
    input_model=DeleteDirectoryInput,
    risk=ToolRisk.DANGEROUS,
    requires_vault=True,
)
def delete_directory(ctx: ToolContext, params: DeleteDirectoryInput) -> dict[str, Any]:
    """Delete a vault directory."""
    return _delete_directory(params.vault, params.path, recursive=params.recursive)


@tool(
    name="restore_backup",
    description="Restore a file from one of its backup versions (current version backed up first).",
    input_model=RestoreBackupInput,
    risk=ToolRisk.WRITE,
    requires_vault=True,
)
def restore_backup(ctx: ToolContext, params: RestoreBackupInput) -> dict[str, Any]:
    """Restore a vault file from a backup."""
    return _restore_backup(params.vault, params.path, params.version)