ClawdDotNet/src/ClawdDotNet.Web/Endpoints/EventEndpoints.cs

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;
using Microsoft.Extensions.Hosting;

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, IHostApplicationLifetime lifetime, CancellationToken ct)
    {
        using var leben = Verbinde(lifetime, ct);
        ct = leben.Token;

        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
            {
                agentId = result.AgentId,
                model,
                status = result.Status.ToString(),
                steps = result.StepCount,
                tokens = result.TokensUsed
            });

        // #711: the team overview — who starts, who stops, who reached whom.
        void OnRunStarted(Core.Activity.ActiveRun run)
            => Publish("activity", new { kind = "started", run = run.ToDto() });

        void OnRunFinished(Core.Activity.ActiveRun run, string status)
            => Publish("activity", new { kind = "finished", run = run.ToDto(), status });

        void OnLink(Core.Activity.AgentLink link)
            => Publish("activity", new { kind = "link", link = link.ToDto() });

        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;
            engine.Activity.RunStarted += OnRunStarted;
            engine.Activity.RunFinished += OnRunFinished;
            engine.Activity.LinkObserved += OnLink;
        }

        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;
                e.Activity.RunStarted -= OnRunStarted;
                e.Activity.RunFinished -= OnRunFinished;
                e.Activity.LinkObserved -= OnLink;
            }

            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, IHostApplicationLifetime lifetime,
        CancellationToken ct, string? module = null, string? level = null)
    {
        using var leben = Verbinde(lifetime, ct);
        ct = leben.Token;

        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();
        }
    }

    /// <summary>
    /// Verbindet den Abbruch der Anfrage mit dem Herunterfahren der Anwendung.
    ///
    /// <para><b>Ohne das hängt das Beenden an jedem offenen Ereignisstrom.</b>
    /// <c>RequestAborted</c> feuert, wenn der Browser auflegt — beim geordneten
    /// Herunterfahren tut er das aber nicht. Kestrel wartet dann darauf, dass die
    /// Anfrage von selbst endet; ein Ereignisstrom endet nie von selbst. Erst nach der
    /// Frist von <c>StopAsync</c> wird abgeschnitten, und so lange steht die Anwendung.</para>
    ///
    /// <para>Sichtbar wurde es beim Instanzwechsel: Zwischen „Wechseln" und dem Neustart
    /// lagen fünf Sekunden, in denen nichts geschah. Mit einem offenen Browserfenster
    /// jedes Mal.</para>
    /// </summary>
    private static CancellationTokenSource Verbinde(
        IHostApplicationLifetime lifetime, CancellationToken requestAborted)
        => CancellationTokenSource.CreateLinkedTokenSource(
            requestAborted, lifetime.ApplicationStopping);

    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";
    }

    /// <summary>
    /// Ein Ereignis in SSE-Form.
    ///
    /// <para>Mit den Web-Vorgaben (camelCase) wie jede REST-Antwort (#591). Ohne sie kamen
    /// Protokollzeilen als <c>Module/Level/Text</c> an, die Ansicht las <c>module/level/text</c>
    /// — und zeigte von jeder Zeile nur den Instanznamen. Den anonymen Objekten der übrigen
    /// Ereignisse sah man das nicht an, weil ihre Namen ohnehin klein geschrieben sind.</para>
    /// </summary>
    internal static string Format(string type, object payload)
        => $"event: {type}\ndata: {JsonSerializer.Serialize(payload, JsonSerializerOptions.Web)}\n\n";
}