Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 79 additions & 0 deletions openhands-agent-server/openhands/agent_server/file_router.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import os
from pathlib import Path
from typing import Annotated
from uuid import UUID
Expand All @@ -11,6 +12,7 @@
status,
)
from fastapi.responses import FileResponse
from pydantic import BaseModel
from starlette.background import BackgroundTask

from openhands.agent_server.bash_service import get_default_bash_event_service
Expand All @@ -21,6 +23,20 @@
from openhands.sdk.logger import get_logger


class SubdirectoryEntry(BaseModel):
name: str
path: str


class ListSubdirsResponse(BaseModel):
path: str
subdirs: list[SubdirectoryEntry]


class HomeResponse(BaseModel):
home: str


logger = get_logger(__name__)
file_router = APIRouter(prefix="/file", tags=["Files"])
config = get_default_config()
Expand Down Expand Up @@ -116,6 +132,69 @@ async def download_file_query(
return await _download_file(path)


@file_router.get("/home")
async def get_home_directory() -> HomeResponse:
"""Return the agent-server user's home directory.

Used by the GUI to start a folder-browser at a sensible default location.
"""
return HomeResponse(home=str(Path.home()))


@file_router.get("/list_subdirs")
async def list_subdirs(
path: Annotated[
str,
Query(description="Absolute directory path to list subdirectories of"),
],
) -> ListSubdirsResponse:
"""List immediate subdirectories of `path`.

Used by the GUI's workspace picker. Hidden entries (names starting with '.')
and symlinks are skipped. Files are skipped. Returns absolute paths so the
GUI can use a result directly as ``workspace.working_dir``.
"""
target = Path(path)
if not target.is_absolute():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Path must be absolute",
)
if not target.exists():
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Directory not found",
)
if not target.is_dir():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Path is not a directory",
)

entries: list[SubdirectoryEntry] = []
try:
with os.scandir(target) as scanner:
for entry in scanner:
if entry.name.startswith("."):
continue
try:
if not entry.is_dir(follow_symlinks=False):
continue
except OSError:
continue
entries.append(
SubdirectoryEntry(name=entry.name, path=str(target / entry.name))
)
except PermissionError as e:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Permission denied: {e}",
)

entries.sort(key=lambda e: e.name.lower())
return ListSubdirsResponse(path=str(target), subdirs=entries)


@file_router.get("/download-trajectory/{conversation_id}")
async def download_trajectory(
conversation_id: UUID,
Expand Down
52 changes: 52 additions & 0 deletions tests/agent_server/test_file_router.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Tests for file_router.py endpoints."""

import io
from pathlib import Path

import pytest
from fastapi.testclient import TestClient
Expand Down Expand Up @@ -246,3 +247,54 @@ def test_file_legacy_routes_are_removed_from_openapi(client):
openapi_paths = response.json()["paths"]
assert "/api/file/upload/{path}" not in openapi_paths
assert "/api/file/download/{path}" not in openapi_paths


# =============================================================================
# list_subdirs Tests
# =============================================================================


def test_list_subdirs_returns_only_directories_with_absolute_paths(client, tmp_path):
"""Return subdirs with absolute paths; skip files and hidden entries."""
(tmp_path / "repo1").mkdir()
(tmp_path / "repo2").mkdir()
(tmp_path / ".hidden_dir").mkdir()
(tmp_path / "README.md").write_text("hi")

response = client.get("/api/file/list_subdirs", params={"path": str(tmp_path)})

assert response.status_code == 200
body = response.json()
assert body["path"] == str(tmp_path)
names = [entry["name"] for entry in body["subdirs"]]
paths = [entry["path"] for entry in body["subdirs"]]
assert names == ["repo1", "repo2"]
assert paths == [str(tmp_path / "repo1"), str(tmp_path / "repo2")]


def test_list_subdirs_relative_path_returns_400(client):
response = client.get("/api/file/list_subdirs", params={"path": "relative/path"})
assert response.status_code == 400
assert "must be absolute" in response.json()["detail"]


def test_list_subdirs_missing_directory_returns_404(client, tmp_path):
response = client.get(
"/api/file/list_subdirs",
params={"path": str(tmp_path / "does-not-exist")},
)
assert response.status_code == 404


def test_list_subdirs_path_is_a_file_returns_400(client, tmp_path):
file_path = tmp_path / "file.txt"
file_path.write_text("hi")
response = client.get("/api/file/list_subdirs", params={"path": str(file_path)})
assert response.status_code == 400
assert "not a directory" in response.json()["detail"]


def test_get_home_returns_user_home(client):
response = client.get("/api/file/home")
assert response.status_code == 200
assert response.json()["home"] == str(Path.home())
Loading