using System.IO.Pipes; using System.Text; using System.Text.Json; namespace XFE.SeAgent.Cli; public sealed class AgentClientException(string code, string message, Exception? inner = null) : Exception(message, inner) { public string Code { get; } = code; } public sealed record AgentResponse(JsonElement Envelope) { public bool IsError => Envelope.TryGetProperty("error", out _); } public sealed class AgentPipeClient { public const int MaximumRequestBytes = 1048576; public const int MaximumResponseBytes = 4194304; public async Task CallAsync(string pipeName, string method, JsonElement parameters, int timeoutMs = 30000, CancellationToken cancellationToken = default) { if (string.IsNullOrWhiteSpace(pipeName) || pipeName.IndexOfAny(['\\', '/', '\0']) >= 0) throw new AgentClientException("invalid_pipe", "管道名称必须是本机管道的名称,不能包含路径。"); if (string.IsNullOrWhiteSpace(method) || method.Length > 128 || parameters.ValueKind != JsonValueKind.Object) throw new AgentClientException("invalid_request", "方法名不能为空,params 必须是 JSON 对象。"); if (timeoutMs is < 1 or > 30000) throw new AgentClientException("invalid_timeout", "请求超时必须在 1 至 30000 毫秒之间。"); string id = Guid.NewGuid().ToString("N"); byte[] request = JsonSerializer.SerializeToUtf8Bytes(new { id, method, @params = parameters, timeoutMs }); if (request.Length + 1 > MaximumRequestBytes) throw new AgentClientException("request_too_large", "请求超过 1 MiB 协议限制。"); using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); timeout.CancelAfter(timeoutMs); await using var pipe = new NamedPipeClientStream(".", pipeName, PipeDirection.InOut, PipeOptions.Asynchronous); try { await pipe.ConnectAsync(timeout.Token); await pipe.WriteAsync(request, timeout.Token); await pipe.WriteAsync(new byte[] { (byte)'\n' }, timeout.Token); await pipe.FlushAsync(timeout.Token); byte[] response = await ReadResponseAsync(pipe, timeout.Token); JsonElement envelope; try { using var json = JsonDocument.Parse(response, new JsonDocumentOptions { MaxDepth = 64 }); envelope = json.RootElement.Clone(); } catch (JsonException ex) { throw new AgentClientException("invalid_response", "Agent 返回的内容不是有效 JSON。", ex); } if (envelope.ValueKind != JsonValueKind.Object || !envelope.TryGetProperty("id", out var responseId) || responseId.ValueKind != JsonValueKind.String || responseId.GetString() != id) throw new AgentClientException("response_id_mismatch", "Agent 响应 ID 与本次请求不匹配。"); bool hasResult = envelope.TryGetProperty("result", out _), hasError = envelope.TryGetProperty("error", out var error); if (hasResult == hasError || hasError && (error.ValueKind != JsonValueKind.Object || !error.TryGetProperty("code", out _) || !error.TryGetProperty("message", out _))) throw new AgentClientException("invalid_response", "Agent 响应必须包含 result 或有效 error。"); return new(envelope); } catch (OperationCanceledException ex) when (!cancellationToken.IsCancellationRequested) { throw new AgentClientException("timeout", "等待 Agent 响应超时;操作可能已经开始,请读取当前状态后再决定是否重试。", ex); } catch (IOException ex) { throw new AgentClientException("pipe_error", "Agent 管道连接中断:" + ex.Message, ex); } } private static async Task ReadResponseAsync(Stream stream, CancellationToken token) { using var output = new MemoryStream(); var buffer = new byte[8192]; while (true) { int read = await stream.ReadAsync(buffer, token); if (read == 0) throw new AgentClientException("incomplete_response", "Agent 在完整响应结束前关闭了管道。"); int newline = Array.IndexOf(buffer, (byte)'\n', 0, read); int count = newline >= 0 ? newline : read; if (output.Length + count + 1 > MaximumResponseBytes) throw new AgentClientException("response_too_large", "Agent 响应超过 4 MiB 协议限制。"); output.Write(buffer, 0, count); if (newline < 0) continue; for (int index = newline + 1; index < read; index++) if (buffer[index] is not ((byte)' ' or (byte)'\r' or (byte)'\n' or (byte)'\t')) throw new AgentClientException("invalid_response", "一个连接只能返回一条 JSON 响应。"); return output.ToArray(); } } }