Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(webui): add native workspace folder picker

Browse Source
This commit is contained in:
Xubin Ren
2026-08-14 04:03:54 +09:00
parent 410e5e5121
commit 26c9687b80
11 changed files with 422 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/channels/websocket/tests/test_websocket_http_routes.py
+79
View File
@@ -3266,6 +3266,85 @@ async def _webui_mutate(
)
@pytest.mark.asyncio
async def test_workspace_folder_picker_is_local_authenticated_mutation(
bus: MagicMock,
tmp_path: Path,
monkeypatch,
) -> None:
selected = tmp_path / "project"
selected.mkdir()
pick_folder = AsyncMock(return_value=str(selected))
monkeypatch.setattr(
"nanobot.webui.ws_http.native_folder_picker_available",
lambda: True,
)
monkeypatch.setattr("nanobot.webui.ws_http.pick_native_folder", pick_folder)
channel = _ch(bus)
response = await _webui_mutate(channel, "workspace.pick_folder")
assert response.status_code == 200
assert response.json() == {"path": str(selected)}
pick_folder.assert_awaited_once_with()
@pytest.mark.asyncio
async def test_workspace_folder_picker_rejects_direct_http(
bus: MagicMock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
pick_folder = AsyncMock(return_value="/tmp")
monkeypatch.setattr(
"nanobot.webui.ws_http.native_folder_picker_available",
lambda: True,
)
monkeypatch.setattr("nanobot.webui.ws_http.pick_native_folder", pick_folder)
channel = _ch(bus)
response = await channel.gateway.http.dispatch(
_LOCAL,
_FakeReq(
{"Host": "127.0.0.1:8765"},
path="/api/workspaces/pick-folder",
),
)
assert response is not None
assert response.status_code == 405
assert b"authenticated WebSocket" in response.body
pick_folder.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
("connection", "host"),
[(_REMOTE, "127.0.0.1"), (_LOCAL, "0.0.0.0")],
)
async def test_workspace_folder_picker_rejects_nonlocal_surfaces(
bus: MagicMock,
monkeypatch,
connection: _FakeConn,
host: str,
) -> None:
pick_folder = AsyncMock(return_value="/tmp")
monkeypatch.setattr(
"nanobot.webui.ws_http.native_folder_picker_available",
lambda: True,
)
monkeypatch.setattr("nanobot.webui.ws_http.pick_native_folder", pick_folder)
channel = _ch(bus, host=host, token="test-token" if host == "0.0.0.0" else "")
response = await _webui_mutate(
channel,
"workspace.pick_folder",
connection=connection,
)
assert response.status_code == 403
pick_folder.assert_not_awaited()
def test_local_browser_request_requires_loopback_host_and_forwarded_origin() -> None:
from nanobot.webui.http_utils import is_local_browser_request
nanobot/webui/native_folder_picker.py
+149
View File
@@ -0,0 +1,149 @@
"""Native directory picker used by a locally hosted WebUI."""
from __future__ import annotations
import asyncio
import os
import shutil
import sys
from contextlib import suppress
from dataclasses import dataclass
from pathlib import Path
_PICKER_TIMEOUT_SECONDS = 300
class NativeFolderPickerError(RuntimeError):
"""Raised when an available native folder picker cannot complete."""
@dataclass(frozen=True)
class _PickerCommand:
argv: tuple[str, ...]
cancel_codes: frozenset[int]
cancel_markers: tuple[str, ...] = ()
def _picker_command() -> _PickerCommand | None:
if sys.platform == "darwin":
executable = shutil.which("osascript")
if executable is None:
return None
return _PickerCommand(
argv=(
executable,
"-e",
'set selectedFolder to choose folder with prompt "Select Workspace Directory"',
"-e",
"POSIX path of selectedFolder",
),
cancel_codes=frozenset({1}),
cancel_markers=("user canceled", "(-128)"),
)
if sys.platform == "win32":
executable = shutil.which("powershell.exe") or shutil.which("powershell")
if executable is None:
return None
script = (
"Add-Type -AssemblyName System.Windows.Forms;"
"$dialog=New-Object System.Windows.Forms.FolderBrowserDialog;"
"$dialog.Description='Select Workspace Directory';"
"$dialog.ShowNewFolderButton=$true;"
"if($dialog.ShowDialog() -eq [System.Windows.Forms.DialogResult]::OK){"
"[Console]::OutputEncoding=[System.Text.UTF8Encoding]::new();"
"[Console]::Out.Write($dialog.SelectedPath)}"
)
return _PickerCommand(
argv=(
executable,
"-NoProfile",
"-NonInteractive",
"-STA",
"-Command",
script,
),
cancel_codes=frozenset(),
)
if sys.platform.startswith("linux"):
if not (os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")):
return None
zenity = shutil.which("zenity")
if zenity is not None:
return _PickerCommand(
argv=(
zenity,
"--file-selection",
"--directory",
"--title=Select Workspace Directory",
),
cancel_codes=frozenset({1}),
)
kdialog = shutil.which("kdialog")
if kdialog is not None:
return _PickerCommand(
argv=(kdialog, "--getexistingdirectory", str(Path.home())),
cancel_codes=frozenset({1}),
)
return None
def native_folder_picker_available() -> bool:
"""Return whether this host can display a native directory picker."""
return _picker_command() is not None
async def _stop_process(process: asyncio.subprocess.Process) -> None:
if process.returncode is not None:
return
with suppress(ProcessLookupError):
process.terminate()
try:
await asyncio.wait_for(process.wait(), timeout=2)
except TimeoutError:
with suppress(ProcessLookupError):
process.kill()
await process.wait()
async def pick_native_folder() -> str | None:
"""Open the platform directory picker and return an existing absolute path."""
command = _picker_command()
if command is None:
raise NativeFolderPickerError("native folder picker is unavailable on this host")
process = await asyncio.create_subprocess_exec(
*command.argv,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(
process.communicate(),
timeout=_PICKER_TIMEOUT_SECONDS,
)
except asyncio.CancelledError:
await _stop_process(process)
raise
except TimeoutError as exc:
await _stop_process(process)
raise NativeFolderPickerError("native folder picker timed out") from exc
error_text = stderr.decode("utf-8", errors="replace").strip()
normalized_error = error_text.lower()
if process.returncode != 0:
if process.returncode in command.cancel_codes and (
not command.cancel_markers
or any(marker in normalized_error for marker in command.cancel_markers)
):
return None
raise NativeFolderPickerError("native folder picker failed")
selected = stdout.decode("utf-8", errors="replace").strip()
if not selected:
return None
path = Path(selected).expanduser()
if not path.is_absolute() or not path.is_dir():
raise NativeFolderPickerError("native folder picker returned an invalid directory")
return str(path)
nanobot/webui/workspaces.py
+9 -1
View File
@@ -149,6 +149,7 @@ def workspaces_payload(
default_workspace: Path,
default_restrict_to_workspace: bool,
controls_available: bool,
folder_picker_available: bool = False,
) -> dict[str, Any]:
default_access_mode = read_webui_default_access_mode()
default_scope = (
@@ -167,6 +168,7 @@ def workspaces_payload(
"controls": {
"can_change_project": controls_available,
"can_use_full_access": controls_available,
"can_pick_folder": folder_picker_available,
},
}
@@ -241,11 +243,17 @@ class WebUIWorkspaceController:
cast(object, metadata_data.get(WORKSPACE_SCOPE_METADATA_KEY))
)
def payload(self, *, controls_available: bool) -> dict[str, Any]:
def payload(
self,
*,
controls_available: bool,
folder_picker_available: bool = False,
) -> dict[str, Any]:
return workspaces_payload(
default_workspace=self._default_workspace,
default_restrict_to_workspace=self._default_restrict_to_workspace,
controls_available=controls_available,
folder_picker_available=folder_picker_available,
)
def scope_from_envelope(
nanobot/webui/ws_http.py
+45 -1
View File
@@ -62,6 +62,7 @@ from nanobot.webui.http_utils import (
from nanobot.webui.http_utils import (
is_localhost as _is_localhost,
)
from nanobot.webui.http_utils import is_loopback_host as _is_loopback_host
from nanobot.webui.http_utils import (
is_trusted_proxy_authenticated_request as _is_trusted_proxy_authenticated_request,
)
@@ -85,6 +86,11 @@ from nanobot.webui.http_utils import (
)
from nanobot.webui.ingress_policy import WebUIIngressPolicy
from nanobot.webui.media_gateway import WebUIMediaGateway
from nanobot.webui.native_folder_picker import (
NativeFolderPickerError,
native_folder_picker_available,
pick_native_folder,
)
from nanobot.webui.session_automations import (
all_automations_payload,
serialize_automation_jobs,
@@ -133,6 +139,7 @@ _WEBUI_MUTATION_PATHS = {
"skill.update": "/api/webui/skills/update",
"skill.delete": "/api/webui/skills/delete",
"sidebar.update": "/api/webui/sidebar-state/update",
"workspace.pick_folder": "/api/workspaces/pick-folder",
"settings.agent.update": "/api/settings/update",
"settings.model_configuration.create": "/api/settings/model-configurations/create",
"settings.model_configuration.update": "/api/settings/model-configurations/update",
@@ -329,6 +336,7 @@ class GatewayHTTPHandler:
)
self.skill_state_action = skill_state_action
self._skill_install_lock = asyncio.Lock()
self._folder_picker_lock = asyncio.Lock()
self.cron_service = cron_service
self.local_trigger_store = local_trigger_store
self.cron_pending_job_ids = cron_pending_job_ids
@@ -360,6 +368,17 @@ class GatewayHTTPHandler:
def workspace_controls_available(self, connection: Any) -> bool:
return self._runtime_surface == "native" or _is_localhost(connection)
def workspace_folder_picker_available(
self,
connection: Any,
request: WsRequest,
) -> bool:
return (
_is_loopback_host(self.config.host)
and _is_local_browser_request(connection, request.headers)
and native_folder_picker_available()
)
# -- Token management ---------------------------------------------------
def check_api_token(self, request: WsRequest) -> bool:
@@ -435,6 +454,7 @@ class GatewayHTTPHandler:
"/api/webui/skills/update",
"/api/webui/skills/delete",
"/api/webui/sidebar-state/update",
"/api/workspaces/pick-folder",
}
@staticmethod
@@ -1054,6 +1074,8 @@ class GatewayHTTPHandler:
return await self._handle_sessions_list(request)
if got == "/api/commands":
return self._handle_commands(request)
if got == "/api/workspaces/pick-folder":
return await self._handle_workspace_folder_picker(connection, request)
if got == "/api/workspaces":
return self._handle_workspaces(connection, request)
if got == "/api/webui/skills/search":
@@ -1089,10 +1111,32 @@ class GatewayHTTPHandler:
return _http_error(401, "Unauthorized")
return _http_json_response(
self.workspaces.payload(
controls_available=self.workspace_controls_available(connection)
controls_available=self.workspace_controls_available(connection),
folder_picker_available=self.workspace_folder_picker_available(
connection,
request,
),
)
)
async def _handle_workspace_folder_picker(
self,
connection: Any,
request: WsRequest,
) -> Response:
if not self.check_api_token(request):
return _http_error(401, "Unauthorized")
if not self.workspace_folder_picker_available(connection, request):
return _http_error(403, "native folder picker is unavailable for this connection")
if self._folder_picker_lock.locked():
return _http_error(409, "native folder picker is already open")
try:
async with self._folder_picker_lock:
path = await pick_native_folder()
except NativeFolderPickerError as exc:
return _http_error(503, str(exc))
return _http_json_response({"path": path})
def _handle_webui_skills(self, request: WsRequest) -> Response:
if not self.check_api_token(request):
return _http_error(401, "Unauthorized")