From 7ffd07d6091076a6bfe552e27249cd0791443941 Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Fri, 2 Oct 2026 22:41:45 +0800
Subject: [PATCH 01/15] feat: add signed Flue bridge with durable Queue reply
delivery
---
.github/workflows/code-quality.yml | 2 +-
HuaJiBot.NET.slnx | 1 +
build_plugins.fsx | 7 +
deploy/flue-bridge/README.md | 73 ++++++
deploy/flue-bridge/config.example.json | 16 ++
deploy/flue-bridge/deployment.json | 38 +++
.../BridgeEngine.cs | 101 ++++++++
.../BridgeStore.cs | 125 ++++++++++
.../CloudflareQueue.cs | 40 +++
.../Contracts.cs | 26 ++
.../EntryPoint.cs | 3 +
.../HuaJiBot.NET.Plugin.FlueBridge.csproj | 14 ++
.../IngressClient.cs | 33 +++
.../PluginConfig.cs | 36 +++
.../PluginMain.cs | 144 +++++++++++
src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs | 227 ++++++++++++++++++
.../HuaJiBot.NET.UnitTest.csproj | 1 +
.../PluginManager/PluginManager.cs | 8 +-
18 files changed, 892 insertions(+), 3 deletions(-)
create mode 100644 deploy/flue-bridge/README.md
create mode 100644 deploy/flue-bridge/config.example.json
create mode 100644 deploy/flue-bridge/deployment.json
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/CloudflareQueue.cs
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/EntryPoint.cs
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/HuaJiBot.NET.Plugin.FlueBridge.csproj
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/IngressClient.cs
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/PluginConfig.cs
create mode 100644 src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
create mode 100644 src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
diff --git a/.github/workflows/code-quality.yml b/.github/workflows/code-quality.yml
index bb03925..ee01a08 100644
--- a/.github/workflows/code-quality.yml
+++ b/.github/workflows/code-quality.yml
@@ -16,7 +16,7 @@ jobs:
dotnet-version: '10.x'
- name: Run focused tests
- run: dotnet test src/HuaJiBot.NET.UnitTest --filter 'FullyQualifiedName~DailySummaryTest|FullyQualifiedName~AsciiArtToolsTest|FullyQualifiedName~WebDataServiceTest|FullyQualifiedName~VisionInputTest'
+ run: dotnet test src/HuaJiBot.NET.UnitTest --filter 'FullyQualifiedName~DailySummaryTest|FullyQualifiedName~AsciiArtToolsTest|FullyQualifiedName~WebDataServiceTest|FullyQualifiedName~VisionInputTest|FullyQualifiedName~FlueBridgeTest'
format-check:
runs-on: ubuntu-latest
diff --git a/HuaJiBot.NET.slnx b/HuaJiBot.NET.slnx
index 1f45f95..cf3f89b 100644
--- a/HuaJiBot.NET.slnx
+++ b/HuaJiBot.NET.slnx
@@ -18,6 +18,7 @@
+X-HuaJiBot-Signature: v1=
+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.
+
+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.
+
+## Storage, build and rollback
+
+The plugin stores `flue_bridge.db` in its normal plugin data directory. Preserve that directory across restarts and container replacement. The default 30-day retention should exceed the Queue's message retention. Old mappings and sent rows are pruned hourly. SQLite uses WAL, FULL synchronization, parameterized SQL, and atomic mapping/ledger commits.
+
+`dotnet fsi build_plugins.fsx` includes SQLite's `runtimes` native assets in `plugins/libs/runtimes`. The bridge resolves its SQLite native library there when loaded from a plugin archive/container. Keep this directory when manually installing a plugin.
+
+Build and test:
+
+```powershell
+dotnet build HuaJiBot.NET.slnx
+dotnet test src/HuaJiBot.NET.UnitTest --filter FullyQualifiedName~FlueBridgeTest
+dotnet fsi build_plugins.fsx
+```
+
+Canary acceptance must use real adapter traffic: mention, reply to each returned bot message ID, separate threads, and Telegram topic routing. Stop the bot after a reply is queued, restart it, and verify delivery. Verify duplicate `jobId` acknowledgment, expired leases, send failure retries and eventual DLQ receipt. Unit tests simulate deduplication, restart, ack loss, adapter failures, poison jobs, protocol signing and topic IDs; they do not prove real Cloudflare DLQ behavior or platform delivery.
+
+Roll back by disabling FlueBridge, removing the canary destination from the Worker's allowlist, and restoring that group to AIChat. Keep the SQLite database and Queue for audit/recovery. Do not delete Queue messages or rewrite Durable Object migration history during rollback.
+
+## Reliability boundaries
+
+Cloudflare delivery is at least once. A crash after adapter acceptance but before recording the returned receipt can produce a duplicate reply. If an adapter partially sends multiple messages and then fails, retrying can also duplicate the successful portion. There is no shared provider-side idempotency key.
+
+Inbound forwarding has three attempts with the same payload/event ID. Its bounded in-memory buffer is not a durable inbox: a bot crash before Worker admission, overflow, or exhausted ingress retries can lose an incoming turn. Such failures emit sanitized diagnostics; they are not represented as successful admissions. Queue replies already admitted to Cloudflare survive bot downtime according to the configured Queue retention.
+
+References: [issue #64](https://github.com/nbtca/HuaJiBot.NET/issues/64), [Cloudflare HTTP pull](https://developers.cloudflare.com/queues/configuration/pull-consumers/), [Flue channels](https://flueframework.com/docs/guide/channels/).
diff --git a/deploy/flue-bridge/config.example.json b/deploy/flue-bridge/config.example.json
new file mode 100644
index 0000000..d81bc4d
--- /dev/null
+++ b/deploy/flue-bridge/config.example.json
@@ -0,0 +1,16 @@
+{
+ "Flue桥接": {
+ "Enabled": false,
+ "BridgeInstance": "main",
+ "IngressUrl": "https://huaji-agent.nbtca.workers.dev/channels/huajibot/events",
+ "AccountId": "a415a3ae7a197093d02233755d38800b",
+ "QueueId": "3181cdb9b0054a4a8f0d0bc70b8a48c5",
+ "Destinations": [{ "RobotId": "YOUR_BOT_ID", "GroupId": "YOUR_CANARY_GROUP_ID" }],
+ "MaxMessageBytes": 32768,
+ "RetentionDays": 30,
+ "ActiveWindowSeconds": 120,
+ "IdlePollSeconds": 15,
+ "RetryDelaySeconds": 10,
+ "VisibilityTimeoutSeconds": 120
+ }
+}
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
new file mode 100644
index 0000000..5358ba1
--- /dev/null
+++ b/deploy/flue-bridge/deployment.json
@@ -0,0 +1,38 @@
+{
+ "verifiedAt": "2026-10-02T22:41:45+08:00",
+ "accountEmail": "nbtca@protonmail.com",
+ "accountId": "a415a3ae7a197093d02233755d38800b",
+ "worker": "huaji-agent",
+ "workerUrl": "https://huaji-agent.nbtca.workers.dev",
+ "workerVersionId": "274df599-02f5-48b0-ae0a-4e2225ad6ae3",
+ "queue": {
+ "name": "huajibot-outbound-main",
+ "id": "3181cdb9b0054a4a8f0d0bc70b8a48c5"
+ },
+ "deadLetterQueue": {
+ "name": "huajibot-outbound-main-dlq",
+ "id": "e7c2fe8f8002433fb679df48a19e1611"
+ },
+ "httpPullConsumer": {
+ "enabled": true,
+ "visibilityTimeoutSeconds": 120,
+ "maxRetries": 5,
+ "retryDelaySeconds": 10
+ },
+ "hmacSecretConfigured": true,
+ "destinationAllowlist": [],
+ "botEnabled": false,
+ "verified": [
+ "Worker health HTTP 200",
+ "Unsigned ingress HTTP 401",
+ "Valid signed unauthorized ingress HTTP 403",
+ "Public agent route HTTP 404"
+ ],
+ "pending": [
+ "Canary robot/group selection",
+ "Bot scoped Queue API token",
+ "Bot deployment",
+ "Real model/Queue/adapter acceptance",
+ "GitHub publication of separate agent repository"
+ ]
+}
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
new file mode 100644
index 0000000..66edde0
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
@@ -0,0 +1,101 @@
+using System.Text;
+using System.Text.Json;
+using HuaJiBot.NET.Bot;
+using HuaJiBot.NET.Events;
+using HuaJiBot.NET.Interfaces;
+
+namespace HuaJiBot.NET.Plugin.FlueBridge;
+
+internal sealed class BridgeEngine(PluginConfig config, BridgeStore store, IPluginService service,
+ IReplyQueue queue, Action warn)
+{
+ internal IngressEvent? Normalize(GroupMessageEventArgs e)
+ {
+ // A missing account ID must not route a reply to an arbitrary robot.
+ if (e.RobotId is not { Length: > 0 } robot || !config.Allows(robot, e.GroupId)
+ || e.SenderId == robot || string.IsNullOrWhiteSpace(e.MessageId)) return null;
+
+ var replyReader = e.CommandReader;
+ var reply = replyReader.Reply(out var context) ? context : null;
+ var mentionReader = e.CommandReader;
+ var mentionsBot = mentionReader.At(out var at) && at == robot;
+ var conversation = store.Find(robot, e.GroupId, e.MessageId);
+ if (conversation is null && reply?.messageId is { Length: > 0 } parent)
+ conversation = store.Find(robot, e.GroupId, parent);
+ if (conversation is null && !mentionsBot) return null;
+ conversation ??= "huajibot:" + string.Join(':', new[] { config.BridgeInstance, robot, e.GroupId, e.MessageId }.Select(Uri.EscapeDataString));
+
+ var textReader = mentionsBot ? mentionReader : replyReader;
+ var text = textReader.Input(out var rest, true) ? rest : e.TextMessage;
+ if (string.IsNullOrWhiteSpace(text)) return null;
+ var eventId = string.Join(':', new[] { config.BridgeInstance, robot, e.GroupId, e.MessageId }.Select(Uri.EscapeDataString));
+ var message = new IngressEvent(1, eventId, conversation,
+ new Destination(config.BridgeInstance, robot, e.GroupId),
+ new IncomingMessage(e.MessageId, e.SenderId, e.SenderMemberCard, text),
+ reply?.messageId is { Length: > 0 } id ? new ReplyContext(id, reply.senderId, reply.content) : null);
+ if (JsonSerializer.SerializeToUtf8Bytes(message, BridgeJson.Options).Length > config.MaxMessageBytes)
+ throw new InvalidDataException("Ingress message exceeds size limit.");
+ store.Map(robot, e.GroupId, e.MessageId, conversation);
+ return message;
+ }
+
+ private bool IsValid(ReplyJob? job)
+ {
+ if (job is not { Version: 1, Destination: not null, Content: 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.Destination.ReplyToMessageId)
+ || job.Content.Format != "markdown" || string.IsNullOrWhiteSpace(job.Content.Text)
+ || Encoding.UTF8.GetByteCount(job.Content.Text) > config.MaxMessageBytes) return false;
+ // The Queue credential is not sufficient authority to select any conversation.
+ return store.Find(job.Destination.RobotId, job.Destination.GroupId, job.Destination.ReplyToMessageId) == job.ConversationId;
+ }
+
+ internal async Task DeliverAsync(QueueLease lease, CancellationToken ct)
+ {
+ ReplyJob? job = null;
+ if (Encoding.UTF8.GetByteCount(lease.Body) <= config.MaxMessageBytes)
+ {
+ try { job = JsonSerializer.Deserialize(lease.Body, BridgeJson.Options); }
+ catch (JsonException) { }
+ }
+ // A sent job remains deduplicated even if its old thread mapping has expired.
+ if (job is { Version: 1, Destination: not null } && job.Destination.BridgeInstance == config.BridgeInstance
+ && !string.IsNullOrWhiteSpace(job.JobId) && store.WasSent(job.JobId))
+ {
+ await queue.AckAsync(lease.LeaseId, ct);
+ return;
+ }
+ if (!IsValid(job))
+ {
+ warn("Discarding malformed or unauthorized Queue job.");
+ await queue.AckAsync(lease.LeaseId, ct);
+ return;
+ }
+
+ try
+ {
+ ct.ThrowIfCancellationRequested();
+ var validated = job!;
+ var destination = validated.Destination;
+ var content = new RichContent(validated.Content!.Text, destination!.ReplyToMessageId);
+ // Do not abandon a non-cancellable adapter send on shutdown: persist its result first.
+ var ids = await service.SendRichMessageAsync(destination.RobotId, destination.GroupId, content,
+ () => Task.FromResult([content.ToPlainText()]));
+ if (ids.Length == 0 || ids.Any(string.IsNullOrWhiteSpace))
+ throw new InvalidOperationException("Adapter returned no delivery receipt.");
+ store.Complete(validated, ids);
+ }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
+ catch (Exception)
+ {
+ warn("Adapter delivery failed; retrying Queue lease.");
+ await queue.RetryAsync(lease.LeaseId, ct);
+ return;
+ }
+ // Keep ack failures outside the send catch. A later pull sees the durable sent ledger.
+ await queue.AckAsync(lease.LeaseId, ct);
+ }
+}
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs
new file mode 100644
index 0000000..1f45dca
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs
@@ -0,0 +1,125 @@
+using Microsoft.Data.Sqlite;
+using System.Text.Json;
+using System.Runtime.InteropServices;
+
+namespace HuaJiBot.NET.Plugin.FlueBridge;
+
+internal sealed class BridgeStore : IDisposable
+{
+ private readonly SqliteConnection _db;
+ private readonly object _gate = new();
+ private static readonly Lazy NativeResolver = new(() =>
+ {
+ var provider = typeof(SQLitePCL.SQLite3Provider_e_sqlite3).Assembly;
+ NativeLibrary.SetDllImportResolver(provider, (name, assembly, searchPath) =>
+ {
+ if (name != "e_sqlite3") return IntPtr.Zero;
+ var filename = OperatingSystem.IsWindows() ? "e_sqlite3.dll"
+ : OperatingSystem.IsMacOS() ? "libe_sqlite3.dylib" : "libe_sqlite3.so";
+ var root = Path.GetDirectoryName(assembly.Location)!;
+ var path = Path.Combine(root, "runtimes", RuntimeInformation.RuntimeIdentifier, "native", filename);
+ return File.Exists(path) ? NativeLibrary.Load(path) : IntPtr.Zero;
+ });
+ return true;
+ });
+
+ internal BridgeStore(string path)
+ {
+ _ = NativeResolver.Value;
+ Directory.CreateDirectory(Path.GetDirectoryName(Path.GetFullPath(path))!);
+ _db = new SqliteConnection(new SqliteConnectionStringBuilder { DataSource = path, Pooling = false }.ToString());
+ _db.Open();
+ using var command = _db.CreateCommand();
+ command.CommandText = """
+ PRAGMA journal_mode=WAL;
+ PRAGMA synchronous=FULL;
+ CREATE TABLE IF NOT EXISTS message_conversations (
+ robot_id TEXT NOT NULL, group_id TEXT NOT NULL, message_id TEXT NOT NULL,
+ conversation_id TEXT NOT NULL, created_at INTEGER NOT NULL,
+ PRIMARY KEY (robot_id, group_id, message_id));
+ CREATE TABLE IF NOT EXISTS queue_deliveries (
+ job_id TEXT PRIMARY KEY, status TEXT NOT NULL, conversation_id TEXT NOT NULL,
+ resulting_message_ids TEXT NOT NULL, updated_at INTEGER NOT NULL);
+ CREATE INDEX IF NOT EXISTS mapping_age ON message_conversations(created_at);
+ CREATE INDEX IF NOT EXISTS delivery_age ON queue_deliveries(updated_at);
+ """;
+ command.ExecuteNonQuery();
+ }
+
+ internal string? Find(string robot, string group, string message)
+ {
+ lock (_gate)
+ {
+ using var cmd = _db.CreateCommand();
+ cmd.CommandText = "SELECT conversation_id FROM message_conversations WHERE robot_id=$r AND group_id=$g AND message_id=$m";
+ cmd.Parameters.AddWithValue("$r", robot);
+ cmd.Parameters.AddWithValue("$g", group);
+ cmd.Parameters.AddWithValue("$m", message);
+ return cmd.ExecuteScalar() as string;
+ }
+ }
+
+ private void MapCore(string robot, string group, string message, string conversation, SqliteTransaction? tx)
+ {
+ using var cmd = _db.CreateCommand();
+ cmd.Transaction = tx;
+ cmd.CommandText = """
+ INSERT INTO message_conversations VALUES ($r,$g,$m,$c,$t)
+ ON CONFLICT(robot_id,group_id,message_id) DO UPDATE SET created_at=excluded.created_at
+ """;
+ cmd.Parameters.AddWithValue("$r", robot);
+ cmd.Parameters.AddWithValue("$g", group);
+ cmd.Parameters.AddWithValue("$m", message);
+ cmd.Parameters.AddWithValue("$c", conversation);
+ cmd.Parameters.AddWithValue("$t", DateTimeOffset.UtcNow.ToUnixTimeSeconds());
+ cmd.ExecuteNonQuery();
+ }
+
+ internal void Map(string robot, string group, string message, string conversation)
+ {
+ lock (_gate) MapCore(robot, group, message, conversation, null);
+ }
+
+ internal bool WasSent(string job)
+ {
+ lock (_gate)
+ {
+ using var cmd = _db.CreateCommand();
+ cmd.CommandText = "SELECT 1 FROM queue_deliveries WHERE job_id=$j AND status='sent'";
+ cmd.Parameters.AddWithValue("$j", job);
+ return cmd.ExecuteScalar() is not null;
+ }
+ }
+
+ internal void Complete(ReplyJob job, string[] ids)
+ {
+ lock (_gate)
+ {
+ using var tx = _db.BeginTransaction();
+ foreach (var id in ids)
+ MapCore(job.Destination.RobotId, job.Destination.GroupId, id, job.ConversationId, tx);
+ using var cmd = _db.CreateCommand();
+ cmd.Transaction = tx;
+ cmd.CommandText = "INSERT INTO queue_deliveries VALUES ($j,'sent',$c,$i,$t) ON CONFLICT(job_id) DO NOTHING";
+ cmd.Parameters.AddWithValue("$j", job.JobId);
+ cmd.Parameters.AddWithValue("$c", job.ConversationId);
+ cmd.Parameters.AddWithValue("$i", JsonSerializer.Serialize(ids));
+ cmd.Parameters.AddWithValue("$t", DateTimeOffset.UtcNow.ToUnixTimeSeconds());
+ cmd.ExecuteNonQuery();
+ tx.Commit();
+ }
+ }
+
+ internal void Prune(DateTimeOffset before)
+ {
+ lock (_gate)
+ {
+ using var cmd = _db.CreateCommand();
+ cmd.CommandText = "DELETE FROM message_conversations WHERE created_at<$t; DELETE FROM queue_deliveries WHERE status='sent' AND updated_at<$t";
+ cmd.Parameters.AddWithValue("$t", before.ToUnixTimeSeconds());
+ cmd.ExecuteNonQuery();
+ }
+ }
+
+ public void Dispose() { lock (_gate) _db.Dispose(); }
+}
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/CloudflareQueue.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/CloudflareQueue.cs
new file mode 100644
index 0000000..3576257
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/CloudflareQueue.cs
@@ -0,0 +1,40 @@
+using System.Net.Http.Headers;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace HuaJiBot.NET.Plugin.FlueBridge;
+
+internal sealed class CloudflareQueue(HttpClient http, PluginConfig config, string token) : IReplyQueue
+{
+ private readonly string _base = $"https://api.cloudflare.com/client/v4/accounts/{Uri.EscapeDataString(config.AccountId)}/queues/{Uri.EscapeDataString(config.QueueId)}/messages/";
+ private sealed record Envelope(bool Success, PullResult? Result);
+ private sealed record PullResult(List Messages);
+ private sealed record PulledMessage(string Body, [property: JsonPropertyName("lease_id")] string LeaseId);
+
+ private async Task CallAsync(string action, object payload, CancellationToken ct)
+ {
+ using var request = new HttpRequestMessage(HttpMethod.Post, _base + action);
+ request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token);
+ request.Content = new StringContent(JsonSerializer.Serialize(payload), System.Text.Encoding.UTF8, "application/json");
+ using var response = await http.SendAsync(request, ct);
+ if (!response.IsSuccessStatusCode)
+ throw new HttpRequestException("Queue API request failed.", null, response.StatusCode);
+ var result = JsonSerializer.Deserialize(await response.Content.ReadAsStringAsync(ct), BridgeJson.Options);
+ if (result?.Success != true) throw new HttpRequestException("Queue API reported failure.");
+ return result;
+ }
+
+ public async Task> PullAsync(CancellationToken ct)
+ {
+ // Process one lease at a time so adapter rendering cannot expire later leases in a batch.
+ var envelope = await CallAsync("pull", new { batch_size = 1, visibility_timeout = config.VisibilityTimeoutSeconds * 1000 }, ct);
+ if (envelope.Result?.Messages is null) throw new InvalidDataException("Invalid Queue pull response.");
+ return envelope.Result.Messages.Select(m => new QueueLease(m.Body, m.LeaseId)).ToArray();
+ }
+
+ public async Task AckAsync(string leaseId, CancellationToken ct) =>
+ await CallAsync("ack", new { acks = new[] { new { lease_id = leaseId } }, retries = Array.Empty() }, ct);
+
+ public async Task RetryAsync(string leaseId, CancellationToken ct) =>
+ await CallAsync("ack", new { acks = Array.Empty(), retries = new[] { new { lease_id = leaseId, delay_seconds = config.RetryDelaySeconds } } }, ct);
+}
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs
new file mode 100644
index 0000000..b82c0f6
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs
@@ -0,0 +1,26 @@
+using System.Text.Json;
+
+namespace HuaJiBot.NET.Plugin.FlueBridge;
+
+internal static class BridgeJson
+{
+ internal static readonly JsonSerializerOptions Options = new(JsonSerializerDefaults.Web);
+}
+
+internal record Destination(string BridgeInstance, string RobotId, string GroupId);
+internal record IncomingMessage(string MessageId, string SenderId, string SenderName, string Text);
+internal record ReplyContext(string MessageId, string? SenderId, string? Content);
+internal record IngressEvent(int Version, string EventId, string ConversationId,
+ Destination Destination, IncomingMessage Message, ReplyContext? ReplyTo);
+internal record ReplyDestination(string BridgeInstance, string RobotId, string GroupId, string ReplyToMessageId);
+internal record ReplyContent(string Format, string Text);
+internal record ReplyJob(int Version, string JobId, string ConversationId,
+ ReplyDestination Destination, ReplyContent Content, DateTimeOffset CreatedAt);
+internal record QueueLease(string Body, string LeaseId);
+
+internal interface IReplyQueue
+{
+ Task> PullAsync(CancellationToken ct);
+ Task AckAsync(string leaseId, CancellationToken ct);
+ Task RetryAsync(string leaseId, CancellationToken ct);
+}
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/EntryPoint.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/EntryPoint.cs
new file mode 100644
index 0000000..27ee31a
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/EntryPoint.cs
@@ -0,0 +1,3 @@
+using HuaJiBot.NET.Plugin.FlueBridge;
+
+[assembly: HuaJiBot.NET.PluginEntryPoint("Flue桥接", "通过签名HTTPS和Cloudflare Queue连接独立Agent")]
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/HuaJiBot.NET.Plugin.FlueBridge.csproj b/src/HuaJiBot.NET.Plugin.FlueBridge/HuaJiBot.NET.Plugin.FlueBridge.csproj
new file mode 100644
index 0000000..05826b7
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/HuaJiBot.NET.Plugin.FlueBridge.csproj
@@ -0,0 +1,14 @@
+
+
+ net10.0
+ enable
+ enable
+ true
+
+
+
+
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/IngressClient.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/IngressClient.cs
new file mode 100644
index 0000000..2adfd28
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/IngressClient.cs
@@ -0,0 +1,33 @@
+using System.Globalization;
+using System.Net.Http.Headers;
+using System.Security.Cryptography;
+using System.Text;
+using System.Text.Json;
+
+namespace HuaJiBot.NET.Plugin.FlueBridge;
+
+internal sealed class IngressClient(HttpClient http, Uri url, string secret, int maxBytes)
+{
+ internal static string Sign(string secret, string timestamp, byte[] body)
+ {
+ var prefix = Encoding.UTF8.GetBytes($"v1:{timestamp}:");
+ byte[] signed = [.. prefix, .. body];
+ return Convert.ToHexStringLower(HMACSHA256.HashData(Encoding.UTF8.GetBytes(secret), signed));
+ }
+
+ internal async Task SendAsync(IngressEvent message, CancellationToken ct)
+ {
+ var body = JsonSerializer.SerializeToUtf8Bytes(message, BridgeJson.Options);
+ if (body.Length > maxBytes) throw new InvalidDataException("Ingress message exceeds size limit.");
+ var timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds().ToString(CultureInfo.InvariantCulture);
+ using var request = new HttpRequestMessage(HttpMethod.Post, url);
+ request.Content = new ByteArrayContent(body);
+ request.Content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
+ request.Headers.Add("X-HuaJiBot-Timestamp", timestamp);
+ request.Headers.Add("X-HuaJiBot-Signature", $"v1={Sign(secret, timestamp, body)}");
+ using var response = await http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, ct);
+ // Never include a response body, request URL, or credentials in diagnostics.
+ if (!response.IsSuccessStatusCode)
+ throw new HttpRequestException("Agent ingress rejected the request.", null, response.StatusCode);
+ }
+}
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/PluginConfig.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/PluginConfig.cs
new file mode 100644
index 0000000..a05cf61
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/PluginConfig.cs
@@ -0,0 +1,36 @@
+namespace HuaJiBot.NET.Plugin.FlueBridge;
+
+public class PluginConfig : ConfigBase
+{
+ public PluginConfig() => Enabled = false;
+ public string BridgeInstance { get; set; } = "main";
+ public string IngressUrl { get; set; } = "";
+ public string AccountId { get; set; } = "";
+ public string QueueId { get; set; } = "";
+ public List Destinations { get; set; } = [];
+ public int MaxMessageBytes { get; set; } = 32768;
+ public int RetentionDays { get; set; } = 30;
+ public int ActiveWindowSeconds { get; set; } = 120;
+ public int IdlePollSeconds { get; set; } = 15;
+ public int RetryDelaySeconds { get; set; } = 10;
+ public int VisibilityTimeoutSeconds { get; set; } = 120;
+
+ internal bool Allows(string robotId, string groupId) =>
+ Destinations.Any(d => d.RobotId == robotId && d.GroupId == groupId);
+
+ internal void Validate()
+ {
+ if (!Uri.TryCreate(IngressUrl, UriKind.Absolute, out var uri) || uri.Scheme != "https"
+ || !string.IsNullOrEmpty(uri.UserInfo) || !string.IsNullOrEmpty(uri.Fragment))
+ throw new InvalidOperationException("FlueBridge requires an HTTPS ingress URL.");
+ if (string.IsNullOrWhiteSpace(BridgeInstance) || string.IsNullOrWhiteSpace(AccountId)
+ || string.IsNullOrWhiteSpace(QueueId) || Destinations.Count == 0
+ || Destinations.Any(d => string.IsNullOrWhiteSpace(d.RobotId) || string.IsNullOrWhiteSpace(d.GroupId))
+ || MaxMessageBytes is < 1 or > 65536 || RetentionDays < 1
+ || ActiveWindowSeconds < 1 || IdlePollSeconds < 1 || RetryDelaySeconds is < 1 or > 43200
+ || VisibilityTimeoutSeconds is < 30 or > 43200)
+ throw new InvalidOperationException("Invalid FlueBridge configuration.");
+ }
+}
+
+public record BridgeDestination(string RobotId, string GroupId);
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
new file mode 100644
index 0000000..2535114
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
@@ -0,0 +1,144 @@
+using System.Threading.Channels;
+using HuaJiBot.NET.Events;
+using Newtonsoft.Json.Linq;
+
+namespace HuaJiBot.NET.Plugin.FlueBridge;
+
+public sealed class PluginMain : PluginBase, IPluginWithConfig
+{
+ public PluginConfig Config { get; } = new();
+ private readonly CancellationTokenSource _stop = new();
+ private readonly SemaphoreSlim _wake = new(0, 1);
+ private readonly Channel _incoming = Channel.CreateBounded(100);
+ private BridgeStore? _store;
+ private HttpClient? _http;
+ private BridgeEngine? _engine;
+ private IngressClient? _ingress;
+ private IReplyQueue? _queue;
+ private Task? _running;
+ private long _lastActiveTicks;
+
+ protected override Task InitializeAsync()
+ {
+ Config.Validate();
+ var hmac = Environment.GetEnvironmentVariable("HUAJIBOT_BRIDGE_HMAC_SECRET");
+ var token = Environment.GetEnvironmentVariable("HUAJIBOT_QUEUE_API_TOKEN");
+ if (string.IsNullOrWhiteSpace(hmac) || string.IsNullOrWhiteSpace(token))
+ throw new InvalidOperationException("Set HUAJIBOT_BRIDGE_HMAC_SECRET and HUAJIBOT_QUEUE_API_TOKEN in the environment.");
+ if (Service.Config.TryGetLive("AIChat", out var ai) && ai.Enabled)
+ {
+ var groups = JObject.FromObject(ai)["GroupIds"]?.Values().ToArray() ?? [];
+ if (Config.Destinations.Any(d => groups.Contains(d.GroupId)))
+ throw new InvalidOperationException("AIChat and FlueBridge groups must be mutually exclusive.");
+ }
+ _store = new BridgeStore(Path.Combine(Service.GetPluginDataPath(), "flue_bridge.db"));
+ // Signed requests and Queue credentials must never follow a redirect to another host.
+ _http = new HttpClient(new HttpClientHandler { AllowAutoRedirect = false }) { Timeout = TimeSpan.FromSeconds(30) };
+ _ingress = new IngressClient(_http, new Uri(Config.IngressUrl), hmac, Config.MaxMessageBytes);
+ _queue = new CloudflareQueue(_http, Config, token);
+ _engine = new BridgeEngine(Config, _store, Service, _queue, message => Warn(message));
+ Service.Events.OnGroupMessageReceived += Receive;
+ _running = Task.WhenAll(ForwardLoopAsync(_stop.Token), PullLoopAsync(_stop.Token));
+ Info("Flue桥接已启动");
+ return Task.CompletedTask;
+ }
+
+ private void Receive(object? sender, GroupMessageEventArgs e)
+ {
+ if (e.RobotId is { } robot && Config.Allows(robot, e.GroupId) && !_incoming.Writer.TryWrite(e))
+ Warn("Flue ingress buffer is full; message was not forwarded.");
+ }
+
+ private void Activate()
+ {
+ Interlocked.Exchange(ref _lastActiveTicks, DateTimeOffset.UtcNow.UtcTicks);
+ try { _wake.Release(); } catch (SemaphoreFullException) { }
+ }
+
+ private async Task ForwardLoopAsync(CancellationToken ct)
+ {
+ try
+ {
+ await foreach (var e in _incoming.Reader.ReadAllAsync(ct))
+ {
+ try
+ {
+ var message = _engine!.Normalize(e);
+ if (message is null) continue;
+ for (var attempt = 0; ; attempt++)
+ {
+ try
+ {
+ await _ingress!.SendAsync(message, ct);
+ Activate();
+ break;
+ }
+ catch (HttpRequestException ex) when (attempt < 2 && (ex.StatusCode is null
+ || (int)ex.StatusCode >= 500 || (int)ex.StatusCode == 429))
+ { await Task.Delay(TimeSpan.FromSeconds(Config.RetryDelaySeconds), ct); }
+ catch (OperationCanceledException) when (!ct.IsCancellationRequested && attempt < 2)
+ { await Task.Delay(TimeSpan.FromSeconds(Config.RetryDelaySeconds), ct); }
+ }
+ }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
+ catch (Exception) { Warn("Agent ingress forwarding failed."); }
+ }
+ }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested) { }
+ }
+
+ private async Task PullLoopAsync(CancellationToken ct)
+ {
+ var nextPrune = DateTimeOffset.MinValue;
+ var recovering = false;
+ try
+ {
+ while (!ct.IsCancellationRequested)
+ {
+ try
+ {
+ if (DateTimeOffset.UtcNow >= nextPrune)
+ {
+ _store!.Prune(DateTimeOffset.UtcNow.AddDays(-Config.RetentionDays));
+ nextPrune = DateTimeOffset.UtcNow.AddHours(1);
+ }
+ var leases = await _queue!.PullAsync(ct);
+ if (recovering || leases.Count > 0) Activate();
+ recovering = false;
+ foreach (var lease in leases) await _engine!.DeliverAsync(lease, ct);
+ if (leases.Count > 0) continue;
+ }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
+ catch (Exception)
+ {
+ recovering = true;
+ Warn("Queue polling failed; leases remain available for redelivery.");
+ await Task.Delay(TimeSpan.FromSeconds(Config.RetryDelaySeconds), ct);
+ }
+ var age = (DateTimeOffset.UtcNow.UtcTicks - Interlocked.Read(ref _lastActiveTicks)) / (double)TimeSpan.TicksPerSecond;
+ var delay = age < Config.ActiveWindowSeconds ? 1
+ : Math.Min(Config.IdlePollSeconds, 1 + (int)Math.Min(Config.IdlePollSeconds, (age - Config.ActiveWindowSeconds) / 10));
+ await _wake.WaitAsync(TimeSpan.FromSeconds(delay), ct);
+ }
+ }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested) { }
+ }
+
+ protected override void Unload()
+ {
+ Service.Events.OnGroupMessageReceived -= Receive;
+ _incoming.Writer.TryComplete();
+ _stop.Cancel();
+ // Defer disposal until any adapter send has returned and persisted its receipt.
+ if (_running is { } running) _ = running.ContinueWith(_ => Cleanup(), TaskScheduler.Default);
+ else Cleanup();
+ }
+
+ private void Cleanup()
+ {
+ _http?.Dispose();
+ _store?.Dispose();
+ _wake.Dispose();
+ _stop.Dispose();
+ }
+}
diff --git a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
new file mode 100644
index 0000000..3a6b632
--- /dev/null
+++ b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
@@ -0,0 +1,227 @@
+using System.Net;
+using System.Text;
+using System.Text.Json;
+using HuaJiBot.NET.Bot;
+using HuaJiBot.NET.Commands;
+using HuaJiBot.NET.Events;
+using HuaJiBot.NET.Interfaces;
+using HuaJiBot.NET.Plugin.FlueBridge;
+using Moq;
+
+namespace HuaJiBot.NET.UnitTest;
+
+public class FlueBridgeTest
+{
+ private string _path = null!;
+ private BridgeStore _store = null!;
+ private PluginConfig _config = null!;
+ private FakeQueue _queue = null!;
+ private Mock _service = null!;
+ private BridgeEngine _engine = null!;
+
+ [SetUp]
+ public void Setup()
+ {
+ _path = Path.Combine(Path.GetTempPath(), "flue-bridge-test-" + Guid.NewGuid(), "bridge.db");
+ _store = new BridgeStore(_path);
+ _config = new PluginConfig { Destinations = [new("bot", "-100:42")] };
+ _queue = new FakeQueue();
+ _service = new Mock();
+ _service.Setup(s => s.SendRichMessageAsync("bot", "-100:42", It.IsAny(), It.IsAny>>()))
+ .ReturnsAsync(["sent-1", "sent-2"]);
+ _engine = new BridgeEngine(_config, _store, _service.Object, _queue, _ => { });
+ }
+
+ [TearDown]
+ public void Cleanup()
+ {
+ _store.Dispose();
+ Directory.Delete(Path.GetDirectoryName(_path)!, true);
+ }
+
+ private static GroupMessageEventArgs Message(string id, params CommonCommandReader.ReaderEntity[] entities) =>
+ new(() => new DefaultCommandReader(entities), () => ValueTask.FromResult("group"))
+ {
+ Service = new TestAdapter(),
+ RobotId = "bot",
+ MessageId = id,
+ GroupId = "-100:42",
+ SenderId = "user",
+ SenderMemberCard = "Alice",
+ TextMessageLazy = new(() => "question")
+ };
+
+ private ReplyJob Job(string id = "job-1")
+ {
+ var conversation = _engine.Normalize(Message("root", new CommonCommandReader.ReaderAt("bot"), "Question"))!.ConversationId;
+ return new ReplyJob(1, id, conversation, new("main", "bot", "-100:42", "root"),
+ new("markdown", "**answer**"), DateTimeOffset.UtcNow);
+ }
+
+ private static QueueLease Lease(ReplyJob job) => new(JsonSerializer.Serialize(job, BridgeJson.Options), "lease-1");
+
+ [Test]
+ public void MentionsCreateIsolatedThreadsAndRepliesResolveMappedMessages()
+ {
+ var first = _engine.Normalize(Message("root1", new CommonCommandReader.ReaderAt("bot"), "first"))!;
+ var second = _engine.Normalize(Message("root2", new CommonCommandReader.ReaderAt("bot"), "second"))!;
+ var reply = _engine.Normalize(Message("follow", new CommonCommandReader.ReaderReply(new("root1", senderId: "user", content: "first")), "follow-up"))!;
+ Assert.Multiple(() =>
+ {
+ Assert.That(second.ConversationId, Is.Not.EqualTo(first.ConversationId));
+ Assert.That(reply.ConversationId, Is.EqualTo(first.ConversationId));
+ Assert.That(reply.ReplyTo!.Content, Is.EqualTo("first"));
+ Assert.That(reply.Message.Text, Is.EqualTo("follow-up"));
+ Assert.That(first.Destination.GroupId, Is.EqualTo("-100:42"));
+ Assert.That(first.ConversationId, Does.Contain("-100%3A42"));
+ Assert.That(_engine.Normalize(Message("other", "ordinary text")), Is.Null);
+ Assert.That(_engine.Normalize(Message("unknown", new CommonCommandReader.ReaderReply(new("missing")), "unknown")), Is.Null);
+ });
+ }
+
+ [Test]
+ public async Task DeliveryPersistsAllIdsAndDuplicateLeaseIsAckedWithoutResending()
+ {
+ var job = Job();
+ await _engine.DeliverAsync(Lease(job), CancellationToken.None);
+ await _engine.DeliverAsync(Lease(job) with { LeaseId = "lease-2" }, CancellationToken.None);
+ Assert.Multiple(() =>
+ {
+ Assert.That(_queue.Acks, Is.EqualTo(new[] { "lease-1", "lease-2" }));
+ Assert.That(_store.Find("bot", "-100:42", "sent-1"), Is.EqualTo(job.ConversationId));
+ Assert.That(_store.Find("bot", "-100:42", "sent-2"), Is.EqualTo(job.ConversationId));
+ Assert.That(_store.WasSent(job.JobId), Is.True);
+ });
+ _service.Verify(s => s.SendRichMessageAsync("bot", "-100:42", new RichContent("**answer**", "root"),
+ It.IsAny>>()), Times.Once);
+ }
+
+ [Test]
+ public async Task LostAcknowledgmentAndRestartDoNotResend()
+ {
+ var job = Job();
+ _queue.FailAck = true;
+ Assert.ThrowsAsync(() => _engine.DeliverAsync(Lease(job), CancellationToken.None));
+ _store.Dispose();
+ _store = new BridgeStore(_path);
+ _engine = new BridgeEngine(_config, _store, _service.Object, _queue, _ => { });
+ _queue.FailAck = false;
+ await _engine.DeliverAsync(Lease(job) with { LeaseId = "new-lease" }, CancellationToken.None);
+ Assert.That(_queue.Retries, Is.Empty);
+ _service.Verify(s => s.SendRichMessageAsync(It.IsAny(), It.IsAny(), It.IsAny(),
+ It.IsAny>>()), Times.Once);
+ }
+
+ [Test]
+ public async Task AdapterFailureRetriesWithoutMarkingSentOrAcknowledging()
+ {
+ var job = Job();
+ _service.Setup(s => s.SendRichMessageAsync(It.IsAny(), It.IsAny(), It.IsAny(),
+ It.IsAny>>())).ThrowsAsync(new IOException("offline"));
+ await _engine.DeliverAsync(Lease(job), CancellationToken.None);
+ Assert.Multiple(() =>
+ {
+ Assert.That(_queue.Acks, Is.Empty);
+ Assert.That(_queue.Retries, Is.EqualTo(new[] { "lease-1" }));
+ Assert.That(_store.WasSent(job.JobId), Is.False);
+ });
+ }
+
+ [Test]
+ public async Task MalformedUnauthorizedAndOversizedJobsAreAckedAsPoison()
+ {
+ var job = Job();
+ foreach (var poisoned in new[]
+ {
+ job with { Version = 2 }, job with { Destination = job.Destination with { BridgeInstance = "other" } },
+ job with { Destination = job.Destination with { GroupId = "other" } },
+ job with { ConversationId = "other" }, job with { Content = new("markdown", new string('x', 40000)) }
+ }) await _engine.DeliverAsync(Lease(poisoned), CancellationToken.None);
+ await _engine.DeliverAsync(new("not-json", "bad"), CancellationToken.None);
+ Assert.That(_queue.Acks, Has.Count.EqualTo(6));
+ _service.Verify(s => s.SendRichMessageAsync(It.IsAny(), It.IsAny(), It.IsAny(),
+ It.IsAny>>()), Times.Never);
+ }
+
+ [Test]
+ public void PruneRemovesOldMappingsAndCompletedRows()
+ {
+ var job = Job();
+ _store.Complete(job, ["sent"]);
+ _store.Prune(DateTimeOffset.UtcNow.AddDays(1));
+ Assert.Multiple(() =>
+ {
+ Assert.That(_store.Find("bot", "-100:42", "root"), Is.Null);
+ Assert.That(_store.WasSent(job.JobId), Is.False);
+ });
+ }
+
+ [Test]
+ public void PluginLoaderSkipsNativeRuntimeDlls()
+ {
+ var plugins = Path.Combine(Path.GetDirectoryName(_path)!, "plugins");
+ var native = Path.Combine(plugins, "libs", "runtimes", "win-x64", "native");
+ Directory.CreateDirectory(native);
+ File.WriteAllText(Path.Combine(native, "e_sqlite3.dll"), "not a managed assembly");
+ var manager = new HuaJiBot.NET.PluginManager.PluginManager();
+ var load = manager.GetType().GetMethod("LoadAllPlugins", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!;
+ Assert.DoesNotThrow(() => load.Invoke(manager, [_service.Object, new DirectoryInfo(plugins)]));
+ }
+
+ [Test]
+ public void SignatureMatchesCrossLanguageGoldenVector()
+ {
+ Assert.That(IngressClient.Sign("test-secret", "1790942400", Encoding.UTF8.GetBytes("{\"text\":\"中文\"}")),
+ Is.EqualTo("5e856ed96ab263401de4b3ae79b66202e00fb2c7d759596d8aa78d625a538030"));
+ }
+
+ [Test]
+ public async Task HttpClientsUseExactSignedBodyAndCloudflareLeaseProtocol()
+ {
+ var requests = new List<(Uri Uri, string Body, string? Signature, string? Timestamp)>();
+ using var http = new HttpClient(new CaptureHandler(async request =>
+ {
+ var body = await request.Content!.ReadAsStringAsync();
+ requests.Add((request.RequestUri!, body,
+ request.Headers.TryGetValues("X-HuaJiBot-Signature", out var sig) ? sig.Single() : null,
+ request.Headers.TryGetValues("X-HuaJiBot-Timestamp", out var ts) ? ts.Single() : null));
+ return new(HttpStatusCode.OK) { Content = new StringContent("{\"success\":true,\"result\":{\"messages\":[{\"body\":\"hello\",\"lease_id\":\"lease\"}]}}") };
+ }));
+ var message = _engine.Normalize(Message("签名", new CommonCommandReader.ReaderAt("bot"), "中文问题"))!;
+ await new IngressClient(http, new Uri("https://agent.example/channels/huajibot/events"), "secret", 32768)
+ .SendAsync(message, CancellationToken.None);
+ var signed = requests[0];
+ Assert.That(signed.Signature, Is.EqualTo("v1=" + IngressClient.Sign("secret", signed.Timestamp!, Encoding.UTF8.GetBytes(signed.Body))));
+ var queue = new CloudflareQueue(http, new PluginConfig { AccountId = "account", QueueId = "queue" }, "token");
+ await queue.PullAsync(CancellationToken.None);
+ await queue.AckAsync("lease", CancellationToken.None);
+ await queue.RetryAsync("lease", CancellationToken.None);
+ Assert.Multiple(() =>
+ {
+ Assert.That(requests[1].Uri.AbsolutePath, Is.EqualTo("/client/v4/accounts/account/queues/queue/messages/pull"));
+ Assert.That(JsonDocument.Parse(requests[1].Body).RootElement.GetProperty("visibility_timeout").GetInt32(), Is.EqualTo(120000));
+ Assert.That(requests[2].Body, Does.Contain("\"acks\":[{\"lease_id\":\"lease\"}]"));
+ Assert.That(requests[3].Body, Does.Contain("\"delay_seconds\":10"));
+ });
+ }
+
+ private sealed class FakeQueue : IReplyQueue
+ {
+ public List Acks { get; } = [];
+ public List Retries { get; } = [];
+ public bool FailAck { get; set; }
+ public Task> PullAsync(CancellationToken ct) => Task.FromResult>([]);
+ public Task AckAsync(string leaseId, CancellationToken ct)
+ {
+ if (FailAck) throw new HttpRequestException("Expired lease / lost ack");
+ Acks.Add(leaseId);
+ return Task.CompletedTask;
+ }
+ public Task RetryAsync(string leaseId, CancellationToken ct) { Retries.Add(leaseId); return Task.CompletedTask; }
+ }
+
+ private sealed class CaptureHandler(Func> handle) : HttpMessageHandler
+ {
+ protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) => handle(request);
+ }
+}
diff --git a/src/HuaJiBot.NET.UnitTest/HuaJiBot.NET.UnitTest.csproj b/src/HuaJiBot.NET.UnitTest/HuaJiBot.NET.UnitTest.csproj
index c25559e..5103f79 100644
--- a/src/HuaJiBot.NET.UnitTest/HuaJiBot.NET.UnitTest.csproj
+++ b/src/HuaJiBot.NET.UnitTest/HuaJiBot.NET.UnitTest.csproj
@@ -30,6 +30,7 @@
("self_id") ?? QQ,
MessageId = msgId,
GroupId = groupId,
SenderId = userId,
diff --git a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
index 3a6b632..1a7bfc1 100644
--- a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
+++ b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
@@ -60,6 +60,27 @@ private ReplyJob Job(string id = "job-1")
private static QueueLease Lease(ReplyJob job) => new(JsonSerializer.Serialize(job, BridgeJson.Options), "lease-1");
+ [Test]
+ public async Task OneBotGroupEventCarriesAccountIdAndReachesBridge()
+ {
+ var adapter = new HuaJiBot.NET.Adapter.OneBot.OneBotAdapter("ws://unused", null)
+ { Logger = new HuaJiBot.NET.Logger.ConsoleLogger() };
+ GroupMessageEventArgs? received = null;
+ adapter.Events.OnGroupMessageReceived += (_, e) => received = e;
+ var handler = new HuaJiBot.NET.Adapter.OneBot.OneBotMessageHandler(
+ new HuaJiBot.NET.Adapter.OneBot.OneBotApi(adapter, _ => { }), adapter);
+ await handler.ProcessMessageAsync(Newtonsoft.Json.Linq.JObject.Parse("""
+ {"post_type":"message","message_type":"group","self_id":"bot",
+ "group_id":"-100:42","user_id":"user","message_id":"real-event",
+ "sender":{"nickname":"Alice","role":"member"},
+ "message":[{"type":"at","data":{"qq":"bot"}},
+ {"type":"text","data":{"text":"question"}}]}
+ """));
+ Assert.That(received, Is.Not.Null);
+ Assert.That(received!.RobotId, Is.EqualTo("bot"));
+ Assert.That(_engine.Normalize(received), Is.Not.Null);
+ }
+
[Test]
public void MentionsCreateIsolatedThreadsAndRepliesResolveMappedMessages()
{
From fb9ed9e0bd568635dbb1a802b43535699f86e424 Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Fri, 2 Oct 2026 23:53:19 +0800
Subject: [PATCH 05/15] docs: record verified Workers AI follow-up fix
---
deploy/flue-bridge/deployment.json | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
index 2007ee0..4ea80a4 100644
--- a/deploy/flue-bridge/deployment.json
+++ b/deploy/flue-bridge/deployment.json
@@ -1,10 +1,10 @@
{
- "verifiedAt": "2026-10-02T23:23:05.0733730+08:00",
+ "verifiedAt": "2026-10-02T23:53:19.4205030+08:00",
"accountEmail": "nbtca@protonmail.com",
"accountId": "a415a3ae7a197093d02233755d38800b",
"worker": "huaji-agent",
"workerUrl": "https://huaji-agent.nbtca.workers.dev",
- "workerVersionId": "b6dbc039-6d7c-4699-806f-ca0c8b6cf27b",
+ "workerVersionId": "13cc55d9-58e7-4585-9590-e033e6b6884c",
"queue": {
"name": "huajibot-outbound-main",
"id": "3181cdb9b0054a4a8f0d0bc70b8a48c5"
@@ -39,10 +39,14 @@
"Signed ingress HTTP 202 followed by real model reply and Queue acknowledgment",
"Linux bridge SQLite startup, adapter login, running container with zero restarts",
"AIChat canary disabled; DailySummary and other services preserved",
- "OneBot real-event regression: account ID populated and mention reaches bridge; 10 bridge tests passed"
+ "OneBot real-event regression: account ID populated and mention reaches bridge; 10 bridge tests passed",
+ "User mention received and first reply sent; original reply target mapped in SQLite",
+ "Follow-up mapping verified; native model tool-history HTTP 400 diagnosed and fixed",
+ "Real synthetic two-turn model/tool execution passed, one reply tool per turn; 9 Worker tests passed",
+ "Temporary inspection route removed: HTTP 404"
],
"pending": [
- "Real group mention/reply-thread acceptance",
+ "User verification of repaired real group follow-up",
"Real platform failure/restart/DLQ acceptance",
"GitHub publication of separate agent repository"
],
From d1d86a79c101f0910e3f132543bb70408955d1b2 Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Sat, 3 Oct 2026 01:56:20 +0800
Subject: [PATCH 06/15] docs: record Agnes secret verification and Worker
origin limit
---
deploy/flue-bridge/deployment.json | 17 +++++++++++++----
1 file changed, 13 insertions(+), 4 deletions(-)
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
index 4ea80a4..3300292 100644
--- a/deploy/flue-bridge/deployment.json
+++ b/deploy/flue-bridge/deployment.json
@@ -1,10 +1,10 @@
{
- "verifiedAt": "2026-10-02T23:53:19.4205030+08:00",
+ "verifiedAt": "2026-10-03T01:56:20.1868874+08:00",
"accountEmail": "nbtca@protonmail.com",
"accountId": "a415a3ae7a197093d02233755d38800b",
"worker": "huaji-agent",
"workerUrl": "https://huaji-agent.nbtca.workers.dev",
- "workerVersionId": "13cc55d9-58e7-4585-9590-e033e6b6884c",
+ "workerVersionId": "bc91eef8-be78-42ea-9191-9bee467ccc54",
"queue": {
"name": "huajibot-outbound-main",
"id": "3181cdb9b0054a4a8f0d0bc70b8a48c5"
@@ -48,10 +48,19 @@
"pending": [
"User verification of repaired real group follow-up",
"Real platform failure/restart/DLQ acceptance",
- "GitHub publication of separate agent repository"
+ "GitHub publication of separate agent repository",
+ "Resolve Agnes Worker-origin rate limit or authorize rate-limit fallback"
],
"model": "cloudflare/@cf/meta/llama-3.3-70b-instruct-fp8-fast",
"activationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z",
"currentImage": "huajibot-local:flue-canary-onebot-fix",
- "latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T152218Z"
+ "latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T152218Z",
+ "agnesPrimary": {
+ "model": "agnes-3.0-flash",
+ "workerError": "HTTP 429 / edge code 1015",
+ "secretConfigured": true,
+ "fallbackPolicy": "explicit quota exhaustion only",
+ "directToolsStreamingAndRecallVerified": true,
+ "enabled": false
+ }
}
From a692a62b4ed4b6f0c2652b80d3b536ac337eb69d Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Sat, 3 Oct 2026 02:08:19 +0800
Subject: [PATCH 07/15] docs: record successful Agnes primary Worker Queue
retest
---
deploy/flue-bridge/deployment.json | 17 ++++++++++-------
1 file changed, 10 insertions(+), 7 deletions(-)
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
index 3300292..dfa4261 100644
--- a/deploy/flue-bridge/deployment.json
+++ b/deploy/flue-bridge/deployment.json
@@ -1,10 +1,10 @@
{
- "verifiedAt": "2026-10-03T01:56:20.1868874+08:00",
+ "verifiedAt": "2026-10-03T02:08:19.4038430+08:00",
"accountEmail": "nbtca@protonmail.com",
"accountId": "a415a3ae7a197093d02233755d38800b",
"worker": "huaji-agent",
"workerUrl": "https://huaji-agent.nbtca.workers.dev",
- "workerVersionId": "bc91eef8-be78-42ea-9191-9bee467ccc54",
+ "workerVersionId": "7ced73d2-a203-49a1-868b-66dba2fe28a1",
"queue": {
"name": "huajibot-outbound-main",
"id": "3181cdb9b0054a4a8f0d0bc70b8a48c5"
@@ -43,13 +43,15 @@
"User mention received and first reply sent; original reply target mapped in SQLite",
"Follow-up mapping verified; native model tool-history HTTP 400 diagnosed and fixed",
"Real synthetic two-turn model/tool execution passed, one reply tool per turn; 9 Worker tests passed",
- "Temporary inspection route removed: HTTP 404"
+ "Temporary inspection route removed: HTTP 404",
+ "Agnes stateless Worker tool call/streaming/recall HTTP 200",
+ "Primary-enabled real two-turn Worker/model/Queue recall passed; both synthetic replies acknowledged",
+ "Temporary Agnes diagnostic removed; HTTP 404"
],
"pending": [
"User verification of repaired real group follow-up",
"Real platform failure/restart/DLQ acceptance",
- "GitHub publication of separate agent repository",
- "Resolve Agnes Worker-origin rate limit or authorize rate-limit fallback"
+ "GitHub publication of separate agent repository"
],
"model": "cloudflare/@cf/meta/llama-3.3-70b-instruct-fp8-fast",
"activationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z",
@@ -57,10 +59,11 @@
"latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T152218Z",
"agnesPrimary": {
"model": "agnes-3.0-flash",
- "workerError": "HTTP 429 / edge code 1015",
+ "workerError": "Intermittent HTTP 429 / edge code 1015; subsequent two-turn test passed",
"secretConfigured": true,
"fallbackPolicy": "explicit quota exhaustion only",
"directToolsStreamingAndRecallVerified": true,
- "enabled": false
+ "enabled": true,
+ "workerTwoTurnQueueRecallVerified": true
}
}
From f39046ed6c8025f061db6300325a8d61dabbc9d3 Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Sat, 3 Oct 2026 02:14:33 +0800
Subject: [PATCH 08/15] docs: record verified rate-limit fallback and invalid
secondary key
---
deploy/flue-bridge/deployment.json | 17 +++++++++++------
1 file changed, 11 insertions(+), 6 deletions(-)
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
index dfa4261..9611ce4 100644
--- a/deploy/flue-bridge/deployment.json
+++ b/deploy/flue-bridge/deployment.json
@@ -1,10 +1,10 @@
{
- "verifiedAt": "2026-10-03T02:08:19.4038430+08:00",
+ "verifiedAt": "2026-10-03T02:14:33.8441915+08:00",
"accountEmail": "nbtca@protonmail.com",
"accountId": "a415a3ae7a197093d02233755d38800b",
"worker": "huaji-agent",
"workerUrl": "https://huaji-agent.nbtca.workers.dev",
- "workerVersionId": "7ced73d2-a203-49a1-868b-66dba2fe28a1",
+ "workerVersionId": "ffd675ae-becb-4d8e-baa2-67ba54c3036f",
"queue": {
"name": "huajibot-outbound-main",
"id": "3181cdb9b0054a4a8f0d0bc70b8a48c5"
@@ -46,12 +46,15 @@
"Temporary inspection route removed: HTTP 404",
"Agnes stateless Worker tool call/streaming/recall HTTP 200",
"Primary-enabled real two-turn Worker/model/Queue recall passed; both synthetic replies acknowledged",
- "Temporary Agnes diagnostic removed; HTTP 404"
+ "Temporary Agnes diagnostic removed; HTTP 404",
+ "18 Worker tests including primary/secondary key order and rate-limit fallback passed",
+ "Two-turn real Worker Queue test passed with cloudflare-limit-fallback observed in both DO runs"
],
"pending": [
"User verification of repaired real group follow-up",
"Real platform failure/restart/DLQ acceptance",
- "GitHub publication of separate agent repository"
+ "GitHub publication of separate agent repository",
+ "Valid secondary Agnes API key (or its correct endpoint)"
],
"model": "cloudflare/@cf/meta/llama-3.3-70b-instruct-fp8-fast",
"activationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z",
@@ -61,9 +64,11 @@
"model": "agnes-3.0-flash",
"workerError": "Intermittent HTTP 429 / edge code 1015; subsequent two-turn test passed",
"secretConfigured": true,
- "fallbackPolicy": "explicit quota exhaustion only",
+ "fallbackPolicy": "Any HTTP 429 or explicit quota exhaustion: primary Agnes, secondary if valid/configured, then Cloudflare",
"directToolsStreamingAndRecallVerified": true,
"enabled": true,
- "workerTwoTurnQueueRecallVerified": true
+ "workerTwoTurnQueueRecallVerified": true,
+ "secondarySecretConfigured": false,
+ "secondaryValidation": "HTTP 401 / Invalid token twice; supplied secondary secret removed"
}
}
From 3ac34b58530736eb9ae325e5606210ff7d501c2c Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Sat, 3 Oct 2026 14:24:31 +0800
Subject: [PATCH 09/15] feat: migrate group archive and AI feature routing to
Worker
---
deploy/flue-bridge/README.md | 22 ++-
deploy/flue-bridge/activate-features.sh | 148 ++++++++++++++++++
deploy/flue-bridge/config.example.json | 10 +-
deploy/flue-bridge/deployment.json | 36 ++++-
.../history-export/HistoryExport.csproj | 4 +
deploy/flue-bridge/history-export/Program.cs | 24 +++
.../BridgeEngine.cs | 32 +++-
.../BridgeStore.cs | 38 +++++
.../Contracts.cs | 7 +-
.../PluginMain.cs | 29 +++-
src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs | 30 +++-
11 files changed, 348 insertions(+), 32 deletions(-)
create mode 100644 deploy/flue-bridge/activate-features.sh
create mode 100644 deploy/flue-bridge/history-export/HistoryExport.csproj
create mode 100644 deploy/flue-bridge/history-export/Program.cs
diff --git a/deploy/flue-bridge/README.md b/deploy/flue-bridge/README.md
index d05b954..41c3c3c 100644
--- a/deploy/flue-bridge/README.md
+++ b/deploy/flue-bridge/README.md
@@ -2,12 +2,24 @@
HuaJiBot.NET forwards messages to the separate `huaji-agent` Worker over signed HTTPS and pulls replies from Cloudflare Queue. The bridge contains no model or Flue agent implementation. OneBot, Satori, and Telegram keep their existing adapter APIs. ServerlessMQ is unchanged.
-The Worker and primary Queue/DLQ were deployed to `nbtca@protonmail.com` on 2026-10-02. The bridge is enabled for robot `3623498320`, group `466691612`; that group was removed from the old AIChat plugin. DailySummary and other services are preserved. A signed synthetic request produced a real model reply through Queue; group delivery and reply-thread acceptance await user testing. See `deployment.json` for the deployed resources and validation boundaries.
+The Worker and primary Queue/DLQ were deployed to `nbtca@protonmail.com` on 2026-10-02. The bridge is enabled for robot `3623498320`, group `466691612`; that group was removed from the old AIChat plugin. The local DailySummary runtime is now disabled; its original database/configuration and other services are preserved. A signed synthetic request produced a real model reply through Queue; group delivery and reply-thread acceptance await user testing. See `deployment.json` for the deployed resources and validation boundaries.
-The account's Workers Free plan rejects Kimi K2.6. The deployed model is `cloudflare/@cf/meta/llama-3.3-70b-instruct-fp8-fast`, verified through real inference and Queue delivery.
+Agnes 3.0 Flash is primary. Text fallback is Cloudflare Llama 3.3; image threads use Cloudflare Llama 4 Scout. Rate limits and explicit exhausted allowances advance through configured Agnes keys, then Cloudflare.
On the current Windows machine, .NET SDK 10.0.401 was installed at `%USERPROFILE%\.codex\dependencies\dotnet`. If an existing terminal still resolves the older system runtime, prepend that directory to its `PATH` (or invoke its `dotnet.exe` directly) before building. The generated shared HMAC secret is stored in the user's `HUAJIBOT_BRIDGE_HMAC_SECRET` environment variable and Cloudflare's Worker secret store; inject it into the bot's deployment environment without putting it in configuration/Git.
+## Worker feature migration (2026-10-03)
+
+The Worker owns D1 archiving, UTC+8 daily scheduling, durable summary generation, WebSearch MCP, image downloads and reply-thread vision. The bot retains platform adapters, message receipts, scheduled-summary anchors and a durable forwarding outbox. Local AIChat and DailySummary are disabled; the old MCP/SearXNG artifacts are retained for rollback and are no longer required by the Agent.
+
+`总结` / `summary` accepts 今天、昨天、前天 (default yesterday). The Worker preserves the ten-message threshold, short fact-based Chinese brief, ten-second sender cooldown and per-group concurrency guard. Cron starts yesterday's summary at 00:00 Beijing time. `kind: summary` jobs use a validated local date anchor for scheduled posts, without asking the platform to reply to a nonexistent message ID.
+
+`activate-features.sh` builds an overlay and disables only local DailySummary/AIChat. Its first migration backup is `/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T052735Z`; the final image update backup is `worker-features-20261003T054443Z`. `history-export` reads the backed-up LiteDB in read-only mode. 494 recent records were imported idempotently into the Worker's D1 database. Private exports remain outside Git with mode 0600 on the server.
+
+Rollback the whole migration by disabling `SUMMARY_ENABLED` and revoking the Worker destination before restoring config/compose from the first migration backup. Preserve the original LiteDB files, bridge SQLite receipts and D1 audit data. Restore the old Worker version if needed; do not rewrite deployed migrations.
+
+Verification passed: 12 bridge tests, 23 Worker tests, real MCP search, native image recognition and image follow-up, D1 archival/import, Workflow summary execution and Queue replies. Tests used an isolated destination and sent no synthetic production-group message. User acceptance should include `总结 今天`, an image mention, an image follow-up and a live search question in group 466691612.
+
## Configuration and secrets
Merge `config.example.json` into the bot's `Plugins` configuration. Enable only after configuring the independent Worker, its primary Queue and DLQ, and a canary destination. `Destinations` is an exact robot/group allowlist. Telegram `chatId:topicId` is an opaque group value.
@@ -32,7 +44,7 @@ conversationId = huajibot:main:bot:-100%3A42:root
eventId = main:bot:-100%3A42:root
```
-`eventId` includes robot and group because platform message IDs can be scoped to a chat. Redelivery uses exactly the same ID and payload. A new mention creates a conversation; a reply to a mapped user or bot message continues it. An unknown reply without a direct bot mention is ignored. The initial contract forwards text and available reply context; images and media are not included.
+`eventId` includes robot and group because platform message IDs can be scoped to a chat. Redelivery uses exactly the same ID and payload. A new mention creates a conversation; a reply to a mapped user or bot message continues it. An unknown reply without a direct mention is archived without invoking a model. Optional `kind` is chat/archive/summary; `message.images` contains native adapter image URLs and `message.observedAt` is UTC epoch milliseconds. The Worker downloads approved QQ images and stores native attachments in the conversation.
Ingress headers:
@@ -50,7 +62,7 @@ Cloudflare HTTP pull's `visibility_timeout` is expressed in milliseconds; the pl
## Storage, build and rollback
-The canary overlay image is `huajibot-local:flue-canary-onebot-fix` on `/home/yunacelisse/stacks/huajibot`. `Dockerfile` preserves existing plugins while replacing the CLI and adding bridge dependencies. `activate.sh` checks artifact hashes, backs up config/compose/plugin data, stops only the bot, checks the intended configuration delta, and automatically restores the old config/image on startup failure. `bridge.env` is a private mode-0600 file. Supply sudo authentication through the operator's terminal; never save it in scripts.
+The canary overlay image is `huajibot-local:worker-features-v2` on `/home/yunacelisse/stacks/huajibot`. `Dockerfile` preserves existing plugins while replacing the CLI and adding bridge dependencies. `activate.sh` checks artifact hashes, backs up config/compose/plugin data, stops only the bot, checks the intended configuration delta, and automatically restores the old config/image on startup failure. `bridge.env` is a private mode-0600 file. Supply sudo authentication through the operator's terminal; never save it in scripts.
The activation backup is `/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z`. For a deployment rollback, stop only `huaji-bot-dotnet`, restore `config.json` and `compose.yaml` from that directory, and run `docker compose up -d --no-deps huaji-bot-dotnet`. Keep the bridge database and Queue. Revoke the Worker canary allowlist after the switch.
@@ -74,6 +86,6 @@ Roll back by disabling FlueBridge, removing the canary destination from the Work
Cloudflare delivery is at least once. A crash after adapter acceptance but before recording the returned receipt can produce a duplicate reply. If an adapter partially sends multiple messages and then fails, retrying can also duplicate the successful portion. There is no shared provider-side idempotency key.
-Inbound forwarding has three attempts with the same payload/event ID. Its bounded in-memory buffer is not a durable inbox: a bot crash before Worker admission, overflow, or exhausted ingress retries can lose an incoming turn. Such failures emit sanitized diagnostics; they are not represented as successful admissions. Queue replies already admitted to Cloudflare survive bot downtime according to the configured Queue retention.
+Inbound events are committed to `ingress_outbox` before entering the bounded channel. The original payload/event ID survives restart and retries; rows are removed only after Worker admission. Messages the platform never delivers while the bot is offline cannot be recovered by this outbox. Queue replies survive downtime according to Queue retention.
References: [issue #64](https://github.com/nbtca/HuaJiBot.NET/issues/64), [Cloudflare HTTP pull](https://developers.cloudflare.com/queues/configuration/pull-consumers/), [Flue channels](https://flueframework.com/docs/guide/channels/).
diff --git a/deploy/flue-bridge/activate-features.sh b/deploy/flue-bridge/activate-features.sh
new file mode 100644
index 0000000..130bcce
--- /dev/null
+++ b/deploy/flue-bridge/activate-features.sh
@@ -0,0 +1,148 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+umask 077
+
+STACK=/home/yunacelisse/stacks/huajibot
+STAGE="$STACK/deploy/flue-feature"
+IMAGE=huajibot-local:worker-features-v2
+STATUS="$STAGE/activation-status.json"
+BACKUP="$STACK/backups/worker-features-$(date -u +%Y%m%dT%H%M%SZ)"
+STOPPED=0
+SWITCHED=0
+
+if [[ $EUID -ne 0 ]]; then
+ echo 'Run this script with sudo in your own terminal.' >&2
+ exit 1
+fi
+
+write_status() {
+ python3 - "$STATUS" "$1" "$BACKUP" "$IMAGE" <<'PY'
+import datetime,json,os,sys
+path,status,backup,image=sys.argv[1:]
+with open(path+'.tmp','w') as f:
+ json.dump({'status':status,'backup':backup,'image':image,
+ 'updatedAt':datetime.datetime.now(datetime.timezone.utc).isoformat()},f)
+os.chmod(path+'.tmp',0o600)
+os.chown(path+'.tmp',1000,1000)
+os.replace(path+'.tmp',path)
+PY
+}
+
+rollback() {
+ local result=$?
+ trap - ERR
+ if [[ $STOPPED -eq 1 ]]; then
+ docker compose -f "$STACK/compose.yaml" stop huaji-bot-dotnet || true
+ cp -a "$BACKUP/config.json" "$STACK/config.json"
+ cp -a "$BACKUP/compose.yaml" "$STACK/compose.yaml"
+ docker compose -f "$STACK/compose.yaml" up -d --no-deps huaji-bot-dotnet || true
+ write_status rolled_back
+ echo "Activation failed; previous configuration/image restored. Backup: $BACKUP" >&2
+ else
+ write_status failed_before_switch
+ echo 'Activation failed before touching the running bot.' >&2
+ fi
+ exit "$result"
+}
+trap rollback ERR
+
+cd "$STACK"
+test -s "$STAGE/bridge.env"
+test "$(stat -c %a "$STAGE/bridge.env")" = 600
+test -s "$STAGE/cli/HuaJiBot.NET.CLI"
+test -s "$STAGE/bridge/HuaJiBot.NET.Plugin.FlueBridge.dll"
+test -s "$STAGE/dependencies/runtimes/linux-x64/native/libe_sqlite3.so"
+(cd "$STAGE" && sha256sum -c SHA256SUMS)
+write_status building
+
+BASE_ID=$(docker inspect huajibot --format '{{.Image}}')
+BASE=$(python3 - "$STACK/compose.yaml" <<'PY'
+import sys,yaml
+print(yaml.safe_load(open(sys.argv[1]))['services']['huaji-bot-dotnet']['image'])
+PY
+)
+test "$(docker image inspect "$BASE" --format '{{.Id}}')" = "$BASE_ID"
+docker build --build-arg "BASE_IMAGE=$BASE" -t "$IMAGE" "$STAGE" > "$STAGE/image-build.log" 2>&1
+chown 1000:1000 "$STAGE/image-build.log"
+
+mkdir -p "$BACKUP"
+cp -a compose.yaml "$BACKUP/compose.yaml"
+docker tag "$BASE_ID" "huajibot-local:rollback-flue-$(date -u +%Y%m%dT%H%M%SZ)"
+docker inspect huajibot --format '{{.Image}}' > "$BACKUP/previous-image-id.txt"
+docker compose stop huaji-bot-dotnet
+STOPPED=1
+cp -a config.json "$BACKUP/config.json"
+if [[ -d plugins/data ]]; then cp -a plugins/data "$BACKUP/plugin-data"; fi
+write_status switching
+
+python3 - "$STACK" "$STAGE" "$IMAGE" <<'PY'
+import copy,json,os,sys,yaml
+from pathlib import Path
+stack,stage,image=map(Path,sys.argv[1:])
+config_path=stack/'config.json'
+config=json.loads(config_path.read_text())
+ai=config.setdefault('Plugins',{}).get('AIChat')
+if ai:
+ ai['GroupIds']=[g for g in ai.get('GroupIds',[]) if str(g)!='466691612']
+ if not ai['GroupIds']: ai['Enabled']=False
+config['Plugins']['每日聊天总结']['Enabled']=False
+config['Plugins']['Flue桥接']={
+ 'Enabled':True,'BridgeInstance':'main',
+ 'IngressUrl':'https://huaji-agent.nbtca.workers.dev/channels/huajibot/events',
+ 'AccountId':'a415a3ae7a197093d02233755d38800b','QueueId':'3181cdb9b0054a4a8f0d0bc70b8a48c5',
+ 'Destinations':[{'RobotId':'3623498320','GroupId':'466691612'}],
+ 'MaxMessageBytes':32768,'RetentionDays':30,'ActiveWindowSeconds':120,
+ 'IdlePollSeconds':15,'RetryDelaySeconds':10,'VisibilityTimeoutSeconds':120}
+
+def atomic(path,text):
+ original=path.stat()
+ temp=path.with_name(path.name+'.flue-tmp')
+ temp.write_text(text)
+ os.chmod(temp,original.st_mode & 0o777)
+ os.chown(temp,original.st_uid,original.st_gid)
+ os.replace(temp,path)
+
+atomic(config_path,json.dumps(config,ensure_ascii=False,indent=2)+'\n')
+os.chmod(config_path,0o600)
+compose_path=stack/'compose.yaml'
+compose=yaml.safe_load(compose_path.read_text())
+service=compose['services']['huaji-bot-dotnet']
+service['image']=str(image)
+env=service.setdefault('env_file',[])
+if isinstance(env,str): env=[env]
+if './deploy/flue-feature/bridge.env' not in env: env.append('./deploy/flue-feature/bridge.env')
+service['env_file']=env
+atomic(compose_path,yaml.safe_dump(compose,sort_keys=False,allow_unicode=True))
+PY
+
+# Check exact intended edits before starting the new container; no secrets printed.
+python3 - "$STACK" "$BACKUP" <<'PY'
+import json,sys,yaml
+from pathlib import Path
+stack,backup=map(Path,sys.argv[1:])
+old=json.loads((backup/'config.json').read_text()); new=json.loads((stack/'config.json').read_text())
+for c in (old,new):
+ c.get('Plugins',{}).pop('AIChat',None); c.get('Plugins',{}).pop('Flue桥接',None); c.get('Plugins',{}).pop('每日聊天总结',None)
+assert old==new,'Unexpected bot config changes'
+old=yaml.safe_load((backup/'compose.yaml').read_text()); new=yaml.safe_load((stack/'compose.yaml').read_text())
+for c in (old,new):
+ s=c['services']['huaji-bot-dotnet']; s.pop('image',None); s.pop('env_file',None)
+assert old==new,'Unexpected compose changes'
+print('Adapter credentials, summary settings and other services preserved; local AI summary disabled.')
+PY
+
+docker compose up -d --no-deps huaji-bot-dotnet
+SWITCHED=1
+for attempt in $(seq 1 30); do
+ if docker logs --since 3m huajibot 2>&1 | grep -q 'Flue桥接已启动'; then break; fi
+ sleep 2
+done
+docker logs --since 3m huajibot > "$STAGE/runtime.log" 2>&1
+chown 1000:1000 "$STAGE/runtime.log"
+grep -q 'Flue桥接已启动' "$STAGE/runtime.log"
+test "$(docker inspect huajibot --format '{{.State.Status}}')" = running
+test "$(docker inspect huajibot --format '{{.RestartCount}}')" = 0
+write_status enabled_pending_validation
+chown -R 1000:1000 "$BACKUP"
+echo "Flue bridge is enabled for group 466691612. Backup: $BACKUP"
+echo 'Wait for Codex to verify readiness before testing in the group.'
diff --git a/deploy/flue-bridge/config.example.json b/deploy/flue-bridge/config.example.json
index 22ff0d3..16f0e98 100644
--- a/deploy/flue-bridge/config.example.json
+++ b/deploy/flue-bridge/config.example.json
@@ -5,12 +5,20 @@
"IngressUrl": "https://huaji-agent.nbtca.workers.dev/channels/huajibot/events",
"AccountId": "a415a3ae7a197093d02233755d38800b",
"QueueId": "3181cdb9b0054a4a8f0d0bc70b8a48c5",
- "Destinations": [{ "RobotId": "3623498320", "GroupId": "466691612" }],
+ "Destinations": [
+ {
+ "RobotId": "3623498320",
+ "GroupId": "466691612"
+ }
+ ],
"MaxMessageBytes": 32768,
"RetentionDays": 30,
"ActiveWindowSeconds": 120,
"IdlePollSeconds": 15,
"RetryDelaySeconds": 10,
"VisibilityTimeoutSeconds": 120
+ },
+ "每日聊天总结": {
+ "Enabled": false
}
}
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
index 9611ce4..5b55fa8 100644
--- a/deploy/flue-bridge/deployment.json
+++ b/deploy/flue-bridge/deployment.json
@@ -1,10 +1,10 @@
{
- "verifiedAt": "2026-10-03T02:14:33.8441915+08:00",
+ "verifiedAt": "2026-10-03T14:24:31.4773546+08:00",
"accountEmail": "nbtca@protonmail.com",
"accountId": "a415a3ae7a197093d02233755d38800b",
"worker": "huaji-agent",
"workerUrl": "https://huaji-agent.nbtca.workers.dev",
- "workerVersionId": "ffd675ae-becb-4d8e-baa2-67ba54c3036f",
+ "workerVersionId": "2a4e1461-eb53-48e6-9963-f5ba2e82d880",
"queue": {
"name": "huajibot-outbound-main",
"id": "3181cdb9b0054a4a8f0d0bc70b8a48c5"
@@ -48,18 +48,22 @@
"Primary-enabled real two-turn Worker/model/Queue recall passed; both synthetic replies acknowledged",
"Temporary Agnes diagnostic removed; HTTP 404",
"18 Worker tests including primary/secondary key order and rate-limit fallback passed",
- "Two-turn real Worker Queue test passed with cloudflare-limit-fallback observed in both DO runs"
+ "Two-turn real Worker Queue test passed with cloudflare-limit-fallback observed in both DO runs",
+ "Real Agent MCP search, image recognition, image follow-up without prior color disclosure, summary Workflow and Queue replies passed",
+ "494 read-only LiteDB records imported; live D1 archive increased to 521 records",
+ "Local AIChat/DailySummary disabled; bot running with zero restarts; OneBot identity/member verified",
+ "Temporary acceptance destination revoked and its D1 fixture records removed"
],
"pending": [
- "User verification of repaired real group follow-up",
+ "User acceptance of summary/search/image mention/image follow-up in QQ group",
"Real platform failure/restart/DLQ acceptance",
- "GitHub publication of separate agent repository",
- "Valid secondary Agnes API key (or its correct endpoint)"
+ "GitHub publication of independent huaji-agent repository",
+ "Valid secondary Agnes key or its correct endpoint"
],
"model": "cloudflare/@cf/meta/llama-3.3-70b-instruct-fp8-fast",
"activationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z",
- "currentImage": "huajibot-local:flue-canary-onebot-fix",
- "latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T152218Z",
+ "currentImage": "huajibot-local:worker-features-v2",
+ "latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T054443Z",
"agnesPrimary": {
"model": "agnes-3.0-flash",
"workerError": "Intermittent HTTP 429 / edge code 1015; subsequent two-turn test passed",
@@ -70,5 +74,21 @@
"workerTwoTurnQueueRecallVerified": true,
"secondarySecretConfigured": false,
"secondaryValidation": "HTTP 401 / Invalid token twice; supplied secondary secret removed"
+ },
+ "workerFeatures": {
+ "vision": true,
+ "latestLiveArchiveRecords": 521,
+ "searchMcp": true,
+ "visionFollowup": true,
+ "d1DatabaseId": "f017db6e-cd05-4d34-af55-9d13f7337538",
+ "historyImportedRecords": 494,
+ "cronUtc": "*/5 16 * * *",
+ "workflow": "huaji-daily-summary",
+ "localDailySummaryEnabled": false,
+ "dailySummary": true,
+ "originalMigrationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T052735Z",
+ "bridgeTests": 12,
+ "imageFallbackModel": "cloudflare/@cf/meta/llama-4-scout-17b-16e-instruct",
+ "workerTests": 23
}
}
diff --git a/deploy/flue-bridge/history-export/HistoryExport.csproj b/deploy/flue-bridge/history-export/HistoryExport.csproj
new file mode 100644
index 0000000..49f529c
--- /dev/null
+++ b/deploy/flue-bridge/history-export/HistoryExport.csproj
@@ -0,0 +1,4 @@
+
+ Exe net10.0 enable enable
diff --git a/deploy/flue-bridge/history-export/Program.cs b/deploy/flue-bridge/history-export/Program.cs
new file mode 100644
index 0000000..849c00e
--- /dev/null
+++ b/deploy/flue-bridge/history-export/Program.cs
@@ -0,0 +1,24 @@
+using LiteDB;
+using System.Text.Json;
+
+if (args.Length != 4) throw new ArgumentException("database, output, robot, group required");
+using var db = new LiteDatabase(new ConnectionString { Filename = args[0], ReadOnly = true });
+using var output = new StreamWriter(new FileStream(args[1], FileMode.CreateNew));
+var messages = db.GetCollection("messages").FindAll().Where(m => m["GroupId"].AsString == args[3])
+ .OrderBy(m => m["Timestamp"].AsDateTime);
+var count = 0;
+foreach (var message in messages)
+{
+ var observed = new DateTimeOffset(message["Timestamp"].AsDateTime.ToUniversalTime());
+ if (observed < DateTimeOffset.UtcNow.AddDays(-3)) continue;
+ var id = message["_id"].AsString;
+ var components = new[] { "main", args[2], args[3], id }.Select(Uri.EscapeDataString);
+ var text = message["Content"].IsString ? message["Content"].AsString : "（非文本消息）";
+ if (string.IsNullOrWhiteSpace(text)) text = "（非文本消息）";
+ var payload = new { version = 1, eventId = string.Join(':', components), conversationId = "huajibot:" + string.Join(':', components),
+ destination = new { bridgeInstance = "main", robotId = args[2], groupId = args[3] }, kind = "archive",
+ message = new { messageId = id, senderId = message["SenderId"].IsString ? message["SenderId"].AsString : "bot",
+ senderName = message["SenderName"].AsString, text, images = Array.Empty(), observedAt = observed.ToUnixTimeMilliseconds() }, replyTo = (object?)null };
+ output.WriteLine(System.Text.Json.JsonSerializer.Serialize(payload)); count++;
+}
+Console.WriteLine($"Exported {count} history records; original database opened read-only.");
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
index 66edde0..97b4ade 100644
--- a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
@@ -14,6 +14,7 @@ internal sealed class BridgeEngine(PluginConfig config, BridgeStore store, IPlug
// A missing account ID must not route a reply to an arbitrary robot.
if (e.RobotId is not { Length: > 0 } robot || !config.Allows(robot, e.GroupId)
|| e.SenderId == robot || string.IsNullOrWhiteSpace(e.MessageId)) return null;
+ SeedSummaryAnchors(robot, e.GroupId);
var replyReader = e.CommandReader;
var reply = replyReader.Reply(out var context) ? context : null;
@@ -22,23 +23,37 @@ internal sealed class BridgeEngine(PluginConfig config, BridgeStore store, IPlug
var conversation = store.Find(robot, e.GroupId, e.MessageId);
if (conversation is null && reply?.messageId is { Length: > 0 } parent)
conversation = store.Find(robot, e.GroupId, parent);
- if (conversation is null && !mentionsBot) return null;
+ var chat = conversation is not null || mentionsBot;
conversation ??= "huajibot:" + string.Join(':', new[] { config.BridgeInstance, robot, e.GroupId, e.MessageId }.Select(Uri.EscapeDataString));
var textReader = mentionsBot ? mentionReader : replyReader;
- var text = textReader.Input(out var rest, true) ? rest : e.TextMessage;
- if (string.IsNullOrWhiteSpace(text)) return null;
+ var text = chat && textReader.Input(out var rest, true) ? rest : e.TextMessage;
+ if (string.IsNullOrWhiteSpace(text)) text = e.ImageUrls.Length > 0 ? "请描述这张图片。" : "（非文本消息）";
+ var summary = System.Text.RegularExpressions.Regex.IsMatch(text.Trim(), @"^(总结|summary)(\s+(今天|昨天|前天))?$");
+ var kind = summary ? "summary" : chat ? "chat" : "archive";
var eventId = string.Join(':', new[] { config.BridgeInstance, robot, e.GroupId, e.MessageId }.Select(Uri.EscapeDataString));
var message = new IngressEvent(1, eventId, conversation,
new Destination(config.BridgeInstance, robot, e.GroupId),
- new IncomingMessage(e.MessageId, e.SenderId, e.SenderMemberCard, text),
- reply?.messageId is { Length: > 0 } id ? new ReplyContext(id, reply.senderId, reply.content) : null);
+ new IncomingMessage(e.MessageId, e.SenderId, e.SenderMemberCard, text,
+ kind == "chat" ? e.ImageUrls : [], DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()),
+ reply?.messageId is { Length: > 0 } id ? new ReplyContext(id, reply.senderId, reply.content) : null, kind);
if (JsonSerializer.SerializeToUtf8Bytes(message, BridgeJson.Options).Length > config.MaxMessageBytes)
throw new InvalidDataException("Ingress message exceeds size limit.");
- store.Map(robot, e.GroupId, e.MessageId, conversation);
+ if (kind != "archive") store.Map(robot, e.GroupId, e.MessageId, conversation);
return message;
}
+ internal void SeedSummaryAnchors(string robot, string group)
+ {
+ for (var days = 0; days < 3; days++)
+ {
+ var day = DateTimeOffset.UtcNow.ToOffset(TimeSpan.FromHours(8)).AddDays(-days).ToString("yyyy-MM-dd");
+ var anchor = "summary:" + day;
+ var conversation = "huajibot:" + string.Join(':', new[] { config.BridgeInstance, robot, group }.Select(Uri.EscapeDataString)) + ":" + anchor;
+ store.Map(robot, group, anchor, conversation);
+ }
+ }
+
private bool IsValid(ReplyJob? job)
{
if (job is not { Version: 1, Destination: not null, Content: not null }
@@ -47,6 +62,7 @@ private bool IsValid(ReplyJob? job)
|| job.Destination.BridgeInstance != config.BridgeInstance
|| !config.Allows(job.Destination.RobotId, job.Destination.GroupId)
|| string.IsNullOrWhiteSpace(job.Destination.ReplyToMessageId)
+ || job.Kind is not ("chat" or "summary")
|| job.Content.Format != "markdown" || string.IsNullOrWhiteSpace(job.Content.Text)
|| Encoding.UTF8.GetByteCount(job.Content.Text) > config.MaxMessageBytes) return false;
// The Queue credential is not sufficient authority to select any conversation.
@@ -80,7 +96,9 @@ internal async Task DeliverAsync(QueueLease lease, CancellationToken ct)
ct.ThrowIfCancellationRequested();
var validated = job!;
var destination = validated.Destination;
- var content = new RichContent(validated.Content!.Text, destination!.ReplyToMessageId);
+ var content = new RichContent(validated.Content!.Text,
+ validated.Kind == "summary" && destination!.ReplyToMessageId.StartsWith("summary:", StringComparison.Ordinal)
+ ? null : destination!.ReplyToMessageId);
// Do not abandon a non-cancellable adapter send on shutdown: persist its result first.
var ids = await service.SendRichMessageAsync(destination.RobotId, destination.GroupId, content,
() => Task.FromResult([content.ToPlainText()]));
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs
index 1f45dca..08e361d 100644
--- a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs
@@ -42,6 +42,7 @@ CREATE TABLE IF NOT EXISTS queue_deliveries (
resulting_message_ids TEXT NOT NULL, updated_at INTEGER NOT NULL);
CREATE INDEX IF NOT EXISTS mapping_age ON message_conversations(created_at);
CREATE INDEX IF NOT EXISTS delivery_age ON queue_deliveries(updated_at);
+ CREATE TABLE IF NOT EXISTS ingress_outbox (event_id TEXT PRIMARY KEY, body TEXT NOT NULL, created_at INTEGER NOT NULL);
""";
command.ExecuteNonQuery();
}
@@ -80,6 +81,43 @@ internal void Map(string robot, string group, string message, string conversatio
lock (_gate) MapCore(robot, group, message, conversation, null);
}
+ internal void SaveIngress(IngressEvent message)
+ {
+ lock (_gate)
+ {
+ using var cmd = _db.CreateCommand();
+ cmd.CommandText = "INSERT OR IGNORE INTO ingress_outbox VALUES($id,$body,$t)";
+ cmd.Parameters.AddWithValue("$id", message.EventId);
+ cmd.Parameters.AddWithValue("$body", JsonSerializer.Serialize(message, BridgeJson.Options));
+ cmd.Parameters.AddWithValue("$t", DateTimeOffset.UtcNow.ToUnixTimeSeconds());
+ cmd.ExecuteNonQuery();
+ }
+ }
+
+ internal IngressEvent[] PendingIngress()
+ {
+ lock (_gate)
+ {
+ using var cmd = _db.CreateCommand();
+ cmd.CommandText = "SELECT body FROM ingress_outbox ORDER BY created_at,rowid LIMIT 50";
+ using var reader = cmd.ExecuteReader();
+ var messages = new List();
+ while (reader.Read()) messages.Add(JsonSerializer.Deserialize(reader.GetString(0), BridgeJson.Options)!);
+ return messages.ToArray();
+ }
+ }
+
+ internal void RemoveIngress(string id)
+ {
+ lock (_gate)
+ {
+ using var cmd = _db.CreateCommand();
+ cmd.CommandText = "DELETE FROM ingress_outbox WHERE event_id=$id";
+ cmd.Parameters.AddWithValue("$id", id);
+ cmd.ExecuteNonQuery();
+ }
+ }
+
internal bool WasSent(string job)
{
lock (_gate)
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs
index b82c0f6..c616273 100644
--- a/src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs
@@ -8,14 +8,15 @@ internal static class BridgeJson
}
internal record Destination(string BridgeInstance, string RobotId, string GroupId);
-internal record IncomingMessage(string MessageId, string SenderId, string SenderName, string Text);
+internal record IncomingMessage(string MessageId, string SenderId, string SenderName, string Text,
+ string[]? Images = null, long? ObservedAt = null);
internal record ReplyContext(string MessageId, string? SenderId, string? Content);
internal record IngressEvent(int Version, string EventId, string ConversationId,
- Destination Destination, IncomingMessage Message, ReplyContext? ReplyTo);
+ Destination Destination, IncomingMessage Message, ReplyContext? ReplyTo, string Kind = "chat");
internal record ReplyDestination(string BridgeInstance, string RobotId, string GroupId, string ReplyToMessageId);
internal record ReplyContent(string Format, string Text);
internal record ReplyJob(int Version, string JobId, string ConversationId,
- ReplyDestination Destination, ReplyContent Content, DateTimeOffset CreatedAt);
+ ReplyDestination Destination, ReplyContent Content, DateTimeOffset CreatedAt, string Kind = "chat");
internal record QueueLease(string Body, string LeaseId);
internal interface IReplyQueue
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
index 2535114..5037001 100644
--- a/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
@@ -1,4 +1,5 @@
using System.Threading.Channels;
+using System.Collections.Concurrent;
using HuaJiBot.NET.Events;
using Newtonsoft.Json.Linq;
@@ -9,7 +10,8 @@ public sealed class PluginMain : PluginBase, IPluginWithConfig
public PluginConfig Config { get; } = new();
private readonly CancellationTokenSource _stop = new();
private readonly SemaphoreSlim _wake = new(0, 1);
- private readonly Channel _incoming = Channel.CreateBounded(100);
+ private readonly Channel _incoming = Channel.CreateBounded(100);
+ private readonly ConcurrentDictionary _queued = new();
private BridgeStore? _store;
private HttpClient? _http;
private BridgeEngine? _engine;
@@ -37,6 +39,7 @@ protected override Task InitializeAsync()
_ingress = new IngressClient(_http, new Uri(Config.IngressUrl), hmac, Config.MaxMessageBytes);
_queue = new CloudflareQueue(_http, Config, token);
_engine = new BridgeEngine(Config, _store, Service, _queue, message => Warn(message));
+ 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));
Info("Flue桥接已启动");
@@ -45,8 +48,21 @@ protected override Task InitializeAsync()
private void Receive(object? sender, GroupMessageEventArgs e)
{
- if (e.RobotId is { } robot && Config.Allows(robot, e.GroupId) && !_incoming.Writer.TryWrite(e))
- Warn("Flue ingress buffer is full; message was not forwarded.");
+ if (e.RobotId is not { } robot || !Config.Allows(robot, e.GroupId)) return;
+ try
+ {
+ var message = _engine!.Normalize(e);
+ if (message is null) return;
+ if (message.Kind == "chat" && e.ImageUrls.Length > 4) { _ = e.Reply("单条消息最多支持4张图片，请分开发送。"); return; }
+ _store!.SaveIngress(message);
+ Enqueue(message);
+ }
+ catch (Exception) { Warn("Flue ingress recording failed."); }
+ }
+
+ private void Enqueue(IngressEvent message)
+ {
+ if (_queued.TryAdd(message.EventId, 0) && !_incoming.Writer.TryWrite(message)) _queued.TryRemove(message.EventId, out _);
}
private void Activate()
@@ -59,17 +75,16 @@ private async Task ForwardLoopAsync(CancellationToken ct)
{
try
{
- await foreach (var e in _incoming.Reader.ReadAllAsync(ct))
+ await foreach (var message in _incoming.Reader.ReadAllAsync(ct))
{
try
{
- var message = _engine!.Normalize(e);
- if (message is null) continue;
for (var attempt = 0; ; attempt++)
{
try
{
await _ingress!.SendAsync(message, ct);
+ _store!.RemoveIngress(message.EventId);
Activate();
break;
}
@@ -82,6 +97,7 @@ private async Task ForwardLoopAsync(CancellationToken ct)
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
catch (Exception) { Warn("Agent ingress forwarding failed."); }
+ finally { _queued.TryRemove(message.EventId, out _); }
}
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { }
@@ -97,6 +113,7 @@ private async Task PullLoopAsync(CancellationToken ct)
{
try
{
+ foreach (var message in _store!.PendingIngress()) Enqueue(message);
if (DateTimeOffset.UtcNow >= nextPrune)
{
_store!.Prune(DateTimeOffset.UtcNow.AddDays(-Config.RetentionDays));
diff --git a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
index 1a7bfc1..011ea26 100644
--- a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
+++ b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
@@ -95,11 +95,37 @@ public void MentionsCreateIsolatedThreadsAndRepliesResolveMappedMessages()
Assert.That(reply.Message.Text, Is.EqualTo("follow-up"));
Assert.That(first.Destination.GroupId, Is.EqualTo("-100:42"));
Assert.That(first.ConversationId, Does.Contain("-100%3A42"));
- Assert.That(_engine.Normalize(Message("other", "ordinary text")), Is.Null);
- Assert.That(_engine.Normalize(Message("unknown", new CommonCommandReader.ReaderReply(new("missing")), "unknown")), Is.Null);
+ Assert.That(_engine.Normalize(Message("other", "ordinary text"))!.Kind, Is.EqualTo("archive"));
+ Assert.That(_engine.Normalize(Message("unknown", new CommonCommandReader.ReaderReply(new("missing")), "unknown"))!.Kind, Is.EqualTo("archive"));
});
}
+ [Test]
+ public void ArchiveOutboxSurvivesRestartAndKeepsOriginalPayload()
+ {
+ var message = _engine.Normalize(Message("archive", "ordinary text"))!;
+ _store.SaveIngress(message);
+ _store.SaveIngress(message with { Message = message.Message with { Text = "changed" } });
+ _store.Dispose();
+ _store = new BridgeStore(_path);
+ var pending = _store.PendingIngress();
+ Assert.That(pending, Has.Length.EqualTo(1));
+ Assert.That(pending[0].Message.Text, Is.EqualTo(message.Message.Text));
+ _store.RemoveIngress(message.EventId);
+ Assert.That(_store.PendingIngress(), Is.Empty);
+ }
+
+ [Test]
+ public async Task ScheduledSummaryUsesKnownAnchorWithoutPlatformReplyId()
+ {
+ _engine.SeedSummaryAnchors("bot", "-100:42");
+ var anchor = "summary:" + DateTimeOffset.UtcNow.ToOffset(TimeSpan.FromHours(8)).ToString("yyyy-MM-dd");
+ var conversation = _store.Find("bot", "-100:42", anchor)!;
+ var job = new ReplyJob(1, "scheduled", conversation, new("main", "bot", "-100:42", anchor), new("markdown", "summary"), DateTimeOffset.UtcNow, "summary");
+ await _engine.DeliverAsync(Lease(job), CancellationToken.None);
+ Assert.That(_store.WasSent("scheduled"), Is.True);
+ }
+
[Test]
public async Task DeliveryPersistsAllIdsAndDuplicateLeaseIsAckedWithoutResending()
{
From c91413dd019da907f5504b8d577fa194c785752b Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Sat, 3 Oct 2026 14:48:10 +0800
Subject: [PATCH 10/15] deploy: recover group archive across migration downtime
---
deploy/flue-bridge/probe_onebot.py | 43 ++++++++++++++++++++++++++++--
1 file changed, 41 insertions(+), 2 deletions(-)
diff --git a/deploy/flue-bridge/probe_onebot.py b/deploy/flue-bridge/probe_onebot.py
index ea271d1..5f21e68 100644
--- a/deploy/flue-bridge/probe_onebot.py
+++ b/deploy/flue-bridge/probe_onebot.py
@@ -11,7 +11,7 @@
from pathlib import Path
-def probe(config_path, group_id):
+def probe(config_path, group_id, history_output=None, since=0):
config = json.loads(Path(config_path).read_text())
endpoint = urllib.parse.urlsplit(config["OneBot"]["Url"])
if endpoint.scheme not in ("ws", "wss"):
@@ -100,13 +100,52 @@ def call(action, params):
robot_id = str(login["user_id"])
group = call("get_group_info", {"group_id": int(group_id)})
member = call("get_group_member_info", {"group_id": int(group_id), "user_id": int(robot_id)})
+ if history_output:
+ records = {}
+ cursor = None
+ covered = False
+ for _ in range(10):
+ params = {"group_id": int(group_id), "count": 100}
+ if cursor is not None:
+ params["message_seq"] = cursor
+ data = call("get_group_msg_history", params)
+ messages = data.get("messages", [])
+ if not messages:
+ covered = True
+ break
+ covered = min(int(item.get("time", 0)) for item in messages) <= since
+ for item in messages:
+ timestamp = int(item.get("time", 0))
+ if timestamp < since or str(item.get("user_id")) == robot_id:
+ continue
+ message_id = str(item["message_id"])
+ segments = item.get("message", [])
+ text = "".join(segment.get("data", {}).get("text", "") for segment in segments if isinstance(segment, dict) and segment.get("type") == "text") if isinstance(segments, list) else "（非文本消息）"
+ sender = item.get("sender", {})
+ components = ["main", robot_id, str(group_id), message_id]
+ event_id = ":".join(urllib.parse.quote(value, safe="") for value in components)
+ records[message_id] = {"version": 1, "eventId": event_id, "conversationId": "huajibot:" + event_id,
+ "kind": "archive", "destination": {"bridgeInstance": "main", "robotId": robot_id, "groupId": str(group_id)},
+ "message": {"messageId": message_id, "senderId": str(item.get("user_id")), "senderName": sender.get("card") or sender.get("nickname") or "成员",
+ "text": text or "（非文本消息）", "images": [], "observedAt": timestamp * 1000}, "replyTo": None}
+ if covered:
+ break
+ sequences = [int(item["message_seq"]) for item in messages if "message_seq" in item]
+ if not sequences or cursor == min(sequences) - 1:
+ break
+ cursor = min(sequences) - 1
+ descriptor = os.open(history_output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+ with os.fdopen(descriptor, "w") as output:
+ for record in sorted(records.values(), key=lambda item: item["message"]["observedAt"]):
+ output.write(json.dumps(record, ensure_ascii=False) + "\n")
+ return {"historyRecords": len(records), "coversMigrationWindow": covered}
return {"robotId": robot_id, "groupId": str(group["group_id"]), "role": member.get("role"),
"muteUntil": member.get("shut_up_timestamp", 0)}
if __name__ == "__main__":
try:
- print(json.dumps(probe(sys.argv[1], sys.argv[2])))
+ print(json.dumps(probe(sys.argv[1], sys.argv[2], sys.argv[3] if len(sys.argv) > 3 else None, int(sys.argv[4]) if len(sys.argv) > 4 else 0)))
except Exception as error:
print("OneBot probe failed: " + type(error).__name__, file=sys.stderr)
sys.exit(1)
From 361a0becd7006b5365feea9708b08eb67b65aa3a Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Sat, 3 Oct 2026 15:00:09 +0800
Subject: [PATCH 11/15] fix: bound image admission and preserve idle Queue
polling
---
deploy/flue-bridge/README.md | 2 +-
deploy/flue-bridge/activate-features.sh | 2 +-
deploy/flue-bridge/deployment.json | 11 +++++++----
src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs | 10 +++++++---
4 files changed, 16 insertions(+), 9 deletions(-)
diff --git a/deploy/flue-bridge/README.md b/deploy/flue-bridge/README.md
index 41c3c3c..94d528f 100644
--- a/deploy/flue-bridge/README.md
+++ b/deploy/flue-bridge/README.md
@@ -62,7 +62,7 @@ Cloudflare HTTP pull's `visibility_timeout` is expressed in milliseconds; the pl
## Storage, build and rollback
-The canary overlay image is `huajibot-local:worker-features-v2` on `/home/yunacelisse/stacks/huajibot`. `Dockerfile` preserves existing plugins while replacing the CLI and adding bridge dependencies. `activate.sh` checks artifact hashes, backs up config/compose/plugin data, stops only the bot, checks the intended configuration delta, and automatically restores the old config/image on startup failure. `bridge.env` is a private mode-0600 file. Supply sudo authentication through the operator's terminal; never save it in scripts.
+The canary overlay image is `huajibot-local:worker-features-v4` on `/home/yunacelisse/stacks/huajibot`. `Dockerfile` preserves existing plugins while replacing the CLI and adding bridge dependencies. `activate.sh` checks artifact hashes, backs up config/compose/plugin data, stops only the bot, checks the intended configuration delta, and automatically restores the old config/image on startup failure. `bridge.env` is a private mode-0600 file. Supply sudo authentication through the operator's terminal; never save it in scripts.
The activation backup is `/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z`. For a deployment rollback, stop only `huaji-bot-dotnet`, restore `config.json` and `compose.yaml` from that directory, and run `docker compose up -d --no-deps huaji-bot-dotnet`. Keep the bridge database and Queue. Revoke the Worker canary allowlist after the switch.
diff --git a/deploy/flue-bridge/activate-features.sh b/deploy/flue-bridge/activate-features.sh
index 130bcce..f12c3c0 100644
--- a/deploy/flue-bridge/activate-features.sh
+++ b/deploy/flue-bridge/activate-features.sh
@@ -4,7 +4,7 @@ umask 077
STACK=/home/yunacelisse/stacks/huajibot
STAGE="$STACK/deploy/flue-feature"
-IMAGE=huajibot-local:worker-features-v2
+IMAGE=huajibot-local:worker-features-v4
STATUS="$STAGE/activation-status.json"
BACKUP="$STACK/backups/worker-features-$(date -u +%Y%m%dT%H%M%SZ)"
STOPPED=0
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
index 5b55fa8..26d50a6 100644
--- a/deploy/flue-bridge/deployment.json
+++ b/deploy/flue-bridge/deployment.json
@@ -1,5 +1,5 @@
{
- "verifiedAt": "2026-10-03T14:24:31.4773546+08:00",
+ "verifiedAt": "2026-10-03T15:00:09.6761576+08:00",
"accountEmail": "nbtca@protonmail.com",
"accountId": "a415a3ae7a197093d02233755d38800b",
"worker": "huaji-agent",
@@ -52,7 +52,9 @@
"Real Agent MCP search, image recognition, image follow-up without prior color disclosure, summary Workflow and Queue replies passed",
"494 read-only LiteDB records imported; live D1 archive increased to 521 records",
"Local AIChat/DailySummary disabled; bot running with zero restarts; OneBot identity/member verified",
- "Temporary acceptance destination revoked and its D1 fixture records removed"
+ "Temporary acceptance destination revoked and its D1 fixture records removed",
+ "Authenticated OneBot history export covered the migration window; 50 records re-imported with deduplication",
+ "Image ingress timeout separated from Queue HTTP timeout; archive forwarding leaves Queue polling idle"
],
"pending": [
"User acceptance of summary/search/image mention/image follow-up in QQ group",
@@ -62,7 +64,7 @@
],
"model": "cloudflare/@cf/meta/llama-3.3-70b-instruct-fp8-fast",
"activationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z",
- "currentImage": "huajibot-local:worker-features-v2",
+ "currentImage": "huajibot-local:worker-features-v4",
"latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T054443Z",
"agnesPrimary": {
"model": "agnes-3.0-flash",
@@ -89,6 +91,7 @@
"originalMigrationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T052735Z",
"bridgeTests": 12,
"imageFallbackModel": "cloudflare/@cf/meta/llama-4-scout-17b-16e-instruct",
- "workerTests": 23
+ "workerTests": 23,
+ "recoveredPlatformHistoryRecords": 50
}
}
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
index 5037001..3aaaa05 100644
--- a/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/PluginMain.cs
@@ -14,6 +14,7 @@ public sealed class PluginMain : PluginBase, IPluginWithConfig
private readonly ConcurrentDictionary _queued = new();
private BridgeStore? _store;
private HttpClient? _http;
+ private HttpClient? _queueHttp;
private BridgeEngine? _engine;
private IngressClient? _ingress;
private IReplyQueue? _queue;
@@ -35,9 +36,11 @@ protected override Task InitializeAsync()
}
_store = new BridgeStore(Path.Combine(Service.GetPluginDataPath(), "flue_bridge.db"));
// Signed requests and Queue credentials must never follow a redirect to another host.
- _http = new HttpClient(new HttpClientHandler { AllowAutoRedirect = false }) { Timeout = TimeSpan.FromSeconds(30) };
+ // Admission may download four images (20 seconds each) before dispatch.
+ _http = new HttpClient(new HttpClientHandler { AllowAutoRedirect = false }) { Timeout = TimeSpan.FromSeconds(100) };
+ _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(_http, Config, token);
+ _queue = new CloudflareQueue(_queueHttp, Config, token);
_engine = new BridgeEngine(Config, _store, Service, _queue, message => Warn(message));
foreach (var destination in Config.Destinations) _engine.SeedSummaryAnchors(destination.RobotId, destination.GroupId);
Service.Events.OnGroupMessageReceived += Receive;
@@ -85,7 +88,7 @@ private async Task ForwardLoopAsync(CancellationToken ct)
{
await _ingress!.SendAsync(message, ct);
_store!.RemoveIngress(message.EventId);
- Activate();
+ if (message.Kind != "archive") Activate();
break;
}
catch (HttpRequestException ex) when (attempt < 2 && (ex.StatusCode is null
@@ -154,6 +157,7 @@ protected override void Unload()
private void Cleanup()
{
_http?.Dispose();
+ _queueHttp?.Dispose();
_store?.Dispose();
_wake.Dispose();
_stop.Dispose();
From 75b4ee8bea89c3d742e919a59a22bcf6bddad766 Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Sat, 3 Oct 2026 15:01:36 +0800
Subject: [PATCH 12/15] docs: record final Worker feature activation backup
---
deploy/flue-bridge/deployment.json | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
index 26d50a6..1b7962a 100644
--- a/deploy/flue-bridge/deployment.json
+++ b/deploy/flue-bridge/deployment.json
@@ -65,7 +65,7 @@
"model": "cloudflare/@cf/meta/llama-3.3-70b-instruct-fp8-fast",
"activationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z",
"currentImage": "huajibot-local:worker-features-v4",
- "latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T054443Z",
+ "latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T065807Z",
"agnesPrimary": {
"model": "agnes-3.0-flash",
"workerError": "Intermittent HTTP 429 / edge code 1015; subsequent two-turn test passed",
From 3a9a48ab89bd4120d30e31f2680c07b271e6c2c7 Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <29951517@qq.com>
Date: Sat, 3 Oct 2026 15:43:26 +0800
Subject: [PATCH 13/15] fix: route summary day aliases with optional whitespace
---
deploy/flue-bridge/activate-features.sh | 2 +-
deploy/flue-bridge/deployment.json | 20 +++++++++-------
.../BridgeEngine.cs | 2 +-
src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs | 24 +++++++++++++++++++
4 files changed, 38 insertions(+), 10 deletions(-)
diff --git a/deploy/flue-bridge/activate-features.sh b/deploy/flue-bridge/activate-features.sh
index f12c3c0..398770e 100644
--- a/deploy/flue-bridge/activate-features.sh
+++ b/deploy/flue-bridge/activate-features.sh
@@ -4,7 +4,7 @@ umask 077
STACK=/home/yunacelisse/stacks/huajibot
STAGE="$STACK/deploy/flue-feature"
-IMAGE=huajibot-local:worker-features-v4
+IMAGE=huajibot-local:worker-features-v5
STATUS="$STAGE/activation-status.json"
BACKUP="$STACK/backups/worker-features-$(date -u +%Y%m%dT%H%M%SZ)"
STOPPED=0
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
index 1b7962a..1a7a184 100644
--- a/deploy/flue-bridge/deployment.json
+++ b/deploy/flue-bridge/deployment.json
@@ -1,10 +1,10 @@
{
- "verifiedAt": "2026-10-03T15:00:09.6761576+08:00",
+ "verifiedAt": "2026-10-03T15:43:25.9701362+08:00",
"accountEmail": "nbtca@protonmail.com",
"accountId": "a415a3ae7a197093d02233755d38800b",
"worker": "huaji-agent",
"workerUrl": "https://huaji-agent.nbtca.workers.dev",
- "workerVersionId": "2a4e1461-eb53-48e6-9963-f5ba2e82d880",
+ "workerVersionId": "3c6202d2-2d9f-4f32-9c50-df5e790c0878",
"queue": {
"name": "huajibot-outbound-main",
"id": "3181cdb9b0054a4a8f0d0bc70b8a48c5"
@@ -54,7 +54,10 @@
"Local AIChat/DailySummary disabled; bot running with zero restarts; OneBot identity/member verified",
"Temporary acceptance destination revoked and its D1 fixture records removed",
"Authenticated OneBot history export covered the migration window; 50 records re-imported with deduplication",
- "Image ingress timeout separated from Queue HTTP timeout; archive forwarding leaves Queue polling idle"
+ "Image ingress timeout separated from Queue HTTP timeout; archive forwarding leaves Queue polling idle",
+ "Summary grammar covers 今/昨/前 + 天/日 with optional whitespace in bot and Worker",
+ "26 Worker tests, 13 bridge tests; live Worker weather RPC returned numeric data",
+ "Real Agent weather answer contained temperatures/probability/source; spaced 今日 alias produced a summary Workflow reply"
],
"pending": [
"User acceptance of summary/search/image mention/image follow-up in QQ group",
@@ -64,8 +67,8 @@
],
"model": "cloudflare/@cf/meta/llama-3.3-70b-instruct-fp8-fast",
"activationBackup": "/home/yunacelisse/stacks/huajibot/backups/flue-canary-20261002T151502Z",
- "currentImage": "huajibot-local:worker-features-v4",
- "latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T065807Z",
+ "currentImage": "huajibot-local:worker-features-v5",
+ "latestActivationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T073322Z",
"agnesPrimary": {
"model": "agnes-3.0-flash",
"workerError": "Intermittent HTTP 429 / edge code 1015; subsequent two-turn test passed",
@@ -89,9 +92,10 @@
"localDailySummaryEnabled": false,
"dailySummary": true,
"originalMigrationBackup": "/home/yunacelisse/stacks/huajibot/backups/worker-features-20261003T052735Z",
- "bridgeTests": 12,
+ "bridgeTests": 13,
"imageFallbackModel": "cloudflare/@cf/meta/llama-4-scout-17b-16e-instruct",
- "workerTests": 23,
- "recoveredPlatformHistoryRecords": 50
+ "workerTests": 26,
+ "recoveredPlatformHistoryRecords": 50,
+ "numericWeather": "Open-Meteo via authenticated MCP get_weather"
}
}
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
index 97b4ade..e36629e 100644
--- a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
@@ -29,7 +29,7 @@ internal sealed class BridgeEngine(PluginConfig config, BridgeStore store, IPlug
var textReader = mentionsBot ? mentionReader : replyReader;
var text = chat && textReader.Input(out var rest, true) ? rest : e.TextMessage;
if (string.IsNullOrWhiteSpace(text)) text = e.ImageUrls.Length > 0 ? "请描述这张图片。" : "（非文本消息）";
- var summary = System.Text.RegularExpressions.Regex.IsMatch(text.Trim(), @"^(总结|summary)(\s+(今天|昨天|前天))?$");
+ var summary = System.Text.RegularExpressions.Regex.IsMatch(text.Trim(), @"^(总结|summary)(\s*[今昨前]\s*[天日])?$");
var kind = summary ? "summary" : chat ? "chat" : "archive";
var eventId = string.Join(':', new[] { config.BridgeInstance, robot, e.GroupId, e.MessageId }.Select(Uri.EscapeDataString));
var message = new IngressEvent(1, eventId, conversation,
diff --git a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
index 011ea26..1b3132d 100644
--- a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
+++ b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
@@ -60,6 +60,30 @@ private ReplyJob Job(string id = "job-1")
private static QueueLease Lease(ReplyJob job) => new(JsonSerializer.Serialize(job, BridgeJson.Options), "lease-1");
+ [Test]
+ public void SummaryDateAliasesWithOptionalSpacesUseSummaryRouting()
+ {
+ var index = 0;
+ foreach (var day in new[] { "今", "昨", "前" })
+ foreach (var suffix in new[] { "天", "日" })
+ foreach (var gap in new[] { "", " ", " " })
+ foreach (var inner in new[] { "", " ", " " })
+ {
+ var text = "总结" + gap + day + inner + suffix;
+ var message = new GroupMessageEventArgs(() => new DefaultCommandReader([text]), () => ValueTask.FromResult("group"))
+ {
+ Service = new TestAdapter(),
+ RobotId = "bot",
+ GroupId = "-100:42",
+ MessageId = "alias-" + index++,
+ SenderId = "user",
+ SenderMemberCard = "Alice",
+ TextMessageLazy = new(() => text)
+ };
+ Assert.That(_engine.Normalize(message)!.Kind, Is.EqualTo("summary"), text);
+ }
+ }
+
[Test]
public async Task OneBotGroupEventCarriesAccountIdAndReachesBridge()
{
From 85f08868144b4e19ad6198fcfe054cb866b74907 Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <180939944+Yuna-Celisse@users.noreply.github.com>
Date: Mon, 12 Oct 2026 08:56:10 +0800
Subject: [PATCH 14/15] Require TLS 1.2 or newer in the OneBot diagnostic probe
---
deploy/flue-bridge/probe_onebot.py | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/deploy/flue-bridge/probe_onebot.py b/deploy/flue-bridge/probe_onebot.py
index 5f21e68..08a8976 100644
--- a/deploy/flue-bridge/probe_onebot.py
+++ b/deploy/flue-bridge/probe_onebot.py
@@ -19,7 +19,9 @@ def probe(config_path, group_id, history_output=None, since=0):
port = endpoint.port or (443 if endpoint.scheme == "wss" else 80)
connection = socket.create_connection((endpoint.hostname, port), timeout=10)
if endpoint.scheme == "wss":
- connection = ssl.create_default_context().wrap_socket(connection, server_hostname=endpoint.hostname)
+ context = ssl.create_default_context()
+ context.minimum_version = ssl.TLSVersion.TLSv1_2
+ connection = context.wrap_socket(connection, server_hostname=endpoint.hostname)
with connection:
key = base64.b64encode(os.urandom(16)).decode()
path = endpoint.path or "/"
From 19f30aa37e8ee0d8c9033f9d01f44a72b3066b67 Mon Sep 17 00:00:00 2001
From: Yuna-Celisse <180939944+Yuna-Celisse@users.noreply.github.com>
Date: Mon, 12 Oct 2026 08:35:03 +0800
Subject: [PATCH 15/15] Verify bridge records against shared Worker protocol
fixtures
---
.gitattributes | 1 +
.../FlueBridgeProtocolTest.cs | 44 ++++++
src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs | 2 +-
.../HuaJiBot.NET.UnitTest.csproj | 1 +
.../AgentProtocol/examples/ingress.json | 20 +++
.../AgentProtocol/examples/reply-job.json | 17 ++
.../AgentProtocol/ingress.schema.json | 149 ++++++++++++++++++
.../AgentProtocol/reply-job.schema.json | 89 +++++++++++
.../TestData/AgentProtocol/test-vectors.json | 22 +++
9 files changed, 344 insertions(+), 1 deletion(-)
create mode 100644 src/HuaJiBot.NET.UnitTest/FlueBridgeProtocolTest.cs
create mode 100644 src/HuaJiBot.NET.UnitTest/TestData/AgentProtocol/examples/ingress.json
create mode 100644 src/HuaJiBot.NET.UnitTest/TestData/AgentProtocol/examples/reply-job.json
create mode 100644 src/HuaJiBot.NET.UnitTest/TestData/AgentProtocol/ingress.schema.json
create mode 100644 src/HuaJiBot.NET.UnitTest/TestData/AgentProtocol/reply-job.schema.json
create mode 100644 src/HuaJiBot.NET.UnitTest/TestData/AgentProtocol/test-vectors.json
diff --git a/.gitattributes b/.gitattributes
index c9c7dbe..723273b 100644
--- a/.gitattributes
+++ b/.gitattributes
@@ -13,4 +13,5 @@
*.docx diff=astextplain
*.DOCX diff=astextplain
+src/HuaJiBot.NET.UnitTest/TestData/AgentProtocol/** text eol=lf
diff --git a/src/HuaJiBot.NET.UnitTest/FlueBridgeProtocolTest.cs b/src/HuaJiBot.NET.UnitTest/FlueBridgeProtocolTest.cs
new file mode 100644
index 0000000..4f227c7
--- /dev/null
+++ b/src/HuaJiBot.NET.UnitTest/FlueBridgeProtocolTest.cs
@@ -0,0 +1,44 @@
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Nodes;
+using HuaJiBot.NET.Plugin.FlueBridge;
+
+namespace HuaJiBot.NET.UnitTest;
+
+public partial class FlueBridgeTest
+{
+ private static string ProtocolFixture(string name) =>
+ File.ReadAllText(Path.Combine(TestContext.CurrentContext.TestDirectory, "TestData", "AgentProtocol", name));
+
+ [Test]
+ public void SharedProtocolExamplesRoundTripThroughActualBridgeRecords()
+ {
+ var ingressJson = ProtocolFixture("examples/ingress.json");
+ var ingress = JsonSerializer.Deserialize(ingressJson, BridgeJson.Options)!;
+ var replyJson = ProtocolFixture("examples/reply-job.json");
+ var reply = JsonSerializer.Deserialize(replyJson, 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(ingress.Kind, Is.EqualTo("chat"));
+ Assert.That(reply.Destination.ReplyToMessageId, Is.EqualTo(ingress.Message.MessageId));
+ });
+ }
+
+ [Test]
+ public void SharedProtocolVectorsMatchRawBodySignatureAndIdentifierEncoding()
+ {
+ using var vectors = JsonDocument.Parse(ProtocolFixture("test-vectors.json"));
+ foreach (var vector in vectors.RootElement.GetProperty("hmac").EnumerateArray())
+ {
+ var actual = "v1=" + IngressClient.Sign(vector.GetProperty("secret").GetString()!,
+ vector.GetProperty("timestamp").GetString()!, Encoding.UTF8.GetBytes(vector.GetProperty("body").GetString()!));
+ Assert.That(actual, Is.EqualTo(vector.GetProperty("signature").GetString()));
+ }
+ foreach (var vector in vectors.RootElement.GetProperty("identifiers").EnumerateArray())
+ {
+ Assert.That(Uri.EscapeDataString(vector.GetProperty("raw").GetString()!), Is.EqualTo(vector.GetProperty("encoded").GetString()));
+ }
+ }
+}
diff --git a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
index 1b3132d..894f147 100644
--- a/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
+++ b/src/HuaJiBot.NET.UnitTest/FlueBridgeTest.cs
@@ -10,7 +10,7 @@
namespace HuaJiBot.NET.UnitTest;
-public class FlueBridgeTest
+public partial class FlueBridgeTest
{
private string _path = null!;
private BridgeStore _store = null!;
diff --git a/src/HuaJiBot.NET.UnitTest/HuaJiBot.NET.UnitTest.csproj b/src/HuaJiBot.NET.UnitTest/HuaJiBot.NET.UnitTest.csproj
index 5103f79..7f04a91 100644
--- a/src/HuaJiBot.NET.UnitTest/HuaJiBot.NET.UnitTest.csproj
+++ b/src/HuaJiBot.NET.UnitTest/HuaJiBot.NET.UnitTest.csproj
@@ -29,6 +29,7 @@
+