diff --git a/marimo/_server/api/endpoints/assets.py b/marimo/_server/api/endpoints/assets.py index ad24df8db65..ed2640e7d60 100644 --- a/marimo/_server/api/endpoints/assets.py +++ b/marimo/_server/api/endpoints/assets.py @@ -144,6 +144,15 @@ async def index(request: Request) -> HTMLResponse: app_manager = app_state.session_manager.app_manager(file_key) app_config = app_manager.app.config + # Make filename relative to file router's directory if possible + filename = app_manager.filename + directory = app_state.session_manager.file_router.directory + if filename and directory: + try: + filename = str(Path(filename).relative_to(directory)) + except ValueError: + pass # Keep absolute if not under directory + html = notebook_page_template( html=html, base_url=app_state.base_url, @@ -151,7 +160,7 @@ async def index(request: Request) -> HTMLResponse: config_overrides=config_manager.get_config_overrides(), server_token=app_state.skew_protection_token, app_config=app_config, - filename=app_manager.filename, + filename=filename, mode=app_state.mode, runtime_config=[{"url": app_state.remote_url}] if app_state.remote_url diff --git a/marimo/_server/api/endpoints/file_explorer.py b/marimo/_server/api/endpoints/file_explorer.py index ac51db862c4..7c7f0a67897 100644 --- a/marimo/_server/api/endpoints/file_explorer.py +++ b/marimo/_server/api/endpoints/file_explorer.py @@ -65,8 +65,11 @@ async def list_files( schema: $ref: "#/components/schemas/FileListResponse" """ + app_state = AppState(request) body = await parse_request(request, cls=FileListRequest) - root = body.path or file_system.get_root() + # Use file router's directory as default, fall back to cwd + directory = app_state.session_manager.file_router.directory + root = body.path or directory or file_system.get_root() files = file_system.list_files(root) return FileListResponse(files=files, root=root) diff --git a/marimo/_server/api/endpoints/files.py b/marimo/_server/api/endpoints/files.py index c9044a026cf..d7080670dba 100644 --- a/marimo/_server/api/endpoints/files.py +++ b/marimo/_server/api/endpoints/files.py @@ -1,6 +1,7 @@ # Copyright 2026 Marimo. All rights reserved. from __future__ import annotations +from pathlib import Path from typing import TYPE_CHECKING from starlette.authentication import requires @@ -109,9 +110,15 @@ async def rename_file( """ body = await parse_request(request, cls=RenameNotebookRequest) app_state = AppState(request) + + # Resolve relative filenames against the file router's directory + if not Path(body.filename).is_absolute(): + directory = app_state.session_manager.file_router.directory + if directory: + body.filename = str(Path(directory) / body.filename) + filename = await abspath(body.filename) - # Convert to absolute path app_state.require_current_session().put_control_request( RenameNotebookCommand(filename=filename), from_consumer_id=ConsumerId(app_state.require_current_session_id()), @@ -152,6 +159,13 @@ async def save( """ app_state = AppState(request) body = await parse_request(request, cls=SaveNotebookRequest) + + # Resolve relative filenames against the file router's directory + if body.filename and not Path(body.filename).is_absolute(): + directory = app_state.session_manager.file_router.directory + if directory: + body.filename = str(Path(directory) / body.filename) + session = app_state.require_current_session() contents = session.app_file_manager.save(body) @@ -186,6 +200,13 @@ async def copy( """ app_state = AppState(request) body = await parse_request(request, cls=CopyNotebookRequest) + + # Resolve relative filenames against the file router's directory + if body.destination and not Path(body.destination).is_absolute(): + directory = app_state.session_manager.file_router.directory + if directory: + body.destination = str(Path(directory) / body.destination) + session = app_state.require_current_session() contents = session.app_file_manager.copy(body) diff --git a/marimo/_server/api/endpoints/home.py b/marimo/_server/api/endpoints/home.py index a54ac3f1c0f..807d99e7dd6 100644 --- a/marimo/_server/api/endpoints/home.py +++ b/marimo/_server/api/endpoints/home.py @@ -3,6 +3,7 @@ import asyncio import os +import pathlib import tempfile from typing import TYPE_CHECKING @@ -58,7 +59,12 @@ async def read_code( $ref: "#/components/schemas/RecentFilesResponse" """ app_state = AppState(request) - files = app_state.session_manager.recents.get_recents() + # Pass the file router's directory to filter and relativize paths + directory = None + dir_str = app_state.session_manager.file_router.directory + if dir_str: + directory = pathlib.Path(dir_str) + files = app_state.session_manager.recents.get_recents(directory) return RecentFilesResponse(files=files) diff --git a/marimo/_server/file_router.py b/marimo/_server/file_router.py index 4c299dc596c..3410d9d20ed 100644 --- a/marimo/_server/file_router.py +++ b/marimo/_server/file_router.py @@ -167,15 +167,16 @@ def maybe_get_single_file(self) -> Optional[MarimoFile]: class LazyListOfFilesAppFileRouter(AppFileRouter): def __init__(self, directory: str, include_markdown: bool) -> None: - # pass through Path to canonicalize, strips trailing slashes - self._directory = str(Path(directory)) + # Make directory absolute but don't resolve symlinks to preserve user paths + abs_directory = Path(directory).absolute() + self._directory = str(abs_directory) self.include_markdown = include_markdown self._lazy_files: Optional[list[FileInfo]] = None # Use PathValidator for security validation - self._validator = PathValidator(Path(directory)) - # Use DirectoryScanner for file discovery - self._scanner = DirectoryScanner(directory, include_markdown) + self._validator = PathValidator(abs_directory) + # Use DirectoryScanner for file discovery (use absolute path) + self._scanner = DirectoryScanner(str(abs_directory), include_markdown) @property def directory(self) -> str: @@ -233,23 +234,22 @@ def get_file_manager( directory = Path(self._directory) filepath = Path(key) + # Resolve filepath for use + # If filepath is relative, resolve it relative to directory + if not filepath.is_absolute(): + filepath = directory / filepath + # Use PathValidator for security validation # Check if file is in an allowed temp directory (e.g., for tutorials) # If so, skip the directory validation is_in_allowed_temp_dir = self._validator.is_file_in_allowed_temp_dir( - key + str(filepath) ) if not is_in_allowed_temp_dir: # Validate that filepath is inside directory self._validator.validate_inside_directory(directory, filepath) - # Resolve filepath for use - # If directory is absolute and filepath is relative, resolve relative to directory - if directory.is_absolute() and not filepath.is_absolute(): - filepath = directory / filepath - # Note: We don't call resolve() here to preserve the original path format - if filepath.exists(): return AppFileManager(str(filepath), defaults=defaults) diff --git a/marimo/_server/files/directory_scanner.py b/marimo/_server/files/directory_scanner.py index 66df3453b47..1ea9b36b861 100644 --- a/marimo/_server/files/directory_scanner.py +++ b/marimo/_server/files/directory_scanner.py @@ -3,6 +3,7 @@ import os import time +from pathlib import Path from typing import Optional from marimo import _loggers @@ -183,10 +184,14 @@ def recurse( continue children = recurse(entry.path, depth + 1) if children: + entry_path = Path(entry.path) + relative_path = str( + entry_path.relative_to(self.directory) + ) folders.append( FileInfo( - id=entry.path, - path=entry.path, + id=relative_path, + path=relative_path, name=entry.name, is_directory=True, is_marimo_file=False, @@ -196,9 +201,13 @@ def recurse( elif entry.name.endswith(self.allowed_extensions): if is_marimo_app(entry.path): file_count[0] += 1 + entry_path = Path(entry.path) + relative_path = str( + entry_path.relative_to(self.directory) + ) file_info = FileInfo( - id=entry.path, - path=entry.path, + id=relative_path, + path=relative_path, name=entry.name, is_directory=False, is_marimo_file=True, diff --git a/marimo/_server/recents.py b/marimo/_server/recents.py index 2c10cee2c2f..7037c45ff24 100644 --- a/marimo/_server/recents.py +++ b/marimo/_server/recents.py @@ -1,14 +1,12 @@ # Copyright 2026 Marimo. All rights reserved. from __future__ import annotations -import os import pathlib from dataclasses import dataclass, field from marimo import _loggers from marimo._server.models.home import MarimoFile from marimo._utils.config.config import ConfigReader -from marimo._utils.paths import pretty_path @dataclass @@ -77,7 +75,9 @@ def rename(self, old_filename: str, new_filename: str) -> None: self.config.write_toml(state) - def get_recents(self) -> list[MarimoFile]: + def get_recents( + self, directory: pathlib.Path | None = None + ) -> list[MarimoFile]: if not self.config: return [] @@ -86,19 +86,21 @@ def get_recents(self) -> list[MarimoFile]: ) files: list[MarimoFile] = [] - cwd = pathlib.Path.cwd() + base_dir = directory or pathlib.Path.cwd() limited_files = state.files[: self.MAX_FILES] for file in limited_files: file_path = pathlib.Path(file) - if _is_tmp_file(file) or cwd not in file_path.parents: + if _is_tmp_file(file) or not file_path.is_relative_to(base_dir): continue - if not os.path.exists(file): + if not file_path.exists(): continue + # Return path relative to base_dir + relative_path = file_path.relative_to(base_dir) files.append( MarimoFile( - name=os.path.basename(file), - path=pretty_path(file), - last_modified=os.path.getmtime(file), + name=file_path.name, + path=str(relative_path), + last_modified=file_path.stat().st_mtime, ) ) diff --git a/marimo/_server/session_manager.py b/marimo/_server/session_manager.py index 84e72d0c73a..a4a6139632e 100644 --- a/marimo/_server/session_manager.py +++ b/marimo/_server/session_manager.py @@ -109,6 +109,7 @@ def _get_code() -> str: # Add recents tracking listener self.recents = RecentFilesManager() self._event_bus = SessionEventBus() + self._event_bus.subscribe(RecentsTrackerListener(self.recents)) # Initialize file watching components self._watcher_manager = FileWatcherManager() @@ -160,9 +161,7 @@ def create_session( # Create the session from marimo._runtime.commands import AppMetadata - extensions: list[SessionExtension] = [ - RecentsTrackerListener(self.recents) - ] + extensions: list[SessionExtension] = [] if self.watch: extensions.append( SessionFileWatcherExtension( diff --git a/tests/_server/test_file_manager_absolute_path.py b/tests/_server/test_file_manager_absolute_path.py index 029ad4af758..cb8b8de53e6 100644 --- a/tests/_server/test_file_manager_absolute_path.py +++ b/tests/_server/test_file_manager_absolute_path.py @@ -4,6 +4,7 @@ from __future__ import annotations import os +import sys from pathlib import Path import pytest @@ -11,6 +12,8 @@ from marimo._server.file_router import AppFileRouter from marimo._utils.http import HTTPException, HTTPStatus +is_windows = sys.platform == "win32" + class TestAbsoluteDirectoryPath: """Test that absolute directory paths work correctly.""" @@ -43,19 +46,21 @@ def __(): absolute_dir = str(test_dir.absolute()) router = AppFileRouter.from_directory(absolute_dir) - # The directory should be stored correctly + # The directory should be stored correctly (always absolute) assert router.directory == absolute_dir - # Get the files - they should have absolute paths + # Get the files - they have relative paths (relative to directory) files = router.files assert len(files) > 0 file_info = files[0] assert file_info.is_marimo_file - assert file_info.path == str(test_file) + # Path is relative to the directory + assert file_info.path == "notebook.py" - # Try to get a file manager using the file path + # Try to get a file manager using the relative path from files list file_manager = router.get_file_manager(file_info.path) assert file_manager is not None + # File manager resolves to absolute path assert file_manager.filename == str(test_file) assert file_manager.is_notebook_named @@ -92,19 +97,22 @@ def __(): relative_dir = "test_dir" router = AppFileRouter.from_directory(relative_dir) - # The directory should be stored correctly - assert router.directory == relative_dir + # The directory is converted to absolute for consistency + assert router.directory == str(test_dir.absolute()) - # Get the files + # Get the files - paths are relative to the directory files = router.files assert len(files) > 0 file_info = files[0] assert file_info.is_marimo_file + assert file_info.path == "notebook.py" - # Try to get a file manager using the file path + # Try to get a file manager using the relative file path file_manager = router.get_file_manager(file_info.path) assert file_manager is not None assert file_manager.is_notebook_named + # File manager resolves to absolute path + assert file_manager.filename == str(test_file.absolute()) finally: os.chdir(original_cwd) @@ -406,7 +414,7 @@ def __(): file_manager_from_basename = router.get_file_manager(basename) assert file_manager_from_basename is not None assert file_manager_from_basename.is_notebook_named - except HTTPException as e: + except HTTPException: # This is the bug - it should have resolved relative to router.directory pytest.fail( f"Should be able to open file with basename '{basename}' " @@ -459,13 +467,13 @@ def __(): os.chdir(other_dir) # Try to open with relative path - relative_path = "subdir/notebook.py" + relative_path = os.path.join("subdir", "notebook.py") try: file_manager = router.get_file_manager(relative_path) assert file_manager is not None assert file_manager.is_notebook_named - except HTTPException as e: + except HTTPException: pytest.fail( f"Should be able to open file with relative path '{relative_path}' " f"when router has absolute directory '{absolute_dir}'" @@ -494,12 +502,15 @@ def test_files_list_shows_correct_paths_for_absolute_dir( files = router.files assert len(files) == 2 - # All paths should be absolute + # All paths are now relative to the directory for file_info in files: - assert Path(file_info.path).is_absolute() - assert file_info.path.startswith(absolute_dir) + # Paths are relative (just the filename for top-level files) + assert not Path(file_info.path).is_absolute() + # When joined with directory, should form a valid path + full_path = Path(absolute_dir) / file_info.path + assert full_path.exists() - # Verify each file can be opened using its path + # Verify each file can be opened using its relative path for file_info in files: file_manager = router.get_file_manager(file_info.path) assert file_manager is not None @@ -562,3 +573,259 @@ def test_absolute_path_with_symlink_attack_denied( router.get_file_manager(str(secret_file.absolute())) assert exc_info.value.status_code == HTTPStatus.FORBIDDEN + + +class TestRelativePathsInFileListing: + """Test that files list returns correct relative paths for various structures.""" + + def test_nested_directory_returns_relative_paths( + self, tmp_path: Path + ) -> None: + """Test that files in subdirectories have correct relative paths.""" + # Create nested structure + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + subdir = test_dir / "subdir" + subdir.mkdir() + + # Create files at different levels + root_file = test_dir / "root.py" + root_file.write_text("import marimo\napp = marimo.App()") + + nested_file = subdir / "nested.py" + nested_file.write_text("import marimo\napp = marimo.App()") + + # Create router + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + files = router.files + file_paths = _collect_file_paths(files) + + # Should have both files with relative paths + assert "root.py" in file_paths + assert os.path.join("subdir", "nested.py") in file_paths + + # Both should be openable using their relative paths + for path in _collect_file_paths(files): + file_manager = router.get_file_manager(path) + assert file_manager is not None + assert file_manager.is_notebook_named + + def test_deep_nesting_returns_correct_paths(self, tmp_path: Path) -> None: + """Test deeply nested directories have correct relative paths.""" + # Create deep structure: test_dir/a/b/c/notebook.py + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + dir_a = test_dir / "a" + dir_a.mkdir() + dir_b = dir_a / "b" + dir_b.mkdir() + dir_c = dir_b / "c" + dir_c.mkdir() + + deep_file = dir_c / "notebook.py" + deep_file.write_text("import marimo\napp = marimo.App()") + + # Create router + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + files = router.files + file_paths = _collect_file_paths(files) + + # Should have the deep file with correct relative path + expected_path = os.path.join("a", "b", "c", "notebook.py") + assert expected_path in file_paths + + # Should be openable using the actual path from the files list + actual_paths = _collect_file_paths(files) + file_manager = router.get_file_manager(actual_paths[0]) + assert file_manager is not None + assert file_manager.filename == str(deep_file.absolute()) + + def test_both_relative_and_absolute_paths_work( + self, tmp_path: Path + ) -> None: + """Test that both relative and absolute paths can open the same file.""" + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + + test_file = test_dir / "notebook.py" + test_file.write_text("import marimo\napp = marimo.App()") + + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + # Get the relative path from files list + files = router.files + assert len(files) == 1 + relative_path = files[0].path + assert relative_path == "notebook.py" + + # Open with relative path + manager_from_relative = router.get_file_manager(relative_path) + assert manager_from_relative is not None + + # Open with absolute path + absolute_path = str(test_file.absolute()) + manager_from_absolute = router.get_file_manager(absolute_path) + assert manager_from_absolute is not None + + # Both should point to the same file + assert manager_from_relative.filename == manager_from_absolute.filename + + def test_path_traversal_in_relative_path_blocked( + self, tmp_path: Path + ) -> None: + """Test that path traversal attacks using relative paths are blocked.""" + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + + # Create a file outside test_dir + outside_file = tmp_path / "outside.py" + outside_file.write_text("import marimo\napp = marimo.App()") + + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + # Try to access file outside using path traversal + with pytest.raises(HTTPException) as exc_info: + router.get_file_manager("../outside.py") + + assert exc_info.value.status_code == HTTPStatus.FORBIDDEN + + @pytest.mark.skipif(is_windows, reason="Non-windows tests") + def test_path_traversal_multiple_levels_blocked( + self, tmp_path: Path + ) -> None: + """Test that multi-level path traversal is blocked.""" + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + # Try various path traversal attempts + traversal_attempts = [ + "../../../etc/passwd", + "subdir/../../outside.py", + "a/../../../etc/passwd", + ] + + for attempt in traversal_attempts: + with pytest.raises(HTTPException) as exc_info: + router.get_file_manager(str(attempt)) + assert exc_info.value.status_code == HTTPStatus.FORBIDDEN, ( + f"Path traversal '{attempt}' should be blocked" + ) + + +class TestFileManagerPathResolution: + """Test file manager path resolution edge cases.""" + + @pytest.mark.skipif(is_windows, reason="Non-windows tests") + def test_file_manager_with_dot_path(self, tmp_path: Path) -> None: + """Test that '.' in paths is handled correctly.""" + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + + test_file = test_dir / "notebook.py" + test_file.write_text("import marimo\napp = marimo.App()") + + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + # Try to open with ./ prefix + file_manager = router.get_file_manager("./notebook.py") + assert file_manager is not None + assert file_manager.filename == str(test_file.absolute()) + + @pytest.mark.skipif(is_windows, reason="// has special meaning on Windows") + def test_file_manager_with_redundant_slashes(self, tmp_path: Path) -> None: + """Test that redundant slashes in paths are handled.""" + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + subdir = test_dir / "subdir" + subdir.mkdir() + + test_file = subdir / "notebook.py" + test_file.write_text("import marimo\napp = marimo.App()") + + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + # Path normalization should handle redundant slashes + file_manager = router.get_file_manager("subdir//notebook.py") + assert file_manager is not None + assert file_manager.filename == str(test_file.absolute()) + + def test_nonexistent_file_returns_not_found(self, tmp_path: Path) -> None: + """Test that accessing a nonexistent file returns NOT_FOUND.""" + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + with pytest.raises(HTTPException) as exc_info: + router.get_file_manager("nonexistent.py") + + assert exc_info.value.status_code == HTTPStatus.NOT_FOUND + + def test_directory_stored_as_absolute(self, tmp_path: Path) -> None: + """Test that directory is always stored as absolute path.""" + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + + original_cwd = os.getcwd() + try: + os.chdir(tmp_path) + + # Create router with relative path + router = AppFileRouter.from_directory("test_dir") + + # Directory should be stored as absolute + assert router.directory is not None + assert Path(router.directory).is_absolute() + assert router.directory == str(test_dir.absolute()) + finally: + os.chdir(original_cwd) + + def test_files_accessible_after_cwd_change(self, tmp_path: Path) -> None: + """Test that files remain accessible after changing working directory.""" + test_dir = tmp_path / "test_dir" + test_dir.mkdir() + other_dir = tmp_path / "other_dir" + other_dir.mkdir() + + test_file = test_dir / "notebook.py" + test_file.write_text("import marimo\napp = marimo.App()") + + original_cwd = os.getcwd() + try: + # Create router while in tmp_path + os.chdir(tmp_path) + router = AppFileRouter.from_directory(str(test_dir.absolute())) + + # Get files list + files = router.files + assert len(files) == 1 + relative_path = files[0].path + + # Change to a completely different directory + os.chdir(other_dir) + + # Should still be able to open file using relative path + file_manager = router.get_file_manager(relative_path) + assert file_manager is not None + assert file_manager.filename == str(test_file.absolute()) + + # Should still be able to read content + content = file_manager.read_file() + assert "marimo.App" in content + finally: + os.chdir(original_cwd) + + +# Find all file paths (including in nested children) +def _collect_file_paths(items: list) -> list[str]: + paths = [] + for item in items: + if not item.is_directory: + paths.append(item.path) + if item.children: + paths.extend(_collect_file_paths(item.children)) + return paths diff --git a/tests/_server/test_file_router.py b/tests/_server/test_file_router.py index 323ca503366..74c818e0d29 100644 --- a/tests/_server/test_file_router.py +++ b/tests/_server/test_file_router.py @@ -145,7 +145,7 @@ def test_lazy_list_of_get_app_file_manager(self): filename = self.test_file1.name assert os.path.exists(filename), f"File {filename} does not exist" file_manager = router.get_file_manager(key=filename) - assert file_manager.filename == os.path.join(self.test_dir, filename) + assert file_manager.filename == filename def test_lazy_list_of_get_app_file_manager_nested(self): router = LazyListOfFilesAppFileRouter( @@ -510,7 +510,8 @@ def test_lazy_router_skips_common_dirs(tmp_path: Path): # Should only find the root file, not files in skipped directories file_paths = [f.path for f in files if not f.is_directory] assert len(file_paths) == 1 - assert str(root_file) in file_paths + # Paths are now relative to the router's directory + assert root_file.name in file_paths def test_lazy_router_counts_nested_files(tmp_path: Path): diff --git a/tests/_server/test_session_manager.py b/tests/_server/test_session_manager.py index fb9a6e869ed..2f3149047f0 100644 --- a/tests/_server/test_session_manager.py +++ b/tests/_server/test_session_manager.py @@ -9,6 +9,7 @@ from marimo._config.manager import get_default_config_manager from marimo._server.file_router import AppFileRouter from marimo._server.lsp import LspServer +from marimo._server.session.listeners import RecentsTrackerListener from marimo._server.session_manager import SessionManager from marimo._server.tokens import AuthToken, SkewProtectionToken from marimo._session import ( @@ -395,3 +396,72 @@ def test_cell(): assert str(session_manager.skew_protection_token) == str( session_manager2.skew_protection_token ) + + +def test_recents_listener_subscribed_to_event_bus( + session_manager: SessionManager, +) -> None: + """Test that RecentsTrackerListener is subscribed to session manager's event bus. + + This is a regression test for a bug where the listener was moved to be a + session extension (subscribing to the session's event bus) but the + emit_session_created event was still fired on the session manager's event bus, + causing recent files to not be tracked. + """ + # Verify that RecentsTrackerListener is in the event bus listeners + listeners = session_manager._event_bus._listeners + recents_listeners = [ + listen + for listen in listeners + if isinstance(listen, RecentsTrackerListener) + ] + assert len(recents_listeners) == 1, ( + "RecentsTrackerListener should be subscribed to session manager's event bus" + ) + + +async def test_recents_touch_called_on_session_create( + session_manager: SessionManager, + mock_session_consumer: SessionConsumer, + tmp_path: Path, +) -> None: + """Test that recents.touch() is called when a session is created with a file. + + This verifies the full integration: when a session is created for a file, + the RecentsTrackerListener receives the event and calls touch(). + """ + # Create a temp marimo file + tmp_file = tmp_path / "test_recents.py" + tmp_file.write_text( + "import marimo\napp = marimo.App()\n@app.cell\ndef _(): pass" + ) + + # Track calls to touch() + original_touch = session_manager.recents.touch + touched_files: list[str] = [] + + def mock_touch(filename: str) -> None: + touched_files.append(filename) + original_touch(filename) + + session_manager.recents.touch = mock_touch # type: ignore + + # Create a session + session = session_manager.create_session( + SessionId("recents_test_session"), + mock_session_consumer, + query_params={}, + file_key=str(tmp_file), + auto_instantiate=False, + ) + + # Allow async event to process + import asyncio + + await asyncio.sleep(0.1) + + # Verify touch was called with the file path + assert len(touched_files) == 1 + assert str(tmp_file) in touched_files[0] + + session.close()