"""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"]*>.*?", "", raw, flags=re.DOTALL | re.IGNORECASE) text = re.sub(r"]*>.*?", "", 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]