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/.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 @@
+
diff --git a/build_plugins.fsx b/build_plugins.fsx
index 511ff7b..9361e3e 100644
--- a/build_plugins.fsx
+++ b/build_plugins.fsx
@@ -108,6 +108,13 @@ for dir in binDirs do
let dest = Path.Combine(libsDir, fileName)
File.Copy(file, dest, true)
printfn "Copy %s to %s" file dest
+ // Preserve native dependencies (e.g. SQLite) alongside the managed plugin libraries.
+ let runtimes = Path.Combine(dir, "runtimes")
+ if Directory.Exists runtimes then
+ for file in Directory.GetFiles(runtimes, "*", SearchOption.AllDirectories) do
+ let dest = Path.Combine(libsDir, "runtimes", Path.GetRelativePath(runtimes, file))
+ Directory.CreateDirectory(Path.GetDirectoryName dest) |> ignore
+ File.Copy(file, dest, true)
//remove original output
for dir in binDirs do
Directory.Delete(dir, true)
diff --git a/deploy/flue-bridge/Dockerfile b/deploy/flue-bridge/Dockerfile
new file mode 100644
index 0000000..97b7eff
--- /dev/null
+++ b/deploy/flue-bridge/Dockerfile
@@ -0,0 +1,7 @@
+# Overlay the current production image; preserve its other plugins and MCP service.
+ARG BASE_IMAGE
+FROM ${BASE_IMAGE}
+COPY cli/ /app/bin/
+COPY --chmod=755 cli/HuaJiBot.NET.CLI /app/bin/HuaJiBot.NET.CLI
+COPY bridge/HuaJiBot.NET.Plugin.FlueBridge.dll /app/plugins/HuaJiBot.NET.Plugin.FlueBridge.dll
+COPY dependencies/ /app/plugins/libs/
diff --git a/deploy/flue-bridge/README.md b/deploy/flue-bridge/README.md
new file mode 100644
index 0000000..94d528f
--- /dev/null
+++ b/deploy/flue-bridge/README.md
@@ -0,0 +1,91 @@
+# Flue / Cloudflare Queue bridge (issue #64)
+
+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. 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.
+
+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.
+
+Keep the canary group out of AIChat's `GroupIds`. The bridge refuses startup if an enabled AIChat configuration overlaps. Stop and restart the bot after changing routing or secrets; live panel edits do not reinitialize this plugin.
+
+Supply these environment variables to the bot container/process:
+
+- `HUAJIBOT_BRIDGE_HMAC_SECRET`: the same secret configured on the Worker.
+- `HUAJIBOT_QUEUE_API_TOKEN`: an account-scoped Cloudflare token with Queue read/write permission, limited to the intended account.
+
+Do not put either secret in plugin configuration or Git. The Worker must independently allowlist the same bridge instance, robot and group. The bot needs outbound HTTPS only; no public listener or Tunnel is used.
+
+## Version 1 wire contract
+
+Bot records in `src/HuaJiBot.NET.Plugin.FlueBridge/Contracts.cs` and Worker schemas in `huaji-agent/src/contracts.ts` use the ingress and reply shapes from issue #64. All property names on the wire are camelCase. Optional reply context is represented by `replyTo: null`; available sender/content fields may also be null.
+
+Both sides use RFC3986 component encoding when forming identifiers. This avoids delimiter collisions with Telegram topics. For example:
+
+```text
+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 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:
+
+```text
+X-HuaJiBot-Timestamp:
+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 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.
+
+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 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..398770e
--- /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-v5
+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/activate.sh b/deploy/flue-bridge/activate.sh
new file mode 100644
index 0000000..bf5b22c
--- /dev/null
+++ b/deploy/flue-bridge/activate.sh
@@ -0,0 +1,147 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+umask 077
+
+STACK=/home/yunacelisse/stacks/huajibot
+STAGE="$STACK/deploy/flue-canary"
+IMAGE=huajibot-local:flue-canary-onebot-fix
+STATUS="$STAGE/activation-status.json"
+BACKUP="$STACK/backups/flue-canary-$(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']['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-canary/bridge.env' not in env: env.append('./deploy/flue-canary/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)
+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, DailySummary and other services preserved.')
+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
new file mode 100644
index 0000000..16f0e98
--- /dev/null
+++ b/deploy/flue-bridge/config.example.json
@@ -0,0 +1,24 @@
+{
+ "Flue桥接": {
+ "Enabled": false,
+ "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
+ },
+ "每日聊天总结": {
+ "Enabled": false
+ }
+}
diff --git a/deploy/flue-bridge/deployment.json b/deploy/flue-bridge/deployment.json
new file mode 100644
index 0000000..1a7a184
--- /dev/null
+++ b/deploy/flue-bridge/deployment.json
@@ -0,0 +1,101 @@
+{
+ "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": "3c6202d2-2d9f-4f32-9c50-df5e790c0878",
+ "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": [
+ {
+ "robotId": "3623498320",
+ "bridgeInstance": "main",
+ "groupId": "466691612"
+ }
+ ],
+ "botEnabled": true,
+ "canaryGroupId": "466691612",
+ "verified": [
+ "Worker health HTTP 200",
+ "Unsigned ingress HTTP 401",
+ "Valid signed unauthorized ingress HTTP 403",
+ "Public agent route HTTP 404",
+ "Queue scoped token pull/ack HTTP 200",
+ "OneBot identity and canary membership verified",
+ "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",
+ "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",
+ "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",
+ "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",
+ "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",
+ "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",
+ "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",
+ "Real platform failure/restart/DLQ acceptance",
+ "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: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",
+ "secretConfigured": true,
+ "fallbackPolicy": "Any HTTP 429 or explicit quota exhaustion: primary Agnes, secondary if valid/configured, then Cloudflare",
+ "directToolsStreamingAndRecallVerified": true,
+ "enabled": true,
+ "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": 13,
+ "imageFallbackModel": "cloudflare/@cf/meta/llama-4-scout-17b-16e-instruct",
+ "workerTests": 26,
+ "recoveredPlatformHistoryRecords": 50,
+ "numericWeather": "Open-Meteo via authenticated MCP get_weather"
+ }
+}
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 @@
+
+ Exenet10.0enableenable
+
+
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/deploy/flue-bridge/probe_onebot.py b/deploy/flue-bridge/probe_onebot.py
new file mode 100644
index 0000000..08a8976
--- /dev/null
+++ b/deploy/flue-bridge/probe_onebot.py
@@ -0,0 +1,153 @@
+"""Read the configured OneBot account/group without printing credentials or messages."""
+import base64
+import hashlib
+import json
+import os
+import socket
+import ssl
+import struct
+import sys
+import urllib.parse
+from pathlib import Path
+
+
+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"):
+ raise RuntimeError("Expected a OneBot WebSocket endpoint")
+ port = endpoint.port or (443 if endpoint.scheme == "wss" else 80)
+ connection = socket.create_connection((endpoint.hostname, port), timeout=10)
+ if endpoint.scheme == "wss":
+ 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 "/"
+ if endpoint.query:
+ path += "?" + endpoint.query
+ lines = [f"GET {path} HTTP/1.1", f"Host: {endpoint.hostname}:{port}", "Upgrade: websocket",
+ "Connection: Upgrade", f"Sec-WebSocket-Key: {key}", "Sec-WebSocket-Version: 13"]
+ token = config["OneBot"].get("Token")
+ if token:
+ lines.append("Authorization: Bearer " + token)
+ connection.sendall(("\r\n".join(lines) + "\r\n\r\n").encode())
+ buffered = bytearray()
+ while b"\r\n\r\n" not in buffered:
+ chunk = connection.recv(4096)
+ if not chunk:
+ raise RuntimeError("OneBot handshake closed")
+ buffered.extend(chunk)
+ if len(buffered) > 65536:
+ raise RuntimeError("OneBot handshake too large")
+ raw_headers, remaining = bytes(buffered).split(b"\r\n\r\n", 1)
+ expected = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()).decode()
+ if b" 101 " not in raw_headers.split(b"\r\n", 1)[0] or expected.lower().encode() not in raw_headers.lower():
+ raise RuntimeError("OneBot rejected the authenticated handshake")
+ buffered = bytearray(remaining)
+
+ def read_exact(count):
+ while len(buffered) < count:
+ chunk = connection.recv(max(4096, count - len(buffered)))
+ if not chunk:
+ raise RuntimeError("OneBot connection closed")
+ buffered.extend(chunk)
+ result = bytes(buffered[:count])
+ del buffered[:count]
+ return result
+
+ def send(payload, opcode=1):
+ mask = os.urandom(4)
+ size = len(payload)
+ header = bytes([0x80 | opcode, 0x80 | size]) if size < 126 else bytes([0x80 | opcode, 0xFE]) + struct.pack("!H", size)
+ connection.sendall(header + mask + bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload)))
+
+ def call(action, params):
+ echo = os.urandom(8).hex()
+ send(json.dumps({"action": action, "params": params, "echo": echo}).encode())
+ fragments = bytearray()
+ for _ in range(200):
+ first, second = read_exact(2)
+ opcode, size = first & 15, second & 127
+ if size == 126:
+ size = struct.unpack("!H", read_exact(2))[0]
+ elif size == 127:
+ size = struct.unpack("!Q", read_exact(8))[0]
+ if size > 1024 * 1024:
+ raise RuntimeError("OneBot frame too large")
+ mask = read_exact(4) if second & 128 else None
+ payload = read_exact(size)
+ if mask:
+ payload = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload))
+ if opcode == 9:
+ send(payload, 10)
+ continue
+ if opcode == 8:
+ raise RuntimeError("OneBot closed the connection")
+ if opcode not in (0, 1):
+ continue
+ fragments.extend(payload)
+ if not first & 128:
+ continue
+ response = json.loads(fragments)
+ fragments.clear()
+ if response.get("echo") == echo:
+ if response.get("status") != "ok" or response.get("retcode", 0) != 0:
+ raise RuntimeError("OneBot query failed: " + action)
+ return response["data"]
+ raise RuntimeError("No matching OneBot query response")
+
+ login = call("get_login_info", {})
+ 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], 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)
diff --git a/deploy/flue-bridge/stage_env.py b/deploy/flue-bridge/stage_env.py
new file mode 100644
index 0000000..b657bc5
--- /dev/null
+++ b/deploy/flue-bridge/stage_env.py
@@ -0,0 +1,21 @@
+"""Accept bridge secrets on stdin and stage a private container env file."""
+import json
+import os
+import sys
+from pathlib import Path
+
+path = Path(sys.argv[1])
+values = json.load(sys.stdin)
+names = ("HUAJIBOT_BRIDGE_HMAC_SECRET", "HUAJIBOT_QUEUE_API_TOKEN")
+if set(values) != set(names) or any(not isinstance(values[name], str) or not values[name]
+ or any(c in values[name] for c in "\r\n\x00") for name in names):
+ raise SystemExit("Invalid secret envelope")
+if path.exists():
+ raise SystemExit("Secret file already exists; refusing to overwrite")
+fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+with os.fdopen(fd, "w") as output:
+ for name in names:
+ output.write(name + "=" + values[name] + "\n")
+ output.flush()
+ os.fsync(output.fileno())
+print("Bridge environment staged with mode 0600; values hidden.")
diff --git a/src/HuaJiBot.NET.Adapter.OneBot/OneBotMessageHandler.cs b/src/HuaJiBot.NET.Adapter.OneBot/OneBotMessageHandler.cs
index eea7b15..6f5609e 100644
--- a/src/HuaJiBot.NET.Adapter.OneBot/OneBotMessageHandler.cs
+++ b/src/HuaJiBot.NET.Adapter.OneBot/OneBotMessageHandler.cs
@@ -197,6 +197,7 @@ public async Task ProcessMessageAsync(JObject json)
)
{
Service = service,
+ RobotId = json.Value("self_id") ?? QQ,
MessageId = msgId,
GroupId = groupId,
SenderId = userId,
diff --git a/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
new file mode 100644
index 0000000..e36629e
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeEngine.cs
@@ -0,0 +1,119 @@
+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;
+ SeedSummaryAnchors(robot, e.GroupId);
+
+ 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);
+ 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 = 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*[今昨前]\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,
+ 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.");
+ 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 }
+ || 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.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.
+ 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,
+ 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()]));
+ 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..08e361d
--- /dev/null
+++ b/src/HuaJiBot.NET.Plugin.FlueBridge/BridgeStore.cs
@@ -0,0 +1,163 @@
+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);
+ CREATE TABLE IF NOT EXISTS ingress_outbox (event_id TEXT PRIMARY KEY, body TEXT NOT NULL, created_at INTEGER NOT NULL);
+ """;
+ 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 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)
+ {
+ 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