ObsiGate/backend/tools/schemas.py

"""Pydantic input/output models for the AI tool layer.

Input models double as JSON Schemas advertised to LLMs (via
``model_json_schema``) and as validation for arguments received over MCP.
"""

from __future__ import annotations

from typing import Any

from pydantic import BaseModel, Field


class ListVaultsInput(BaseModel):
    """No parameters — lists the vaults the caller can access."""


class ListDirectoryInput(BaseModel):
    """Browse a directory inside a vault."""

    vault: str = Field(..., description="Vault name")
    path: str = Field("", description="Vault-relative directory path (empty = vault root)")


class ReadFileInput(BaseModel):
    """Read a text file's content from a vault."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")


class SearchFulltextInput(BaseModel):
    """Full-text search across one or all accessible vaults."""

    q: str = Field(..., min_length=1, description="Search query")
    vault: str = Field("all", description="Vault name or 'all'")
    tag: str | None = Field(None, description="Optional comma-separated tag filter")
    limit: int = Field(50, ge=1, le=200, description="Maximum number of results")


class ListTagsInput(BaseModel):
    """List tags and their occurrence counts."""

    vault: str | None = Field(None, description="Vault name or 'all' (default: all accessible)")


class ListAllFilesInput(BaseModel):
    """List every file of a vault (optionally under a subdirectory)."""

    vault: str = Field(..., description="Vault name")
    dir: str = Field("", description="Vault-relative directory path (empty = vault root)")
    limit: int = Field(200, ge=1, le=2000, description="Maximum number of files")
    recursive: bool = Field(True, description="Recurse into subdirectories")


class ReadFileRawInput(BaseModel):
    """Read a text file's raw content (no HTML rendering)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")


class GetBacklinksInput(BaseModel):
    """List files linking to a target file via wikilinks."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the target file")


class ListBackupsInput(BaseModel):
    """List available backup versions of a file."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")


class DiffBackupInput(BaseModel):
    """Unified diff between a backup version and another version or the current file."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")
    version: int = Field(..., description="Backup timestamp used as the old/left side")
    compare_with: int | None = Field(
        None,
        description="Backup timestamp used as the new/right side (omit to compare with the current file)",
    )


class GetGraphInput(BaseModel):
    """Graph data (nodes and edges) for a vault or directory."""

    vault: str = Field(..., description="Vault name")
    path: str = Field("", description="Vault-relative directory path to focus on (empty = root)")
    depth: int = Field(1, ge=0, le=3, description="Expansion depth")
    scope: str = Field("directory", description="'directory' for a subtree, 'full' for the whole vault")
    tag: str = Field("", description="Only include files carrying this tag")


class SearchAdvancedInput(BaseModel):
    """Advanced full-text search (TF-IDF, facets, operators)."""

    q: str = Field("", description="Query (supports tag:, vault:, title:, path:, ext: operators)")
    vault: str = Field("all", description="Vault name or 'all'")
    tag: str | None = Field(None, description="Optional comma-separated tag filter")
    limit: int = Field(50, ge=1, le=200, description="Maximum number of results")
    offset: int = Field(0, ge=0, description="Pagination offset")
    sort: str = Field("relevance", description="Sort by 'relevance' or 'modified'")
    case_sensitive: bool = Field(False, description="Match case")
    whole_word: bool = Field(False, description="Match whole words only")
    regex: bool = Field(False, description="Treat query as a regular expression")
    include_paths: str | None = Field(None, description="Comma-separated glob patterns to include")
    exclude_paths: str | None = Field(None, description="Comma-separated glob patterns to exclude")
    created: str | None = Field(None, description="Created date filter (>date, <date, date..date)")
    modified: str | None = Field(None, description="Modified date filter (>date, <date, date..date, <Nd)")
    size: str | None = Field(None, description="Size filter (>size, <size, size..size, e.g. >1MB)")


class SearchPathsInput(BaseModel):
    """Search files and directories by path substring."""

    q: str = Field(..., min_length=1, description="Path substring to search for")
    vault: str = Field("all", description="Vault name or 'all'")


class SuggestTagsInput(BaseModel):
    """Suggest tags matching a prefix."""

    q: str = Field(..., min_length=1, description="Tag prefix (with or without leading '#')")
    vault: str = Field("all", description="Vault name or 'all'")
    limit: int = Field(10, ge=1, le=50, description="Maximum number of suggestions")


class ListRecentInput(BaseModel):
    """List recently opened (or modified) files."""

    vault: str | None = Field(None, description="Optional single-vault filter")
    limit: int = Field(20, ge=1, le=200, description="Maximum number of files")
    mode: str = Field("opened", description="'opened' (history) or 'modified'")


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


class CreateFileInput(BaseModel):
    """Create a new text file in a vault."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the new file")
    content: str = Field("", description="Initial file content")


class CreateDirectoryInput(BaseModel):
    """Create a new directory in a vault."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the new directory")


class EditFileInput(BaseModel):
    """Overwrite the full content of an existing file."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")
    content: str = Field(..., description="New full content of the file")


class AppendToFileInput(BaseModel):
    """Append text to the end of an existing file."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")
    content: str = Field(..., description="Text to append")


class RenameFileInput(BaseModel):
    """Rename a file in place (same parent directory)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Current vault-relative file path")
    new_name: str = Field(..., description="New file name (no directory separator)")


class RenameDirectoryInput(BaseModel):
    """Rename a directory in place (same parent directory)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Current vault-relative directory path")
    new_name: str = Field(..., description="New directory name (no directory separator)")


class MovePathInput(BaseModel):
    """Move a file or directory to another directory within the same vault."""

    vault: str = Field(..., description="Vault name")
    source_path: str = Field(..., description="Current vault-relative path of the file/directory")
    destination_dir: str = Field("", description="Target directory path (empty = vault root)")


class ReplaceInFilesInput(BaseModel):
    """Find and replace text across vault files (destructive, dry-run by default)."""

    find: str = Field(..., min_length=1, description="Text or pattern to search for")
    replace: str = Field("", description="Replacement text")
    vault: str = Field("all", description="Vault name or 'all'")
    case_sensitive: bool = Field(False, description="Match case")
    whole_word: bool = Field(False, description="Match whole words only")
    regex: bool = Field(False, description="Treat 'find' as a regular expression")
    include_paths: str | None = Field(None, description="Comma-separated glob patterns to include")
    exclude_paths: str | None = Field(None, description="Comma-separated glob patterns to exclude")
    replace_all: bool = Field(False, description="Apply the replacement (otherwise only preview)")
    dry_run: bool | None = Field(
        None,
        description="Preview matches without writing. Defaults to the opposite of replace_all.",
    )


class DeleteFileInput(BaseModel):
    """Delete a file from a vault (destructive)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")


class DeleteDirectoryInput(BaseModel):
    """Delete a directory from a vault (destructive)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative directory path")
    recursive: bool = Field(True, description="Delete non-empty directories recursively")


class RestoreBackupInput(BaseModel):
    """Restore a file from one of its backup versions."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")
    version: int = Field(..., description="Backup timestamp to restore")


class ToolResult(BaseModel):
    """Uniform result returned by :func:`backend.tools.registry.call_tool`."""

    ok: bool = True
    data: Any = None
    error: str | None = None