summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--README.md2
-rw-r--r--doc/Design.md3
-rw-r--r--src/Agent.App/Logging/McpHostLogObserver.cs74
-rw-r--r--src/Agent.App/Logging/RollingFileLog.cs6
-rw-r--r--tests/Agent.Core.Tests/ObservabilityTests.cs52
5 files changed, 126 insertions, 11 deletions
diff --git a/README.md b/README.md
index e66f2c9..d83b5bb 100644
--- a/README.md
+++ b/README.md
@@ -133,6 +133,8 @@ log directoryは起動ユーザーだけがアクセスできるprotected DACL
HTTP logは`X-Forwarded-For`を接続元として信用せず、Authorization、Cookie、query、body、prompt、画像、Base64を保存しません。stateless HTTPでは永続接続を主張せず、request started/completedとして記録します。
+HTTP request処理はconsole/file出力と同期しません。運用logはbounded queue経由でbackground出力し、consoleが一時停止してもHTTP requestを停止させません。queue飽和時はrequest継続を優先し、drop件数をapplication logへ警告します。
+
正常停止後、`agent-mcp.exe` descendant、追加ACL、WFP filter、loopback exemption、`state/backups`、`state/journals` が残っていないことを確認してください。loopback exemptionは `CheckNetIsolation.exe LoopbackExempt -s` で確認できます。
## よくあるエラー
diff --git a/doc/Design.md b/doc/Design.md
index 1ca8cbe..1bcbc99 100644
--- a/doc/Design.md
+++ b/doc/Design.md
@@ -218,7 +218,8 @@ result: <final natural-language answer; completed only>
- Appはtransport構築前に`WindowsPlatformCheck`を実行する。administrator/BFE/WFP probeが1つでも失敗した場合はhostをbuild/startせずfatalとする。
- `StartAsync`完了後にだけ`IMcpHostObserver.Ready`を通知するため、「起動中」と「接続可能」を区別できる。
-- HTTP middlewareはsocket endpointとrouting metadataだけを`logs/http.jsonl`へJSON Linesで保存する。forwarded header、credential header、query、payloadは保存しない。
+- HTTP middlewareはsocket endpointとrouting metadataだけをbounded in-memory queueへ`TryWrite`し、専用background writerがstderr、application log、`logs/http.jsonl`へ出力する。request threadはconsole/file I/Oを待たず、queue飽和時はrequestを止めずlog entryをdropして件数を警告する。
+- forwarded header、credential header、query、payloadは保存しない。
- `RollingFileWriter`をapplication/http logで共用し、同じsize/retention policyでrotationする。stdioの運用出力はstderrだけへ送る。
---
## 6. Ollama統合設計
diff --git a/src/Agent.App/Logging/McpHostLogObserver.cs b/src/Agent.App/Logging/McpHostLogObserver.cs
index b1410e8..37f32ec 100644
--- a/src/Agent.App/Logging/McpHostLogObserver.cs
+++ b/src/Agent.App/Logging/McpHostLogObserver.cs
@@ -3,16 +3,22 @@ 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)
@@ -21,6 +27,14 @@ public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable
_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)
@@ -35,21 +49,21 @@ public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable
{
["protocol"] = info.ProtocolVersion, ["pid"] = info.ProcessId, ["logs"] = _logDirectory
});
- _application.Operational(message);
+ Enqueue(() => _application.Operational(message));
}
public void HttpRequestStarted(McpHttpRequestStartInfo info)
{
if (_observedConnections.TryAdd(info.ConnectionId, 0))
- _application.Information($"HTTP connection observed connection={info.ConnectionId} remote={info.RemoteIp}:{info.RemotePort}");
- _application.Information($"HTTP request started connection={info.ConnectionId} trace={info.TraceId} remote={info.RemoteIp}:{info.RemotePort} method={info.Method} path={info.Path}");
+ 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)
{
- _application.Information(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}"));
- _http?.WriteLine(JsonSerializer.Serialize(new
+ 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,
@@ -68,9 +82,53 @@ public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable
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() => _http?.Dispose();
+
+ public void Dispose()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
+ _queue.Writer.TryComplete();
+ try { _writerTask.Wait(TimeSpan.FromSeconds(5)); }
+ catch (AggregateException) { }
+ }
}
diff --git a/src/Agent.App/Logging/RollingFileLog.cs b/src/Agent.App/Logging/RollingFileLog.cs
index 8bdbbb4..f471a9e 100644
--- a/src/Agent.App/Logging/RollingFileLog.cs
+++ b/src/Agent.App/Logging/RollingFileLog.cs
@@ -5,12 +5,14 @@ namespace Agent.App.Logging;
public sealed class RollingFileLog : IDisposable
{
private readonly RollingFileWriter _writer;
+ private readonly TextWriter _consoleWriter;
private readonly int _minimumLevel;
- public RollingFileLog(string directory, long maxBytes, int retainedFiles, string minimumLevel = "information")
+ public RollingFileLog(string directory, long maxBytes, int retainedFiles, string minimumLevel = "information", TextWriter? consoleWriter = null)
{
Directory.CreateDirectory(directory);
_writer = new RollingFileWriter(Path.Combine(directory, "application.log"), maxBytes, retainedFiles);
+ _consoleWriter = consoleWriter ?? Console.Error;
_minimumLevel = ParseLevel(minimumLevel);
}
@@ -24,7 +26,7 @@ public sealed class RollingFileLog : IDisposable
{
if (!force && Level(level) < _minimumLevel) return;
string line = string.Create(CultureInfo.InvariantCulture, $"{DateTimeOffset.UtcNow:O} [{level}] {message}");
- Console.Error.WriteLine(line);
+ _consoleWriter.WriteLine(line);
_writer.WriteLine(line);
}
diff --git a/tests/Agent.Core.Tests/ObservabilityTests.cs b/tests/Agent.Core.Tests/ObservabilityTests.cs
index cd86bc2..c259eb0 100644
--- a/tests/Agent.Core.Tests/ObservabilityTests.cs
+++ b/tests/Agent.Core.Tests/ObservabilityTests.cs
@@ -1,6 +1,8 @@
using Agent.App.Logging;
using Agent.Core.Localization;
using Agent.Mcp;
+using System.Diagnostics;
+using System.Text;
namespace Agent.Core.Tests;
@@ -43,4 +45,54 @@ public sealed class ObservabilityTests
Assert.DoesNotContain("Cookie", httpText, StringComparison.OrdinalIgnoreCase);
Assert.DoesNotContain("prompt", httpText, StringComparison.OrdinalIgnoreCase);
}
+
+ [Fact]
+ public void Http_request_callbacks_do_not_wait_for_a_blocked_console_writer()
+ {
+ using var directory = new TestDirectory();
+ using var console = new BlockingTextWriter();
+ LocaleCatalog locale = LocaleCatalog.Load(Path.GetFullPath("config/locales"), "en");
+ using var application = new RollingFileLog(directory.Path, 1024 * 1024, 2, consoleWriter: console);
+ using var observer = new McpHostLogObserver(application, locale, directory.Path, true, 1024 * 1024, 2);
+
+ observer.HttpRequestStarted(new McpHttpRequestStartInfo("connection", "trace-1", "127.0.0.1", 50123, "POST", "/mcp"));
+ Assert.True(console.WaitUntilBlocked(TimeSpan.FromSeconds(5)), "Background writer did not reach the blocking console.");
+
+ var stopwatch = Stopwatch.StartNew();
+ observer.HttpRequest(new McpHttpRequestInfo(DateTimeOffset.UtcNow, "connection", "trace-2", "127.0.0.1", 50124,
+ "127.0.0.1", 8080, "POST", "/mcp", 200, 12.5, "application/json", 100, "test-client",
+ "2026-07-28", "tools/list", null));
+ stopwatch.Stop();
+
+ Assert.True(stopwatch.Elapsed < TimeSpan.FromMilliseconds(250), $"HTTP callback blocked for {stopwatch.Elapsed}.");
+ console.Release();
+ }
+
+ private sealed class BlockingTextWriter : TextWriter
+ {
+ private readonly ManualResetEventSlim _entered = new(false);
+ private readonly ManualResetEventSlim _release = new(false);
+
+ public override Encoding Encoding => Encoding.UTF8;
+
+ public override void WriteLine(string? value)
+ {
+ _entered.Set();
+ _release.Wait(TimeSpan.FromSeconds(30));
+ }
+
+ public bool WaitUntilBlocked(TimeSpan timeout) => _entered.Wait(timeout);
+ public void Release() => _release.Set();
+
+ protected override void Dispose(bool disposing)
+ {
+ if (disposing)
+ {
+ _release.Set();
+ _entered.Dispose();
+ _release.Dispose();
+ }
+ base.Dispose(disposing);
+ }
+ }
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-19 15:13:32 +0000