Directory search improvements

VERSION

@@ -1 +1 @@
-2.0.29
+2.0.30

backend/app/api/browser.py

@@ -45,6 +45,7 @@
     ContentTransferResult,
     CopyMoveRequest,
     CreateItemRequest,
+    DirectoryIssueDetail,
     DirectoryListing,
     DirectorySearchResult,
     FileInfo,
@@ -1075,6 +1076,7 @@ async def search_directories(
     connection_id: uuid.UUID,
     q: str = Query("", max_length=MAX_QUERY_BYTES, description="Directory search text; shared UTF-8 byte and term limits apply"),
     include_dot_directories: bool = Query(False, description="Whether to include directories whose path contains dot-prefixed segments"),
+    include_issue_details: bool = Query(False, description="Include bounded directory issue details for the status dialog"),
     current_user: User = Depends(get_current_user_with_auth_check),
     session: Session = Depends(get_session),
 ) -> DirectorySearchResult:
@@ -1122,6 +1124,9 @@ async def search_directories(
             except asyncio.CancelledError:
                 cancelled.set()
                 raise
+            details, incomplete = (
+                cache.directory_issue_details() if response.issue_count and response.coverage == "partial" else ([], False)
+            )
             return DirectorySearchResult(
                 results=[row.path for row in response.results],
                 match_ranks={row.path: row.match_rank for row in response.results},
@@ -1133,6 +1138,9 @@ async def search_directories(
                 accepted_revision=response.accepted_revision,
                 coverage=response.coverage,
                 coverage_issue_count=response.issue_count,
+                issue_details_available=bool(details),
+                issue_details=[DirectoryIssueDetail.model_validate(detail) for detail in details] if include_issue_details else None,
+                issue_details_incomplete=incomplete,
                 last_verified_at_ms=response.last_verified_at_ms,
                 status_message=cache.scan_error,
                 monitoring_state=cache.monitoring_state,

backend/app/models/file.py

@@ -144,6 +144,13 @@ def validate_factual_outcome(self) -> "ContentTransferResult":
         return self
 
 
+class DirectoryIssueDetail(BaseModel):
+    """A directory that could not be fully read during verification."""
+
+    path: str
+    code: Literal["access_denied", "unsupported_encoding", "unsafe_traversal", "scan_failed"]
+
+
 class DirectorySearchResult(BaseModel):
     """Response model for directory search (quick navigate)."""
 
@@ -157,6 +164,9 @@ class DirectorySearchResult(BaseModel):
     accepted_revision: int | None = Field(default=None, ge=0)
     coverage: str | None = None
     coverage_issue_count: int = Field(default=0, ge=0)
+    issue_details_available: bool = False
+    issue_details: list[DirectoryIssueDetail] | None = None
+    issue_details_incomplete: bool = False
     last_verified_at_ms: int | None = Field(default=None, ge=0)
     status_message: str | None = None
     monitoring_state: Literal["starting", "healthy", "recovering", "unavailable"] | None = None

backend/app/services/directory_cache.py

@@ -86,6 +86,7 @@ class CacheState(str, Enum):
     """State of a connection's directory cache."""
 
     EMPTY = "empty"  # No scan started yet
+    OPENING = "opening"
     BUILDING = "building"  # Initial scan in progress
     READY = "ready"  # Scan complete, cache is usable
     UPDATING = "updating"  # Rescan in progress, stale data served
@@ -254,9 +255,26 @@ def __init__(
         self._streamed_parent_reads: set[int] = set()
         self._completed_scan_reads: set[int] = set()
         self._pending_scan_issues: dict[int, dict[str, int]] = {}
+        self.issue_details: set[tuple[str, str]] = set()
+        self.issue_details_incomplete = False
         self.scan_commits = 0
         self.on_scan_commit: Callable[[int], None] | None = None
 
+    def _publish_issue_details(self, parent: str, counts: dict[str, int]) -> None:
+        """Publish diagnostics only after the owning parent transaction succeeds."""
+        self.issue_details = {detail for detail in self.issue_details if detail[0] != parent}
+        for code in counts:
+            if len(self.issue_details) < MAX_LOGGED_SCAN_ISSUES_PER_PASS:
+                self.issue_details.add((parent, code))
+            else:
+                self.issue_details_incomplete = True
+
+    def _discard_missing_issue_details(self, generation_id: int) -> None:
+        """Remove samples for subtrees pruned by committed parent snapshots."""
+        self.issue_details = {
+            detail for detail in self.issue_details if not detail[0] or self.index.contains_directory(generation_id, detail[0])
+        }
+
     def start_scan_groups(self, *, reconcile: bool = False) -> None:
         with self.lock:
             self._scan_group = {}
@@ -341,6 +359,9 @@ def flush_scan_group(self) -> None:
                     if file_ids[generation]:
                         options["file_ids"] = file_ids[generation]
                     self.index.commit_scan_parents(generation, results, **options)
+                    for parent, _children in results:
+                        self._publish_issue_details(parent, issues.get(parent, {}))
+                    self._discard_missing_issue_details(generation)
                     self.scan_commits += 1
                 if self.on_scan_commit is not None:
                     for generation in changed:
@@ -579,6 +600,10 @@ def apply_pending(self) -> int:
                     self.index.mutate_observed_path("add", mutation.path)
                 elif mutation.action == FileAction.FILE_ACTION_REMOVED:
                     self.index.mutate_observed_path("remove", mutation.path)
+                    prefix = mutation.path + "/"
+                    self.issue_details = {
+                        detail for detail in self.issue_details if detail[0] != mutation.path and not detail[0].startswith(prefix)
+                    }
                 elif mutation.action == FileAction.FILE_ACTION_RENAMED_NEW_NAME:
                     self.index.mutate_observed_path("rename", mutation.path, mutation.new_path)
                 else:
@@ -656,6 +681,9 @@ def record_issue(self, token: ScanReadToken, generation_id: int, code: str, *, r
                     self.flush_scan_group_if_due()
             else:
                 self.index.record_issue(generation_id, code, reconcile=reconcile)
+                counts = {previous_code: 1 for parent, previous_code in self.issue_details if parent == token.path}
+                counts[code] = 1
+                self._publish_issue_details(token.path, counts)
 
     def _record_scan_issue(self, token: ScanReadToken, generation_id: int, record: Callable[[int], None]) -> None:
         with self.lock:
@@ -689,6 +717,8 @@ def finish_parent(self, token: ScanReadToken, generation_id: int, *, reconcile:
                     if token.producer_id in self._pending_scan_issues
                     else None,
                 )
+                self._publish_issue_details(token.path, self._pending_scan_issues.get(token.producer_id, {}))
+                self._discard_missing_issue_details(generation_id)
             elif self._scan_group is not None:
                 existing = self._scan_group.get(token.producer_id)
                 snapshot_children = existing.children if existing is not None else []
@@ -1543,7 +1573,7 @@ async def start_scan(self) -> None:
         if self._scan_task is not None and not self._scan_task.done():
             return
         self._owning_loop = asyncio.get_running_loop()
-        self._state = CacheState.BUILDING
+        self._state = CacheState.OPENING
         self._scan_task = asyncio.create_task(self._initialize())
 
     async def _initialize(self) -> None:
@@ -1706,6 +1736,12 @@ def _classify_source_error(error: Exception) -> str | None:
             return "unsupported_encoding"
         return "scan_failed"
 
+    def directory_issue_details(self) -> tuple[list[dict[str, str]], bool]:
+        """Return bounded, source-local diagnostics, never a persistent error history."""
+        with self.coordinator.lock:
+            details = [{"path": parent or "/", "code": code} for parent, code in sorted(self.coordinator.issue_details)]
+            return details, self.coordinator.issue_details_incomplete
+
     def _record_item_issue(self, token: ScanReadToken, generation: int, code: str, *, reconcile: bool, path: str | None = None) -> None:
         self.coordinator.record_issue(token, generation, code, reconcile=reconcile)
         self._scan_issue_counts[code] = self._scan_issue_counts.get(code, 0) + 1
@@ -2293,6 +2329,8 @@ def _scan_disk_queue(self) -> None:
                     resuming = self.index.has_resumable_scan()
                     generation, _selected = self.index.prepare_scan()
                     if not resuming:
+                        self.coordinator.issue_details.clear()
+                        self.coordinator.issue_details_incomplete = False
                         self._scan_epoch = (generation, baseline)
                     if self.coordinator.observation_lost:
                         while self.index.requeue_completed_scan_batch(generation):

backend/app/services/directory_index.py

@@ -954,6 +954,17 @@ def scan_parent_file_id(self, generation_id: int, parent: str) -> bytes | None:
             )
             return None if row is None or row[0] is None else bytes(row[0])
 
+    def contains_directory(self, generation_id: int, path: str) -> bool:
+        """Check whether a sampled directory remains in the committed generation."""
+        validate_directory_path(path)
+        with self._writer_lock:
+            return (
+                self._connection()
+                .execute("SELECT 1 FROM directories WHERE generation_id = ? AND path = ?", (generation_id, path))
+                .fetchone()
+                is not None
+            )
+
     def write_scan_children(
         self, generation_id: int, children: Iterable[tuple[str, bool]], *, file_ids: dict[str, bytes] | None = None
     ) -> None:

backend/tests/test_directory_cache.py

@@ -350,6 +350,7 @@ def test_enum_values(self):
         """Verify all expected cache states exist with correct values."""
 
         assert CacheState.EMPTY.value == "empty"
+        assert CacheState.OPENING.value == "opening"
         assert CacheState.BUILDING.value == "building"
         assert CacheState.READY.value == "ready"
         assert CacheState.UPDATING.value == "updating"
@@ -363,7 +364,14 @@ def test_enum_is_string(self):
     def test_enum_member_count(self):
         """Ensure no unexpected states were added."""
 
-        assert len(CacheState) == 4
+        expected_states = (
+            CacheState.EMPTY,
+            CacheState.OPENING,
+            CacheState.BUILDING,
+            CacheState.READY,
+            CacheState.UPDATING,
+        )
+        assert len(CacheState) == len(expected_states)
 
 
 # ============================================================================

backend/tests/test_directory_index_contract.py

@@ -2205,6 +2205,31 @@ def unavailable():
     assert "Wait for active workers" in sqlite_cache.scan_error
     assert started == [True]
     assert not sqlite_cache._opened.is_set()
+    assert sqlite_cache.state.value == "empty"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("next_state", ["building", "updating", "ready"])
+async def test_cold_scan_reports_opening_until_initialization_finishes(sqlite_cache, monkeypatch, next_state):
+    from app.services.directory_cache import CacheState
+
+    entered = asyncio.Event()
+    release = asyncio.Event()
+
+    async def initialize():
+        entered.set()
+        await release.wait()
+        sqlite_cache._state = CacheState(next_state)
+
+    monkeypatch.setattr(sqlite_cache, "_initialize", initialize)
+    await sqlite_cache.start_scan()
+    try:
+        await entered.wait()
+        assert sqlite_cache.state is CacheState.OPENING
+    finally:
+        release.set()
+        await sqlite_cache._scan_task
+    assert sqlite_cache.state.value == next_state
 
 
 @pytest.mark.asyncio
@@ -5253,6 +5278,32 @@ async def get_source(**kwargs):
     assert response.json()["coverage"] == "partial"
 
 
+@pytest.mark.parametrize("include_details", [False, True])
+def test_directory_api_returns_issue_details_only_on_request(
+    sqlite_cache, monkeypatch, client, auth_headers_user, test_connection, include_details
+):
+    async def get_source(**kwargs):
+        return sqlite_cache
+
+    sqlite_cache.index.seed_scan(1)
+    sqlite_cache.coordinator.start_scan_groups()
+    token = sqlite_cache.coordinator.begin_read("")
+    sqlite_cache.coordinator.record_issue(token, 1, "unsupported_encoding")
+    sqlite_cache.coordinator.finish_parent(token, 1)
+    sqlite_cache.coordinator.flush_scan_group()
+    assert sqlite_cache.index.finish_generation(1)
+    monkeypatch.setattr("app.services.directory_cache.get_directory_cache_manager", lambda: SimpleNamespace(get_or_create_cache=get_source))
+    response = client.get(
+        f"/api/browse/{test_connection.id}/directories",
+        headers=auth_headers_user,
+        params={"q": "", "include_issue_details": str(include_details).lower()},
+    )
+    assert response.status_code == 200
+    assert response.json()["issue_details_available"] is True
+    assert response.json()["issue_details"] == ([{"path": "/", "code": "unsupported_encoding"}] if include_details else None)
+    assert response.json()["issue_details_incomplete"] is False
+
+
 def test_directory_api_rejects_excess_terms_before_source_creation(monkeypatch, client, auth_headers_user, test_connection):
     def unexpected_manager():
         raise AssertionError("Invalid queries must not initialize a directory source.")
@@ -5452,6 +5503,14 @@ def scan(path, **kwargs):
     assert sqlite_cache.index.pending_scan_paths(1) == []
     assert sqlite_cache._smb_base_path + "\\readable\\child" in reads
     assert sqlite_cache.index._connection().execute("SELECT code, count FROM coverage_issues").fetchall() == [("access_denied", 1)]
+    assert sqlite_cache.directory_issue_details() == ([{"path": "denied", "code": "access_denied"}], False)
+
+    def empty_scan(*_args, **_kwargs):
+        yield from ()
+
+    monkeypatch.setattr("smbclient.scandir", empty_scan)
+    sqlite_cache._scan_disk_queue()
+    assert sqlite_cache.directory_issue_details() == ([], False)
 
 
 @pytest.mark.parametrize("failure", ["separator", "surrogate", "oversized", "folder_io", "denied"])
@@ -7376,12 +7435,14 @@ def test_scan_group_buffers_parent_issues(index, early_flush, reconcile):
         coordinator.flush_scan_group()
     coordinator.record_issue(token, 1, "unsupported_encoding", reconcile=reconcile)
     assert index.status().issue_count == 0
+    assert not coordinator.issue_details
     coordinator.finish_parent(token, 1, reconcile=reconcile)
     coordinator.flush_scan_group()
     assert index.status().issue_count == 2
     assert coordinator.scan_commits == 1
     assert not coordinator._pending_scan_issues
     assert not coordinator._producers
+    assert coordinator.issue_details == {("", "unsupported_encoding")}
 
 
 @pytest.mark.parametrize("reconcile", [False, True])
@@ -7405,6 +7466,86 @@ def test_stale_scan_group_discards_parent_issues(index, reconcile):
     assert index.scan_parent_pending(1, "")
     assert index._connection().execute("SELECT count(*) FROM coverage_issues WHERE code = 'unsupported_encoding'").fetchone() == (0,)
     assert not coordinator._pending_scan_issues
+    assert not coordinator.issue_details
+
+
+@pytest.mark.parametrize("parent_hint", [False, True])
+def test_issue_details_remove_deleted_subtree_after_reconciliation(index, parent_hint):
+    from smbprotocol.change_notify import FileAction
+
+    from app.services.directory_cache import READ_PARENT_ACTION, DirectoryScanCoordinator, _DirectoryMutation
+
+    index.seed_scan(1)
+    coordinator = DirectoryScanCoordinator(index)
+    coordinator.start_scan_groups()
+    for parent, children in [
+        ("", [("private", True), ("private-backup", True)]),
+        ("private", [("private/child", True)]),
+        ("private/child", []),
+        ("private-backup", []),
+    ]:
+        with coordinator.read(parent) as token:
+            coordinator.commit_children(token, 1, children)
+            coordinator.record_issue(token, 1, "access_denied")
+            coordinator.finish_parent(token, 1)
+        coordinator.flush_scan_group()
+    coordinator.cancel_scan_groups()
+    assert index.finish_generation(1)
+
+    mutation = _DirectoryMutation(READ_PARENT_ACTION, "") if parent_hint else _DirectoryMutation(FileAction.FILE_ACTION_REMOVED, "private")
+    assert coordinator.enqueue(mutation)
+    coordinator.apply_pending()
+    coordinator.start_scan_groups(reconcile=True)
+    with coordinator.read("") as token:
+        coordinator.commit_children(token, 1, [("private-backup", True)], reconcile=True)
+        coordinator.record_issue(token, 1, "access_denied", reconcile=True)
+        coordinator.finish_parent(token, 1, reconcile=True)
+    coordinator.flush_scan_group()
+    coordinator.cancel_scan_groups()
+
+    assert not index.search("private/child").results
+    assert coordinator.issue_details == {("", "access_denied"), ("private-backup", "access_denied")}
+
+
+def test_issue_details_preserved_when_subtree_deletion_rolls_back(index):
+    from smbprotocol.change_notify import FileAction
+
+    from app.services.directory_cache import DirectoryScanCoordinator, _DirectoryMutation
+
+    index.write_paths(["private", "private/child"])
+    assert index.finish_generation(1)
+    coordinator = DirectoryScanCoordinator(index)
+    coordinator._publish_issue_details("private", {"access_denied": 1})
+    coordinator._publish_issue_details("private/child", {"access_denied": 1})
+    samples = coordinator.issue_details.copy()
+    index._connection().execute(
+        "CREATE TRIGGER reject_sample_deletion BEFORE DELETE ON directories "
+        "WHEN OLD.path = 'private' BEGIN SELECT RAISE(ABORT, 'injected deletion failure'); END"
+    )
+    assert coordinator.enqueue(_DirectoryMutation(FileAction.FILE_ACTION_REMOVED, "private"))
+    with pytest.raises(sqlite3.IntegrityError, match="injected deletion failure"):
+        coordinator.apply_pending()
+    assert coordinator.issue_details == samples
+    assert index.contains_directory(1, "private/child")
+    assert coordinator.pending
+    index._connection().execute("DROP TRIGGER reject_sample_deletion")
+    coordinator.apply_pending()
+    assert not coordinator.issue_details
+
+
+def test_issue_detail_samples_are_bounded_and_replaced_by_clean_reads(index):
+    from app.services.directory_cache import MAX_LOGGED_SCAN_ISSUES_PER_PASS, DirectoryScanCoordinator
+
+    coordinator = DirectoryScanCoordinator(index)
+    for number in range(MAX_LOGGED_SCAN_ISSUES_PER_PASS + 1):
+        coordinator._publish_issue_details(f"parent-{number}", {"access_denied": 1})
+    assert len(coordinator.issue_details) == MAX_LOGGED_SCAN_ISSUES_PER_PASS
+    assert coordinator.issue_details_incomplete
+    coordinator._publish_issue_details("parent-0", {})
+    assert not any(parent == "parent-0" for parent, _code in coordinator.issue_details)
+    coordinator._publish_issue_details("parent-1", {"scan_failed": 1})
+    assert ("parent-1", "scan_failed") in coordinator.issue_details
+    assert ("parent-1", "access_denied") not in coordinator.issue_details
 
 
 def test_reconciliation_groups_independent_pages(smb_source_cache, smb_directory_server, monkeypatch):