1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
using System.Globalization;
using System.Text.Json;
using Agent.Core.Localization;
using Agent.Mcp;
using System.Collections.Concurrent;
using System.Threading.Channels;
namespace Agent.App.Logging;
public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable
{
private const int QueueCapacity = 4096;
private readonly RollingFileLog _application;
private readonly RollingFileWriter? _http;
private readonly LocaleCatalog _locale;
private readonly string _logDirectory;
private readonly ConcurrentDictionary<string, byte> _observedConnections = new(StringComparer.Ordinal);
private readonly Channel<Action> _queue;
private readonly Task _writerTask;
private long _droppedEntries;
private int _disposed;
public McpHostLogObserver(RollingFileLog application, LocaleCatalog locale, string logDirectory,
bool httpRequests, long maxBytes, int retainedFiles)
{
_application = application;
_locale = locale;
_logDirectory = logDirectory;
if (httpRequests) _http = new RollingFileWriter(Path.Combine(logDirectory, "http.jsonl"), maxBytes, retainedFiles);
_queue = Channel.CreateBounded<Action>(new BoundedChannelOptions(QueueCapacity)
{
SingleReader = true,
SingleWriter = false,
FullMode = BoundedChannelFullMode.Wait,
AllowSynchronousContinuations = false
});
_writerTask = Task.Run(WriteLoopAsync);
}
public void Ready(McpReadyInfo info)
{
string message = info.Transport == "http"
? _locale.Format("app.ready.http", new Dictionary<string, object?>
{
["endpoint"] = info.Endpoint, ["protocol"] = info.ProtocolVersion, ["stateless"] = info.Stateless.ToString().ToLowerInvariant(),
["fallback"] = info.LegacyFallback.ToString().ToLowerInvariant(), ["pid"] = info.ProcessId, ["logs"] = _logDirectory
})
: _locale.Format("app.ready.stdio", new Dictionary<string, object?>
{
["protocol"] = info.ProtocolVersion, ["pid"] = info.ProcessId, ["logs"] = _logDirectory
});
Enqueue(() => _application.Operational(message));
}
public void HttpRequestStarted(McpHttpRequestStartInfo info)
{
if (_observedConnections.TryAdd(info.ConnectionId, 0))
Enqueue(() => _application.Information($"HTTP connection observed connection={info.ConnectionId} remote={info.RemoteIp}:{info.RemotePort}"));
Enqueue(() => _application.Information($"HTTP request started connection={info.ConnectionId} trace={info.TraceId} remote={info.RemoteIp}:{info.RemotePort} method={info.Method} path={info.Path}"));
}
public void HttpRequest(McpHttpRequestInfo info)
{
string applicationLine = string.Create(CultureInfo.InvariantCulture,
$"HTTP request completed connection={info.ConnectionId} remote={info.RemoteIp}:{info.RemotePort} method={info.Method} path={info.Path} status={info.Status} duration_ms={info.DurationMs:F3}");
string? httpLine = _http is null ? null : JsonSerializer.Serialize(new
{
timestamp = info.Timestamp,
connection_id = info.ConnectionId,
trace_id = info.TraceId,
remote_ip = info.RemoteIp,
remote_port = info.RemotePort,
local_ip = info.LocalIp,
local_port = info.LocalPort,
method = info.Method,
path = info.Path,
status = info.Status,
duration_ms = info.DurationMs,
content_type = info.ContentType,
content_length = info.ContentLength,
user_agent = info.UserAgent,
mcp_protocol_version = NullIfEmpty(info.ProtocolVersion),
mcp_method = NullIfEmpty(info.McpMethod),
mcp_name = NullIfEmpty(info.McpName)
});
Enqueue(() =>
{
if (httpLine is not null) _http!.WriteLine(httpLine);
_application.Information(applicationLine);
});
}
private void Enqueue(Action write)
{
if (Volatile.Read(ref _disposed) != 0) return;
if (!_queue.Writer.TryWrite(write)) Interlocked.Increment(ref _droppedEntries);
}
private async Task WriteLoopAsync()
{
try
{
await foreach (Action write in _queue.Reader.ReadAllAsync().ConfigureAwait(false))
{
try { write(); }
catch (Exception ex) when (ex is not (OutOfMemoryException or StackOverflowException or AccessViolationException))
{
System.Diagnostics.Debug.WriteLine($"MCP observability write failed: {ex.Message}");
}
long dropped = Interlocked.Exchange(ref _droppedEntries, 0);
if (dropped > 0)
{
try { _application.Warning($"MCP observability queue dropped {dropped} log entries."); }
catch (Exception ex) when (ex is not (OutOfMemoryException or StackOverflowException or AccessViolationException))
{
System.Diagnostics.Debug.WriteLine($"MCP observability drop report failed: {ex.Message}");
}
}
}
}
finally { _http?.Dispose(); }
}
private static string? NullIfEmpty(string? value) => string.IsNullOrWhiteSpace(value) ? null : value;
public void Dispose()
{
if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
_queue.Writer.TryComplete();
try { _writerTask.Wait(TimeSpan.FromSeconds(5)); }
catch (AggregateException) { }
}
}
|
