diff --git a/app/MindWork AI Studio/Tools/PluginSystem/PluginFactory.HotReload.cs b/app/MindWork AI Studio/Tools/PluginSystem/PluginFactory.HotReload.cs index 0505787c..06c3cf1e 100644 --- a/app/MindWork AI Studio/Tools/PluginSystem/PluginFactory.HotReload.cs +++ b/app/MindWork AI Studio/Tools/PluginSystem/PluginFactory.HotReload.cs @@ -19,7 +19,6 @@ public static partial class PluginFactory HOT_RELOAD_WATCHER.NotifyFilter = NotifyFilters.CreationTime | NotifyFilters.DirectoryName | NotifyFilters.FileName - | NotifyFilters.LastAccess | NotifyFilters.LastWrite | NotifyFilters.Size; @@ -102,4 +101,4 @@ public static partial class PluginFactory LOG.LogError(e, $"Error while handling hot reload event for file '{args.FullPath}' with change type '{args.ChangeType}'."); } } -} \ No newline at end of file +} diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Files.cs b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Files.cs index 9f10af6f..5c0d6de5 100644 --- a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Files.cs +++ b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Files.cs @@ -221,11 +221,7 @@ public sealed partial class DataSourceEmbeddingService private bool IsSkippedRagFile(FileInfo file) { - var extension = file.Extension.TrimStart('.'); - if (SKIPPED_RAG_FILE_EXTENSIONS.Contains(extension, StringComparer.OrdinalIgnoreCase)) - return true; - - if (file.Name.StartsWith(OFFICE_LOCK_FILE_PREFIX, StringComparison.Ordinal)) + if (IsSkippedRagFileName(file.Name)) return true; try @@ -242,6 +238,13 @@ public sealed partial class DataSourceEmbeddingService } } + private static bool IsSkippedRagFileName(string fileName) + { + var extension = Path.GetExtension(fileName).TrimStart('.'); + return SKIPPED_RAG_FILE_EXTENSIONS.Contains(extension, StringComparer.OrdinalIgnoreCase) + || fileName.StartsWith(OFFICE_LOCK_FILE_PREFIX, StringComparison.Ordinal); + } + private bool IsSkippedRagDirectory(string path) { try diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Watchers.cs b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Watchers.cs index ad6302b3..ba213f91 100644 --- a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Watchers.cs +++ b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Watchers.cs @@ -67,16 +67,16 @@ public sealed partial class DataSourceEmbeddingService NotifyFilter = NotifyFilters.FileName | NotifyFilters.DirectoryName | NotifyFilters.LastWrite | NotifyFilters.CreationTime | NotifyFilters.Size, }; - watcher.Changed += (_, _) => this.OnWatchedDataSourceChanged(dataSourceId); - watcher.Deleted += (_, _) => this.OnWatchedDataSourceChanged(dataSourceId); - watcher.Created += (_, _) => this.OnWatchedDataSourceChanged(dataSourceId); - watcher.Renamed += (_, _) => this.OnWatchedDataSourceChanged(dataSourceId); + watcher.Changed += (_, args) => this.OnWatchedDataSourceChanged(dataSourceId, configuration, args); + watcher.Deleted += (_, args) => this.OnWatchedDataSourceChanged(dataSourceId, configuration, args); + watcher.Created += (_, args) => this.OnWatchedDataSourceChanged(dataSourceId, configuration, args); + watcher.Renamed += (_, args) => this.OnWatchedDataSourceChanged(dataSourceId, configuration, args); watcher.Error += (_, args) => { logger.LogWarning(args.GetException(), "The file watcher for data source '{DataSourceId}' failed. Recreating it.", dataSourceId); this.RemoveWatcher(dataSourceId); this.EnsureWatcher(dataSourceId); - this.OnWatchedDataSourceChanged(dataSourceId); + this.ScheduleWatchedDataSourceRefresh(dataSourceId); }; watcher.EnableRaisingEvents = true; return watcher; @@ -112,12 +112,32 @@ public sealed partial class DataSourceEmbeddingService this.watchers.Clear(); } - private void OnWatchedDataSourceChanged(string dataSourceId) + private void OnWatchedDataSourceChanged(string dataSourceId, DataSourceWatcherConfiguration configuration, FileSystemEventArgs args) + { + if (!this.IsRelevantWatcherEvent(configuration, args)) + { + logger.LogDebug( + "Ignoring file system change for data source '{DataSourceId}' at '{Path}' (event={ChangeType}) because the path is not part of the RAG index.", + dataSourceId, + args.FullPath, + args.ChangeType); + return; + } + + logger.LogDebug( + "Detected relevant file system change for data source '{DataSourceId}' at '{Path}' (event={ChangeType}). Scheduling a debounced embedding run.", + dataSourceId, + args.FullPath, + args.ChangeType); + + this.ScheduleWatchedDataSourceRefresh(dataSourceId); + } + + private void ScheduleWatchedDataSourceRefresh(string dataSourceId) { if (!settingsManager.ConfigurationData.DataSourceIndexing.AutomaticRefresh) return; - logger.LogDebug("Detected file system change for data source '{DataSourceId}'. Scheduling a debounced embedding run.", dataSourceId); var debounceToken = new CancellationTokenSource(); lock (this.watcherDebounceLock) @@ -142,7 +162,7 @@ public sealed partial class DataSourceEmbeddingService if (dataSource is not null) { logger.LogInformation("Queueing data source '{DataSourceName}' ({DataSourceId}) after file system changes settled.", dataSource.Name, dataSource.Id); - await this.QueueDataSourceAsync(dataSource); + await this.QueueDataSourceAsync(dataSource, true); } } catch (OperationCanceledException) @@ -200,6 +220,42 @@ public sealed partial class DataSourceEmbeddingService } } + private bool IsRelevantWatcherEvent(DataSourceWatcherConfiguration configuration, FileSystemEventArgs args) + { + if (args is RenamedEventArgs renamedArgs) + { + return this.IsRelevantWatcherPath(configuration, renamedArgs.FullPath, args.ChangeType) + || this.IsRelevantWatcherPath(configuration, renamedArgs.OldFullPath, args.ChangeType); + } + + return this.IsRelevantWatcherPath(configuration, args.FullPath, args.ChangeType); + } + + private bool IsRelevantWatcherPath(DataSourceWatcherConfiguration configuration, string path, WatcherChangeTypes changeType) + { + if (string.IsNullOrWhiteSpace(path)) + return false; + + var fileName = Path.GetFileName(path); + if (string.IsNullOrWhiteSpace(fileName)) + return true; + + if (!configuration.IncludeSubdirectories && !string.Equals(fileName, configuration.Filter, StringComparison.OrdinalIgnoreCase)) + return false; + + if (Directory.Exists(path)) + return true; + + if (IsSkippedRagFileName(fileName)) + return false; + + if (this.IsSupportedRagFilePath(path)) + return true; + + return changeType is WatcherChangeTypes.Deleted or WatcherChangeTypes.Renamed + && string.IsNullOrWhiteSpace(Path.GetExtension(path)); + } + private static DataSourceWatcherConfiguration? GetWatchConfiguration(IDataSource dataSource) => dataSource switch { DataSourceLocalDirectory localDirectory when Directory.Exists(localDirectory.Path) => new DataSourceWatcherConfiguration( diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs index c74decbf..f5434e44 100644 --- a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs +++ b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs @@ -22,10 +22,21 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM private readonly Channel queue = Channel.CreateUnbounded(); private readonly ConcurrentDictionary queuedIds = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentDictionary runningIds = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentDictionary pendingQueueIds = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary statuses = new(StringComparer.OrdinalIgnoreCase); + private readonly object queueStateLock = new(); private static string TB(string fallbackEN) => I18N.I.T(fallbackEN, typeof(DataSourceEmbeddingService).Namespace, nameof(DataSourceEmbeddingService)); + private enum DataSourceQueueRequestResult + { + QUEUED, + ALREADY_QUEUED, + RUNNING, + RUNNING_MARKED_PENDING, + } + public IReadOnlyList GetStatuses() { return this.statuses.Values @@ -61,12 +72,17 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM } public Task QueueAllInternalDataSourcesAsync() + { + return this.QueueAllInternalDataSourcesAsync(true); + } + + private Task QueueAllInternalDataSourcesAsync(bool queueAfterCurrentRun) { this.RefreshWatchers(); var tasks = settingsManager.ConfigurationData.DataSources .Where(this.IsSupportedInternalDataSource) - .Select(this.QueueDataSourceAsync); + .Select(dataSource => this.QueueDataSourceAsync(dataSource, queueAfterCurrentRun)); return Task.WhenAll(tasks); } @@ -79,7 +95,7 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM return Task.CompletedTask; } - return this.QueueAllInternalDataSourcesAsync(); + return this.QueueAllInternalDataSourcesAsync(false); } public void RefreshAutomaticWatchers() @@ -87,20 +103,40 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM this.RefreshWatchers(); } - public async Task QueueDataSourceAsync(IDataSource dataSource) + public Task QueueDataSourceAsync(IDataSource dataSource) + { + return this.QueueDataSourceAsync(dataSource, true); + } + + private async Task QueueDataSourceAsync(IDataSource dataSource, bool queueAfterCurrentRun) { if (!this.IsSupportedInternalDataSource(dataSource)) return; - logger.LogInformation("Queueing data source '{DataSourceName}' ({DataSourceId}) for background embeddings.", dataSource.Name, dataSource.Id); this.RefreshWatchers(); - logger.LogDebug("Adding watcher for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); + logger.LogDebug("Ensured watcher for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); + var queueRequestResult = this.TryReserveDataSourceQueueSlot(dataSource.Id, queueAfterCurrentRun); + switch (queueRequestResult) + { + case DataSourceQueueRequestResult.ALREADY_QUEUED: + logger.LogDebug("Data source '{DataSourceName}' ({DataSourceId}) is already queued for background embeddings. Ignoring duplicate queue request.", dataSource.Name, dataSource.Id); + return; + + case DataSourceQueueRequestResult.RUNNING: + logger.LogDebug("Data source '{DataSourceName}' ({DataSourceId}) is already being embedded. Ignoring duplicate queue request.", dataSource.Name, dataSource.Id); + return; + + case DataSourceQueueRequestResult.RUNNING_MARKED_PENDING: + logger.LogDebug("Data source '{DataSourceName}' ({DataSourceId}) is already being embedded. Scheduled one follow-up embedding run.", dataSource.Name, dataSource.Id); + return; + } + + logger.LogInformation("Queueing data source '{DataSourceName}' ({DataSourceId}) for background embeddings.", dataSource.Name, dataSource.Id); if (!this.statuses.TryGetValue(dataSource.Id, out var currentStatus) || currentStatus.State is not DataSourceEmbeddingState.RUNNING) this.UpsertStatus(this.CreateStatus(dataSource, DataSourceEmbeddingState.QUEUED, currentStatus?.TotalFiles ?? 0, currentStatus?.IndexedFiles ?? 0, currentStatus?.FailedFiles ?? 0)); logger.LogDebug("Upserting status for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); - if (this.queuedIds.TryAdd(dataSource.Id, 0)) - await this.queue.Writer.WriteAsync(dataSource.Id); + await this.queue.Writer.WriteAsync(dataSource.Id); logger.LogDebug("Queued data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); } @@ -122,16 +158,18 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM while (!stoppingToken.IsCancellationRequested) { var dataSourceId = await this.queue.Reader.ReadAsync(stoppingToken); - this.queuedIds.TryRemove(dataSourceId, out _); + this.MarkDataSourceRunStarted(dataSourceId); - var dataSource = settingsManager.ConfigurationData.DataSources - .FirstOrDefault(source => source.Id.Equals(dataSourceId, StringComparison.OrdinalIgnoreCase)); - - if (dataSource is null || !this.IsSupportedInternalDataSource(dataSource)) - continue; + IDataSource? dataSource = null; try { + dataSource = settingsManager.ConfigurationData.DataSources + .FirstOrDefault(source => source.Id.Equals(dataSourceId, StringComparison.OrdinalIgnoreCase)); + + if (dataSource is null || !this.IsSupportedInternalDataSource(dataSource)) + continue; + await this.ProcessDataSourceAsync(dataSource, stoppingToken); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) @@ -140,8 +178,19 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM } catch (Exception exception) { - logger.LogError(exception, "Background embedding failed for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); - this.UpsertStatus(this.GetFallbackStatus(dataSource, exception.Message)); + if (dataSource is null) + { + logger.LogError(exception, "Background embedding failed for data source '{DataSourceId}'.", dataSourceId); + } + else + { + logger.LogError(exception, "Background embedding failed for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); + this.UpsertStatus(this.GetFallbackStatus(dataSource, exception.Message)); + } + } + finally + { + await this.QueuePendingDataSourceRunAsync(dataSourceId, stoppingToken); } } } @@ -613,6 +662,92 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM return this.CreateStatus(dataSource, DataSourceEmbeddingState.FAILED, 0, 0, 1, lastError: errorMessage); } + private DataSourceQueueRequestResult TryReserveDataSourceQueueSlot(string dataSourceId, bool queueAfterCurrentRun) + { + lock (this.queueStateLock) + { + if (this.runningIds.ContainsKey(dataSourceId)) + { + if (queueAfterCurrentRun && this.pendingQueueIds.TryAdd(dataSourceId, 0)) + return DataSourceQueueRequestResult.RUNNING_MARKED_PENDING; + + return DataSourceQueueRequestResult.RUNNING; + } + + if (!this.queuedIds.TryAdd(dataSourceId, 0)) + return DataSourceQueueRequestResult.ALREADY_QUEUED; + + return DataSourceQueueRequestResult.QUEUED; + } + } + + private void MarkDataSourceRunStarted(string dataSourceId) + { + lock (this.queueStateLock) + { + this.queuedIds.TryRemove(dataSourceId, out _); + this.runningIds.TryAdd(dataSourceId, 0); + } + } + + private bool TryCompleteDataSourceRun(string dataSourceId, bool allowPendingRequeue) + { + lock (this.queueStateLock) + { + this.runningIds.TryRemove(dataSourceId, out _); + + if (!this.pendingQueueIds.TryRemove(dataSourceId, out _)) + return false; + + return allowPendingRequeue && this.queuedIds.TryAdd(dataSourceId, 0); + } + } + + private void ReleaseQueuedDataSourceRun(string dataSourceId) + { + lock (this.queueStateLock) + { + this.queuedIds.TryRemove(dataSourceId, out _); + } + } + + private async Task QueuePendingDataSourceRunAsync(string dataSourceId, CancellationToken token) + { + var dataSource = token.IsCancellationRequested + ? null + : settingsManager.ConfigurationData.DataSources + .FirstOrDefault(source => source.Id.Equals(dataSourceId, StringComparison.OrdinalIgnoreCase)); + + if (!this.TryCompleteDataSourceRun(dataSourceId, dataSource is not null && this.IsSupportedInternalDataSource(dataSource))) + return; + + if (dataSource is null) + { + this.ReleaseQueuedDataSourceRun(dataSourceId); + return; + } + + logger.LogInformation("Queueing one follow-up embedding run for data source '{DataSourceName}' ({DataSourceId}) after changes arrived during the previous run.", dataSource.Name, dataSource.Id); + + this.statuses.TryGetValue(dataSource.Id, out var currentStatus); + this.UpsertStatus(this.CreateStatus( + dataSource, + DataSourceEmbeddingState.QUEUED, + currentStatus?.TotalFiles ?? 0, + currentStatus?.IndexedFiles ?? 0, + currentStatus?.FailedFiles ?? 0, + lastError: currentStatus?.LastError ?? string.Empty)); + + try + { + await this.queue.Writer.WriteAsync(dataSourceId, token); + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + this.ReleaseQueuedDataSourceRun(dataSourceId); + } + } + private void UpsertStatus(DataSourceEmbeddingStatus status) { this.statuses[status.DataSourceId] = status;