diff --git a/src/pineagents/agents/tools/__init__.py b/src/pineagents/agents/tools/__init__.py index 0121495..9b72c7b 100644 --- a/src/pineagents/agents/tools/__init__.py +++ b/src/pineagents/agents/tools/__init__.py @@ -41,6 +41,7 @@ from .file_io import ( # noqa: E402 write_file, edit_file, append_file, + list_directory, ) from .file_search import grep_search, glob_search # noqa: E402 from .shell import execute_shell_command # noqa: E402 diff --git a/src/pineagents/agents/tools/file_io.py b/src/pineagents/agents/tools/file_io.py index d8e90a0..6f45a1f 100644 --- a/src/pineagents/agents/tools/file_io.py +++ b/src/pineagents/agents/tools/file_io.py @@ -527,3 +527,136 @@ async def append_file( ), ], ) + + +# --------------------------------------------------------------------------- +# list_directory +# --------------------------------------------------------------------------- + +# Entries the file browser and this tool agree are never worth surfacing to +# the model (mirrors upstream qwenpaw.services.workspace_files._SKIPPED_NAMES). +_LIST_DIRECTORY_SKIPPED_NAMES = frozenset( + { + ".git", + ".hypothesis", + ".mypy_cache", + ".pytest_cache", + ".ruff_cache", + ".venv", + "__pycache__", + "node_modules", + }, +) + +# Hard cap so a huge directory never floods the model's context. The tool +# reports the count of omitted entries so the model can ask for a subtree. +_LIST_DIRECTORY_MAX_ENTRIES = 200 + + +@tool_descriptor( + requires_sandbox=("file_read",), + async_execution=True, + tool_type="file", + target_param="file_path", + policy_name="ListDirectory", + default_policy="allow", + policy_reason="Read-only directory listing (global)", + ui_description="List directory contents", + ui_icon="📁", +) +async def list_directory( + file_path: str = ".", + max_entries: int = _LIST_DIRECTORY_MAX_ENTRIES, +) -> ToolChunk: + """List the immediate children of a directory. + + Use this to discover what a directory contains before reading or + editing files in it. Returns one line per entry, directories first, + each prefixed with ``D`` (directory) or ``F`` (file) plus the file + size in bytes. + + Args: + file_path (`str`): + Directory to list. An absolute path is used as-is; a + relative path resolves from the agent workspace. A trailing + ``/`` or ``\\`` is allowed. + max_entries (`int`, optional): + Maximum number of entries to return (default 200). The tool + reports how many entries were omitted so you can drill into + a subtree instead of requesting a bigger page. + """ + try: + resolved = Path(_resolve_file_path(file_path)) + except Exception as exc: # pragma: no cover - defensive + return ToolChunk( + is_last=True, + state=ToolResultState.ERROR, + content=[ + TextBlock( + type="text", + text=f"Error: Failed to resolve directory {file_path!r}: {exc}", + ), + ], + ) + + if not resolved.exists(): + return ToolChunk( + is_last=True, + state=ToolResultState.ERROR, + content=[ + TextBlock( + type="text", + text=f"Error: The directory {resolved} does not exist.", + ), + ], + ) + if not resolved.is_dir(): + return ToolChunk( + is_last=True, + state=ToolResultState.ERROR, + content=[ + TextBlock( + type="text", + text=f"Error: The path {resolved} is not a directory.", + ), + ], + ) + + limit = min(max(1, int(max_entries)), 500) + + entries: list[tuple[str, str, int]] = [] # (kind, name, size) + with os.scandir(resolved) as scanner: + for entry in scanner: + name = entry.name + if name.startswith(".") or name in _LIST_DIRECTORY_SKIPPED_NAMES: + continue + try: + is_dir = entry.is_dir(follow_symlinks=False) + size = 0 if is_dir else entry.stat(follow_symlinks=False).st_size + except OSError: + # Broken symlink / permission — still surface the name so the + # model knows it exists, marked as a file with unknown size. + is_dir = False + size = -1 + entries.append(("D" if is_dir else "F", name, size)) + + entries.sort(key=lambda item: (item[0] != "D", item[2], item[1])) + + omitted = max(0, len(entries) - limit) + visible = entries[:limit] + + lines = [f"{resolved}/ — {len(entries)} entries"] + for kind, name, size in visible: + if kind == "D": + lines.append(f"D {name}/") + else: + size_str = f"{size} B" if size >= 0 else "?" + lines.append(f"F {name} {size_str}") + if omitted: + lines.append(f"(… {omitted} more entries omitted; list a subdirectory to drill in)") + + return ToolChunk( + is_last=True, + state=ToolResultState.SUCCESS, + content=[TextBlock(type="text", text="\n".join(lines))], + )