Forge
search
csharpdeeb25a2
1using System.Collections.Concurrent;
2using System.Linq;
3using System.Threading.Channels;
4using CascadeIDE.Cockpit.ComputingUnits.HybridIndex;
5using CascadeIDE.Cockpit.DataBus;
6using CascadeIDE.Contracts;
7using HybridCodebaseIndex.Core;
8using HybridCodebaseIndex.Core.Indexing;
9
10namespace CascadeIDE.Features.HybridIndex.Application;
11
12/// <summary>Оркестратор in-proc Hybrid Codebase Index (ADR 0106): watcher + debounce + публикация статуса в DataBus.</summary>
13[ApplicationOrchestrator("hybrid-index-in-proc")]
14[DataBusPublisher("hybrid-index-status")]
15public sealed class HybridIndexOrchestrator : IDisposable, IHybridIndexOrchestratorSearch
16{
17 private sealed record WatchKey(string WorkspaceRoot, string? SolutionPath)
18 {
19 public override string ToString() =>
20 string.IsNullOrWhiteSpace(SolutionPath) ? WorkspaceRoot : $"{WorkspaceRoot} | {SolutionPath}";
21 }
22
23 private sealed class WatcherState : IDisposable
24 {
25 private readonly CodebaseIndexService _service;
26 private readonly IDataBus _dataBus;
27 private readonly string _workspaceRoot;
28 private readonly string? _solutionPath;
29 private readonly int _debounceMs;
30 private readonly string _indexDirRelativeMarker;
31 private readonly string _indexDirRelative;
32
33 private readonly Channel<int> _poke;
34 private readonly CancellationTokenSource _cts;
35 private readonly Task _loop;
36 private readonly FileSystemWatcher _fsw;
37
38 public WatcherState(
39 CodebaseIndexService service,
40 IDataBus dataBus,
41 string workspaceRoot,
42 string? solutionPath,
43 int debounceMs,
44 string indexDirectoryRelativeForNoiseFilter)
45 {
46 _service = service;
47 _dataBus = dataBus;
48 _workspaceRoot = workspaceRoot;
49 _solutionPath = solutionPath;
50 _debounceMs = Math.Clamp(debounceMs, 50, 60_000);
51 _indexDirRelative = indexDirectoryRelativeForNoiseFilter ?? "";
52 _indexDirRelativeMarker = _indexDirRelative.Trim().TrimEnd(Path.DirectorySeparatorChar, Path.AltDirectorySeparatorChar);
53
54 _cts = new CancellationTokenSource();
55 _poke = Channel.CreateUnbounded<int>(new UnboundedChannelOptions
56 {
57 SingleReader = true,
58 SingleWriter = false,
59 AllowSynchronousContinuations = false,
60 });
61
62 _fsw = new FileSystemWatcher(_workspaceRoot)
63 {
64 IncludeSubdirectories = true,
65 EnableRaisingEvents = true,
66 NotifyFilter =
67 NotifyFilters.FileName
68 | NotifyFilters.DirectoryName
69 | NotifyFilters.LastWrite
70 | NotifyFilters.Size
71 | NotifyFilters.CreationTime,
72 };
73
74 _fsw.Changed += OnAny;
75 _fsw.Created += OnAny;
76 _fsw.Deleted += OnAny;
77 _fsw.Renamed += OnAny;
78 _fsw.Error += OnError;
79
80 _loop = Task.Run(LoopAsync, _cts.Token);
81 }
82
83 private void OnAny(object sender, FileSystemEventArgs e)
84 {
85 // Best-effort: ignore self-induced churn from index directory (matches configured `index_dir`).
86 // A false-positive poke is ok because reindex is incremental and debounced.
87 if (IsIndexArtifactPath(e.FullPath, _indexDirRelativeMarker))
88 return;
89 _poke.Writer.TryWrite(0);
90 }
91
92 private static bool IsIndexArtifactPath(string? fullPath, string indexDirRelativeMarker)
93 {
94 if (string.IsNullOrEmpty(fullPath) || string.IsNullOrEmpty(indexDirRelativeMarker))
95 return false;
96 return fullPath.Contains(indexDirRelativeMarker, StringComparison.OrdinalIgnoreCase);
97 }
98
99 private void OnError(object sender, ErrorEventArgs e)
100 {
101 // FileSystemWatcher can drop events when buffer overflows; trigger a catch-up reindex.
102 _poke.Writer.TryWrite(0);
103 }
104
105 private async Task LoopAsync()
106 {
107 var ct = _cts.Token;
108
109 // Proactively publish initial status (best-effort).
110 await PublishStatusAsync(CancellationToken.None).ConfigureAwait(false);
111
112 var nextDelay = Task.Delay(Timeout.Infinite, ct);
113 var hasPending = false;
114
115 while (!ct.IsCancellationRequested)
116 {
117 var read = _poke.Reader.ReadAsync(ct).AsTask();
118 var completed = await Task.WhenAny(read, nextDelay).ConfigureAwait(false);
119
120 if (completed == read)
121 {
122 hasPending = true;
123 _ = await read.ConfigureAwait(false);
124 nextDelay = Task.Delay(_debounceMs, ct);
125 continue;
126 }
127
128 if (!hasPending)
129 {
130 nextDelay = Task.Delay(Timeout.Infinite, ct);
131 continue;
132 }
133
134 hasPending = false;
135 nextDelay = Task.Delay(Timeout.Infinite, ct);
136
137 try
138 {
139 var observers = HybridIndexIntercomReindexObservers.Create(_solutionPath, _indexDirRelative);
140 await _service.FullReindexAsync(_workspaceRoot, _solutionPath, observers, ct).ConfigureAwait(false);
141 }
142 catch (OperationCanceledException) when (ct.IsCancellationRequested)
143 {
144 return;
145 }
146 catch
147 {
148 // best-effort background loop; status tool surfaces last error
149 }
150
151 await PublishStatusAsync(CancellationToken.None).ConfigureAwait(false);
152 }
153 }
154
155 private async Task PublishStatusAsync(CancellationToken ct)
156 {
157 try
158 {
159 var st = await _service.GetStatusAsync(_workspaceRoot, _solutionPath, ct).ConfigureAwait(false);
160 _dataBus.Publish(HybridIndexStateChangedUnit.FromCoreStatus(st, _workspaceRoot, _solutionPath));
161 }
162 catch
163 {
164 // ignore
165 }
166 }
167
168 public void Poke() => _poke.Writer.TryWrite(0);
169
170 public void Dispose()
171 {
172 _fsw.EnableRaisingEvents = false;
173 _fsw.Changed -= OnAny;
174 _fsw.Created -= OnAny;
175 _fsw.Deleted -= OnAny;
176 _fsw.Renamed -= OnAny;
177 _fsw.Error -= OnError;
178 _fsw.Dispose();
179
180 _cts.Cancel();
181 _poke.Writer.TryComplete();
182 try { _loop.GetAwaiter().GetResult(); } catch { /* ignore */ }
183 _cts.Dispose();
184 }
185 }
186
187 private CodebaseIndexService _service;
188 private readonly IDataBus _dataBus;
189 private string _indexDirectoryRelative;
190 private readonly ConcurrentDictionary<WatchKey, WatcherState> _watchers = new();
191
192 public HybridIndexOrchestrator(IDataBus dataBus, string indexDirectoryRelative)
193 {
194 _dataBus = dataBus;
195 _indexDirectoryRelative = HybridIndexIndexDirectoryRelative.ResolveOrDefault(indexDirectoryRelative);
196 _service = new CodebaseIndexService(indexDirectoryRelative: _indexDirectoryRelative);
197 }
198
199 /// <summary>
200 /// Recreates the in-proc <see cref="CodebaseIndexService"/> and drops active watchers.
201 /// Callers should re-apply <see cref="SetEnabled"/> for the current workspace/solution.
202 /// </summary>
203 public void SetIndexDirectoryRelative(string indexDirectoryRelative)
204 {
205 var normalized = HybridIndexIndexDirectoryRelative.ResolveOrDefault(indexDirectoryRelative);
206 if (string.Equals(_indexDirectoryRelative, normalized, StringComparison.Ordinal))
207 return;
208
209 foreach (var key in _watchers.Keys.ToArray())
210 {
211 if (_watchers.TryRemove(key, out var st))
212 st.Dispose();
213 }
214
215 _indexDirectoryRelative = normalized;
216 _service = new CodebaseIndexService(indexDirectoryRelative: normalized);
217 }
218
219 public void SetEnabled(string workspaceRoot, string? solutionPath, bool enabled, int debounceMs = 750)
220 {
221 if (string.IsNullOrWhiteSpace(workspaceRoot))
222 return;
223
224 var root = CanonicalFilePath.Normalize(workspaceRoot.TrimEnd(Path.DirectorySeparatorChar));
225 var key = new WatchKey(root, string.IsNullOrWhiteSpace(solutionPath) ? null : solutionPath.Trim());
226
227 if (!enabled)
228 {
229 if (_watchers.TryRemove(key, out var st))
230 st.Dispose();
231 return;
232 }
233
234 _watchers.AddOrUpdate(
235 key,
236 static (k, arg) => new WatcherState(
237 arg.service,
238 arg.bus,
239 arg.root,
240 arg.solutionPath,
241 arg.debounceMs,
242 arg.indexDir),
243 static (k, existing, arg) =>
244 {
245 existing.Dispose();
246 return new WatcherState(
247 arg.service,
248 arg.bus,
249 arg.root,
250 arg.solutionPath,
251 arg.debounceMs,
252 arg.indexDir);
253 },
254 (service: _service, bus: _dataBus, root, solutionPath: key.SolutionPath, debounceMs, indexDir: _indexDirectoryRelative));
255 }
256
257 public void Poke(string workspaceRoot, string? solutionPath)
258 {
259 var root = CanonicalFilePath.Normalize((workspaceRoot ?? "").TrimEnd(Path.DirectorySeparatorChar));
260 var key = new WatchKey(root, string.IsNullOrWhiteSpace(solutionPath) ? null : solutionPath.Trim());
261 if (_watchers.TryGetValue(key, out var st))
262 st.Poke();
263 }
264
265 public Task<IndexStatus> GetIndexStatusAsync(string workspaceRoot, string? solutionPath, CancellationToken cancellationToken = default) =>
266 _service.GetStatusAsync(workspaceRoot, solutionPath, cancellationToken);
267
268 public Task<(SearchResponse Response, string? Error)> SearchHybridAsync(
269 string workspaceRoot,
270 string? solutionPath,
271 string query,
272 int topN,
273 string? pathPrefix,
274 IReadOnlyList<string>? excludePathPrefixes,
275 IReadOnlyList<string>? extensions,
276 bool semantic,
277 double alpha,
278 double beta,
279 int vecTopK,
280 CancellationToken cancellationToken = default) =>
281 _service.SearchHybridAsync(workspaceRoot, solutionPath, query, topN, pathPrefix, excludePathPrefixes, extensions, semantic, alpha, beta, vecTopK, cancellationToken);
282
283 public Task<ExplainHitResponse> ExplainHitAsync(string workspaceRoot, string? solutionPath, long hitId, CancellationToken cancellationToken = default) =>
284 _service.ExplainHitAsync(workspaceRoot, solutionPath, hitId, cancellationToken);
285
286 /// <summary>Инкрементальный или полный reindex через Core; затем снимок в DataBus (страница HIS).</summary>
287 public async Task<ReindexSummary> RunReindexWithPublishAsync(
288 string workspaceRoot,
289 string? solutionPath,
290 bool fullRebuild,
291 CancellationToken cancellationToken = default)
292 {
293 if (string.IsNullOrWhiteSpace(workspaceRoot))
294 throw new ArgumentException("workspace_root required", nameof(workspaceRoot));
295
296 var root = CanonicalFilePath.Normalize(workspaceRoot.TrimEnd(Path.DirectorySeparatorChar));
297 var sln = string.IsNullOrWhiteSpace(solutionPath) ? null : solutionPath.Trim();
298 var observers = HybridIndexIntercomReindexObservers.Create(sln, _indexDirectoryRelative);
299 ReindexSummary summary = fullRebuild
300 ? await _service.FullRebuildAsync(root, sln, observers, cancellationToken).ConfigureAwait(false)
301 : await _service.FullReindexAsync(root, sln, observers, cancellationToken).ConfigureAwait(false);
302 await PublishHybridIndexSnapshotAsync(root, sln, cancellationToken).ConfigureAwait(false);
303 return summary;
304 }
305
306 private async Task PublishHybridIndexSnapshotAsync(string rootNormalized, string? solutionPathTrimmedOrNull, CancellationToken cancellationToken)
307 {
308 try
309 {
310 var st = await _service.GetStatusAsync(rootNormalized, solutionPathTrimmedOrNull, cancellationToken).ConfigureAwait(false);
311 _dataBus.Publish(HybridIndexStateChangedUnit.FromCoreStatus(st, rootNormalized, solutionPathTrimmedOrNull));
312 }
313 catch
314 {
315 // ignore
316 }
317 }
318
319 /// <summary>
320 /// Full reindex outside the watcher loop (например watcher выключен или master switch HCI off).
321 /// Публикует <see cref="HybridIndexStateChanged"/> в DataBus по завершении.
322 /// </summary>
323 public async Task RunFullReindexAndPublishStatusAsync(
324 string workspaceRoot,
325 string? solutionPath,
326 CancellationToken cancellationToken = default)
327 {
328 if (string.IsNullOrWhiteSpace(workspaceRoot))
329 return;
330
331 var root = CanonicalFilePath.Normalize(workspaceRoot.TrimEnd(Path.DirectorySeparatorChar));
332 var sln = string.IsNullOrWhiteSpace(solutionPath) ? null : solutionPath.Trim();
333 try
334 {
335 var observers = HybridIndexIntercomReindexObservers.Create(sln, _indexDirectoryRelative);
336 await _service.FullReindexAsync(root, sln, observers, cancellationToken).ConfigureAwait(false);
337 }
338 catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
339 {
340 return;
341 }
342 catch
343 {
344 // surfaced via GetStatusAsync below
345 }
346
347 await PublishHybridIndexSnapshotAsync(root, sln, cancellationToken).ConfigureAwait(false);
348 }
349
350 public void Dispose()
351 {
352 foreach (var key in _watchers.Keys.ToArray())
353 {
354 if (_watchers.TryRemove(key, out var st))
355 st.Dispose();
356 }
357 }
358}
359
360
View only · write via MCP/CIDE