From 788df5e1d86c1bd91571d2b9cd1559599b65c9d6 Mon Sep 17 00:00:00 2001 From: tanaka takashi Date: Sun, 9 Aug 2026 23:19:24 +0900 Subject: feat(mcp): migrate server to MCP 2026-07-28 Replace the MCP 1.4.1 stateful HTTP design with the official C# SDK 2.0.0 stateless transport. The SDK now owns protocol negotiation, server/discover, standard MCP headers, GET and DELETE handling, error serialization, caching hints, and legacy initialize compatibility; production code no longer maintains a custom wire-protocol workaround for MCP Inspector. Separate transport state from application conversation state. HTTP does not issue Mcp-Session-Id, while the existing session_id remains an Agent conversation identifier. Preserve agent_run and agent_status for clients without Tasks support and add the official ModelContextProtocol.Extensions.Tasks 2.0.0 path for capable clients. Add persistent MCP Task storage with atomic JSON replacement, user-only state location, configurable TTL and poll interval, restart recovery, cancellation correlation, and public Agent-to-MCP state mapping. Keep conformance-only tools and protocol exception fixtures in a test host so they cannot enter the Release Bundle. Run administrator, BFE, WFP transaction, and filter add/delete preflight before binding. Fail fast with exit code 2 and actionable localized diagnostics when sandbox enforcement is unavailable. Emit ready only after the listener starts and keep stdio stdout protocol-pure. Add shared rotating application and HTTP request logs. Record real socket endpoints, connection and trace identifiers, method, path, result, timing, media metadata, User-Agent, and MCP metadata. Do not trust X-Forwarded-For and do not log authorization, cookies, query data, request or response bodies, prompts, images, or Base64 payloads. Add typed logging and Tasks configuration, Japanese and English messages, protocol and operations documentation, design/specification deltas, traceability, and exact conformance acquisition/offline runner scripts for @modelcontextprotocol/conformance 0.2.0-alpha.11. Node remains test-only and is not a runtime dependency. Conformance evidence: server-stateless passed 28 of 28 checks with zero unexpected failures; DNS rebinding and multiple POST stream scenarios passed. The alpha Tasks lifecycle/cancel/error/required-task checks passed, while its wire-schema validator misclassifies CreateTaskResult and its MRTR composition remains incompatible with SDK 2.0.0. The aggregate conformance Gate is therefore recorded as PARTIAL, not PASS. Test-host fixture gaps for tools-list descriptions, JSON Schema diagnostics, caching fixtures, and a header-validation/required-Task capability conflict are also recorded without expected-failure masking. Verification: Release build passed with 0 warnings and 0 errors; Planner tests 10/10 and Core tests 47/47 passed with skip 0. The last Windows suite run passed 19/25; five WFP cases require an elevated token and one symlink case requires unavailable host privilege, so the elevated security Gate remains open. A win-x64 self-contained 1.0.0 Bundle and ZIP were regenerated under build/Release/publish; runtime check was intentionally skipped in this non-elevated session. Evidence remains under ignored artifacts/evidence/mcp-2026-07-28. Harden release regeneration when another shell holds the previous Bundle directory as its current directory: validate the target and remove only its children instead of deleting the directory itself. build/ and artifacts/ remain excluded from Git. --- src/Agent.App/Configuration/AgentConfiguration.cs | 4 +- src/Agent.App/Configuration/ConfigurationLoader.cs | 2 + src/Agent.App/Logging/McpHostLogObserver.cs | 76 +++++++++ src/Agent.App/Logging/RollingFileLog.cs | 55 +++--- src/Agent.App/Logging/RollingFileWriter.cs | 50 ++++++ src/Agent.App/Program.cs | 30 +++- src/Agent.Core/Tasks/TaskRegistry.cs | 36 +++- src/Agent.Mcp/Agent.Mcp.csproj | 1 + src/Agent.Mcp/AgentTools.cs | 65 ++++++- src/Agent.Mcp/McpHost.cs | 93 ++++++++-- src/Agent.Mcp/McpHostObservability.cs | 37 ++++ src/Agent.Mcp/PersistentMcpTaskStore.cs | 190 +++++++++++++++++++++ src/Agent.Mcp/QueuedTaskWorker.cs | 10 +- 13 files changed, 587 insertions(+), 62 deletions(-) create mode 100644 src/Agent.App/Logging/McpHostLogObserver.cs create mode 100644 src/Agent.App/Logging/RollingFileWriter.cs create mode 100644 src/Agent.Mcp/McpHostObservability.cs create mode 100644 src/Agent.Mcp/PersistentMcpTaskStore.cs (limited to 'src') diff --git a/src/Agent.App/Configuration/AgentConfiguration.cs b/src/Agent.App/Configuration/AgentConfiguration.cs index 800286b..362b889 100644 --- a/src/Agent.App/Configuration/AgentConfiguration.cs +++ b/src/Agent.App/Configuration/AgentConfiguration.cs @@ -23,10 +23,12 @@ public sealed class McpOptions public string Transport { get; set; } = "stdio"; public HttpOptions Http { get; set; } = new(); public ImageOptions Images { get; set; } = new(); + public McpTaskOptions Tasks { get; set; } = new(); } public sealed class HttpOptions { public string Host { get; set; } = "127.0.0.1"; public int Port { get; set; } = 8080; } public sealed class ImageOptions { public int MaxCount { get; set; } = 4; public long MaxBytesEach { get; set; } = 10 * 1024 * 1024; public long MaxTotalBytes { get; set; } = 20 * 1024 * 1024; } +public sealed class McpTaskOptions { public int TimeToLiveHours { get; set; } = 24; public int PollIntervalMs { get; set; } = 1000; } public sealed class RuntimeOptions { public string UserSystemPrompt { get; set; } = "prompts/user-system.md"; public int MaxIterations { get; set; } = 128; public int SameFailureLimit { get; set; } = 5; } public sealed class OllamaConfiguration @@ -73,4 +75,4 @@ public sealed class FileEditOptions { public int MemoryBackupLimitMb { get; set; public sealed class LimitOptions { public long WholeReadBytes { get; set; } = 1024 * 1024; public int SearchMaxResults { get; set; } = 1000; public int TailMaxLines { get; set; } = 10000; public double ContextSummaryRatio { get; set; } = 0.8; } public sealed class LocalizationOptions { public string Directory { get; set; } = "config/locales"; public string Fallback { get; set; } = "en"; } public sealed class SkillOptions { public string Directory { get; set; } = "skills"; } -public sealed class LoggingOptions { public string Directory { get; set; } = "logs"; public long MaxFileBytes { get; set; } = 10 * 1024 * 1024; public int RetainedFiles { get; set; } = 5; } +public sealed class LoggingOptions { public string Directory { get; set; } = "logs"; public string Level { get; set; } = "information"; public bool HttpRequests { get; set; } = true; public long MaxFileBytes { get; set; } = 10 * 1024 * 1024; public int RetainedFiles { get; set; } = 5; } diff --git a/src/Agent.App/Configuration/ConfigurationLoader.cs b/src/Agent.App/Configuration/ConfigurationLoader.cs index b6c4c2d..298a143 100644 --- a/src/Agent.App/Configuration/ConfigurationLoader.cs +++ b/src/Agent.App/Configuration/ConfigurationLoader.cs @@ -44,6 +44,7 @@ public static partial class ConfigurationLoader if (value.Mcp.Http.Port is < 1 or > 65535) Throw("mcp.http.port is outside 1..65535."); if (value.Mcp.Images.MaxCount < 0 || value.Mcp.Images.MaxBytesEach <= 0 || value.Mcp.Images.MaxTotalBytes <= 0) Throw("mcp.images limits must be positive."); if (value.Mcp.Images.MaxTotalBytes < value.Mcp.Images.MaxBytesEach) Throw("mcp.images.max_total_bytes must be >= max_bytes_each."); + if (value.Mcp.Tasks.TimeToLiveHours <= 0 || value.Mcp.Tasks.PollIntervalMs <= 0) Throw("mcp.tasks TTL and poll interval must be positive."); if (value.Agent.MaxIterations <= 0 || value.Agent.SameFailureLimit <= 0) Throw("agent iteration limits must be positive."); if (!Uri.TryCreate(value.Ollama.Endpoint, UriKind.Absolute, out Uri? endpoint) || endpoint.Scheme is not ("http" or "https")) Throw("ollama.endpoint must be an absolute HTTP(S) URI."); if (string.IsNullOrWhiteSpace(value.Ollama.Model)) Throw("ollama.model is required."); @@ -62,6 +63,7 @@ public static partial class ConfigurationLoader } if (value.FileEdit.MemoryBackupLimitMb <= 0 || value.Limits.WholeReadBytes <= 0 || value.Limits.SearchMaxResults <= 0 || value.Limits.TailMaxLines <= 0 || value.Limits.ContextSummaryRatio is <= 0 or >= 1) Throw("file/context limits are invalid."); if (value.Logging.MaxFileBytes <= 0 || value.Logging.RetainedFiles < 1) Throw("logging limits are invalid."); + if (value.Logging.Level is not ("information" or "warning" or "error" or "fatal")) Throw("logging.level must be information, warning, error, or fatal."); RequireFile(baseDirectory, value.Agent.UserSystemPrompt, "agent.user_system_prompt"); RequireDirectory(baseDirectory, value.Localization.Directory, "localization.directory"); diff --git a/src/Agent.App/Logging/McpHostLogObserver.cs b/src/Agent.App/Logging/McpHostLogObserver.cs new file mode 100644 index 0000000..b1410e8 --- /dev/null +++ b/src/Agent.App/Logging/McpHostLogObserver.cs @@ -0,0 +1,76 @@ +using System.Globalization; +using System.Text.Json; +using Agent.Core.Localization; +using Agent.Mcp; +using System.Collections.Concurrent; + +namespace Agent.App.Logging; + +public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable +{ + private readonly RollingFileLog _application; + private readonly RollingFileWriter? _http; + private readonly LocaleCatalog _locale; + private readonly string _logDirectory; + private readonly ConcurrentDictionary _observedConnections = new(StringComparer.Ordinal); + + 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); + } + + public void Ready(McpReadyInfo info) + { + string message = info.Transport == "http" + ? _locale.Format("app.ready.http", new Dictionary + { + ["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 + { + ["protocol"] = info.ProtocolVersion, ["pid"] = info.ProcessId, ["logs"] = _logDirectory + }); + _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}"); + } + + 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 + { + 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) + })); + } + + private static string? NullIfEmpty(string? value) => string.IsNullOrWhiteSpace(value) ? null : value; + public void Dispose() => _http?.Dispose(); +} diff --git a/src/Agent.App/Logging/RollingFileLog.cs b/src/Agent.App/Logging/RollingFileLog.cs index 801768d..8bdbbb4 100644 --- a/src/Agent.App/Logging/RollingFileLog.cs +++ b/src/Agent.App/Logging/RollingFileLog.cs @@ -4,52 +4,43 @@ namespace Agent.App.Logging; public sealed class RollingFileLog : IDisposable { - private readonly object _sync = new(); - private readonly string _path; - private readonly long _maxBytes; - private readonly int _retainedFiles; - private StreamWriter _writer; + private readonly RollingFileWriter _writer; + private readonly int _minimumLevel; - public RollingFileLog(string directory, long maxBytes, int retainedFiles) + public RollingFileLog(string directory, long maxBytes, int retainedFiles, string minimumLevel = "information") { Directory.CreateDirectory(directory); - _path = Path.Combine(directory, "application.log"); - _maxBytes = maxBytes; - _retainedFiles = retainedFiles; - _writer = Open(); + _writer = new RollingFileWriter(Path.Combine(directory, "application.log"), maxBytes, retainedFiles); + _minimumLevel = ParseLevel(minimumLevel); } public void Information(string message) => Write("INF", message); + public void Operational(string message) => Write("INF", message, force: true); + public void Warning(string message) => Write("WRN", message); public void Error(string message) => Write("ERR", message); + public void Fatal(string message) => Write("FTL", message); - private void Write(string level, string message) + private void Write(string level, string message, bool force = false) { - lock (_sync) - { - if (_writer.BaseStream.Length >= _maxBytes) Rotate(); - string line = string.Create(CultureInfo.InvariantCulture, $"{DateTimeOffset.UtcNow:O} [{level}] {message}"); - Console.Error.WriteLine(line); - _writer.WriteLine(line); - _writer.Flush(); - } + if (!force && Level(level) < _minimumLevel) return; + string line = string.Create(CultureInfo.InvariantCulture, $"{DateTimeOffset.UtcNow:O} [{level}] {message}"); + Console.Error.WriteLine(line); + _writer.WriteLine(line); } - private StreamWriter Open() => new(new FileStream(_path, FileMode.Append, FileAccess.Write, FileShare.Read), new System.Text.UTF8Encoding(false)); - - private void Rotate() + private static int ParseLevel(string level) => level.ToLowerInvariant() switch { - _writer.Dispose(); - for (int index = _retainedFiles - 1; index >= 1; index--) - { - string source = $"{_path}.{index}"; - if (File.Exists(source)) File.Move(source, $"{_path}.{index + 1}", true); - } - if (File.Exists(_path)) File.Move(_path, $"{_path}.1", true); - _writer = Open(); - } + "information" or "info" => 0, + "warning" or "warn" => 1, + "error" => 2, + "fatal" => 3, + _ => throw new ArgumentException($"Unknown logging level '{level}'.", nameof(level)) + }; + + private static int Level(string level) => level switch { "INF" => 0, "WRN" => 1, "ERR" => 2, "FTL" => 3, _ => 0 }; public void Dispose() { - lock (_sync) _writer.Dispose(); + _writer.Dispose(); } } diff --git a/src/Agent.App/Logging/RollingFileWriter.cs b/src/Agent.App/Logging/RollingFileWriter.cs new file mode 100644 index 0000000..e155858 --- /dev/null +++ b/src/Agent.App/Logging/RollingFileWriter.cs @@ -0,0 +1,50 @@ +using System.Text; + +namespace Agent.App.Logging; + +internal sealed class RollingFileWriter : IDisposable +{ + private readonly object _sync = new(); + private readonly string _path; + private readonly long _maxBytes; + private readonly int _retainedFiles; + private StreamWriter _writer; + + public RollingFileWriter(string path, long maxBytes, int retainedFiles) + { + _path = path; + _maxBytes = maxBytes; + _retainedFiles = retainedFiles; + Directory.CreateDirectory(Path.GetDirectoryName(path)!); + _writer = Open(); + } + + public void WriteLine(string line) + { + lock (_sync) + { + if (_writer.BaseStream.Length >= _maxBytes) Rotate(); + _writer.WriteLine(line); + _writer.Flush(); + } + } + + private StreamWriter Open() => new(new FileStream(_path, FileMode.Append, FileAccess.Write, FileShare.Read), new UTF8Encoding(false)); + + private void Rotate() + { + _writer.Dispose(); + for (int index = _retainedFiles - 1; index >= 1; index--) + { + string source = $"{_path}.{index}"; + if (File.Exists(source)) File.Move(source, $"{_path}.{index + 1}", true); + } + if (File.Exists(_path)) File.Move(_path, $"{_path}.1", true); + _writer = Open(); + } + + public void Dispose() + { + lock (_sync) _writer.Dispose(); + } +} diff --git a/src/Agent.App/Program.cs b/src/Agent.App/Program.cs index 3fae93f..d9ee17a 100644 --- a/src/Agent.App/Program.cs +++ b/src/Agent.App/Program.cs @@ -35,7 +35,7 @@ internal static class Program string logDirectory = ConfigurationLoader.Resolve(baseDirectory, configuration.Logging.Directory); UserOnlyDirectory.Ensure(logDirectory); - using var log = new RollingFileLog(logDirectory, configuration.Logging.MaxFileBytes, configuration.Logging.RetainedFiles); + using var log = new RollingFileLog(logDirectory, configuration.Logging.MaxFileBytes, configuration.Logging.RetainedFiles, configuration.Logging.Level); if (commandLine.Command == AppCommand.Check) { WindowsPlatformCheckResult platform = WindowsPlatformCheck.Run(); @@ -57,8 +57,30 @@ internal static class Program } log.Information(locale.Get("app.starting")); + WindowsPlatformCheckResult servePlatform = WindowsPlatformCheck.Run(); + if (!servePlatform.Success) + { + string failure = locale.Format("app.preflight_failed", new Dictionary + { + ["operation"] = servePlatform.ErrorOperation ?? "privilege/BFE/WFP probe", + ["error_code"] = servePlatform.ErrorCode, + ["administrator"] = servePlatform.IsAdministrator.ToString().ToLowerInvariant(), + ["bfe"] = servePlatform.BfeEngineOpen.ToString().ToLowerInvariant(), + ["transaction"] = servePlatform.WfpWriteTransaction.ToString().ToLowerInvariant(), + ["filters"] = servePlatform.WfpFilterAddDelete.ToString().ToLowerInvariant() + }); + log.Fatal(failure); + return 2; + } + InternalToolDispatcher toolDispatcher = CreateToolDispatcher(configuration, baseDirectory, logDirectory); var runtimeObserver = new JsonLinesRuntimeObserver(logDirectory); + string taskStateDirectory = Path.Combine(baseDirectory, "state", "mcp-tasks"); + UserOnlyDirectory.Ensure(taskStateDirectory); + var taskStoreSettings = new McpTaskStoreSettings(taskStateDirectory, + TimeSpan.FromHours(configuration.Mcp.Tasks.TimeToLiveHours), configuration.Mcp.Tasks.PollIntervalMs); + using var hostObserver = new McpHostLogObserver(log, locale, logDirectory, configuration.Logging.HttpRequests, + configuration.Logging.MaxFileBytes, configuration.Logging.RetainedFiles); if (configuration.Mcp.Transport == "stdio") { string userPrompt = File.ReadAllText(ConfigurationLoader.Resolve(baseDirectory, configuration.Agent.UserSystemPrompt)); @@ -66,7 +88,8 @@ internal static class Program string[] skills = Directory.EnumerateFiles(skillDirectory, "*.md", SearchOption.TopDirectoryOnly).Select(File.ReadAllText).ToArray(); var runtimeSettings = new AgentRuntimeSettings(userPrompt, language, skills, "Windows AppContainer, ACL, Job Object, child-process restriction, and WFP sandbox capabilities.", configuration.Agent.MaxIterations, configuration.Agent.SameFailureLimit, locale.Get("ollama.vision_unsupported")); var visionLimits = new VisionLimits(configuration.Mcp.Images.MaxCount, configuration.Mcp.Images.MaxBytesEach, configuration.Mcp.Images.MaxTotalBytes); - await McpHost.RunStdioAsync(locale, OllamaSettingsFactory.Create(configuration.Ollama), runtimeSettings, visionLimits, CancellationToken.None, toolDispatcher, runtimeObserver).ConfigureAwait(false); + await McpHost.RunStdioAsync(locale, OllamaSettingsFactory.Create(configuration.Ollama), runtimeSettings, visionLimits, CancellationToken.None, + toolDispatcher, runtimeObserver, taskStoreSettings, hostObserver).ConfigureAwait(false); return 0; } if (configuration.Mcp.Transport == "http") @@ -77,7 +100,8 @@ internal static class Program var runtimeSettings = new AgentRuntimeSettings(userPrompt, language, skills, "Windows AppContainer, ACL, Job Object, child-process restriction, and WFP sandbox capabilities.", configuration.Agent.MaxIterations, configuration.Agent.SameFailureLimit, locale.Get("ollama.vision_unsupported")); var visionLimits = new VisionLimits(configuration.Mcp.Images.MaxCount, configuration.Mcp.Images.MaxBytesEach, configuration.Mcp.Images.MaxTotalBytes); await McpHost.RunHttpAsync(locale, OllamaSettingsFactory.Create(configuration.Ollama), runtimeSettings, visionLimits, - configuration.Mcp.Http.Host, configuration.Mcp.Http.Port, CancellationToken.None, toolDispatcher, runtimeObserver).ConfigureAwait(false); + configuration.Mcp.Http.Host, configuration.Mcp.Http.Port, CancellationToken.None, toolDispatcher, runtimeObserver, + taskStoreSettings, hostObserver).ConfigureAwait(false); return 0; } throw new ConfigurationException("Unsupported MCP transport."); diff --git a/src/Agent.Core/Tasks/TaskRegistry.cs b/src/Agent.Core/Tasks/TaskRegistry.cs index b61e525..b7a4afb 100644 --- a/src/Agent.Core/Tasks/TaskRegistry.cs +++ b/src/Agent.Core/Tasks/TaskRegistry.cs @@ -14,7 +14,7 @@ public sealed class TaskRegistry public SessionStore Sessions => _sessions; - public TaskSnapshot Enqueue(string request, Guid? requestedSessionId, double? temperature, IReadOnlyList? images = null) + public TaskSnapshot Enqueue(string request, Guid? requestedSessionId, double? temperature, IReadOnlyList? images = null, Guid? requestedTaskId = null) { if (string.IsNullOrWhiteSpace(request)) throw new ArgumentException("Request cannot be empty.", nameof(request)); if (temperature is < 0 or > 2) throw new ArgumentOutOfRangeException(nameof(temperature)); @@ -29,7 +29,7 @@ public sealed class TaskRegistry sessionId = _sessions.Create().SessionId; } - var task = new MutableTask(Guid.NewGuid(), sessionId, request, temperature, images ?? []); + var task = new MutableTask(requestedTaskId ?? Guid.NewGuid(), sessionId, request, temperature, images ?? []); if (!_tasks.TryAdd(task.TaskId, task)) throw new InvalidOperationException("Could not register task."); if (!_queue.Writer.TryWrite(task.TaskId)) { @@ -60,6 +60,19 @@ public sealed class TaskRegistry public bool AddUsage(Guid taskId, long inputTokens, long outputTokens) => _tasks.TryGetValue(taskId, out MutableTask? task) && task.AddUsage(inputTokens, outputTokens); + public CancellationToken GetCancellationToken(Guid taskId) => + _tasks.TryGetValue(taskId, out MutableTask? task) ? task.CancellationToken : CancellationToken.None; + + public bool Cancel(Guid taskId) => + _tasks.TryGetValue(taskId, out MutableTask? task) && task.Cancel(); + + public async Task WaitForTerminalAsync(Guid taskId, CancellationToken cancellationToken) + { + if (!_tasks.TryGetValue(taskId, out MutableTask? task)) throw new KeyNotFoundException($"Task '{taskId}' was not found."); + await task.Terminal.WaitAsync(cancellationToken).ConfigureAwait(false); + return task.Snapshot(); + } + private sealed class MutableTask(Guid taskId, Guid sessionId, string request, double? temperature, IReadOnlyList images) { private readonly object _sync = new(); @@ -68,9 +81,13 @@ public sealed class TaskRegistry private string? _result; private long _inputTokens; private long _outputTokens; + private readonly CancellationTokenSource _cancellation = new(); + private readonly TaskCompletionSource _terminal = new(TaskCreationOptions.RunContinuationsAsynchronously); public Guid TaskId { get; } = taskId; public Guid SessionId { get; } = sessionId; + public CancellationToken CancellationToken => _cancellation.Token; + public Task Terminal => _terminal.Task; public TaskSnapshot Snapshot() { @@ -100,6 +117,21 @@ public sealed class TaskRegistry _state = next; _currentTask = currentTask; _result = result; + if (next is AgentTaskState.Completed or AgentTaskState.Failed or AgentTaskState.Cancelled) _terminal.TrySetResult(); + return true; + } + } + + public bool Cancel() + { + lock (_sync) + { + if (_state is AgentTaskState.Completed or AgentTaskState.Failed or AgentTaskState.Cancelled) return false; + _cancellation.Cancel(); + _state = AgentTaskState.Cancelled; + _currentTask = null; + _result = "cancelled"; + _terminal.TrySetResult(); return true; } } diff --git a/src/Agent.Mcp/Agent.Mcp.csproj b/src/Agent.Mcp/Agent.Mcp.csproj index 5b09556..df76e22 100644 --- a/src/Agent.Mcp/Agent.Mcp.csproj +++ b/src/Agent.Mcp/Agent.Mcp.csproj @@ -5,6 +5,7 @@ + diff --git a/src/Agent.Mcp/AgentTools.cs b/src/Agent.Mcp/AgentTools.cs index c44d6ec..6a5759f 100644 --- a/src/Agent.Mcp/AgentTools.cs +++ b/src/Agent.Mcp/AgentTools.cs @@ -3,34 +3,47 @@ using Agent.Core.Localization; using Agent.Core.Tasks; using ModelContextProtocol.Server; using ModelContextProtocol.Protocol; +using ModelContextProtocol.Extensions.Tasks; +using System.Text.Json; +using System.Text.Json.Nodes; namespace Agent.Mcp; [McpServerToolType] -public sealed class AgentTools(TaskRegistry registry, LocaleCatalog locale, VisionLimits visionLimits) +public sealed class AgentTools(TaskRegistry registry, LocaleCatalog locale, VisionLimits visionLimits, PersistentMcpTaskStore taskStore) { [McpServerTool(Name = "agent_run", ReadOnly = false, Destructive = true, OpenWorld = false, UseStructuredContent = false)] [Description("Queue a natural-language request for asynchronous execution.")] - public string AgentRun( + public async Task AgentRun( RequestContext context, [Description("Non-empty natural-language work request.")] string request, [Description("Existing session UUID, or omit to create a session.")] Guid? session_id = null, [Description("Task-scoped temperature from 0.0 through 2.0.")] double? temperature = null, - [Description("Optional PNG, JPEG, or WebP images.")] IReadOnlyList? images = null) + [Description("Optional PNG, JPEG, or WebP images.")] IReadOnlyList? images = null, + CancellationToken cancellationToken = default) { RequireOnly(context, "request", "session_id", "temperature", "images"); ValidateImages(images); - TaskSnapshot task = registry.Enqueue(request, session_id, temperature, images?.Select(static image => image.DataBase64).ToArray()); + Guid? mcpTaskId = GetRelatedTaskId(context.Params?.Meta) ?? GetRelatedTaskId(context.JsonRpcRequest) ?? + taskStore.ResolveTaskId(cancellationToken); + TaskSnapshot task = registry.Enqueue(request, session_id, temperature, images?.Select(static image => image.DataBase64).ToArray(), mcpTaskId); string message = locale.Format("mcp.task_queued", new Dictionary { ["task"] = task.TaskId, ["session"] = task.SessionId }); - return string.Join('\n', + string queued = string.Join('\n', $"task_id: {task.TaskId}", $"session_id: {task.SessionId}", "status: queued", $"message: {message}"); + if (mcpTaskId is null) return queued; + + using CancellationTokenRegistration registration = cancellationToken.Register(() => registry.Cancel(task.TaskId)); + TaskSnapshot terminal = await registry.WaitForTerminalAsync(task.TaskId, cancellationToken).ConfigureAwait(false); + if (terminal.State == AgentTaskState.Failed) throw new InvalidOperationException(terminal.Result ?? "Agent task failed."); + if (terminal.State == AgentTaskState.Cancelled) throw new OperationCanceledException("Agent task was cancelled.", cancellationToken); + return FormatStatus(terminal); } private void ValidateImages(IReadOnlyList? images) @@ -66,6 +79,11 @@ public sealed class AgentTools(TaskRegistry registry, LocaleCatalog locale, Visi $"message: {locale.Format("mcp.task_not_found", new Dictionary { ["task"] = task_id })}"); } + return FormatStatus(task); + } + + private string FormatStatus(TaskSnapshot task) + { string state = task.State.ToString().ToLowerInvariant(); bool terminal = task.State is AgentTaskState.Completed or AgentTaskState.Failed or AgentTaskState.Cancelled; return string.Join('\n', @@ -80,6 +98,43 @@ public sealed class AgentTools(TaskRegistry registry, LocaleCatalog locale, Visi terminal ? $"result: {task.Result}" : "result:"); } + private static Guid? GetRelatedTaskId(object? metadata) + { + if (metadata is null) return null; + JsonElement json = JsonSerializer.SerializeToElement(metadata); + return Find(json); + + static Guid? Find(JsonElement json) + { + if (json.ValueKind == JsonValueKind.Object) + { + foreach (JsonProperty property in json.EnumerateObject()) + { + if (property.NameEquals(TasksProtocol.MetaRelatedTask)) return Parse(property.Value); + Guid? nested = Find(property.Value); + if (nested is not null) return nested; + } + } + else if (json.ValueKind == JsonValueKind.Array) + { + foreach (JsonElement item in json.EnumerateArray()) + { + Guid? nested = Find(item); + if (nested is not null) return nested; + } + } + return null; + } + + static Guid? Parse(JsonElement related) + { + if (related.ValueKind == JsonValueKind.String && Guid.TryParse(related.GetString(), out Guid direct)) return direct; + if (related.ValueKind == JsonValueKind.Object && related.TryGetProperty("taskId", out JsonElement taskId) && + Guid.TryParse(taskId.GetString(), out Guid nested)) return nested; + return null; + } + } + private static void RequireOnly(RequestContext context, params string[] allowed) { if (context.Params?.Arguments is null) return; diff --git a/src/Agent.Mcp/McpHost.cs b/src/Agent.Mcp/McpHost.cs index ff546e0..9a2ba66 100644 --- a/src/Agent.Mcp/McpHost.cs +++ b/src/Agent.Mcp/McpHost.cs @@ -15,12 +15,16 @@ using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Http; using ModelContextProtocol.Server; +using ModelContextProtocol.Extensions.Tasks; +using ModelContextProtocol; +using ModelContextProtocol.Protocol; +using System.Diagnostics; namespace Agent.Mcp; public static class McpHost { - public static async Task RunStdioAsync(LocaleCatalog locale, OllamaSettings ollamaSettings, AgentRuntimeSettings runtimeSettings, VisionLimits visionLimits, CancellationToken cancellationToken, InternalToolDispatcher? toolDispatcher = null, IAgentRuntimeObserver? observer = null) + public static async Task RunStdioAsync(LocaleCatalog locale, OllamaSettings ollamaSettings, AgentRuntimeSettings runtimeSettings, VisionLimits visionLimits, CancellationToken cancellationToken, InternalToolDispatcher? toolDispatcher = null, IAgentRuntimeObserver? observer = null, McpTaskStoreSettings? taskStoreSettings = null, IMcpHostObserver? hostObserver = null, IReadOnlyList? additionalTools = null, IReadOnlySet? taskCapableTools = null, IReadOnlySet? requiredTaskTools = null, Action? configureBuilder = null) { HostApplicationBuilder builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { @@ -37,18 +41,24 @@ public static class McpHost builder.Services.AddSingleton(runtime); builder.Services.AddHostedService(); - McpServerTool[] tools = CreateTools(registry, locale, visionLimits); + var taskStore = new PersistentMcpTaskStore(taskStoreSettings ?? DefaultTaskStoreSettings()); + McpServerTool[] tools = CreateTools(registry, locale, visionLimits, taskStore).Concat(additionalTools ?? []).ToArray(); - builder.Services + IMcpServerBuilder mcpBuilder = builder.Services .AddMcpServer() .WithStdioServerTransport() .WithTools(tools); + ConfigureAgentTaskFilter(mcpBuilder); + configureBuilder?.Invoke(mcpBuilder); + mcpBuilder.WithTasks(taskStore, options => ConfigureTasks(options, taskCapableTools, requiredTaskTools)); using IHost host = builder.Build(); - await host.RunAsync(cancellationToken).ConfigureAwait(false); + await host.StartAsync(cancellationToken).ConfigureAwait(false); + hostObserver?.Ready(new McpReadyInfo("stdio", null, Environment.ProcessId, "2026-07-28", false, true)); + await host.WaitForShutdownAsync(cancellationToken).ConfigureAwait(false); } - public static async Task RunHttpAsync(LocaleCatalog locale, OllamaSettings ollamaSettings, AgentRuntimeSettings runtimeSettings, VisionLimits visionLimits, string hostName, int port, CancellationToken cancellationToken, InternalToolDispatcher? toolDispatcher = null, IAgentRuntimeObserver? observer = null) + public static async Task RunHttpAsync(LocaleCatalog locale, OllamaSettings ollamaSettings, AgentRuntimeSettings runtimeSettings, VisionLimits visionLimits, string hostName, int port, CancellationToken cancellationToken, InternalToolDispatcher? toolDispatcher = null, IAgentRuntimeObserver? observer = null, McpTaskStoreSettings? taskStoreSettings = null, IMcpHostObserver? hostObserver = null, IReadOnlyList? additionalTools = null, IReadOnlySet? taskCapableTools = null, IReadOnlySet? requiredTaskTools = null, Action? configureBuilder = null) { if (!System.Net.IPAddress.TryParse(hostName, out System.Net.IPAddress? address) || !System.Net.IPAddress.IsLoopback(address)) throw new ArgumentException("HTTP transport host must be a loopback IP literal.", nameof(hostName)); @@ -63,29 +73,76 @@ public static class McpHost builder.Services.AddSingleton(registry); builder.Services.AddSingleton(runtime); builder.Services.AddHostedService(); - McpServerTool[] tools = CreateTools(registry, locale, visionLimits); - builder.Services.AddMcpServer().WithHttpTransport().WithTools(tools); + var taskStore = new PersistentMcpTaskStore(taskStoreSettings ?? DefaultTaskStoreSettings()); + McpServerTool[] tools = CreateTools(registry, locale, visionLimits, taskStore).Concat(additionalTools ?? []).ToArray(); + IMcpServerBuilder mcpBuilder = builder.Services.AddMcpServer() + .WithHttpTransport(options => options.Stateless = true) + .WithTools(tools); + ConfigureAgentTaskFilter(mcpBuilder); + configureBuilder?.Invoke(mcpBuilder); + mcpBuilder.WithTasks(taskStore, options => ConfigureTasks(options, taskCapableTools, requiredTaskTools)); await using WebApplication app = builder.Build(); app.Use(async (context, next) => { - if (context.Request.Headers.TryGetValue("Origin", out Microsoft.Extensions.Primitives.StringValues origins) && - origins.Any(origin => !IsLoopbackOrigin(origin))) + var stopwatch = Stopwatch.StartNew(); + DateTimeOffset timestamp = DateTimeOffset.UtcNow; + hostObserver?.HttpRequestStarted(new McpHttpRequestStartInfo(context.Connection.Id, context.TraceIdentifier, + context.Connection.RemoteIpAddress?.ToString(), context.Connection.RemotePort, + context.Request.Method, context.Request.Path.Value ?? string.Empty)); + try { - context.Response.StatusCode = StatusCodes.Status403Forbidden; - return; + if (context.Request.Headers.TryGetValue("Origin", out Microsoft.Extensions.Primitives.StringValues origins) && + origins.Any(origin => !IsLoopbackOrigin(origin))) + { + context.Response.StatusCode = StatusCodes.Status403Forbidden; + } + else await next(context).ConfigureAwait(false); } - if (context.Request.Path == "/mcp" && HttpMethods.IsGet(context.Request.Method)) + finally { - context.Response.StatusCode = StatusCodes.Status405MethodNotAllowed; - return; + stopwatch.Stop(); + hostObserver?.HttpRequest(new McpHttpRequestInfo(timestamp, context.Connection.Id, context.TraceIdentifier, + context.Connection.RemoteIpAddress?.ToString(), context.Connection.RemotePort, + context.Connection.LocalIpAddress?.ToString(), context.Connection.LocalPort, + context.Request.Method, context.Request.Path.Value ?? string.Empty, context.Response.StatusCode, + stopwatch.Elapsed.TotalMilliseconds, context.Request.ContentType, context.Request.ContentLength, + context.Request.Headers.UserAgent.ToString(), context.Request.Headers["MCP-Protocol-Version"].ToString(), + context.Request.Headers["MCP-Method"].ToString(), context.Request.Headers["MCP-Name"].ToString())); } - await next(context).ConfigureAwait(false); }); app.MapMcp("/mcp"); - try { await app.RunAsync(cancellationToken).ConfigureAwait(false); } + try + { + await app.StartAsync(cancellationToken).ConfigureAwait(false); + hostObserver?.Ready(new McpReadyInfo("http", $"http://{listenHost}:{port}/mcp", Environment.ProcessId, "2026-07-28", true, true)); + await app.WaitForShutdownAsync(cancellationToken).ConfigureAwait(false); + } finally { httpClient.Dispose(); } } + private static McpTaskStoreSettings DefaultTaskStoreSettings() => + new(Path.Combine(Path.GetTempPath(), "agent-mcp", "mcp-tasks"), TimeSpan.FromHours(24), 1000); + + private static void ConfigureTasks(McpTasksOptions options, IReadOnlySet? taskCapableTools, IReadOnlySet? requiredTaskTools) + { + IReadOnlySet names = taskCapableTools ?? new HashSet(["agent_run"], StringComparer.Ordinal); + options.ExecutionModeSelector = context => + context.MatchedPrimitive is not McpServerTool tool ? McpTaskExecutionMode.Synchronous : + requiredTaskTools?.Contains(tool.ProtocolTool.Name) == true ? McpTaskExecutionMode.Required : + names.Contains(tool.ProtocolTool.Name) ? McpTaskExecutionMode.Optional : McpTaskExecutionMode.Synchronous; + } + + private static void ConfigureAgentTaskFilter(IMcpServerBuilder builder) + { + builder.WithRequestFilters(filters => filters.AddCallToolFilter(next => async (context, cancellationToken) => + { + CallToolResult result = await next(context, cancellationToken).ConfigureAwait(false); + if (context.Params?.Name == "agent_run" && result.IsError == true) + throw new McpProtocolException("The Agent task failed.", McpErrorCode.InternalError); + return result; + })); + } + private static bool IsLoopbackOrigin(string? origin) { return Uri.TryCreate(origin, UriKind.Absolute, out Uri? uri) && @@ -94,7 +151,7 @@ public static class McpHost (System.Net.IPAddress.TryParse(uri.Host, out System.Net.IPAddress? address) && System.Net.IPAddress.IsLoopback(address))); } - private static McpServerTool[] CreateTools(TaskRegistry registry, LocaleCatalog locale, VisionLimits visionLimits) + private static McpServerTool[] CreateTools(TaskRegistry registry, LocaleCatalog locale, VisionLimits visionLimits, PersistentMcpTaskStore taskStore) { var serializerOptions = new JsonSerializerOptions(JsonSerializerDefaults.Web) { @@ -102,7 +159,7 @@ public static class McpHost TypeInfoResolver = new DefaultJsonTypeInfoResolver(), UnmappedMemberHandling = JsonUnmappedMemberHandling.Disallow }; - var target = new AgentTools(registry, locale, visionLimits); + var target = new AgentTools(registry, locale, visionLimits, taskStore); McpServerTool[] tools = typeof(AgentTools) .GetMethods(BindingFlags.Instance | BindingFlags.Public | BindingFlags.DeclaredOnly) .Where(static method => method.GetCustomAttribute() is not null) diff --git a/src/Agent.Mcp/McpHostObservability.cs b/src/Agent.Mcp/McpHostObservability.cs new file mode 100644 index 0000000..17beb07 --- /dev/null +++ b/src/Agent.Mcp/McpHostObservability.cs @@ -0,0 +1,37 @@ +namespace Agent.Mcp; + +public sealed record McpReadyInfo(string Transport, string? Endpoint, int ProcessId, string ProtocolVersion, bool Stateless, bool LegacyFallback); + +public sealed record McpHttpRequestInfo( + DateTimeOffset Timestamp, + string ConnectionId, + string TraceId, + string? RemoteIp, + int? RemotePort, + string? LocalIp, + int? LocalPort, + string Method, + string Path, + int Status, + double DurationMs, + string? ContentType, + long? ContentLength, + string? UserAgent, + string? ProtocolVersion, + string? McpMethod, + string? McpName); + +public sealed record McpHttpRequestStartInfo( + string ConnectionId, + string TraceId, + string? RemoteIp, + int RemotePort, + string Method, + string Path); + +public interface IMcpHostObserver +{ + void Ready(McpReadyInfo info); + void HttpRequestStarted(McpHttpRequestStartInfo info); + void HttpRequest(McpHttpRequestInfo info); +} diff --git a/src/Agent.Mcp/PersistentMcpTaskStore.cs b/src/Agent.Mcp/PersistentMcpTaskStore.cs new file mode 100644 index 0000000..73608d4 --- /dev/null +++ b/src/Agent.Mcp/PersistentMcpTaskStore.cs @@ -0,0 +1,190 @@ +using System.Collections.Concurrent; +using System.Text.Json; +using ModelContextProtocol.Extensions.Tasks; +using ModelContextProtocol.Protocol; + +namespace Agent.Mcp; + +public sealed record McpTaskStoreSettings(string Directory, TimeSpan TimeToLive, long PollIntervalMs); + +public sealed class PersistentMcpTaskStore : IMcpTaskStore +{ + private readonly ConcurrentDictionary _taskIdsByCancellation = new(); + private readonly ConcurrentDictionary _tasks = new(StringComparer.Ordinal); + private readonly McpTaskStoreSettings _settings; + private readonly JsonSerializerOptions _json = new(JsonSerializerDefaults.Web) { WriteIndented = true }; + + public PersistentMcpTaskStore(McpTaskStoreSettings settings) + { + _settings = settings; + System.IO.Directory.CreateDirectory(settings.Directory); + Recover(); + } + + public event Action? InputResponseReceived; + + public Guid? ResolveTaskId(CancellationToken cancellationToken) + { + if (!_taskIdsByCancellation.TryRemove(cancellationToken, out Guid taskId)) return null; + return taskId; + } + + public async Task CreateTaskAsync(CancellationToken cancellationToken) + { + DateTimeOffset now = DateTimeOffset.UtcNow; + var task = new StoredTask(Guid.NewGuid().ToString("D"), McpTaskStatus.Working, now, now, + _settings.TimeToLive, _settings.PollIntervalMs, "working", null, null, null); + if (!_tasks.TryAdd(task.TaskId, task)) throw new InvalidOperationException("Could not allocate an MCP task ID."); + await SaveAsync(task, cancellationToken).ConfigureAwait(false); + Guid createdId = Guid.Parse(task.TaskId); + _taskIdsByCancellation[cancellationToken] = createdId; + return ToInfo(task); + } + + public Task GetTaskAsync(string taskId, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + if (!_tasks.TryGetValue(taskId, out StoredTask? task)) return Task.FromResult(null); + if (IsExpired(task)) + { + _tasks.TryRemove(taskId, out _); + Delete(taskId); + return Task.FromResult(null); + } + return Task.FromResult(ToInfo(task)); + } + + public Task SetCompletedAsync(string taskId, JsonElement result, CancellationToken cancellationToken) => + UpdateAsync(taskId, McpTaskStatus.Completed, "completed", result, null, null, cancellationToken); + + public Task SetFailedAsync(string taskId, JsonElement error, CancellationToken cancellationToken) => + UpdateAsync(taskId, McpTaskStatus.Failed, "failed", null, error, null, cancellationToken); + + public async Task SetCancelledAsync(string taskId, CancellationToken cancellationToken) + { + if (!_tasks.ContainsKey(taskId)) return false; + await UpdateAsync(taskId, McpTaskStatus.Cancelled, "cancelled", null, null, null, cancellationToken).ConfigureAwait(false); + return true; + } + + public async Task ResolveInputRequestsAsync(string taskId, IDictionary responses, CancellationToken cancellationToken) + { + if (!_tasks.TryGetValue(taskId, out StoredTask? task)) return; + var pending = task.InputRequests is null + ? new Dictionary(StringComparer.Ordinal) + : new Dictionary(task.InputRequests, StringComparer.Ordinal); + var accepted = new List>(); + foreach ((string requestId, InputResponse response) in responses) + { + if (!pending.Remove(requestId)) continue; + accepted.Add(new KeyValuePair(requestId, response)); + } + await UpdateAsync(taskId, pending.Count == 0 ? McpTaskStatus.Working : McpTaskStatus.InputRequired, + pending.Count == 0 ? "working" : "input required", task.Result, task.Error, + pending.Count == 0 ? null : pending, cancellationToken).ConfigureAwait(false); + foreach ((string requestId, InputResponse response) in accepted) + InputResponseReceived?.Invoke(new InputResponseReceivedEventArgs { TaskId = taskId, RequestId = requestId, Response = response }); + } + + public Task SetInputRequestsAsync(string taskId, IDictionary requests, CancellationToken cancellationToken) => + UpdateAsync(taskId, McpTaskStatus.InputRequired, "input required", null, null, + new Dictionary(requests, StringComparer.Ordinal), cancellationToken); + + private async Task UpdateAsync(string taskId, McpTaskStatus status, string message, JsonElement? result, + JsonElement? error, IReadOnlyDictionary? inputRequests, CancellationToken cancellationToken) + { + if (!_tasks.TryGetValue(taskId, out StoredTask? current)) throw new KeyNotFoundException($"MCP task '{taskId}' was not found."); + var updated = current with + { + Status = status, + LastUpdatedAt = DateTimeOffset.UtcNow, + StatusMessage = message, + Result = result, + Error = error, + InputRequests = inputRequests + }; + _tasks[taskId] = updated; + await SaveAsync(updated, cancellationToken).ConfigureAwait(false); + } + + private void Recover() + { + foreach (string path in System.IO.Directory.EnumerateFiles(_settings.Directory, "*.json", SearchOption.TopDirectoryOnly)) + { + try + { + StoredTask? task = JsonSerializer.Deserialize(File.ReadAllText(path), _json); + if (task is null || IsExpired(task)) { File.Delete(path); continue; } + if (task.Status is McpTaskStatus.Working or McpTaskStatus.InputRequired) + { + JsonElement error = JsonSerializer.SerializeToElement(new { code = "server_restarted", message = "The server restarted before the task completed." }); + task = task with { Status = McpTaskStatus.Failed, StatusMessage = "server_restarted", Error = error, LastUpdatedAt = DateTimeOffset.UtcNow, InputRequests = null }; + SaveSynchronously(task); + } + _tasks[task.TaskId] = task; + } + catch (JsonException) + { + File.Move(path, path + ".invalid", true); + } + } + } + + private bool IsExpired(StoredTask task) => task.TimeToLive is { } ttl && DateTimeOffset.UtcNow - task.CreatedAt >= ttl; + + private async ValueTask SaveAsync(StoredTask task, CancellationToken cancellationToken) + { + string path = PathFor(task.TaskId); + string temp = path + ".tmp-" + Guid.NewGuid().ToString("N"); + try + { + await using (var stream = new FileStream(temp, FileMode.CreateNew, FileAccess.Write, FileShare.None, 4096, + FileOptions.Asynchronous | FileOptions.WriteThrough)) + { + await JsonSerializer.SerializeAsync(stream, task, _json, cancellationToken).ConfigureAwait(false); + await stream.FlushAsync(cancellationToken).ConfigureAwait(false); + } + File.Move(temp, path, true); + } + finally { if (File.Exists(temp)) File.Delete(temp); } + } + + private void SaveSynchronously(StoredTask task) + { + string path = PathFor(task.TaskId); + string temp = path + ".tmp-" + Guid.NewGuid().ToString("N"); + try + { + using (var stream = new FileStream(temp, FileMode.CreateNew, FileAccess.Write, FileShare.None, 4096, FileOptions.WriteThrough)) + { + JsonSerializer.Serialize(stream, task, _json); + stream.Flush(true); + } + File.Move(temp, path, true); + } + finally { if (File.Exists(temp)) File.Delete(temp); } + } + + private void Delete(string taskId) + { + string path = PathFor(taskId); + if (File.Exists(path)) File.Delete(path); + } + + private string PathFor(string taskId) => Path.Combine(_settings.Directory, taskId + ".json"); + + private static McpTaskInfo ToInfo(StoredTask task) => new(task.TaskId, task.Status, task.CreatedAt, task.LastUpdatedAt, + task.TimeToLive, task.PollIntervalMs, task.StatusMessage, task.Result, task.Error, task.InputRequests); + + private sealed record StoredTask( + string TaskId, + McpTaskStatus Status, + DateTimeOffset CreatedAt, + DateTimeOffset LastUpdatedAt, + TimeSpan? TimeToLive, + long? PollIntervalMs, + string StatusMessage, + JsonElement? Result, + JsonElement? Error, + IReadOnlyDictionary? InputRequests); +} diff --git a/src/Agent.Mcp/QueuedTaskWorker.cs b/src/Agent.Mcp/QueuedTaskWorker.cs index 23404e3..edea1ad 100644 --- a/src/Agent.Mcp/QueuedTaskWorker.cs +++ b/src/Agent.Mcp/QueuedTaskWorker.cs @@ -12,7 +12,15 @@ public sealed class QueuedTaskWorker(TaskRegistry registry, AgentRuntime runtime { if (!registry.Transition(taskId, AgentTaskState.Planning, "planning")) continue; registry.Transition(taskId, AgentTaskState.Running, "ollama chat"); - await runtime.ExecuteAsync(taskId, stoppingToken).ConfigureAwait(false); + using var linked = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken, registry.GetCancellationToken(taskId)); + try + { + await runtime.ExecuteAsync(taskId, linked.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (linked.IsCancellationRequested) + { + registry.Cancel(taskId); + } } } } -- cgit v1.2.3