mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-07-20 23:21:06 +02:00
feat(filesystem): refactor local filesystem handling to use mounts instead of root paths, enhancing mount management and path normalization
This commit is contained in:
parent
a7a758f26e
commit
30b55a9baa
16 changed files with 421 additions and 80 deletions
|
|
@ -16,9 +16,9 @@ from app.agents.new_chat.middleware.multi_root_local_folder_backend import (
|
|||
|
||||
@lru_cache(maxsize=64)
|
||||
def _cached_multi_root_backend(
|
||||
root_paths: tuple[str, ...],
|
||||
mounts: tuple[tuple[str, str], ...],
|
||||
) -> MultiRootLocalFolderBackend:
|
||||
return MultiRootLocalFolderBackend(root_paths)
|
||||
return MultiRootLocalFolderBackend(mounts)
|
||||
|
||||
|
||||
def build_backend_resolver(
|
||||
|
|
@ -26,10 +26,13 @@ def build_backend_resolver(
|
|||
) -> Callable[[ToolRuntime], StateBackend | MultiRootLocalFolderBackend]:
|
||||
"""Create deepagents backend resolver for the selected filesystem mode."""
|
||||
|
||||
if selection.mode == FilesystemMode.DESKTOP_LOCAL_FOLDER and selection.local_root_paths:
|
||||
if selection.mode == FilesystemMode.DESKTOP_LOCAL_FOLDER and selection.local_mounts:
|
||||
|
||||
def _resolve_local(_runtime: ToolRuntime) -> MultiRootLocalFolderBackend:
|
||||
return _cached_multi_root_backend(selection.local_root_paths)
|
||||
mounts = tuple(
|
||||
(entry.mount_id, entry.root_path) for entry in selection.local_mounts
|
||||
)
|
||||
return _cached_multi_root_backend(mounts)
|
||||
|
||||
return _resolve_local
|
||||
|
||||
|
|
|
|||
|
|
@ -20,13 +20,21 @@ class ClientPlatform(StrEnum):
|
|||
DESKTOP = "desktop"
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class LocalFilesystemMount:
|
||||
"""Canonical mount mapping provided by desktop runtime."""
|
||||
|
||||
mount_id: str
|
||||
root_path: str
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class FilesystemSelection:
|
||||
"""Resolved filesystem selection for a single chat request."""
|
||||
|
||||
mode: FilesystemMode = FilesystemMode.CLOUD
|
||||
client_platform: ClientPlatform = ClientPlatform.WEB
|
||||
local_root_paths: tuple[str, ...] = ()
|
||||
local_mounts: tuple[LocalFilesystemMount, ...] = ()
|
||||
|
||||
@property
|
||||
def is_local_mode(self) -> bool:
|
||||
|
|
|
|||
|
|
@ -48,6 +48,20 @@ class FileIntentPlan(BaseModel):
|
|||
default=None,
|
||||
description="Optional filename (e.g. notes.md) inferred from user request.",
|
||||
)
|
||||
suggested_directory: str | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Optional directory path (e.g. /reports/q2 or reports/q2) inferred from "
|
||||
"user request."
|
||||
),
|
||||
)
|
||||
suggested_path: str | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Optional full file path (e.g. /reports/q2/summary.md). If present, this "
|
||||
"takes precedence over suggested_directory + suggested_filename."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _extract_text_from_message(message: BaseMessage) -> str:
|
||||
|
|
@ -88,6 +102,13 @@ def _sanitize_filename(value: str) -> str:
|
|||
return name
|
||||
|
||||
|
||||
def _sanitize_path_segment(value: str) -> str:
|
||||
segment = re.sub(r"[\\/:*?\"<>|]+", "_", value).strip()
|
||||
segment = re.sub(r"\s+", "_", segment)
|
||||
segment = segment.strip("._-")
|
||||
return segment
|
||||
|
||||
|
||||
def _infer_text_file_extension(user_text: str) -> str:
|
||||
lowered = user_text.lower()
|
||||
if any(token in lowered for token in ("json", ".json")):
|
||||
|
|
@ -119,29 +140,102 @@ def _infer_text_file_extension(user_text: str) -> str:
|
|||
return ".md"
|
||||
|
||||
|
||||
def _fallback_path(suggested_filename: str | None, *, user_text: str) -> str:
|
||||
def _normalize_directory(value: str) -> str:
|
||||
raw = value.strip().replace("\\", "/")
|
||||
raw = raw.strip("/")
|
||||
if not raw:
|
||||
return ""
|
||||
parts = [_sanitize_path_segment(part) for part in raw.split("/") if part.strip()]
|
||||
parts = [part for part in parts if part]
|
||||
return "/".join(parts)
|
||||
|
||||
|
||||
def _normalize_file_path(value: str) -> str:
|
||||
raw = value.strip().replace("\\", "/").strip()
|
||||
if not raw:
|
||||
return ""
|
||||
had_trailing_slash = raw.endswith("/")
|
||||
raw = raw.strip("/")
|
||||
if not raw:
|
||||
return ""
|
||||
parts = [_sanitize_path_segment(part) for part in raw.split("/") if part.strip()]
|
||||
parts = [part for part in parts if part]
|
||||
if not parts:
|
||||
return ""
|
||||
if had_trailing_slash:
|
||||
return f"/{'/'.join(parts)}/"
|
||||
return f"/{'/'.join(parts)}"
|
||||
|
||||
|
||||
def _infer_directory_from_user_text(user_text: str) -> str | None:
|
||||
patterns = (
|
||||
r"\b(?:in|inside|under)\s+(?:the\s+)?([a-zA-Z0-9 _\-/]+?)\s+folder\b",
|
||||
r"\b(?:in|inside|under)\s+([a-zA-Z0-9 _\-/]+?)\b",
|
||||
)
|
||||
lowered = user_text.lower()
|
||||
for pattern in patterns:
|
||||
match = re.search(pattern, lowered, flags=re.IGNORECASE)
|
||||
if not match:
|
||||
continue
|
||||
candidate = match.group(1).strip()
|
||||
if candidate in {"the", "a", "an"}:
|
||||
continue
|
||||
normalized = _normalize_directory(candidate)
|
||||
if normalized:
|
||||
return normalized
|
||||
return None
|
||||
|
||||
|
||||
def _fallback_path(
|
||||
suggested_filename: str | None,
|
||||
*,
|
||||
suggested_directory: str | None = None,
|
||||
suggested_path: str | None = None,
|
||||
user_text: str,
|
||||
) -> str:
|
||||
default_extension = _infer_text_file_extension(user_text)
|
||||
inferred_dir = _infer_directory_from_user_text(user_text)
|
||||
|
||||
sanitized_filename = ""
|
||||
if suggested_filename:
|
||||
sanitized = _sanitize_filename(suggested_filename)
|
||||
if sanitized.lower().endswith(".txt"):
|
||||
sanitized = f"{sanitized[:-4]}.md"
|
||||
if "." not in sanitized:
|
||||
sanitized = f"{sanitized}{default_extension}"
|
||||
return f"/{sanitized}"
|
||||
return f"/notes{default_extension}"
|
||||
sanitized_filename = _sanitize_filename(suggested_filename)
|
||||
if sanitized_filename.lower().endswith(".txt"):
|
||||
sanitized_filename = f"{sanitized_filename[:-4]}.md"
|
||||
if not sanitized_filename:
|
||||
sanitized_filename = f"notes{default_extension}"
|
||||
elif "." not in sanitized_filename:
|
||||
sanitized_filename = f"{sanitized_filename}{default_extension}"
|
||||
|
||||
normalized_suggested_path = (
|
||||
_normalize_file_path(suggested_path) if suggested_path else ""
|
||||
)
|
||||
if normalized_suggested_path:
|
||||
if normalized_suggested_path.endswith("/"):
|
||||
return f"{normalized_suggested_path.rstrip('/')}/{sanitized_filename}"
|
||||
return normalized_suggested_path
|
||||
|
||||
directory = _normalize_directory(suggested_directory or "")
|
||||
if not directory and inferred_dir:
|
||||
directory = inferred_dir
|
||||
if directory:
|
||||
return f"/{directory}/{sanitized_filename}"
|
||||
|
||||
return f"/{sanitized_filename}"
|
||||
|
||||
|
||||
def _build_classifier_prompt(*, recent_conversation: str, user_text: str) -> str:
|
||||
return (
|
||||
"Classify the latest user request into a filesystem intent for an AI agent.\n"
|
||||
"Return JSON only with this exact schema:\n"
|
||||
'{"intent":"chat_only|file_write|file_read","confidence":0.0,"suggested_filename":"string or null"}\n\n'
|
||||
'{"intent":"chat_only|file_write|file_read","confidence":0.0,"suggested_filename":"string or null","suggested_directory":"string or null","suggested_path":"string or null"}\n\n'
|
||||
"Rules:\n"
|
||||
"- Use semantic intent, not literal keywords.\n"
|
||||
"- file_write: user asks to create/save/write/update/edit content as a file.\n"
|
||||
"- file_read: user asks to open/read/list/search existing files.\n"
|
||||
"- chat_only: conversational/analysis responses without required file operations.\n"
|
||||
"- For file_write, choose a concise semantic suggested_filename and match the requested format.\n"
|
||||
"- If the user mentions a folder/directory, populate suggested_directory.\n"
|
||||
"- If user specifies an explicit full path, populate suggested_path.\n"
|
||||
"- Use extensions that match user intent (e.g. .md, .json, .yaml, .csv, .py, .ts, .js, .html, .css, .sql).\n"
|
||||
"- Do not use .txt; prefer .md for generic text notes.\n"
|
||||
"- Do not include dates or timestamps in suggested_filename unless explicitly requested.\n"
|
||||
|
|
@ -217,7 +311,12 @@ class FileIntentMiddleware(AgentMiddleware): # type: ignore[type-arg]
|
|||
return None
|
||||
|
||||
plan = await self._classify_intent(messages=messages, user_text=user_text)
|
||||
suggested_path = _fallback_path(plan.suggested_filename, user_text=user_text)
|
||||
suggested_path = _fallback_path(
|
||||
plan.suggested_filename,
|
||||
suggested_directory=plan.suggested_directory,
|
||||
suggested_path=plan.suggested_path,
|
||||
user_text=user_text,
|
||||
)
|
||||
contract = {
|
||||
"intent": plan.intent.value,
|
||||
"confidence": plan.confidence,
|
||||
|
|
|
|||
|
|
@ -30,19 +30,20 @@ class MultiRootLocalFolderBackend:
|
|||
where `<mount>` is derived from each selected root folder name.
|
||||
"""
|
||||
|
||||
def __init__(self, root_paths: tuple[str, ...]) -> None:
|
||||
if not root_paths:
|
||||
msg = "At least one local root path is required"
|
||||
def __init__(self, mounts: tuple[tuple[str, str], ...]) -> None:
|
||||
if not mounts:
|
||||
msg = "At least one local mount is required"
|
||||
raise ValueError(msg)
|
||||
self._mount_to_backend: dict[str, LocalFolderBackend] = {}
|
||||
for raw_root in root_paths:
|
||||
for raw_mount, raw_root in mounts:
|
||||
mount = raw_mount.strip()
|
||||
if not mount:
|
||||
msg = "Mount id cannot be empty"
|
||||
raise ValueError(msg)
|
||||
if mount in self._mount_to_backend:
|
||||
msg = f"Duplicate mount id: {mount}"
|
||||
raise ValueError(msg)
|
||||
normalized_root = str(Path(raw_root).expanduser().resolve())
|
||||
base_mount = Path(normalized_root).name or "root"
|
||||
mount = base_mount
|
||||
suffix = 2
|
||||
while mount in self._mount_to_backend:
|
||||
mount = f"{base_mount}-{suffix}"
|
||||
suffix += 1
|
||||
self._mount_to_backend[mount] = LocalFolderBackend(normalized_root)
|
||||
self._mount_order = tuple(self._mount_to_backend.keys())
|
||||
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ from sqlalchemy.orm import selectinload
|
|||
|
||||
from app.agents.new_chat.filesystem_selection import (
|
||||
ClientPlatform,
|
||||
LocalFilesystemMount,
|
||||
FilesystemMode,
|
||||
FilesystemSelection,
|
||||
)
|
||||
|
|
@ -42,6 +43,7 @@ from app.db import (
|
|||
)
|
||||
from app.schemas.new_chat import (
|
||||
AgentToolInfo,
|
||||
LocalFilesystemMountPayload,
|
||||
NewChatMessageRead,
|
||||
NewChatRequest,
|
||||
NewChatThreadCreate,
|
||||
|
|
@ -73,7 +75,7 @@ def _resolve_filesystem_selection(
|
|||
*,
|
||||
mode: str,
|
||||
client_platform: str,
|
||||
local_roots: list[str] | None,
|
||||
local_mounts: list[LocalFilesystemMountPayload] | None,
|
||||
) -> FilesystemSelection:
|
||||
"""Validate and normalize filesystem mode settings from request payload."""
|
||||
try:
|
||||
|
|
@ -96,29 +98,37 @@ def _resolve_filesystem_selection(
|
|||
status_code=400,
|
||||
detail="desktop_local_folder mode is only available on desktop runtime.",
|
||||
)
|
||||
normalized_roots: list[str] = []
|
||||
for root in local_roots or []:
|
||||
trimmed = root.strip()
|
||||
if trimmed and trimmed not in normalized_roots:
|
||||
normalized_roots.append(trimmed)
|
||||
if not normalized_roots:
|
||||
normalized_mounts: list[tuple[str, str]] = []
|
||||
seen_mounts: set[str] = set()
|
||||
for mount in local_mounts or []:
|
||||
mount_id = mount.mount_id.strip()
|
||||
root_path = mount.root_path.strip()
|
||||
if not mount_id or not root_path:
|
||||
continue
|
||||
if mount_id in seen_mounts:
|
||||
continue
|
||||
seen_mounts.add(mount_id)
|
||||
normalized_mounts.append((mount_id, root_path))
|
||||
if not normalized_mounts:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=(
|
||||
"local_filesystem_roots must include at least one root for "
|
||||
"local_filesystem_mounts must include at least one mount for "
|
||||
"desktop_local_folder mode."
|
||||
),
|
||||
)
|
||||
return FilesystemSelection(
|
||||
mode=resolved_mode,
|
||||
client_platform=resolved_platform,
|
||||
local_root_paths=tuple(normalized_roots),
|
||||
local_mounts=tuple(
|
||||
LocalFilesystemMount(mount_id=mount_id, root_path=root_path)
|
||||
for mount_id, root_path in normalized_mounts
|
||||
),
|
||||
)
|
||||
|
||||
return FilesystemSelection(
|
||||
mode=FilesystemMode.CLOUD,
|
||||
client_platform=resolved_platform,
|
||||
local_root_paths=(),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -1196,7 +1206,7 @@ async def handle_new_chat(
|
|||
filesystem_selection = _resolve_filesystem_selection(
|
||||
mode=request.filesystem_mode,
|
||||
client_platform=request.client_platform,
|
||||
local_roots=request.local_filesystem_roots,
|
||||
local_mounts=request.local_filesystem_mounts,
|
||||
)
|
||||
|
||||
# Get search space to check LLM config preferences
|
||||
|
|
@ -1318,7 +1328,7 @@ async def regenerate_response(
|
|||
filesystem_selection = _resolve_filesystem_selection(
|
||||
mode=request.filesystem_mode,
|
||||
client_platform=request.client_platform,
|
||||
local_roots=request.local_filesystem_roots,
|
||||
local_mounts=request.local_filesystem_mounts,
|
||||
)
|
||||
|
||||
# Get the checkpointer and state history
|
||||
|
|
@ -1577,7 +1587,7 @@ async def resume_chat(
|
|||
filesystem_selection = _resolve_filesystem_selection(
|
||||
mode=request.filesystem_mode,
|
||||
client_platform=request.client_platform,
|
||||
local_roots=request.local_filesystem_roots,
|
||||
local_mounts=request.local_filesystem_mounts,
|
||||
)
|
||||
|
||||
search_space_result = await session.execute(
|
||||
|
|
|
|||
|
|
@ -168,6 +168,11 @@ class ChatMessage(BaseModel):
|
|||
content: str
|
||||
|
||||
|
||||
class LocalFilesystemMountPayload(BaseModel):
|
||||
mount_id: str
|
||||
root_path: str
|
||||
|
||||
|
||||
class NewChatRequest(BaseModel):
|
||||
"""Request schema for the deep agent chat endpoint."""
|
||||
|
||||
|
|
@ -186,7 +191,7 @@ class NewChatRequest(BaseModel):
|
|||
)
|
||||
filesystem_mode: Literal["cloud", "desktop_local_folder"] = "cloud"
|
||||
client_platform: Literal["web", "desktop"] = "web"
|
||||
local_filesystem_roots: list[str] | None = None
|
||||
local_filesystem_mounts: list[LocalFilesystemMountPayload] | None = None
|
||||
|
||||
|
||||
class RegenerateRequest(BaseModel):
|
||||
|
|
@ -209,7 +214,7 @@ class RegenerateRequest(BaseModel):
|
|||
disabled_tools: list[str] | None = None
|
||||
filesystem_mode: Literal["cloud", "desktop_local_folder"] = "cloud"
|
||||
client_platform: Literal["web", "desktop"] = "web"
|
||||
local_filesystem_roots: list[str] | None = None
|
||||
local_filesystem_mounts: list[LocalFilesystemMountPayload] | None = None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
|
|
@ -235,7 +240,7 @@ class ResumeRequest(BaseModel):
|
|||
decisions: list[ResumeDecision]
|
||||
filesystem_mode: Literal["cloud", "desktop_local_folder"] = "cloud"
|
||||
client_platform: Literal["web", "desktop"] = "web"
|
||||
local_filesystem_roots: list[str] | None = None
|
||||
local_filesystem_mounts: list[LocalFilesystemMountPayload] | None = None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
|
|
|
|||
|
|
@ -190,6 +190,23 @@ def _tool_output_has_error(tool_output: Any) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def _extract_resolved_file_path(*, tool_name: str, tool_output: Any) -> str | None:
|
||||
if isinstance(tool_output, dict):
|
||||
path_value = tool_output.get("path")
|
||||
if isinstance(path_value, str) and path_value.strip():
|
||||
return path_value.strip()
|
||||
text = _tool_output_to_text(tool_output)
|
||||
if tool_name == "write_file":
|
||||
match = re.search(r"Updated file\s+(.+)$", text.strip())
|
||||
if match:
|
||||
return match.group(1).strip()
|
||||
if tool_name == "edit_file":
|
||||
match = re.search(r"in '([^']+)'", text)
|
||||
if match:
|
||||
return match.group(1).strip()
|
||||
return None
|
||||
|
||||
|
||||
def _contract_enforcement_active(result: StreamResult) -> bool:
|
||||
# Keep policy deterministic with no env-driven progression modes:
|
||||
# enforce the file-operation contract only in desktop local-folder mode.
|
||||
|
|
@ -1016,6 +1033,30 @@ async def _stream_agent_events(
|
|||
f"Scrape failed: {error_msg}",
|
||||
"error",
|
||||
)
|
||||
elif tool_name in ("write_file", "edit_file"):
|
||||
resolved_path = _extract_resolved_file_path(
|
||||
tool_name=tool_name,
|
||||
tool_output=tool_output,
|
||||
)
|
||||
result_text = _tool_output_to_text(tool_output)
|
||||
if _tool_output_has_error(tool_output):
|
||||
yield streaming_service.format_tool_output_available(
|
||||
tool_call_id,
|
||||
{
|
||||
"status": "error",
|
||||
"error": result_text,
|
||||
"path": resolved_path,
|
||||
},
|
||||
)
|
||||
else:
|
||||
yield streaming_service.format_tool_output_available(
|
||||
tool_call_id,
|
||||
{
|
||||
"status": "completed",
|
||||
"path": resolved_path,
|
||||
"result": result_text,
|
||||
},
|
||||
)
|
||||
elif tool_name == "generate_report":
|
||||
# Stream the full report result so frontend can render the ReportCard
|
||||
yield streaming_service.format_tool_output_available(
|
||||
|
|
|
|||
|
|
@ -114,3 +114,60 @@ async def test_file_write_txt_suggestion_is_normalized_to_markdown():
|
|||
assert contract["intent"] == FileOperationIntent.FILE_WRITE.value
|
||||
assert contract["suggested_path"] == "/random.md"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file_write_with_suggested_directory_preserves_folder():
|
||||
llm = _FakeLLM(
|
||||
'{"intent":"file_write","confidence":0.86,"suggested_filename":"random.md","suggested_directory":"pc backups","suggested_path":null}'
|
||||
)
|
||||
middleware = FileIntentMiddleware(llm=llm)
|
||||
state = {
|
||||
"messages": [HumanMessage(content="create a random file in pc backups folder")],
|
||||
"turn_id": "turn:4",
|
||||
}
|
||||
|
||||
result = await middleware.abefore_agent(state, runtime=None) # type: ignore[arg-type]
|
||||
|
||||
assert result is not None
|
||||
contract = result["file_operation_contract"]
|
||||
assert contract["intent"] == FileOperationIntent.FILE_WRITE.value
|
||||
assert contract["suggested_path"] == "/pc_backups/random.md"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file_write_with_suggested_path_takes_precedence():
|
||||
llm = _FakeLLM(
|
||||
'{"intent":"file_write","confidence":0.9,"suggested_filename":"ignored.md","suggested_directory":"docs","suggested_path":"/reports/q2/summary.md"}'
|
||||
)
|
||||
middleware = FileIntentMiddleware(llm=llm)
|
||||
state = {
|
||||
"messages": [HumanMessage(content="create report")],
|
||||
"turn_id": "turn:5",
|
||||
}
|
||||
|
||||
result = await middleware.abefore_agent(state, runtime=None) # type: ignore[arg-type]
|
||||
|
||||
assert result is not None
|
||||
contract = result["file_operation_contract"]
|
||||
assert contract["intent"] == FileOperationIntent.FILE_WRITE.value
|
||||
assert contract["suggested_path"] == "/reports/q2/summary.md"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file_write_infers_directory_from_user_text_when_missing():
|
||||
llm = _FakeLLM(
|
||||
'{"intent":"file_write","confidence":0.83,"suggested_filename":"random.md","suggested_directory":null,"suggested_path":null}'
|
||||
)
|
||||
middleware = FileIntentMiddleware(llm=llm)
|
||||
state = {
|
||||
"messages": [HumanMessage(content="create a random file in pc backups folder")],
|
||||
"turn_id": "turn:6",
|
||||
}
|
||||
|
||||
result = await middleware.abefore_agent(state, runtime=None) # type: ignore[arg-type]
|
||||
|
||||
assert result is not None
|
||||
contract = result["file_operation_contract"]
|
||||
assert contract["intent"] == FileOperationIntent.FILE_WRITE.value
|
||||
assert contract["suggested_path"] == "/pc_backups/random.md"
|
||||
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ from app.agents.new_chat.filesystem_selection import (
|
|||
ClientPlatform,
|
||||
FilesystemMode,
|
||||
FilesystemSelection,
|
||||
LocalFilesystemMount,
|
||||
)
|
||||
from app.agents.new_chat.middleware.multi_root_local_folder_backend import (
|
||||
MultiRootLocalFolderBackend,
|
||||
|
|
@ -23,7 +24,7 @@ def test_backend_resolver_returns_multi_root_backend_for_single_root(tmp_path: P
|
|||
selection = FilesystemSelection(
|
||||
mode=FilesystemMode.DESKTOP_LOCAL_FOLDER,
|
||||
client_platform=ClientPlatform.DESKTOP,
|
||||
local_root_paths=(str(tmp_path),),
|
||||
local_mounts=(LocalFilesystemMount(mount_id="tmp", root_path=str(tmp_path)),),
|
||||
)
|
||||
resolver = build_backend_resolver(selection)
|
||||
|
||||
|
|
@ -47,7 +48,10 @@ def test_backend_resolver_returns_multi_root_backend_for_multiple_roots(tmp_path
|
|||
selection = FilesystemSelection(
|
||||
mode=FilesystemMode.DESKTOP_LOCAL_FOLDER,
|
||||
client_platform=ClientPlatform.DESKTOP,
|
||||
local_root_paths=(str(root_one), str(root_two)),
|
||||
local_mounts=(
|
||||
LocalFilesystemMount(mount_id="resume", root_path=str(root_one)),
|
||||
LocalFilesystemMount(mount_id="notes", root_path=str(root_two)),
|
||||
),
|
||||
)
|
||||
resolver = build_backend_resolver(selection)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,5 +1,11 @@
|
|||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.agents.new_chat.middleware.multi_root_local_folder_backend import (
|
||||
MultiRootLocalFolderBackend,
|
||||
)
|
||||
from app.agents.new_chat.filesystem_selection import FilesystemMode
|
||||
from app.agents.new_chat.middleware.filesystem import SurfSenseFilesystemMiddleware
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
|
@ -43,9 +49,9 @@ def test_verify_written_content_prefers_raw_sync() -> None:
|
|||
|
||||
|
||||
def test_contract_suggested_path_falls_back_to_notes_md() -> None:
|
||||
suggested = SurfSenseFilesystemMiddleware._get_contract_suggested_path(
|
||||
_RuntimeNoSuggestedPath()
|
||||
)
|
||||
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
|
||||
middleware._filesystem_mode = FilesystemMode.CLOUD
|
||||
suggested = middleware._get_contract_suggested_path(_RuntimeNoSuggestedPath()) # type: ignore[arg-type]
|
||||
assert suggested == "/notes.md"
|
||||
|
||||
|
||||
|
|
@ -62,3 +68,32 @@ async def test_verify_written_content_prefers_raw_async() -> None:
|
|||
)
|
||||
|
||||
assert verify_error is None
|
||||
|
||||
|
||||
def test_normalize_local_mount_path_prefixes_default_mount(tmp_path: Path) -> None:
|
||||
root = tmp_path / "PC Backups"
|
||||
root.mkdir()
|
||||
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
|
||||
runtime = _RuntimeNoSuggestedPath()
|
||||
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
|
||||
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
|
||||
|
||||
resolved = middleware._normalize_local_mount_path("/random-note.md", runtime) # type: ignore[arg-type]
|
||||
|
||||
assert resolved == "/pc_backups/random-note.md"
|
||||
|
||||
|
||||
def test_normalize_local_mount_path_keeps_explicit_mount(tmp_path: Path) -> None:
|
||||
root = tmp_path / "PC Backups"
|
||||
root.mkdir()
|
||||
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
|
||||
runtime = _RuntimeNoSuggestedPath()
|
||||
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
|
||||
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
|
||||
|
||||
resolved = middleware._normalize_local_mount_path( # type: ignore[arg-type]
|
||||
"/pc_backups/notes/random-note.md",
|
||||
runtime,
|
||||
)
|
||||
|
||||
assert resolved == "/pc_backups/notes/random-note.md"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,28 @@
|
|||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.agents.new_chat.middleware.multi_root_local_folder_backend import (
|
||||
MultiRootLocalFolderBackend,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
|
||||
def test_mount_ids_preserve_client_mapping_order(tmp_path: Path) -> None:
|
||||
root_one = tmp_path / "PC Backups"
|
||||
root_two = tmp_path / "pc_backups"
|
||||
root_three = tmp_path / "notes@2026"
|
||||
root_one.mkdir()
|
||||
root_two.mkdir()
|
||||
root_three.mkdir()
|
||||
|
||||
backend = MultiRootLocalFolderBackend(
|
||||
(
|
||||
("pc_backups", str(root_one)),
|
||||
("pc_backups_2", str(root_two)),
|
||||
("notes_2026", str(root_three)),
|
||||
)
|
||||
)
|
||||
|
||||
assert backend.list_mounts() == ("pc_backups", "pc_backups_2", "notes_2026")
|
||||
Loading…
Add table
Add a link
Reference in a new issue