feat: add natural-language command routing

deploy/flue-bridge/README.md

@@ -56,7 +56,7 @@ signed bytes = UTF8("v1:" + timestamp + ":") || exact HTTP body bytes
 
 The Worker accepts timestamps within 300 seconds and uses `eventId` as Flue's durable dispatch idempotency key. Timestamp is included in the MAC to prevent changing the replay window. Signed requests do not follow redirects.
 
-Replies are JSON strings published with Queue `contentType: 'text'`; `json`/`bytes` content would be Base64-encoded by HTTP pull. The consumer validates the version, destination allowlist, Markdown format, UTF-8 size, and the conversation mapping of `replyToMessageId` before sending. Invalid jobs are acknowledged as poison. Adapter failures explicitly retry with delay. Successful sends atomically save all returned message IDs and the sent ledger before acknowledgment.
+Replies are JSON strings published with Queue `contentType: 'text'`; `json`/`bytes` content would be Base64-encoded by HTTP pull. The consumer validates the version, destination allowlist, Markdown format, UTF-8 size, and the conversation mapping of `replyToMessageId` before sending. Invalid jobs are acknowledged as poison. Adapter failures explicitly retry with delay. Successful sends atomically save all returned message IDs and the sent ledger before acknowledgment.\n\nNatural-language command requests are a separate Queue job shape: kind: command carries a typed command.id, scalar command.arguments, and the original sender ID. The bot resolves the command from its runtime catalog, applies admin/group/cooldown checks, and sends the command result through the normal adapter. Command jobs use the same destination and conversation mapping validation as replies, and are deduplicated by jobId. The agent's summary command remains on the Worker Workflow rather than being sent back to the bot.
 
 Cloudflare HTTP pull's `visibility_timeout` is expressed in milliseconds; the plugin setting is in seconds. The plugin pulls one lease per batch to avoid expiring later leases while adapters render/send an earlier reply. It polls every second during the active window, gradually backs off to the idle interval, and wakes immediately after accepted ingress.
 

src/HuaJiBot.NET.Plugin.Calendar/PluginMain.cs

@@ -170,7 +170,7 @@ public async Task CalendarCommandAsync(GroupMessageEventArgs e)
     [Command("日程", "查看近期日程", Alias = "schedule")]
     // ReSharper disable once UnusedMember.Global
     public async Task CalendarCommandAsync(
-        [CommandArgumentString("时间")] string? content,
+        [CommandArgumentString("时间")] string? weeks,
         GroupMessageEventArgs e
     )
     {
@@ -179,9 +179,9 @@ GroupMessageEventArgs e
             return;
         var now = Utils.NetworkTime.Now;
         var week = 1;
-        if (!string.IsNullOrWhiteSpace(content)) //参数不为空
+        if (!string.IsNullOrWhiteSpace(weeks)) //参数不为空
         {
-            if (!int.TryParse(content, out week)) //尝试转换为数字表示周数
+            if (!int.TryParse(weeks, out week)) //尝试转换为数字表示周数
             {
                 await e.Reply("参数错误");
                 return;

src/HuaJiBot.NET.Plugin.DailySummary/PluginMain.cs

@@ -1,4 +1,4 @@
-using System.Collections.Concurrent;
+using System.Collections.Concurrent;
 using HuaJiBot.NET.AI;
 using HuaJiBot.NET.Bot;
 using HuaJiBot.NET.Commands;
@@ -147,7 +147,7 @@ [new ChatMessage(ChatRole.User, userMessage)],
     // ReSharper disable once UnusedMember.Global
     public async Task SummaryCommand(
         GroupMessageEventArgs e,
-        [CommandArgumentEnum<SummaryTarget>("日期")] SummaryTarget? target = null
+        [CommandArgumentEnum<SummaryTarget>("日期")] SummaryTarget? day = null
     )
     {
         if (!Config.GroupIds.Contains(e.GroupId))
@@ -162,7 +162,7 @@ public async Task SummaryCommand(
         _cooldown[e.SenderId] = now;
         _ = Task.Delay(Cooldown).ContinueWith(_ => _cooldown.TryRemove(e.SenderId, out var _));
 
-        var date = now.Date.AddDays(-(int)(target ?? SummaryTarget.昨天));
+        var date = now.Date.AddDays(-(int)(day ?? SummaryTarget.昨天));
         await e.Reply($"收到，正在总结 {date:yyyy-MM-dd} 的聊天记录…");
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(Config.LlmTimeoutSeconds));
         try

src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs

@@ -1,4 +1,4 @@
-using System.Text;
+using System.Text;
 using System.Text.Json;
 using HuaJiBot.NET.Bot;
 using HuaJiBot.NET.Events;
@@ -7,7 +7,7 @@
 namespace HuaJiBot.NET.Plugin.FlueBridge;
 
 internal sealed class BridgeEngine(PluginConfig config, BridgeStore store, IPluginService service,
-    IReplyQueue queue, Action<string> warn)
+    IReplyQueue queue, Action<string> warn, Func<CommandJob, CancellationToken, Task>? commandExecutor = null)
 {
     internal IngressEvent? Normalize(GroupMessageEventArgs e)
     {
@@ -69,8 +69,51 @@ private bool IsValid(ReplyJob? job)
         return store.Find(job.Destination.RobotId, job.Destination.GroupId, job.Destination.ReplyToMessageId) == job.ConversationId;
     }
 
+    private bool IsValid(CommandJob? job)
+    {
+        if (job is not { Version: 2, Destination: not null, Command: not null }
+            || string.IsNullOrWhiteSpace(job.JobId) || job.JobId.Length > 200
+            || string.IsNullOrWhiteSpace(job.ConversationId) || job.CreatedAt == default
+            || job.Destination.BridgeInstance != config.BridgeInstance
+            || !config.Allows(job.Destination.RobotId, job.Destination.GroupId)
+            || string.IsNullOrWhiteSpace(job.SenderId)
+            || string.IsNullOrWhiteSpace(job.Destination.ReplyToMessageId)
+            || job.Kind != "command"
+            || string.IsNullOrWhiteSpace(job.Command.Id)
+            || job.Command.Arguments.ValueKind != JsonValueKind.Object) return false;
+        var descriptor = service.CommandService?.GetCommandCatalog()
+            .FirstOrDefault(x => string.Equals(x.Id, job.Command.Id, StringComparison.OrdinalIgnoreCase));
+        if (descriptor is null) return false;
+        return store.Find(job.Destination.RobotId, job.Destination.GroupId, job.Destination.ReplyToMessageId) == job.ConversationId;
+    }
+
+    private static string? JobKind(string body)
+    {
+        try
+        {
+            using var doc = JsonDocument.Parse(body);
+            return doc.RootElement.TryGetProperty("kind", out var kind) && kind.ValueKind == JsonValueKind.String
+                ? kind.GetString()
+                : "chat";
+        }
+        catch (JsonException) { return null; }
+    }
+
     internal async Task DeliverAsync(QueueLease lease, CancellationToken ct)
     {
+        if (Encoding.UTF8.GetByteCount(lease.Body) > config.MaxMessageBytes)
+        {
+            warn("Discarding oversized Queue job.");
+            await queue.AckAsync(lease.LeaseId, ct);
+            return;
+        }
+        var kind = JobKind(lease.Body);
+        if (kind == "command")
+        {
+            await DeliverCommandAsync(lease, ct);
+            return;
+        }
+
         ReplyJob? job = null;
         if (Encoding.UTF8.GetByteCount(lease.Body) <= config.MaxMessageBytes)
         {
@@ -116,4 +159,48 @@ internal async Task DeliverAsync(QueueLease lease, CancellationToken ct)
         // Keep ack failures outside the send catch. A later pull sees the durable sent ledger.
         await queue.AckAsync(lease.LeaseId, ct);
     }
-}
+
+    private async Task DeliverCommandAsync(QueueLease lease, CancellationToken ct)
+    {
+        CommandJob? job = null;
+        if (Encoding.UTF8.GetByteCount(lease.Body) <= config.MaxMessageBytes)
+        {
+            try { job = JsonSerializer.Deserialize<CommandJob>(lease.Body, BridgeJson.Options); }
+            catch (JsonException) { }
+        }
+        if (job is { Version: 2 } && !string.IsNullOrWhiteSpace(job.JobId) && store.WasSent(job.JobId))
+        {
+            await queue.AckAsync(lease.LeaseId, ct);
+            return;
+        }
+        if (commandExecutor is null || !IsValid(job))
+        {
+            warn("Discarding malformed or unauthorized command job.");
+            await queue.AckAsync(lease.LeaseId, ct);
+            return;
+        }
+
+        try
+        {
+            ct.ThrowIfCancellationRequested();
+            await commandExecutor(job!, ct);
+            store.CompleteCommand(job!);
+        }
+        catch (ArgumentException ex)
+        {
+            warn($"Command rejected: {ex.Message}");
+            var content = new RichContent($"命令参数无效：{ex.Message}", job!.Destination.ReplyToMessageId);
+            await service.SendRichMessageAsync(job!.Destination.RobotId, job.Destination.GroupId, content,
+                () => Task.FromResult<SendingMessageBase[]>([content.ToPlainText()]));
+            store.CompleteCommand(job);
+        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
+        catch
+        {
+            warn("Command execution failed; retrying Queue lease.");
+            await queue.RetryAsync(lease.LeaseId, ct);
+            return;
+        }
+        await queue.AckAsync(lease.LeaseId, ct);
+    }
+}

src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs

@@ -148,6 +148,19 @@ internal void Complete(ReplyJob job, string[] ids)
         }
     }
 
+    internal void CompleteCommand(CommandJob job)
+    {
+        lock (_gate)
+        {
+            using var cmd = _db.CreateCommand();
+            cmd.CommandText = "INSERT INTO queue_deliveries VALUES ($j,'sent',$c,'[]',$t) ON CONFLICT(job_id) DO NOTHING";
+            cmd.Parameters.AddWithValue("$j", job.JobId);
+            cmd.Parameters.AddWithValue("$c", job.ConversationId);
+            cmd.Parameters.AddWithValue("$t", DateTimeOffset.UtcNow.ToUnixTimeSeconds());
+            cmd.ExecuteNonQuery();
+        }
+    }
+
     internal void Prune(DateTimeOffset before)
     {
         lock (_gate)

src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs

@@ -17,6 +17,10 @@ internal record ReplyDestination(string BridgeInstance, string RobotId, string G
 internal record ReplyContent(string Format, string Text);
 internal record ReplyJob(int Version, string JobId, string ConversationId,
     ReplyDestination Destination, ReplyContent Content, DateTimeOffset CreatedAt, string Kind = "chat");
+internal record CommandRequest(string Id, JsonElement Arguments);
+internal record CommandJob(int Version, string JobId, string ConversationId,
+    ReplyDestination Destination, string SenderId, string SenderName, CommandRequest Command,
+    DateTimeOffset CreatedAt, string Kind = "command");
 internal record QueueLease(string Body, string LeaseId);
 
 internal interface IReplyQueue

src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs

@@ -1,5 +1,7 @@
-using System.Threading.Channels;
+using System.Threading.Channels;
 using System.Collections.Concurrent;
+using HuaJiBot.NET.Bot;
+using HuaJiBot.NET.Commands;
 using HuaJiBot.NET.Events;
 using Newtonsoft.Json.Linq;
 
@@ -41,7 +43,7 @@ protected override Task InitializeAsync()
         _queueHttp = new HttpClient(new HttpClientHandler { AllowAutoRedirect = false }) { Timeout = TimeSpan.FromSeconds(30) };
         _ingress = new IngressClient(_http, new Uri(Config.IngressUrl), hmac, Config.MaxMessageBytes);
         _queue = new CloudflareQueue(_queueHttp, Config, token);
-        _engine = new BridgeEngine(Config, _store, Service, _queue, message => Warn(message));
+        _engine = new BridgeEngine(Config, _store, Service, _queue, message => Warn(message), ExecuteCommandAsync);
         foreach (var destination in Config.Destinations) _engine.SeedSummaryAnchors(destination.RobotId, destination.GroupId);
         Service.Events.OnGroupMessageReceived += Receive;
         _running = Task.WhenAll(ForwardLoopAsync(_stop.Token), PullLoopAsync(_stop.Token));
@@ -144,6 +146,29 @@ private async Task PullLoopAsync(CancellationToken ct)
         catch (OperationCanceledException) when (ct.IsCancellationRequested) { }
     }
 
+    private async Task ExecuteCommandAsync(CommandJob job, CancellationToken ct)
+    {
+        if (Service is not BotService botService)
+            throw new InvalidOperationException("FlueBridge command execution requires a BotService.");
+
+        var e = new GroupMessageEventArgs(
+            () => new DefaultCommandReader([]),
+            () => ValueTask.FromResult(string.Empty)
+        )
+        {
+            Service = botService,
+            RobotId = job.Destination.RobotId,
+            MessageId = job.Destination.ReplyToMessageId,
+            GroupId = job.Destination.GroupId,
+            SenderId = job.SenderId,
+            SenderMemberCard = job.SenderName,
+            SenderMemberType = MemberType.Unknown,
+            TextMessageLazy = new(() => string.Empty),
+        };
+
+        await Service.CommandService.ExecuteCommandAsync(job.Command.Id, job.Command.Arguments, e, ct);
+    }
+
     protected override void Unload()
     {
         Service.Events.OnGroupMessageReceived -= Receive;

src/HuaJiBot.NET.Plugin.MessageBridge/PluginMain.cs

@@ -333,8 +333,8 @@ public string EnumToAttributeName<T>(T value)
     [Command("事件", "开关事件", Alias = "events")]
     // ReSharper disable once UnusedMember.Global
     public void EventControlCommand(
-        [CommandArgumentEnum<ClientEventType>("类型")] ClientEventType? typeOptional,
-        [CommandArgumentString("状态")] string? status,
+        [CommandArgumentEnum<ClientEventType>("类型")] ClientEventType? type,
+        [CommandArgumentString("状态")] string? enabled,
         GroupMessageEventArgs e
     )
     {
@@ -343,7 +343,7 @@ GroupMessageEventArgs e
             e.Reply("只有群管理员可以修改事件转发设置");
             return;
         }
-        if (typeOptional is not { } type)
+        if (type is not { } selectedType)
         {
             var allStatus = new StringBuilder();
             foreach (var clientInfo in Config.Clients)
@@ -373,14 +373,14 @@ select EnumToAttributeName(x)
             );
             return;
         }
-        var name = EnumToAttributeName(type);
-        if (!StringToToggle(status ?? "", out var result))
+        var name = EnumToAttributeName(selectedType);
+        if (!StringToToggle(enabled ?? "", out var result))
         {
             var currentStatusDisabled =
                 Config
                     .Clients.SelectMany(x => x.Groups)
                     .FirstOrDefault(x => x is { Enabled: true })
-                    ?.ForwardFromClientDisabledEvent.Contains(type)
+                    ?.ForwardFromClientDisabledEvent.Contains(selectedType)
                 ?? true;
             e.Reply("状态 可选：true、false\n" + $"当前{name}状态：{!currentStatusDisabled}");
             return;
@@ -393,9 +393,9 @@ select EnumToAttributeName(x)
             )
             {
                 if (result)
-                    group.ForwardFromClientDisabledEvent.Remove(type);
+                    group.ForwardFromClientDisabledEvent.Remove(selectedType);
                 else
-                    group.ForwardFromClientDisabledEvent.Add(type);
+                    group.ForwardFromClientDisabledEvent.Add(selectedType);
                 Service.Config.Save();
                 e.Reply($"已 {(result ? "开启" : "关闭")} 事件 {name} 的转发");
                 return;

src/HuaJiBot.NET.SourceGenerator/CommandSourceGenerator.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
@@ -195,7 +196,7 @@ SemanticModel semanticModel
             sb.AppendLine($"        yield return new CommandInfo(");
             sb.AppendLine($"            \"{key}\",");
             sb.AppendLine($"            \"{description}\",");
-            sb.AppendLine($"            args => ");
+            sb.AppendLine($"            args =>");
             sb.AppendLine($"            {{");
 
             // Check if method returns void
@@ -238,23 +239,21 @@ SemanticModel semanticModel
 
             sb.AppendLine(");");
 
-            if (!isVoid)
+            if (isVoid)
+            {
+                sb.AppendLine($"                return Task.CompletedTask;");
+            }
+            else if (method.ReturnType?.ToString().StartsWith("ValueTask", StringComparison.Ordinal) == true)
+            {
+                sb.AppendLine($"                return result.AsTask();");
+            }
+            else if (method.ReturnType?.ToString().StartsWith("Task", StringComparison.Ordinal) == true)
+            {
+                sb.AppendLine($"                return result;");
+            }
+            else
             {
-                sb.AppendLine($"                if (result is Task task)");
-                sb.AppendLine($"                {{");
-                sb.AppendLine($"                    task.ContinueWith(");
-                sb.AppendLine($"                        t =>");
-                sb.AppendLine($"                        {{");
-                sb.AppendLine($"                            if (t.Exception is not null)");
-                sb.AppendLine($"                            {{");
-                sb.AppendLine(
-                    $"                                Error($\"执行命令 {key} 时出现异常：\", t.Exception);"
-                );
-                sb.AppendLine($"                            }}");
-                sb.AppendLine($"                        }},");
-                sb.AppendLine($"                        TaskContinuationOptions.OnlyOnFaulted");
-                sb.AppendLine($"                    );");
-                sb.AppendLine($"                }}");
+                sb.AppendLine($"                return Task.CompletedTask;");
             }
             sb.AppendLine($"            }},");
 
@@ -323,6 +322,7 @@ StringBuilder sb
         }
 
         sb.AppendLine($"                new CommandArgumentInfo(");
+        sb.AppendLine($"                    \"{param.Identifier.ValueText}\",");
         sb.AppendLine($"                    {attributeCreation},");
         sb.AppendLine($"                    typeof({typeForTypeof}),");
         sb.AppendLine($"                    {isOptional},");

src/HuaJiBot.NET.UnitTest/FlueBridgeProtocolTest.cs

@@ -17,12 +17,16 @@ public void SharedProtocolExamplesRoundTripThroughActualBridgeRecords()
         var ingress = JsonSerializer.Deserialize<IngressEvent>(ingressJson, BridgeJson.Options)!;
         var replyJson = ProtocolFixture("examples/reply-job.json");
         var reply = JsonSerializer.Deserialize<ReplyJob>(replyJson, BridgeJson.Options)!;
+        var commandJson = ProtocolFixture("v2/examples/command-job.json");
+        var command = JsonSerializer.Deserialize<CommandJob>(commandJson, BridgeJson.Options)!;
         Assert.Multiple(() =>
         {
             Assert.That(JsonNode.DeepEquals(JsonNode.Parse(ingressJson), JsonSerializer.SerializeToNode(ingress, BridgeJson.Options)), Is.True);
             Assert.That(JsonNode.DeepEquals(JsonNode.Parse(replyJson), JsonSerializer.SerializeToNode(reply, BridgeJson.Options)), Is.True);
+            Assert.That(JsonNode.DeepEquals(JsonNode.Parse(commandJson), JsonSerializer.SerializeToNode(command, BridgeJson.Options)), Is.True);
             Assert.That(ingress.Kind, Is.EqualTo("chat"));
             Assert.That(reply.Destination.ReplyToMessageId, Is.EqualTo(ingress.Message.MessageId));
+            Assert.That(command.Command.Id, Is.EqualTo("schedule"));
         });
     }