Merge pull request #1299 from AnishSarkar22/feat/swappable-filesystem

feat: introduce swappable filesystem on desktop & monaco editor to edit local files

surfsense_backend/tests/unit/middleware/test_file_intent_middleware.py

@@ -0,0 +1,214 @@
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+
+from app.agents.new_chat.middleware.file_intent import (
+    FileIntentMiddleware,
+    FileOperationIntent,
+    _fallback_path,
+)
+
+pytestmark = pytest.mark.unit
+
+
+class _FakeLLM:
+    def __init__(self, response_text: str):
+        self._response_text = response_text
+
+    async def ainvoke(self, *_args, **_kwargs):
+        return AIMessage(content=self._response_text)
+
+
+@pytest.mark.asyncio
+async def test_file_write_intent_injects_contract_message():
+    llm = _FakeLLM(
+        '{"intent":"file_write","confidence":0.93,"suggested_filename":"ideas.md"}'
+    )
+    middleware = FileIntentMiddleware(llm=llm)
+    state = {
+        "messages": [HumanMessage(content="Create another random note for me")],
+        "turn_id": "123:456",
+    }
+
+    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"] == "/ideas.md"
+    assert contract["turn_id"] == "123:456"
+    assert any(
+        "file_operation_contract" in str(msg.content)
+        for msg in result["messages"]
+        if hasattr(msg, "content")
+    )
+
+
+@pytest.mark.asyncio
+async def test_non_write_intent_does_not_inject_contract_message():
+    llm = _FakeLLM(
+        '{"intent":"file_read","confidence":0.88,"suggested_filename":null}'
+    )
+    middleware = FileIntentMiddleware(llm=llm)
+    original_messages = [HumanMessage(content="Read /notes.md")]
+    state = {"messages": original_messages, "turn_id": "abc:def"}
+
+    result = await middleware.abefore_agent(state, runtime=None)  # type: ignore[arg-type]
+
+    assert result is not None
+    assert result["file_operation_contract"]["intent"] == FileOperationIntent.FILE_READ.value
+    assert "messages" not in result
+
+
+@pytest.mark.asyncio
+async def test_file_write_null_filename_uses_semantic_default_path():
+    llm = _FakeLLM(
+        '{"intent":"file_write","confidence":0.74,"suggested_filename":null}'
+    )
+    middleware = FileIntentMiddleware(llm=llm)
+    state = {
+        "messages": [HumanMessage(content="create a random markdown file")],
+        "turn_id": "turn:1",
+    }
+
+    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"] == "/notes.md"
+
+
+@pytest.mark.asyncio
+async def test_file_write_null_filename_infers_json_extension():
+    llm = _FakeLLM(
+        '{"intent":"file_write","confidence":0.71,"suggested_filename":null}'
+    )
+    middleware = FileIntentMiddleware(llm=llm)
+    state = {
+        "messages": [HumanMessage(content="create a sample json config file")],
+        "turn_id": "turn:2",
+    }
+
+    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"] == "/notes.json"
+
+
+@pytest.mark.asyncio
+async def test_file_write_txt_suggestion_is_normalized_to_markdown():
+    llm = _FakeLLM(
+        '{"intent":"file_write","confidence":0.82,"suggested_filename":"random.txt"}'
+    )
+    middleware = FileIntentMiddleware(llm=llm)
+    state = {
+        "messages": [HumanMessage(content="create a random file")],
+        "turn_id": "turn:3",
+    }
+
+    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"] == "/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"
+
+
+def test_fallback_path_normalizes_windows_slashes() -> None:
+    resolved = _fallback_path(
+        suggested_filename="summary.md",
+        suggested_path=r"\reports\q2\summary.md",
+        user_text="create report",
+    )
+
+    assert resolved == "/reports/q2/summary.md"
+
+
+def test_fallback_path_normalizes_windows_drive_path() -> None:
+    resolved = _fallback_path(
+        suggested_filename=None,
+        suggested_path=r"C:\Users\anish\notes\todo.md",
+        user_text="create note",
+    )
+
+    assert resolved == "/C/Users/anish/notes/todo.md"
+
+
+def test_fallback_path_normalizes_mixed_separators_and_duplicate_slashes() -> None:
+    resolved = _fallback_path(
+        suggested_filename="summary.md",
+        suggested_path=r"\\reports\\q2//summary.md",
+        user_text="create report",
+    )
+
+    assert resolved == "/reports/q2/summary.md"
+
+
+def test_fallback_path_keeps_posix_style_absolute_path_for_linux_and_macos() -> None:
+    resolved = _fallback_path(
+        suggested_filename=None,
+        suggested_path="/var/log/surfsense/notes.md",
+        user_text="create note",
+    )
+
+    assert resolved == "/var/log/surfsense/notes.md"
+

surfsense_backend/tests/unit/middleware/test_filesystem_backends.py

@@ -0,0 +1,59 @@
+from pathlib import Path
+
+import pytest
+
+from app.agents.new_chat.filesystem_backends import build_backend_resolver
+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,
+)
+
+pytestmark = pytest.mark.unit
+
+
+class _RuntimeStub:
+    state = {"files": {}}
+
+
+def test_backend_resolver_returns_multi_root_backend_for_single_root(tmp_path: Path):
+    selection = FilesystemSelection(
+        mode=FilesystemMode.DESKTOP_LOCAL_FOLDER,
+        client_platform=ClientPlatform.DESKTOP,
+        local_mounts=(LocalFilesystemMount(mount_id="tmp", root_path=str(tmp_path)),),
+    )
+    resolver = build_backend_resolver(selection)
+
+    backend = resolver(_RuntimeStub())
+    assert isinstance(backend, MultiRootLocalFolderBackend)
+
+
+def test_backend_resolver_uses_cloud_mode_by_default():
+    resolver = build_backend_resolver(FilesystemSelection())
+    backend = resolver(_RuntimeStub())
+    # StateBackend class name check keeps this test decoupled
+    # from internal deepagents runtime class identity.
+    assert backend.__class__.__name__ == "StateBackend"
+
+
+def test_backend_resolver_returns_multi_root_backend_for_multiple_roots(tmp_path: Path):
+    root_one = tmp_path / "resume"
+    root_two = tmp_path / "notes"
+    root_one.mkdir()
+    root_two.mkdir()
+    selection = FilesystemSelection(
+        mode=FilesystemMode.DESKTOP_LOCAL_FOLDER,
+        client_platform=ClientPlatform.DESKTOP,
+        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)
+
+    backend = resolver(_RuntimeStub())
+    assert isinstance(backend, MultiRootLocalFolderBackend)

surfsense_backend/tests/unit/middleware/test_filesystem_verification.py

@@ -0,0 +1,164 @@
+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
+
+
+class _BackendWithRawRead:
+    def __init__(self, content: str) -> None:
+        self._content = content
+
+    def read(self, file_path: str, offset: int = 0, limit: int = 200000) -> str:
+        del file_path, offset, limit
+        return "     1\tline1\n     2\tline2"
+
+    async def aread(self, file_path: str, offset: int = 0, limit: int = 200000) -> str:
+        return self.read(file_path, offset, limit)
+
+    def read_raw(self, file_path: str) -> str:
+        del file_path
+        return self._content
+
+    async def aread_raw(self, file_path: str) -> str:
+        return self.read_raw(file_path)
+
+
+class _RuntimeNoSuggestedPath:
+    state = {"file_operation_contract": {}}
+
+
+def test_verify_written_content_prefers_raw_sync() -> None:
+    middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
+    expected = "line1\nline2"
+    backend = _BackendWithRawRead(expected)
+
+    verify_error = middleware._verify_written_content_sync(
+        backend=backend,
+        path="/note.md",
+        expected_content=expected,
+    )
+
+    assert verify_error is None
+
+
+def test_contract_suggested_path_falls_back_to_notes_md() -> None:
+    middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
+    middleware._filesystem_mode = FilesystemMode.CLOUD
+    suggested = middleware._get_contract_suggested_path(_RuntimeNoSuggestedPath())  # type: ignore[arg-type]
+    assert suggested == "/notes.md"
+
+
+@pytest.mark.asyncio
+async def test_verify_written_content_prefers_raw_async() -> None:
+    middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
+    expected = "line1\nline2"
+    backend = _BackendWithRawRead(expected)
+
+    verify_error = await middleware._verify_written_content_async(
+        backend=backend,
+        path="/note.md",
+        expected_content=expected,
+    )
+
+    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"
+
+
+def test_normalize_local_mount_path_windows_backslashes(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]
+        r"\notes\random-note.md",
+        runtime,
+    )
+
+    assert resolved == "/pc_backups/notes/random-note.md"
+
+
+def test_normalize_local_mount_path_normalizes_mixed_separators(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]
+        r"\\notes//nested\\random-note.md",
+        runtime,
+    )
+
+    assert resolved == "/pc_backups/notes/nested/random-note.md"
+
+
+def test_normalize_local_mount_path_keeps_explicit_mount_with_backslashes(
+    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]
+        r"\pc_backups\notes\random-note.md",
+        runtime,
+    )
+
+    assert resolved == "/pc_backups/notes/random-note.md"
+
+
+def test_normalize_local_mount_path_prefixes_posix_absolute_path_for_linux_and_macos(
+    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("/var/log/app.log", runtime)  # type: ignore[arg-type]
+
+    assert resolved == "/pc_backups/var/log/app.log"

surfsense_backend/tests/unit/middleware/test_local_folder_backend.py

@@ -0,0 +1,59 @@
+from pathlib import Path
+
+import pytest
+
+from app.agents.new_chat.middleware.local_folder_backend import LocalFolderBackend
+
+pytestmark = pytest.mark.unit
+
+
+def test_local_backend_write_read_edit_roundtrip(tmp_path: Path):
+    backend = LocalFolderBackend(str(tmp_path))
+
+    write = backend.write("/notes/test.md", "line1\nline2")
+    assert write.error is None
+    assert write.path == "/notes/test.md"
+
+    read = backend.read("/notes/test.md", offset=0, limit=20)
+    assert "line1" in read
+    assert "line2" in read
+
+    edit = backend.edit("/notes/test.md", "line2", "updated")
+    assert edit.error is None
+    assert edit.occurrences == 1
+
+    read_after = backend.read("/notes/test.md", offset=0, limit=20)
+    assert "updated" in read_after
+
+
+def test_local_backend_blocks_path_escape(tmp_path: Path):
+    backend = LocalFolderBackend(str(tmp_path))
+
+    result = backend.write("/../../etc/passwd", "bad")
+    assert result.error is not None
+    assert "Invalid path" in result.error
+
+
+def test_local_backend_glob_and_grep(tmp_path: Path):
+    backend = LocalFolderBackend(str(tmp_path))
+    (tmp_path / "docs").mkdir()
+    (tmp_path / "docs" / "a.txt").write_text("hello world\n")
+    (tmp_path / "docs" / "b.md").write_text("hello markdown\n")
+
+    infos = backend.glob_info("**/*.txt", "/docs")
+    paths = {info["path"] for info in infos}
+    assert "/docs/a.txt" in paths
+
+    grep = backend.grep_raw("hello", "/docs", "*.md")
+    assert isinstance(grep, list)
+    assert any(match["path"] == "/docs/b.md" for match in grep)
+
+
+def test_local_backend_read_raw_returns_exact_content(tmp_path: Path):
+    backend = LocalFolderBackend(str(tmp_path))
+    expected = "# Title\n\nline 1\nline 2\n"
+    write = backend.write("/notes/raw.md", expected)
+    assert write.error is None
+
+    raw = backend.read_raw("/notes/raw.md")
+    assert raw == expected

surfsense_backend/tests/unit/middleware/test_multi_root_local_folder_backend.py

@@ -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")

surfsense_backend/tests/unit/test_stream_new_chat_contract.py

@@ -0,0 +1,48 @@
+import pytest
+
+from app.tasks.chat.stream_new_chat import (
+    StreamResult,
+    _contract_enforcement_active,
+    _evaluate_file_contract_outcome,
+    _tool_output_has_error,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def test_tool_output_error_detection():
+    assert _tool_output_has_error("Error: failed to write file")
+    assert _tool_output_has_error({"error": "boom"})
+    assert _tool_output_has_error({"result": "Error: disk is full"})
+    assert not _tool_output_has_error({"result": "Updated file /notes.md"})
+
+
+def test_file_write_contract_outcome_reasons():
+    result = StreamResult(intent_detected="file_write")
+    passed, reason = _evaluate_file_contract_outcome(result)
+    assert not passed
+    assert reason == "no_write_attempt"
+
+    result.write_attempted = True
+    passed, reason = _evaluate_file_contract_outcome(result)
+    assert not passed
+    assert reason == "write_failed"
+
+    result.write_succeeded = True
+    passed, reason = _evaluate_file_contract_outcome(result)
+    assert not passed
+    assert reason == "verification_failed"
+
+    result.verification_succeeded = True
+    passed, reason = _evaluate_file_contract_outcome(result)
+    assert passed
+    assert reason == ""
+
+
+def test_contract_enforcement_local_only():
+    result = StreamResult(filesystem_mode="desktop_local_folder")
+    assert _contract_enforcement_active(result)
+
+    result.filesystem_mode = "cloud"
+    assert not _contract_enforcement_active(result)
+