summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/Agent.App/Configuration/AgentConfiguration.cs4
-rw-r--r--src/Agent.App/Configuration/ConfigurationLoader.cs2
-rw-r--r--src/Agent.App/Logging/McpHostLogObserver.cs76
-rw-r--r--src/Agent.App/Logging/RollingFileLog.cs55
-rw-r--r--src/Agent.App/Logging/RollingFileWriter.cs50
-rw-r--r--src/Agent.App/Program.cs30
-rw-r--r--src/Agent.Core/Tasks/TaskRegistry.cs36
-rw-r--r--src/Agent.Mcp/Agent.Mcp.csproj1
-rw-r--r--src/Agent.Mcp/AgentTools.cs65
-rw-r--r--src/Agent.Mcp/McpHost.cs93
-rw-r--r--src/Agent.Mcp/McpHostObservability.cs37
-rw-r--r--src/Agent.Mcp/PersistentMcpTaskStore.cs190
-rw-r--r--src/Agent.Mcp/QueuedTaskWorker.cs10
13 files changed, 587 insertions, 62 deletions
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<string, byte> _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<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
+ });
+ _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<string, object?>
+ {
+ ["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<string>? images = null)
+ public TaskSnapshot Enqueue(string request, Guid? requestedSessionId, double? temperature, IReadOnlyList<string>? 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<TaskSnapshot> 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<string> 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 @@
<ItemGroup>
<PackageReference Include="ModelContextProtocol" />
<PackageReference Include="ModelContextProtocol.AspNetCore" />
+ <PackageReference Include="ModelContextProtocol.Extensions.Tasks" />
</ItemGroup>
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App" />
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<string> AgentRun(
RequestContext<CallToolRequestParams> 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<McpImageInput>? images = null)
+ [Description("Optional PNG, JPEG, or WebP images.")] IReadOnlyList<McpImageInput>? 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<string, object?>
{
["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<McpImageInput>? images)
@@ -66,6 +79,11 @@ public sealed class AgentTools(TaskRegistry registry, LocaleCatalog locale, Visi
$"message: {locale.Format("mcp.task_not_found", new Dictionary<string, object?> { ["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<CallToolRequestParams> 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<McpServerTool>? additionalTools = null, IReadOnlySet<string>? taskCapableTools = null, IReadOnlySet<string>? requiredTaskTools = null, Action<IMcpServerBuilder>? configureBuilder = null)
{
HostApplicationBuilder builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings
{
@@ -37,18 +41,24 @@ public static class McpHost
builder.Services.AddSingleton(runtime);
builder.Services.AddHostedService<QueuedTaskWorker>();
- 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<McpServerTool>? additionalTools = null, IReadOnlySet<string>? taskCapableTools = null, IReadOnlySet<string>? requiredTaskTools = null, Action<IMcpServerBuilder>? 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<QueuedTaskWorker>();
- 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<string>? taskCapableTools, IReadOnlySet<string>? requiredTaskTools)
+ {
+ IReadOnlySet<string> names = taskCapableTools ?? new HashSet<string>(["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<McpServerToolAttribute>() 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<CancellationToken, Guid> _taskIdsByCancellation = new();
+ private readonly ConcurrentDictionary<string, StoredTask> _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<InputResponseReceivedEventArgs>? InputResponseReceived;
+
+ public Guid? ResolveTaskId(CancellationToken cancellationToken)
+ {
+ if (!_taskIdsByCancellation.TryRemove(cancellationToken, out Guid taskId)) return null;
+ return taskId;
+ }
+
+ public async Task<McpTaskInfo> 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<McpTaskInfo?> GetTaskAsync(string taskId, CancellationToken cancellationToken)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ if (!_tasks.TryGetValue(taskId, out StoredTask? task)) return Task.FromResult<McpTaskInfo?>(null);
+ if (IsExpired(task))
+ {
+ _tasks.TryRemove(taskId, out _);
+ Delete(taskId);
+ return Task.FromResult<McpTaskInfo?>(null);
+ }
+ return Task.FromResult<McpTaskInfo?>(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<bool> 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<string, InputResponse> responses, CancellationToken cancellationToken)
+ {
+ if (!_tasks.TryGetValue(taskId, out StoredTask? task)) return;
+ var pending = task.InputRequests is null
+ ? new Dictionary<string, InputRequest>(StringComparer.Ordinal)
+ : new Dictionary<string, InputRequest>(task.InputRequests, StringComparer.Ordinal);
+ var accepted = new List<KeyValuePair<string, InputResponse>>();
+ foreach ((string requestId, InputResponse response) in responses)
+ {
+ if (!pending.Remove(requestId)) continue;
+ accepted.Add(new KeyValuePair<string, InputResponse>(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<string, InputRequest> requests, CancellationToken cancellationToken) =>
+ UpdateAsync(taskId, McpTaskStatus.InputRequired, "input required", null, null,
+ new Dictionary<string, InputRequest>(requests, StringComparer.Ordinal), cancellationToken);
+
+ private async Task UpdateAsync(string taskId, McpTaskStatus status, string message, JsonElement? result,
+ JsonElement? error, IReadOnlyDictionary<string, InputRequest>? 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<StoredTask>(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<string, InputRequest>? 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);
+ }
}
}
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-19 19:06:18 +0000