- ClawdDotNet.Web: Kestrel im Prozess der Anwendung, Token-Absicherung (TokenGate mit FixedTimeEquals), Endpunkte unter /api, Ereignisstrom unter /events (Server-Sent Events) und fortlaufendes Log-Streaming. - Weboberflaeche (wwwroot/): Vollstaendiges Single-Page-Frontend mit nativem HTML5, CSS3 und ES-Modulen ohne Buildschritt (strikt textContent, #-Routing, sofortiger Login-Sprung bei fehlendem Token, Light/Dark-Theme, lokale Barlow-Schriften). - ClawdDotNet.App: Fachlogik aus den Avalonia-ViewModels in Dienste herausgeloest (AgentAdministration, JobAdministration, ServiceAdministration, RunCoordinator) und ueber InstanceChange synchronisiert. - Tests: ClawdDotNet.Web.Tests und ClawdDotNet.App.Tests hinzugefuegt, alle 684 Tests erfolgreich.
This commit is contained in:
Richard
@@ -0,0 +1,231 @@
|
||||
using System.Text.Json;
|
||||
using System.Threading.Channels;
|
||||
using ClawdDotNet.App;
|
||||
using ClawdDotNet.App.Services;
|
||||
using ClawdDotNet.Web.Contracts;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
|
||||
namespace ClawdDotNet.Web.Endpoints;
|
||||
|
||||
/// <summary>
|
||||
/// Der Ereignisstrom: was im Kern passiert, während niemand danach fragt.
|
||||
///
|
||||
/// <para><b>Server-Sent Events und nicht WebSocket.</b> Die Richtung ist einseitig —
|
||||
/// Befehle gehen über gewöhnliche POSTs —, SSE übersteht Zwischenschichten, bringt das
|
||||
/// Wiederverbinden mit und braucht auf keiner Seite eine zusätzliche Bibliothek.</para>
|
||||
///
|
||||
/// <para><b>Die Ereignisse kommen aus Hintergrundfäden.</b> <c>AgentEngine</c> löst sie
|
||||
/// dort aus, wo der Lauf gerade ist. Deshalb geht hier nichts direkt in die Antwort,
|
||||
/// sondern erst in einen Kanal, den die Anfrage abarbeitet: Ein Ereignisbehandler, der in
|
||||
/// den Antwortstrom schreibt, träfe früher oder später auf einen zweiten, der dasselbe
|
||||
/// tut, und ein <c>HttpResponse</c> verträgt keine zwei Schreiber.</para>
|
||||
/// </summary>
|
||||
internal static class EventEndpoints
|
||||
{
|
||||
/// <summary>
|
||||
/// Wie viele Ereignisse ein langsamer Browser hinter sich herziehen darf, bevor die
|
||||
/// ältesten wegfallen.
|
||||
///
|
||||
/// <para>Begrenzt und verwerfend, nicht unbegrenzt: Ein Tab, der im Hintergrund
|
||||
/// eingefroren ist, würde sonst den Speicher des Prozesses füllen, in dem die
|
||||
/// Agenten laufen. Ein verlorener Zwischenstand ist der bessere Verlust — die Seite
|
||||
/// lädt den Verlauf beim Verbinden ohnehin neu.</para>
|
||||
/// </summary>
|
||||
private const int QueueDepth = 256;
|
||||
|
||||
public static void MapEvents(this WebApplication app)
|
||||
{
|
||||
var events = app.MapGroup("/events");
|
||||
|
||||
events.MapGet("/", StreamCoreEvents);
|
||||
events.MapGet("/logs", StreamLogs);
|
||||
}
|
||||
|
||||
private static async Task StreamCoreEvents(
|
||||
HttpContext context, AppHost host, CancellationToken ct)
|
||||
{
|
||||
var queue = Channel.CreateBounded<string>(new BoundedChannelOptions(QueueDepth)
|
||||
{
|
||||
FullMode = BoundedChannelFullMode.DropOldest,
|
||||
SingleReader = true
|
||||
});
|
||||
|
||||
void Publish(string type, object payload)
|
||||
=> queue.Writer.TryWrite(Format(type, payload));
|
||||
|
||||
// Die Behandler stehen als benannte Delegaten da, weil sie im finally wieder
|
||||
// abgemeldet werden muessen. Ein Lambda direkt am += liesse sich nicht mehr
|
||||
// entfernen, und nach dreissig Neuladen der Seite liefe jede Nachricht
|
||||
// dreissigmal durch.
|
||||
void OnChat(string agentId, string role, string content, string? source)
|
||||
=> Publish("chat", new { agentId, role, content, source, timestamp = DateTime.UtcNow });
|
||||
|
||||
void OnStep(string agentId, string text)
|
||||
=> Publish("step", new { agentId, text });
|
||||
|
||||
void OnRunCompleted(string model, Core.Engine.AgentRunResult result)
|
||||
=> Publish("run", new
|
||||
{
|
||||
model,
|
||||
status = result.Status.ToString(),
|
||||
steps = result.StepCount,
|
||||
tokens = result.TokensUsed
|
||||
});
|
||||
|
||||
void OnInstanceChanged(InstanceChange change)
|
||||
=> Publish("instance", new { kind = change.Kind.ToString(), id = change.Id });
|
||||
|
||||
void OnStatus()
|
||||
=> Publish("status", new { });
|
||||
|
||||
if (host.Engine is { } engine)
|
||||
{
|
||||
engine.OnChatEntryAdded += OnChat;
|
||||
engine.OnStepCompleted += OnStep;
|
||||
engine.OnRunCompleted += OnRunCompleted;
|
||||
}
|
||||
|
||||
host.InstanceChanged += OnInstanceChanged;
|
||||
|
||||
if (host.Status is { } status)
|
||||
status.OnStatusUpdated += OnStatus;
|
||||
|
||||
try
|
||||
{
|
||||
await WriteStream(context, queue.Reader, ct);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (host.Engine is { } e)
|
||||
{
|
||||
e.OnChatEntryAdded -= OnChat;
|
||||
e.OnStepCompleted -= OnStep;
|
||||
e.OnRunCompleted -= OnRunCompleted;
|
||||
}
|
||||
|
||||
host.InstanceChanged -= OnInstanceChanged;
|
||||
|
||||
if (host.Status is { } s)
|
||||
s.OnStatusUpdated -= OnStatus;
|
||||
|
||||
queue.Writer.TryComplete();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Das Protokoll, fortlaufend.
|
||||
///
|
||||
/// <para><b>Eine eigene <see cref="LogTail"/> je Verbindung</b>, und das ist keine
|
||||
/// Sparsamkeitsfrage: <c>ReadNew</c> liefert die Zeilen seit dem letzten Aufruf und
|
||||
/// merkt sich die Stelle. Eine geteilte Ausprägung hieße, dass der erste Browser die
|
||||
/// Zeilen bekommt und alle anderen nichts.</para>
|
||||
/// </summary>
|
||||
private static async Task StreamLogs(
|
||||
HttpContext context, AppHost host, CancellationToken ct, string? module = null, string? level = null)
|
||||
{
|
||||
var tail = new LogTail(host.LogDirectory);
|
||||
|
||||
var filter = Enum.TryParse<LogLevelFilter>(level, ignoreCase: true, out var parsed)
|
||||
? parsed
|
||||
: LogLevelFilter.All;
|
||||
|
||||
var interval = TimeSpan.FromMilliseconds(
|
||||
Math.Clamp(host.Settings.AppSettings.LogRefreshIntervalMs, 200, 5000));
|
||||
|
||||
PrepareResponse(context);
|
||||
|
||||
using var timer = new PeriodicTimer(interval);
|
||||
|
||||
try
|
||||
{
|
||||
while (await timer.WaitForNextTickAsync(ct))
|
||||
{
|
||||
var lines = tail.ReadNew(module, filter);
|
||||
|
||||
if (lines.Count == 0)
|
||||
{
|
||||
// Ein Kommentar haelt die Verbindung offen, ohne ein Ereignis
|
||||
// vorzutaeuschen. Ohne ihn kappt so mancher Proxy nach einer Minute.
|
||||
await context.Response.WriteAsync(": still\n\n", ct);
|
||||
}
|
||||
else
|
||||
{
|
||||
foreach (var line in lines)
|
||||
await context.Response.WriteAsync(
|
||||
Format("log", new LogLineDto(line.Module, line.Level, line.Text)), ct);
|
||||
}
|
||||
|
||||
await context.Response.Body.FlushAsync(ct);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Der Browser hat aufgelegt. Der einzige normale Weg, auf dem dieser Strom endet.
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task WriteStream(
|
||||
HttpContext context, ChannelReader<string> reader, CancellationToken ct)
|
||||
{
|
||||
PrepareResponse(context);
|
||||
|
||||
// Sofort etwas schicken: Erst damit steht die Verbindung fuer den Browser, und
|
||||
// onopen feuert. Ohne das haengt die Seite im Ungewissen, bis das erste Ereignis
|
||||
// von selbst kommt — was Stunden dauern kann.
|
||||
await context.Response.WriteAsync(": bereit\n\n", ct);
|
||||
await context.Response.Body.FlushAsync(ct);
|
||||
|
||||
var heartbeat = new PeriodicTimer(TimeSpan.FromSeconds(20));
|
||||
|
||||
try
|
||||
{
|
||||
var next = reader.ReadAsync(ct).AsTask();
|
||||
var beat = heartbeat.WaitForNextTickAsync(ct).AsTask();
|
||||
|
||||
while (true)
|
||||
{
|
||||
var done = await Task.WhenAny(next, beat);
|
||||
|
||||
if (done == next)
|
||||
{
|
||||
await context.Response.WriteAsync(await next, ct);
|
||||
next = reader.ReadAsync(ct).AsTask();
|
||||
}
|
||||
else
|
||||
{
|
||||
if (!await beat) break;
|
||||
await context.Response.WriteAsync(": still\n\n", ct);
|
||||
beat = heartbeat.WaitForNextTickAsync(ct).AsTask();
|
||||
}
|
||||
|
||||
await context.Response.Body.FlushAsync(ct);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Der Browser hat aufgelegt.
|
||||
}
|
||||
catch (ChannelClosedException)
|
||||
{
|
||||
// Der Strom wird abgeraeumt.
|
||||
}
|
||||
finally
|
||||
{
|
||||
heartbeat.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
private static void PrepareResponse(HttpContext context)
|
||||
{
|
||||
context.Response.Headers.ContentType = "text/event-stream";
|
||||
context.Response.Headers.CacheControl = "no-cache";
|
||||
|
||||
// Ohne das puffert ein vorgelagerter nginx den Strom und liefert ihn erst, wenn
|
||||
// er voll ist — aus einem Ereignisstrom wird dann eine Ereignissammlung.
|
||||
context.Response.Headers["X-Accel-Buffering"] = "no";
|
||||
}
|
||||
|
||||
private static string Format(string type, object payload)
|
||||
=> $"event: {type}\ndata: {JsonSerializer.Serialize(payload)}\n\n";
|
||||
}
|
||||
Reference in New Issue
Block a user