Forge
search
csharpdeeb25a2
1using System.Collections.Concurrent;
2using System.Text.Json;
3using CascadeIDE.Cockpit.DataBus;
4using DotNetBuildTest.Core;
5
6namespace CascadeIDE.Features.Agent.Environment;
7
8/// <summary>Очередь environment tasks с DataBus-событиями (ADR 0148 W1).</summary>
9public sealed class EnvironmentTaskRunner
10{
11 private readonly IDataBus _dataBus;
12 private readonly IEnvironmentJobBackend _backend;
13 private readonly string _hostKind;
14 private readonly ConcurrentDictionary<string, ConcurrentBag<string>> _coreJobsByRun = new();
15
16 /// <summary>Test seam: override job status (return null → <see cref="AgentEnvironmentTaskDied"/>).</summary>
17 public Func<string, object?>? TestJobStatusFactory { get; set; }
18
19 public EnvironmentTaskRunner(IDataBus dataBus, IEnvironmentJobBackend backend)
20 {
21 _dataBus = dataBus;
22 _backend = backend;
23 _hostKind = backend.HostKind;
24 }
25
26 public EnvironmentTaskRunner(IDataBus dataBus, BuildTestJobCoordinator coordinator, string hostKind = "supervised-inproc")
27 : this(dataBus, new InProcessEnvironmentJobBackend(coordinator, hostKind))
28 {
29 }
30
31 public IEnvironmentJobBackend Backend => _backend;
32
33 public async Task<EnvironmentTaskOutcome> RunBuildAsync(
34 string runId,
35 string solutionPath,
36 bool waitForCompletion,
37 CancellationToken cancellationToken = default)
38 {
39 return await RunCoreJobAsync(
40 runId,
41 "msbuild.compile",
42 BuildTestJobKind.BuildStructured,
43 solutionPath,
44 includeRawOutput: true,
45 BuildTestToolRequestParser.DefaultBuildTimeoutSeconds,
46 DotnetExecutionOptions.Empty,
47 waitForCompletion,
48 cancellationToken).ConfigureAwait(false);
49 }
50
51 public async Task<EnvironmentTaskOutcome> RunTestsAsync(
52 string runId,
53 string solutionPath,
54 string? filterExpression,
55 bool waitForCompletion,
56 CancellationToken cancellationToken = default,
57 IReadOnlyDictionary<string, string>? supplementalEnvironmentVariables = null)
58 {
59 var baseOptions = string.IsNullOrWhiteSpace(filterExpression)
60 ? DotnetExecutionOptions.Empty
61 : DotnetExecutionOptions.Empty with { Filter = filterExpression.Trim() };
62
63 var options = supplementalEnvironmentVariables is null
64 ? baseOptions
65 : baseOptions with { SupplementalEnvironmentVariables = supplementalEnvironmentVariables };
66
67 return await RunCoreJobAsync(
68 runId,
69 "dotnet.test",
70 BuildTestJobKind.RunTests,
71 solutionPath,
72 includeRawOutput: true,
73 BuildTestToolRequestParser.DefaultTestTimeoutSeconds,
74 options,
75 waitForCompletion,
76 cancellationToken).ConfigureAwait(false);
77 }
78
79 public bool TryCancelCoreJob(string coreJobId) =>
80 _backend.CancelJob(coreJobId) is { } o
81 && JsonSerializer.Serialize(o).Contains("\"cancelled\":true", StringComparison.Ordinal);
82
83 /// <summary>Implicit cancel predecessor (ADR 0148 §8.1.1): отменить build/test jobs активного run.</summary>
84 public void CancelJobsForRun(string runId)
85 {
86 if (!_coreJobsByRun.TryRemove(runId, out var bag))
87 return;
88
89 foreach (var coreJobId in bag)
90 TryCancelCoreJob(coreJobId);
91 }
92
93 private async Task<EnvironmentTaskOutcome> RunCoreJobAsync(
94 string runId,
95 string kind,
96 BuildTestJobKind coreKind,
97 string solutionPath,
98 bool includeRawOutput,
99 int timeoutSeconds,
100 DotnetExecutionOptions dotnetOptions,
101 bool waitForCompletion,
102 CancellationToken cancellationToken)
103 {
104 var taskId = Guid.NewGuid().ToString("N");
105 PublishTaskChanged(taskId, runId, kind, AgentEnvironmentTaskState.Queued, "queued");
106
107 var enqueued = _backend.TryEnqueue(
108 coreKind,
109 solutionPath,
110 includeRawOutput,
111 timeoutSeconds,
112 dotnetOptions);
113
114 if (!enqueued.Accepted)
115 {
116 PublishTaskChanged(taskId, runId, kind, AgentEnvironmentTaskState.Failed, "worker busy");
117 return new EnvironmentTaskOutcome(taskId, null, false, "busy", null);
118 }
119
120 var coreJobId = enqueued.JobId!;
121 _coreJobsByRun.GetOrAdd(runId, _ => []).Add(coreJobId);
122 _ = WatchJobAsync(taskId, runId, kind, coreJobId, cancellationToken);
123
124 if (!waitForCompletion)
125 return new EnvironmentTaskOutcome(taskId, coreJobId, true, "queued", null);
126
127 var resultJson = await _backend.WaitForCompletionAsync(coreJobId, cancellationToken).ConfigureAwait(false);
128 var success = TryReadSuccess(resultJson);
129 return new EnvironmentTaskOutcome(taskId, coreJobId, success, success ? "completed" : "failed", resultJson);
130 }
131
132 private async Task WatchJobAsync(
133 string taskId,
134 string runId,
135 string kind,
136 string coreJobId,
137 CancellationToken cancellationToken)
138 {
139 var started = DateTimeOffset.UtcNow;
140 PublishTaskChanged(taskId, runId, kind, AgentEnvironmentTaskState.Running, "running");
141
142 while (!cancellationToken.IsCancellationRequested)
143 {
144 object? statusObj = TestJobStatusFactory is not null
145 ? TestJobStatusFactory(coreJobId)
146 : _backend.GetJobStatus(coreJobId);
147 if (statusObj is null)
148 {
149 PublishDied(taskId, runId, kind, null, "job not found");
150 return;
151 }
152
153 var json = JsonSerializer.Serialize(statusObj);
154 using var doc = JsonDocument.Parse(json);
155 var root = doc.RootElement;
156 var state = root.TryGetProperty("status", out var st) ? st.GetString() : null;
157
158 if (state is "done" or "failed" or "cancelled" or "timed_out")
159 {
160 var success = TryReadSuccessFromStatus(root);
161 var durationMs = (int)(DateTimeOffset.UtcNow - started).TotalMilliseconds;
162 var summary = state switch
163 {
164 "cancelled" => "cancelled",
165 "timed_out" => "timed out",
166 _ => success ? "green" : "failed",
167 };
168
169 PublishTaskChanged(
170 taskId,
171 runId,
172 kind,
173 state == "cancelled"
174 ? AgentEnvironmentTaskState.Cancelled
175 : success
176 ? AgentEnvironmentTaskState.Completed
177 : AgentEnvironmentTaskState.Failed,
178 summary);
179
180 _dataBus.Publish(new AgentEnvironmentTaskCompleted(taskId, runId, kind, summary, durationMs));
181 return;
182 }
183
184 await Task.Delay(250, cancellationToken).ConfigureAwait(false);
185 }
186 }
187
188 private void PublishTaskChanged(
189 string taskId,
190 string runId,
191 string kind,
192 AgentEnvironmentTaskState state,
193 string? message) =>
194 _dataBus.Publish(new AgentEnvironmentTaskChanged(taskId, runId, kind, state, message));
195
196 private void PublishDied(string taskId, string runId, string taskKind, int? exitCode, string? tail) =>
197 _dataBus.Publish(new AgentEnvironmentTaskDied(taskId, runId, _hostKind, exitCode, tail ?? taskKind));
198
199 private static bool TryReadSuccess(string? resultJson)
200 {
201 if (string.IsNullOrWhiteSpace(resultJson))
202 return false;
203
204 try
205 {
206 using var doc = JsonDocument.Parse(resultJson);
207 return TryReadSuccessFromStatus(doc.RootElement);
208 }
209 catch (JsonException)
210 {
211 return false;
212 }
213 }
214
215 private static bool TryReadSuccessFromStatus(JsonElement root)
216 {
217 if (root.TryGetProperty("result", out var result)
218 && result.ValueKind == JsonValueKind.Object
219 && result.TryGetProperty("success", out var nested))
220 {
221 return nested.ValueKind == JsonValueKind.True;
222 }
223
224 if (root.TryGetProperty("success", out var direct))
225 return direct.ValueKind == JsonValueKind.True;
226
227 return false;
228 }
229}
230
231public sealed record EnvironmentTaskOutcome(
232 string TaskId,
233 string? CoreJobId,
234 bool Success,
235 string Status,
236 string? ResultJson);
237
View only · write via MCP/CIDE