fix(ephemeral): fix volume browsing — race condition, event filter, sync listing

Three bugs caused ephemeral volume browsing to either show nothing or require
navigation away and back before content appeared.

1. Transport race condition (packages/ts-client/src/transport.ts)
   listen("core-event") was called AFTER invoke("subscribe_to_events"), so any
   events replayed from the 5s EventBuffer during subscription setup were emitted
   before the listener existed and were silently dropped. Fixed by swapping the
   order: listen first, then subscribe.

2. Event filter: direct children not matched (core/src/infra/event/mod.rs)
   affects_path() with include_descendants=false only matched exact path equality
   (file_path == scope_path). A directory listing subscribes with the directory as
   scope, so files inside it never matched. Fixed by also checking
   file_path.parent() == scope_path (direct children).

3. Synchronous ephemeral directory listing (core/src/ops/files/query/directory_listing.rs)
   query_ephemeral_directory_impl() dispatched the indexer job and immediately
   returned empty, relying on events to update the UI. Since ephemeral indexing
   takes <500ms, it is better to wait for completion and return results directly.
   The function now polls cache.is_indexing() every 5ms (10s timeout) and reads
   from cache before returning. Concurrent requests also wait instead of returning
   empty. Subdirectories with no indexed children now trigger their own indexer
   job instead of returning empty (was: parent index has entry but no children →
   treated as "has results").
   Code refactored into read_ephemeral_listing() and wait_for_indexing() helpers
   to eliminate triplication.

4. Removed is_explicitly_indexed() duplicate (core/src/ops/indexing/ephemeral/cache.rs)
   Identical to is_indexed(); callers now use is_indexed() directly.

5. Debug logging infrastructure (packages/ts-client/src/hooks/useNormalizedQuery.ts)
   Added debug: boolean option to useNormalizedQuery. When true, logs subscription
   lifecycle, raw events, batch filter counts, and cache updates to the console.
   Off by default.

core/src/infra/daemon/rpc.rs

@@ -243,7 +243,6 @@ impl RpcServer {
 		// If event_types is empty, forward all events
 		// Otherwise, treat event_types as an INCLUSION list (only forward these)
 		if !event_types.is_empty() {
-			// Use the Event's own variant_name() method - single source of truth!
 			let event_type = event.variant_name();
 
 			if !event_types.contains(&event_type.to_string()) {
@@ -260,7 +259,6 @@ impl RpcServer {
 						return false;
 					}
 				} else {
-					// Event is not a resource event, but filter expects one
 					return false;
 				}
 			}
@@ -268,9 +266,7 @@ impl RpcServer {
 			// Filter by path scope (for resource events)
 			if let Some(path_scope) = &filter.path_scope {
 				let include_descendants = filter.include_descendants.unwrap_or(false);
-				let affects = event.affects_path(path_scope, include_descendants);
-
-				if !affects {
+				if !event.affects_path(path_scope, include_descendants) {
 					return false;
 				}
 			}

core/src/infra/event/mod.rs

@@ -502,12 +502,9 @@ impl Event {
 						return false;
 					}
 
-					// Exact mode: ONLY match the scope directory itself
-					// This indicates files are DIRECTLY in this directory
-					// Subdirectories in affected_paths mean files are in THOSE subdirectories
-					let matches = file_path == scope_path;
-
-					matches
+					// Direct children mode: match files whose parent is the scope directory
+					file_path == scope_path
+						|| file_path.parent().map_or(false, |p| p == scope_path)
 				} else {
 					false
 				}

core/src/ops/files/query/directory_listing.rs

@@ -12,6 +12,7 @@ use crate::{
 		video_media_data,
 	},
 	infra::query::LibraryQuery,
+	ops::indexing::ephemeral::{cache::EphemeralIndexCache, EphemeralIndex},
 };
 use sea_orm::{
 	ColumnTrait, ConnectionTrait, DatabaseConnection, EntityTrait, JoinType, QueryFilter,
@@ -20,6 +21,7 @@ use sea_orm::{
 use serde::{Deserialize, Serialize};
 use specta::Type;
 use std::{collections::HashMap, sync::Arc};
+use tokio::sync::RwLock as TokioRwLock;
 use tracing;
 use uuid::Uuid;
 
@@ -624,7 +626,6 @@ impl DirectoryListingQuery {
 		context: Arc<CoreContext>,
 		library_id: Uuid,
 	) -> QueryResult<DirectoryListingOutput> {
-		use crate::domain::file::File;
 		use crate::ops::indexing::{IndexScope, IndexerJob, IndexerJobConfig};
 
 		// Get the local path for cache lookup
@@ -647,7 +648,7 @@ impl DirectoryListingQuery {
 
 		// Check if we have a cached index that covers this path (or a parent path)
 		if let Some(index) = cache.get_for_search(&local_path) {
-			tracing::info!(
+			tracing::debug!(
 				"Found cached ephemeral index for path: {}",
 				local_path.display()
 			);
@@ -660,69 +661,18 @@ impl DirectoryListingQuery {
 			};
 
 			// Check if the index actually has entries for this directory
+			// BUT: if children is empty and this path wasn't explicitly indexed,
+			// it means a parent was indexed and this subdirectory needs its own indexing job
 			if let Some(children) = children {
-				tracing::debug!(
-					"Cached index has {} children for {}",
-					children.len(),
-					local_path.display()
-				);
-
-				// Convert cached entries to File objects with lazy UUID assignment
-				// Acquire write lock once for the entire batch instead of per-entry
-				let mut index_write = index.write().await;
-				let mut files = Vec::new();
-
-				for child_path in children {
-					if let Some(metadata) = index_write.get_entry_ref(&child_path) {
-						// Apply hidden file filter
-						if !self.input.include_hidden.unwrap_or(false) && metadata.is_hidden {
-							continue;
-						}
-
-						// Get or assign UUID (lazy generation)
-						let entry_uuid = index_write.get_or_assign_uuid(&child_path);
-
-						// Build SdPath for this entry
-						let entry_sd_path = SdPath::Physical {
-							device_slug: match &self.input.path {
-								SdPath::Physical { device_slug, .. } => device_slug.clone(),
-								_ => String::new(),
-							},
-							path: child_path.clone(),
-						};
-
-						// Get content kind from index (identified by extension)
-						let content_kind = index_write.get_content_kind(&child_path);
-
-						// Convert to File
-						let mut file = File::from_ephemeral(entry_uuid, &metadata, entry_sd_path);
-						file.content_kind = content_kind;
-						files.push(file);
-					}
+				if children.is_empty() && !cache.is_indexed(&local_path) {
+					tracing::info!(
+						"Subdirectory has no indexed children, triggering indexing: {}",
+						local_path.display()
+					);
+					// Fall through to spawn indexer job
+				} else if let Some(output) = self.read_ephemeral_listing(&index, &local_path).await {
+					return Ok(output);
 				}
-				drop(index_write);
-
-				// Apply sorting
-				self.sort_files(&mut files);
-
-				// Apply limit
-				let total_count = files.len() as u32;
-				let has_more = if let Some(limit) = self.input.limit {
-					if files.len() > limit as usize {
-						files.truncate(limit as usize);
-						true
-					} else {
-						false
-					}
-				} else {
-					false
-				};
-
-				return Ok(DirectoryListingOutput {
-					files,
-					total_count,
-					has_more,
-				});
 			}
 
 			// Index exists but doesn't have this directory yet
@@ -734,19 +684,22 @@ impl DirectoryListingQuery {
 		}
 
 		// No cached index or index doesn't cover this path
-		// Check if indexing is already in progress
+		// If another request is already indexing this path, wait for it
+		// instead of returning empty (avoids flicker on concurrent requests)
 		if cache.is_indexing(&local_path) {
-			tracing::info!("Indexing already in progress for: {}", local_path.display());
-			// Return empty, UI will get updates via events
-			return Ok(DirectoryListingOutput {
-				files: Vec::new(),
-				total_count: 0,
-				has_more: false,
-			});
+			tracing::debug!("Indexing already in progress, waiting: {}", local_path.display());
+			self.wait_for_indexing(&cache, &local_path).await;
+
+			if let Some(index) = cache.get_for_search(&local_path) {
+				if let Some(output) = self.read_ephemeral_listing(&index, &local_path).await {
+					return Ok(output);
+				}
+			}
+			// Fall through if cache still empty after wait
 		}
 
 		tracing::info!(
-			"No cached index, triggering ephemeral indexing for: {:?}",
+			"Triggering ephemeral indexing for: {:?}",
 			self.input.path
 		);
 
@@ -777,11 +730,20 @@ impl DirectoryListingQuery {
 			// Share the cached index with the job
 			indexer_job.set_ephemeral_index(ephemeral_index);
 
-			// Dispatch job asynchronously
-			// The job will emit ResourceChanged events as files are discovered
+			// Dispatch job and wait for completion so we can return results directly.
+			// Ephemeral indexing is very fast (typically <100ms for a single directory),
+			// so waiting is better than relying on event-based UI updates which have
+			// race conditions with subscription setup timing.
 			match library.jobs().dispatch(indexer_job).await {
-				Ok(_) => {
-					tracing::info!("Dispatched ephemeral indexer for {:?}", self.input.path);
+				Ok(_handle) => {
+					tracing::info!("Dispatched ephemeral indexer for {:?}, waiting", self.input.path);
+					self.wait_for_indexing(&cache, &local_path).await;
+
+					if let Some(index) = cache.get_for_search(&local_path) {
+						if let Some(output) = self.read_ephemeral_listing(&index, &local_path).await {
+							return Ok(output);
+						}
+					}
 				}
 				Err(e) => {
 					tracing::warn!(
@@ -789,14 +751,12 @@ impl DirectoryListingQuery {
 						self.input.path,
 						e
 					);
-					// Mark indexing as not in progress since job failed
 					cache.mark_indexing_complete(&local_path);
 				}
 			}
 		}
 
-		// Return empty result immediately
-		// UI will receive ResourceChanged events and populate incrementally
+		// Fallback: return empty if job dispatch failed or cache is empty
 		Ok(DirectoryListingOutput {
 			files: Vec::new(),
 			total_count: 0,
@@ -804,6 +764,87 @@ impl DirectoryListingQuery {
 		})
 	}
 
+	/// Read children from the ephemeral index and convert to a DirectoryListingOutput.
+	///
+	/// Shared by all ephemeral code paths (cache hit, wait-for-indexing, post-dispatch).
+	async fn read_ephemeral_listing(
+		&self,
+		index: &Arc<TokioRwLock<EphemeralIndex>>,
+		local_path: &std::path::Path,
+	) -> Option<DirectoryListingOutput> {
+
+		let children = {
+			let guard = index.read().await;
+			guard.list_directory(local_path)
+		};
+		let children = children?;
+
+		let mut index_write = index.write().await;
+		let mut files = Vec::new();
+
+		for child_path in children {
+			if let Some(metadata) = index_write.get_entry_ref(&child_path) {
+				if !self.input.include_hidden.unwrap_or(false) && metadata.is_hidden {
+					continue;
+				}
+				let entry_uuid = index_write.get_or_assign_uuid(&child_path);
+				let entry_sd_path = SdPath::Physical {
+					device_slug: match &self.input.path {
+						SdPath::Physical { device_slug, .. } => device_slug.clone(),
+						_ => String::new(),
+					},
+					path: child_path.clone(),
+				};
+				let content_kind = index_write.get_content_kind(&child_path);
+				let mut file = File::from_ephemeral(entry_uuid, &metadata, entry_sd_path);
+				file.content_kind = content_kind;
+				files.push(file);
+			}
+		}
+		drop(index_write);
+
+		self.sort_files(&mut files);
+		let total_count = files.len() as u32;
+		let has_more = if let Some(limit) = self.input.limit {
+			if files.len() > limit as usize {
+				files.truncate(limit as usize);
+				true
+			} else {
+				false
+			}
+		} else {
+			false
+		};
+
+		Some(DirectoryListingOutput {
+			files,
+			total_count,
+			has_more,
+		})
+	}
+
+	/// Poll until ephemeral indexing finishes (or 10s timeout).
+	///
+	/// Ephemeral indexing typically completes in <500ms, so this avoids
+	/// returning empty results and relying on event-based UI updates.
+	async fn wait_for_indexing(
+		&self,
+		cache: &EphemeralIndexCache,
+		local_path: &std::path::Path,
+	) {
+		let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
+		loop {
+			if !cache.is_indexing(local_path) {
+				break;
+			}
+			if tokio::time::Instant::now() >= deadline {
+				tracing::warn!("Ephemeral indexing timed out for {}", local_path.display());
+				break;
+			}
+			tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+		}
+	}
+
 	/// Sort files according to the input options
 	fn sort_files(&self, files: &mut Vec<File>) {
 		use crate::domain::file::EntryKind;

packages/ts-client/src/hooks/useNormalizedQuery.ts

@@ -177,11 +177,19 @@ export function useNormalizedQuery<I, O = any, TSelected = O>(
 		const capturedQueryKey = queryKey;
 
 		const handleEvent = (event: Event) => {
+			const isDebug = optionsRef.current.debug;
+			if (isDebug) {
+				console.log(`[useNormalizedQuery] RAW EVENT received:`, typeof event, event);
+			}
+
 			// Guard: only process events if pathScope hasn't changed since subscription
 			if (
 				JSON.stringify(optionsRef.current.pathScope) !==
 				JSON.stringify(capturedPathScope)
 			) {
+				if (isDebug) {
+					console.log(`[useNormalizedQuery] STALE pathScope, skipping event`);
+				}
 				return;
 			}
 
@@ -194,6 +202,10 @@ export function useNormalizedQuery<I, O = any, TSelected = O>(
 			);
 		};
 
+		if (options.debug) {
+			console.log(`[useNormalizedQuery] SUBSCRIBING: resourceType=${options.resourceType}, pathScope=`, JSON.stringify(options.pathScope), `includeDescendants=${options.includeDescendants ?? false}`);
+		}
+
 		client
 			.subscribeFiltered(
 				{
@@ -206,13 +218,22 @@ export function useNormalizedQuery<I, O = any, TSelected = O>(
 			)
 			.then((unsub) => {
 				if (isCancelled) {
+					if (options.debug) {
+						console.log(`[useNormalizedQuery] Subscription created but already cancelled`);
+					}
 					unsub();
 				} else {
+					if (options.debug) {
+						console.log(`[useNormalizedQuery] Subscription ACTIVE`);
+					}
 					unsubscribe = unsub;
 				}
 			});
 
 		return () => {
+			if (options.debug) {
+				console.log(`[useNormalizedQuery] UNSUBSCRIBING`);
+			}
 			isCancelled = true;
 			unsubscribe?.();
 		};
@@ -514,6 +535,14 @@ export function updateBatchResources<O>(
 	// Apply client-side filtering (safety fallback)
 	const filteredResources = filterBatchResources(resources, options);
 
+	const wireMethod = toWireMethod(options.query);
+	if (options.debug) {
+		console.log(`[useNormalizedQuery] ${wireMethod} BATCH: ${resources.length} total, ${filteredResources.length} after filter`);
+		if (filteredResources.length === 0 && resources.length > 0) {
+			console.log(`[useNormalizedQuery] ${wireMethod} ALL FILTERED OUT! First resource:`, JSON.stringify(resources[0]?.sd_path), `pathScope:`, JSON.stringify(options.pathScope));
+		}
+	}
+
 	// If all resources were filtered out, they may have moved OUT of scope
 	// Remove them from cache if they exist (handles file moves out of current view)
 	if (filteredResources.length === 0) {
@@ -526,6 +555,9 @@ export function updateBatchResources<O>(
 	}
 
 	queryClient.setQueryData<O>(queryKey, (oldData: any) => {
+		if (options.debug) {
+			console.log(`[useNormalizedQuery] ${wireMethod} setQueryData: oldData has`, Array.isArray(oldData) ? oldData.length : Object.keys(oldData || {}).join(','), `adding ${filteredResources.length} resources`);
+		}
 		if (!oldData) return oldData;
 
 		// Handle array responses

packages/ts-client/src/transport.ts

@@ -59,19 +59,20 @@ export class TauriTransport implements Transport {
 		callback: (event: any) => void,
 		options?: SubscriptionOptions,
 	): Promise<() => void> {
-		// Start the event subscription on the backend
-		// Returns subscription ID for cleanup
 		const args = {
 			eventTypes: options?.event_types ?? DEFAULT_EVENT_SUBSCRIPTION,
 			filter: options?.filter ?? null,
 		};
-		const subscriptionId = await this.invoke("subscribe_to_events", args);
 
-		// Listen to forwarded events from Tauri
+		// Listen FIRST so buffered events replayed by the daemon are not lost.
+		// subscribe_to_events creates a TCP connection that may emit events
+		// before the invoke promise resolves.
 		const unlisten = await this.listen("core-event", (tauriEvent: any) => {
 			callback(tauriEvent.payload);
 		});
 
+		const subscriptionId = await this.invoke("subscribe_to_events", args);
+
 		// Return cleanup function that properly unsubscribes
 		return async () => {
 			unlisten();