Newer
Older
navi-1 / navi / tools / share_file.py
@Eugene Sukhodolskiy Eugene Sukhodolskiy on 29 Apr 5 KB Clarify share file publishing boundaries
"""share_file tool — expose a local file to the user as a download link."""

import asyncio
import shutil
from pathlib import Path
from urllib.parse import quote

from navi.config import settings
from navi.session_files import ensure_session_dir

from .base import Tool, ToolResult, current_session_id


def _fmt_size(n: int) -> str:
    if n < 1024:
        return f"{n} B"
    if n < 1024 ** 2:
        return f"{n / 1024:.1f} KB"
    return f"{n / 1024 ** 2:.1f} MB"


class ShareFileTool(Tool):
    name = "share_file"
    description = (
        "Copy an existing local file into the current session directory and return a direct download link. "
        "Use this when the user should receive a file to keep: archives, reports, exports, source bundles, "
        "datasets, PDFs, CSV/JSON files, or other generated artifacts.\n\n"
        "Mechanics: share_file takes an ABSOLUTE source path, copies that file into "
        "SESSION_FILES_DIR/{session_id}/ under the optional clean filename, and returns a URL at "
        "/sessions/{session_id}/files/{filename}. The source file remains where it was. "
        "If a file with the same name already exists in the session directory, share_file creates "
        "a numbered filename instead of overwriting it. Max file size is SHARE_FILE_MAX_SIZE_MB "
        "(default 1024 MB / 1 GB).\n\n"
        "Do not confuse this with content_publish: share_file is for download links and may copy "
        "from elsewhere; content_publish is for inline viewer cards and only registers a file that "
        "already exists in the session directory.\n\n"
        "IMPORTANT — path must be an ABSOLUTE path (e.g. /home/user/file.zip). Relative paths are rejected. "
        "If you only know a relative path, resolve it first: use filesystem(action='info') or "
        "terminal('realpath <relative_path>') to get the absolute path, then call share_file."
    )
    parameters = {
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "description": "Absolute path to the file to share",
            },
            "filename": {
                "type": "string",
                "description": (
                    "Download filename shown to the user. "
                    "Defaults to the original filename. "
                    "Use this to give the file a clean, descriptive name."
                ),
            },
        },
        "required": ["path"],
    }

    async def execute(self, params: dict) -> ToolResult:
        session_id = current_session_id.get()
        if not session_id:
            return ToolResult(success=False, output="No active session context.", error="no_session")

        raw_path = Path(params["path"]).expanduser()
        if not raw_path.is_absolute():
            return ToolResult(
                success=False,
                output=(
                    f"share_file requires an absolute path, got: {params['path']}\n"
                    "Resolve it first with filesystem info or terminal realpath, then call share_file again."
                ),
                error="path_not_absolute",
            )

        src = raw_path.resolve()
        if not src.exists():
            return ToolResult(success=False, output=f"File not found: {src}", error="not_found")
        if not src.is_file():
            return ToolResult(success=False, output=f"Path is not a file: {src}", error="not_a_file")

        max_bytes = settings.share_file_max_size_mb * 1024 * 1024
        src_size = src.stat().st_size
        if src_size > max_bytes:
            return ToolResult(
                success=False,
                output=(
                    f"File is too large to share: {_fmt_size(src_size)}. "
                    f"Limit: {settings.share_file_max_size_mb} MB."
                ),
                error="file_too_large",
            )

        # Determine the download filename
        download_name = Path(params.get("filename") or src.name).name or src.name

        dest_dir = ensure_session_dir(session_id)
        dest = dest_dir / download_name

        # If it's already the same file in the session dir, skip copy
        if dest.resolve() == src.resolve():
            pass
        else:
            # Avoid clobbering an existing different file with the same name
            if dest.exists():
                stem = Path(download_name).stem
                suffix = Path(download_name).suffix
                i = 1
                while dest.exists():
                    dest = dest_dir / f"{stem}_{i}{suffix}"
                    i += 1
            await asyncio.to_thread(shutil.copy2, str(src), str(dest))

        size = dest.stat().st_size
        base_url = settings.public_url.rstrip("/")
        url = f"{base_url}/sessions/{quote(session_id, safe='')}/files/{quote(dest.name, safe='')}"

        return ToolResult(
            success=True,
            output=(
                f"Download ready: {dest.name} ({_fmt_size(size)})\n"
                f"URL: {url}\n\n"
                f"→ Include this link in your reply: [{dest.name}]({url})"
            ),
            metadata={"url": url, "filename": dest.name, "size": size},
        )