Forge
search
csharpdeeb25a2
1using System.Collections.Concurrent;
2using System.Diagnostics;
3using System.Text.Json;
4
5namespace CascadeIDE.Features.Agent.Environment;
6
7/// <summary>JSON-lines клиент к <c>BuildVerifyWorker serve</c> (long-lived daemon).</summary>
8public sealed class BuildVerifyWorkerDaemonClient : IAsyncDisposable
9{
10 private readonly string _workerDllPath;
11 private readonly SemaphoreSlim _writeGate = new(1, 1);
12 private readonly ConcurrentDictionary<string, TaskCompletionSource<JsonElement>> _pending = new();
13 private readonly object _processGate = new();
14 private Process? _process;
15 private StreamWriter? _stdin;
16 private Task? _stdoutPump;
17 private int _nextRequestId;
18 private bool _disposed;
19
20 public BuildVerifyWorkerDaemonClient(string workerDllPath)
21 {
22 _workerDllPath = Path.GetFullPath(workerDllPath);
23 }
24
25 public bool IsRunning
26 {
27 get
28 {
29 lock (_processGate)
30 return _process is { HasExited: false };
31 }
32 }
33
34 public async Task EnsureStartedAsync(CancellationToken cancellationToken = default)
35 {
36 ObjectDisposedException.ThrowIf(_disposed, this);
37
38 if (IsRunning)
39 return;
40
41 lock (_processGate)
42 {
43 if (_process is { HasExited: false })
44 return;
45
46 if (!File.Exists(_workerDllPath))
47 throw new FileNotFoundException("BuildVerifyWorker assembly not found.", _workerDllPath);
48
49 var psi = new ProcessStartInfo("dotnet", $"exec \"{_workerDllPath}\" serve")
50 {
51 RedirectStandardInput = true,
52 RedirectStandardOutput = true,
53 RedirectStandardError = true,
54 UseShellExecute = false,
55 CreateNoWindow = true,
56 };
57
58 _process = Process.Start(psi)
59 ?? throw new InvalidOperationException("Failed to start BuildVerifyWorker serve process.");
60
61 _stdin = new StreamWriter(_process.StandardInput.BaseStream) { AutoFlush = true, NewLine = "\n" };
62 _stdoutPump = Task.Run(() => PumpStdoutAsync(_process.StandardOutput, _process));
63 }
64
65 await PingAsync(cancellationToken).ConfigureAwait(false);
66 }
67
68 public async Task<JsonElement> SendAsync(
69 IReadOnlyDictionary<string, object?> payload,
70 CancellationToken cancellationToken = default)
71 {
72 await EnsureStartedAsync(cancellationToken).ConfigureAwait(false);
73
74 var id = Interlocked.Increment(ref _nextRequestId).ToString();
75 var tcs = new TaskCompletionSource<JsonElement>(TaskCreationOptions.RunContinuationsAsynchronously);
76 if (!_pending.TryAdd(id, tcs))
77 throw new InvalidOperationException("Failed to register IPC request.");
78
79 var body = new Dictionary<string, object?>(payload) { ["id"] = id };
80 var line = JsonSerializer.Serialize(body);
81
82 await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
83 try
84 {
85 if (_stdin is null || _process is { HasExited: true })
86 throw new InvalidOperationException("BuildVerifyWorker daemon is not running.");
87
88 await _stdin.WriteLineAsync(line.AsMemory(), cancellationToken).ConfigureAwait(false);
89 }
90 finally
91 {
92 _writeGate.Release();
93 }
94
95 using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
96 timeout.CancelAfter(TimeSpan.FromMinutes(30));
97 try
98 {
99 return await tcs.Task.WaitAsync(timeout.Token).ConfigureAwait(false);
100 }
101 catch
102 {
103 _pending.TryRemove(id, out _);
104 throw;
105 }
106 }
107
108 public async Task PingAsync(CancellationToken cancellationToken = default)
109 {
110 var resp = await SendAsync(new Dictionary<string, object?> { ["op"] = "ping" }, cancellationToken)
111 .ConfigureAwait(false);
112 if (!resp.TryGetProperty("ok", out var ok) || ok.ValueKind != JsonValueKind.True)
113 throw new InvalidOperationException("BuildVerifyWorker daemon ping failed.");
114 }
115
116 public async ValueTask DisposeAsync()
117 {
118 if (_disposed)
119 return;
120
121 _disposed = true;
122
123 try
124 {
125 if (IsRunning)
126 {
127 await SendAsync(
128 new Dictionary<string, object?> { ["op"] = "shutdown" },
129 CancellationToken.None)
130 .ConfigureAwait(false);
131 }
132 }
133 catch
134 {
135 //
136 }
137
138 lock (_processGate)
139 {
140 try
141 {
142 if (_process is { HasExited: false })
143 _process.Kill(entireProcessTree: true);
144 }
145 catch
146 {
147 //
148 }
149
150 _process?.Dispose();
151 _process = null;
152 _stdin = null;
153 }
154
155 FailAllPending("daemon_disposed");
156 _writeGate.Dispose();
157 }
158
159 private async Task PumpStdoutAsync(StreamReader reader, Process process)
160 {
161 try
162 {
163 while (true)
164 {
165 var line = await reader.ReadLineAsync().ConfigureAwait(false);
166 if (line is null)
167 break;
168
169 if (line.Length == 0)
170 continue;
171
172 JsonDocument doc;
173 try
174 {
175 doc = JsonDocument.Parse(line);
176 }
177 catch (JsonException)
178 {
179 continue;
180 }
181
182 using (doc)
183 {
184 var root = doc.RootElement;
185 if (!root.TryGetProperty("id", out var idEl))
186 continue;
187
188 var id = idEl.GetString();
189 if (string.IsNullOrEmpty(id))
190 continue;
191
192 if (id == "0")
193 continue;
194
195 if (_pending.TryRemove(id, out var tcs))
196 tcs.TrySetResult(root.Clone());
197 }
198 }
199 }
200 finally
201 {
202 if (!process.HasExited)
203 {
204 try
205 {
206 process.Kill(entireProcessTree: true);
207 }
208 catch
209 {
210 //
211 }
212 }
213
214 FailAllPending("daemon_exited");
215 }
216 }
217
218 private void FailAllPending(string reason)
219 {
220 foreach (var pair in _pending.ToArray())
221 {
222 if (_pending.TryRemove(pair.Key, out var tcs))
223 tcs.TrySetException(new InvalidOperationException(reason));
224 }
225 }
226}
227
View only · write via MCP/CIDE