feat(worker): route Serilog Warn/Error to footer + buffer recent logs for overlay
This commit is contained in:
@@ -6,6 +6,7 @@ using ClaudeDo.Data.Repositories;
|
||||
using ClaudeDo.Worker.Agents;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Lifecycle;
|
||||
using ClaudeDo.Worker.Logging;
|
||||
using ClaudeDo.Worker.Online;
|
||||
using ClaudeDo.Worker.Planning;
|
||||
using ClaudeDo.Worker.Prime;
|
||||
@@ -113,6 +114,7 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
private readonly WorkerConfig _cfg;
|
||||
private readonly OnlineInboxConfig _onlineInboxConfig;
|
||||
private readonly OnlineTokenStore _onlineTokenStore;
|
||||
private readonly LogRingBuffer? _logBuffer;
|
||||
|
||||
public WorkerHub(
|
||||
QueueService queue,
|
||||
@@ -136,7 +138,8 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
IRefineRunner refineRunner,
|
||||
WorkerConfig cfg,
|
||||
OnlineInboxConfig onlineInboxConfig,
|
||||
OnlineTokenStore onlineTokenStore)
|
||||
OnlineTokenStore onlineTokenStore,
|
||||
LogRingBuffer? logBuffer = null)
|
||||
{
|
||||
_queue = queue;
|
||||
_waker = waker;
|
||||
@@ -160,8 +163,13 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
_cfg = cfg;
|
||||
_onlineInboxConfig = onlineInboxConfig;
|
||||
_onlineTokenStore = onlineTokenStore;
|
||||
_logBuffer = logBuffer;
|
||||
}
|
||||
|
||||
/// <summary>Recent worker log records (last 30 min, all levels) for the Log Visualizer overlay.</summary>
|
||||
public IReadOnlyList<WorkerLogRecord> GetRecentLogs() =>
|
||||
_logBuffer?.Snapshot() ?? Array.Empty<WorkerLogRecord>();
|
||||
|
||||
// Maps the two exceptions service methods throw into client-facing HubExceptions:
|
||||
// KeyNotFoundException -> notFoundMessage, InvalidOperationException -> its own message.
|
||||
private static async Task HubGuard(Func<Task> action, string notFoundMessage = "task not found")
|
||||
|
||||
100
src/ClaudeDo.Worker/Logging/BroadcastLogSink.cs
Normal file
100
src/ClaudeDo.Worker/Logging/BroadcastLogSink.cs
Normal file
@@ -0,0 +1,100 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using ClaudeDo.Data.Models;
|
||||
using Serilog.Core;
|
||||
using Serilog.Events;
|
||||
|
||||
namespace ClaudeDo.Worker.Logging;
|
||||
|
||||
/// <summary>
|
||||
/// Serilog sink that (a) buffers every event (all levels) into <see cref="LogRingBuffer"/>
|
||||
/// for the Log Visualizer overlay, and (b) broadcasts Warn/Error events to the UI footer
|
||||
/// via the attached delegate — deduped within a rate-limit window, with a loop guard so
|
||||
/// SignalR's own log output cannot feed back into another broadcast.
|
||||
/// </summary>
|
||||
public sealed class BroadcastLogSink : ILogEventSink
|
||||
{
|
||||
private static readonly TimeSpan RateLimitWindow = TimeSpan.FromSeconds(120);
|
||||
|
||||
private readonly LogRingBuffer _buffer;
|
||||
private readonly Func<DateTime> _utcNow;
|
||||
private readonly Dictionary<string, DateTime> _lastBroadcast = new();
|
||||
private readonly object _gate = new();
|
||||
private Func<string, WorkerLogLevel, DateTime, Task>? _broadcast;
|
||||
|
||||
public BroadcastLogSink(LogRingBuffer buffer, Func<DateTime>? utcNow = null)
|
||||
{
|
||||
_buffer = buffer;
|
||||
_utcNow = utcNow ?? (() => DateTime.UtcNow);
|
||||
}
|
||||
|
||||
/// <summary>Wired post-build, once the SignalR hub context exists.</summary>
|
||||
public void Attach(Func<string, WorkerLogLevel, DateTime, Task> broadcast) => _broadcast = broadcast;
|
||||
|
||||
public void Emit(LogEvent logEvent)
|
||||
{
|
||||
var level = Map(logEvent.Level);
|
||||
var message = Render(logEvent);
|
||||
var tsUtc = logEvent.Timestamp.UtcDateTime;
|
||||
|
||||
_buffer.Append(new WorkerLogRecord(message, level, tsUtc));
|
||||
|
||||
if (level is not (WorkerLogLevel.Warn or WorkerLogLevel.Error))
|
||||
return;
|
||||
if (IsPlumbing(logEvent))
|
||||
return;
|
||||
|
||||
var broadcast = _broadcast;
|
||||
if (broadcast is null)
|
||||
return;
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
if (_lastBroadcast.TryGetValue(message, out var last) && _utcNow() - last < RateLimitWindow)
|
||||
return;
|
||||
_lastBroadcast[message] = _utcNow();
|
||||
Prune();
|
||||
}
|
||||
|
||||
// Fire-and-forget. A broadcast failure itself logs (re-entering this sink), so it
|
||||
// must never throw out of Emit nor leave an unobserved faulted task.
|
||||
try { _ = broadcast(message, level, tsUtc).ContinueWith(t => { _ = t.Exception; }, TaskScheduler.Default); }
|
||||
catch { /* swallow — logging must not crash on a transport hiccup */ }
|
||||
}
|
||||
|
||||
private static WorkerLogLevel Map(LogEventLevel level) => level switch
|
||||
{
|
||||
LogEventLevel.Warning => WorkerLogLevel.Warn,
|
||||
LogEventLevel.Error or LogEventLevel.Fatal => WorkerLogLevel.Error,
|
||||
_ => WorkerLogLevel.Info,
|
||||
};
|
||||
|
||||
private static string Render(LogEvent e)
|
||||
{
|
||||
var msg = e.RenderMessage();
|
||||
if (e.Exception is { } ex)
|
||||
{
|
||||
var first = ex.Message.Split('\n', 2)[0].Trim();
|
||||
msg = $"{msg}: {ex.GetType().Name}: {first}";
|
||||
}
|
||||
return msg;
|
||||
}
|
||||
|
||||
private static bool IsPlumbing(LogEvent e)
|
||||
{
|
||||
if (!e.Properties.TryGetValue("SourceContext", out var v) || v is not ScalarValue { Value: string sc })
|
||||
return false;
|
||||
return sc.StartsWith("Microsoft.AspNetCore.SignalR", StringComparison.Ordinal)
|
||||
|| sc.StartsWith("Microsoft.AspNetCore.Http.Connections", StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
private void Prune()
|
||||
{
|
||||
if (_lastBroadcast.Count <= 256) return;
|
||||
var cutoff = _utcNow() - RateLimitWindow;
|
||||
foreach (var key in _lastBroadcast.Where(kv => kv.Value < cutoff).Select(kv => kv.Key).ToList())
|
||||
_lastBroadcast.Remove(key);
|
||||
}
|
||||
}
|
||||
48
src/ClaudeDo.Worker/Logging/LogRingBuffer.cs
Normal file
48
src/ClaudeDo.Worker/Logging/LogRingBuffer.cs
Normal file
@@ -0,0 +1,48 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
|
||||
namespace ClaudeDo.Worker.Logging;
|
||||
|
||||
/// <summary>
|
||||
/// Thread-safe, time-bounded ring of recent log records. On each append it evicts
|
||||
/// records older than <c>window</c> and trims to <c>cap</c>. Feeds the Log Visualizer
|
||||
/// overlay via <see cref="Snapshot"/> (oldest-first). The clock is injectable for tests.
|
||||
/// </summary>
|
||||
public sealed class LogRingBuffer
|
||||
{
|
||||
private readonly TimeSpan _window;
|
||||
private readonly int _cap;
|
||||
private readonly Func<DateTime> _utcNow;
|
||||
private readonly LinkedList<WorkerLogRecord> _items = new();
|
||||
private readonly object _gate = new();
|
||||
|
||||
public LogRingBuffer(TimeSpan window, int cap = 5000, Func<DateTime>? utcNow = null)
|
||||
{
|
||||
_window = window;
|
||||
_cap = cap;
|
||||
_utcNow = utcNow ?? (() => DateTime.UtcNow);
|
||||
}
|
||||
|
||||
public void Append(WorkerLogRecord record)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
_items.AddLast(record);
|
||||
var cutoff = _utcNow() - _window;
|
||||
while (_items.First is { } first &&
|
||||
(first.Value.TimestampUtc < cutoff || _items.Count > _cap))
|
||||
_items.RemoveFirst();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Records still inside the window, oldest-first.</summary>
|
||||
public IReadOnlyList<WorkerLogRecord> Snapshot()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
var cutoff = _utcNow() - _window;
|
||||
return _items.Where(r => r.TimestampUtc >= cutoff).ToList();
|
||||
}
|
||||
}
|
||||
}
|
||||
7
src/ClaudeDo.Worker/Logging/WorkerLogRecord.cs
Normal file
7
src/ClaudeDo.Worker/Logging/WorkerLogRecord.cs
Normal file
@@ -0,0 +1,7 @@
|
||||
using System;
|
||||
using ClaudeDo.Data.Models;
|
||||
|
||||
namespace ClaudeDo.Worker.Logging;
|
||||
|
||||
/// <summary>One captured log line for the in-memory ring buffer / Log Visualizer overlay.</summary>
|
||||
public sealed record WorkerLogRecord(string Message, WorkerLogLevel Level, DateTime TimestampUtc);
|
||||
@@ -7,6 +7,7 @@ using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.External;
|
||||
using ClaudeDo.Worker.Hub;
|
||||
using ClaudeDo.Worker.Lifecycle;
|
||||
using ClaudeDo.Worker.Logging;
|
||||
using ClaudeDo.Worker.Planning;
|
||||
using ClaudeDo.Worker.Queue;
|
||||
using ClaudeDo.Worker.Runner;
|
||||
@@ -34,13 +35,22 @@ var builder = WebApplication.CreateBuilder(args);
|
||||
|
||||
var logRoot = cfg.LogRoot;
|
||||
Directory.CreateDirectory(logRoot);
|
||||
|
||||
// In-memory ring + broadcast sink power the footer log strip and the Log Visualizer
|
||||
// overlay. Created pre-build so Serilog can write to the sink; the SignalR broadcaster
|
||||
// is attached once the host is built (see below).
|
||||
var logBuffer = new LogRingBuffer(TimeSpan.FromMinutes(30));
|
||||
var broadcastSink = new BroadcastLogSink(logBuffer);
|
||||
|
||||
builder.Host.UseSerilog((ctx, lc) => lc
|
||||
.MinimumLevel.Information()
|
||||
.WriteTo.File(
|
||||
System.IO.Path.Combine(logRoot, "worker-.log"),
|
||||
rollingInterval: RollingInterval.Day,
|
||||
retainedFileCountLimit: 7,
|
||||
shared: true));
|
||||
shared: true)
|
||||
.WriteTo.Sink(broadcastSink));
|
||||
builder.Services.AddSingleton(logBuffer);
|
||||
|
||||
builder.Services.AddDbContextFactory<ClaudeDoDbContext>(opt =>
|
||||
opt.UseSqlite($"Data Source={cfg.DbPath}"));
|
||||
@@ -180,6 +190,10 @@ builder.WebHost.UseUrls($"http://127.0.0.1:{cfg.SignalRPort}");
|
||||
|
||||
var app = builder.Build();
|
||||
|
||||
// Now that the hub context exists, let the broadcast sink push Warn/Error to the footer.
|
||||
var logBroadcaster = app.Services.GetRequiredService<HubBroadcaster>();
|
||||
broadcastSink.Attach((message, level, ts) => logBroadcaster.WorkerLog(message, level, ts));
|
||||
|
||||
using (var scope = app.Services.CreateScope())
|
||||
{
|
||||
ClaudeDoDbContext.MigrateAndConfigure(
|
||||
|
||||
Reference in New Issue
Block a user