api/app/agents/filesystem_agent.py

"""Filesystem agent — tools for reading local directories and files on Electron.

These tools delegate to the Electron client via ``execute_on_client()`` using
the same WS tool-call round-trip pattern as CRUD tools.  The Electron app
handles actual disk I/O and responds with ``tool_result`` frames.
"""

from __future__ import annotations

import os
import re
from pathlib import Path
from typing import Any

from langchain_core.tools import tool

from app.core.ws_context import execute_on_client

# Max characters returned by read_file_content in journey (exploration) tools.
# The journey only needs to understand file structure, not full content.
_JOURNEY_READ_MAX_CHARS: int = 4000


def _resolve_path(path: str, base: str) -> str:
    """Resolve *path* against *base* when *path* is relative.

    The LLM often passes ``"."`` meaning "the configured directory".
    Without this, Electron resolves ``"."`` relative to its own CWD instead
    of the user's chosen directory.
    """
    if os.path.isabs(path):
        return path
    return str(Path(base) / path)


@tool
async def list_directory(path: str) -> str:
    """List files and folders in a local directory on the user's device.

    Returns a formatted listing of entries with name, type (file/directory),
    and full path.
    """
    result = await execute_on_client(
        action="list_directory",
        data={"path": path},
    )
    entries: list[dict[str, Any]] = result.get("entries", [])
    if not entries:
        return f"Directory '{path}' is empty or does not exist."
    lines: list[str] = []
    for entry in entries:
        entry_type = entry.get("type", "unknown")
        entry_name = entry.get("name", "")
        entry_path = entry.get("path", "")
        lines.append(f"- [{entry_type}] {entry_name}  ({entry_path})")
    return f"Directory listing for '{path}' ({len(entries)} entries):\n" + "\n".join(lines)


@tool
async def read_file_content(path: str) -> str:
    """Read the text content of a local file on the user's device.

    Returns the file content as a string.  Large files may be truncated
    by the Electron client.
    """
    result = await execute_on_client(
        action="read_file_content",
        data={"path": path},
    )
    content: str = result.get("content", "")
    if not content:
        return f"File '{path}' is empty or could not be read."
    return content


@tool
async def get_file_metadata(path: str) -> str:
    """Get metadata for a local file: size, creation date, modification date, extension.

    Returns a formatted summary of the file's metadata.
    """
    result = await execute_on_client(
        action="get_file_metadata",
        data={"path": path},
    )
    size = result.get("size", "unknown")
    created = result.get("createdAt", "unknown")
    modified = result.get("modifiedAt", "unknown")
    extension = result.get("extension", "unknown")
    name = result.get("name", path)
    return (
        f"File: {name}\n"
        f"  Extension: {extension}\n"
        f"  Size: {size} bytes\n"
        f"  Created: {created}\n"
        f"  Modified: {modified}"
    )


FILESYSTEM_TOOLS: list[Any] = [
    list_directory,
    read_file_content,
    get_file_metadata,
]


def make_directory_tools(base_directory: str) -> list[Any]:
    """Return filesystem tools that resolve relative paths against *base_directory*.

    Use this instead of ``FILESYSTEM_TOOLS`` whenever you know the user's target
    directory upfront (e.g., journey setup sessions).  Relative paths like ``"."``
    from the LLM are resolved to the correct absolute path before being sent to
    the Electron client, preventing it from falling back to its own CWD.
    """

    def _compact_for_journey(raw: str) -> str:
        """Strip HTML noise and truncate for journey exploration.

        The journey LLM only needs to understand file structure (headers,
        first paragraphs).  Full CSS/style blocks are pure noise that eat
        up context window budget.
        """
        text = re.sub(r"<style[^>]*>.*?</style>", "", raw, flags=re.DOTALL | re.IGNORECASE)
        text = re.sub(r"<script[^>]*>.*?</script>", "", text, flags=re.DOTALL | re.IGNORECASE)
        text = re.sub(r"<!--.*?-->", "", text, flags=re.DOTALL)
        if len(text) > _JOURNEY_READ_MAX_CHARS:
            text = text[:_JOURNEY_READ_MAX_CHARS] + "\n[…truncated for exploration]"
        return text

    @tool
    async def list_directory(path: str) -> str:  # noqa: F811
        """List files and folders in a local directory on the user's device.

        Returns a formatted listing of entries with name, type (file/directory),
        and full path.
        """
        resolved = _resolve_path(path, base_directory)
        result = await execute_on_client(
            action="list_directory",
            data={"path": resolved},
        )
        entries: list[dict[str, Any]] = result.get("entries", [])
        if not entries:
            return f"Directory '{resolved}' is empty or does not exist."
        lines: list[str] = []
        for entry in entries:
            entry_type = entry.get("type", "unknown")
            entry_name = entry.get("name", "")
            entry_path = entry.get("path", "")
            lines.append(f"- [{entry_type}] {entry_name}  ({entry_path})")
        return f"Directory listing for '{resolved}' ({len(entries)} entries):\n" + "\n".join(lines)

    @tool
    async def read_file_content(path: str) -> str:  # noqa: F811
        """Read the text content of a local file on the user's device.

        Returns the file content as a string.  Large files may be truncated
        by the Electron client.
        """
        resolved = _resolve_path(path, base_directory)
        result = await execute_on_client(
            action="read_file_content",
            data={"path": resolved},
        )
        content: str = result.get("content", "")
        if not content:
            return f"File '{resolved}' is empty or could not be read."
        return _compact_for_journey(content)

    @tool
    async def get_file_metadata(path: str) -> str:  # noqa: F811
        """Get metadata for a local file: size, creation date, modification date, extension.

        Returns a formatted summary of the file's metadata.
        """
        resolved = _resolve_path(path, base_directory)
        result = await execute_on_client(
            action="get_file_metadata",
            data={"path": resolved},
        )
        size = result.get("size", "unknown")
        created = result.get("createdAt", "unknown")
        modified = result.get("modifiedAt", "unknown")
        extension = result.get("extension", "unknown")
        name = result.get("name", resolved)
        return (
            f"File: {name}\n"
            f"  Extension: {extension}\n"
            f"  Size: {size} bytes\n"
            f"  Created: {created}\n"
            f"  Modified: {modified}"
        )

    return [list_directory, read_file_content, get_file_metadata]