feat(ha): 网关多副本安全 —— 事件订阅改队列组，杜绝重复落库/重复计费

去单点的代码层基础：dispatcher/mcp-go 本就靠队列组可多副本，但网关侧的 eval/usage/
status/config 订阅是广播（nc.Subscribe），多网关副本下每条会被每个副本各处理一遍 →
评测/状态重复写 PG、token 用量重复累加（日预算翻倍）、config 请求多份重复应答。

改为 QueueSubscribe + contract.QueueGateway 队列组：组内每条事件/请求只一个副本处理。
（dispatcher/mcp-go 的 config 变更广播订阅保持 nc.Subscribe 不动——每副本都要热更新。）

验证：
- 单测 TestGatewayQueueDedup：2 网关副本订阅，发 50 条 eval，合计处理 50 次（非 100）。
- live：2 dispatcher 副本提 8 任务，队列组自动 4/4 分摊。

至此进程级全部可水平复制。剩 NATS 集群 / 网关 LB / PG·Redis·Milvus 基础设施 HA 属部署期。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

project_analysis.md

@@ -134,7 +134,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
 - **规模未真实验证**：并发是近期才补的，仅本地小规模测过；无长稳、无容量上限实测曲线。
 - **compose 引擎仍灰度**（默认走自研 graph.go）；compose 路径下 HITL 审批未实现。
 - **OCR 多模态是骨架**：`parse_document` 真支持 txt/md/csv/docx/xlsx/文本 PDF，但 MinerU/PaddleOCR 扫描件路径是 TODO 空实现（`mineru.py` 极小）。
-- **推理模型未适配**：主力 deepseek-v4-pro（及生产 Qwen thinking）答案在 `reasoning_content`，平台只读 `content`；思维链未流式呈现。
+- **推理模型已部分适配**：reasoning 模型的 `reasoning_content` 已捕获并 surface 到 exec 轨迹「推理过程」事件（不污染答案）；尚未在最终答案区把思维链单独流式呈现给前端。
 - **测试偏单元**：多为小函数单测；核心链路（图执行、RAG 管线）集成/e2e 稀疏；无前端 E2E。
 
 ---
@@ -143,13 +143,13 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
 
 | 缺口 | 现状 | 影响 |
 |------|------|------|
-| **高可用 / 单点** | 网关·调度·NATS 均单实例 | 任一挂即中断；无故障转移/自愈 |
+| **高可用 / 单点** | 🟡 代码层多副本安全（调度队列组 + 网关队列组订阅去重，均实测）；NATS 集群/网关 LB/PG·Redis HA 待部署期 | 进程级已可水平复制；基础设施 HA 仍缺 |
 | **备份 / 灾备** | PG/Milvus/Neo4j 无备份恢复方案 | 数据丢失风险 |
 | **安全审计** | 全自述，未渗透/未审计 | "纵深防御"未被验证 |
 | **配额 / 多租户** | 仅按 IP 全局限流；owner 基础隔离 | 无团队/组织/按用户配额 |
 | **可靠性细节** | exec 轨迹流丢事件；~~停机不 drain~~（已优雅停机 drain） | 观测缺口 |
 | **计量计费** | 计价配置有，计量×单价+配额未落地 | 商业化未闭环 |
-| **本地模型** | 仅 OpenAI 兼容在线 API | vLLM/Ollama、模型路由/fallback 未做 |
+| **本地模型** | ✅ vLLM/Ollama 已接（provider 感知，自动补 /v1 + 占位 key）；模型路由/多模型 fallback 仍未做 | |
 
 ---
 
@@ -205,7 +205,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
 |:------:|------|------|
 | ~~P0~~ 🟡 | ~~容量压测曲线~~（首版）：`cmd/loadtest` 闭环加压器 + 平台天花板实测（见下「容量实测」）。剩长稳/真实 LLM 大压 | "生产级并发"需数据背书 |
 | ~~P0~~ ✅ | ~~本地模型(vLLM/Ollama) + reasoning_content 适配~~：Pool provider 感知（Ollama/vLLM 自动补 /v1 + 占位 key），统一走 OpenAI 兼容路径；ChatStream 加 onReasoning，思考过程 surface 到 exec 轨迹（不污染答案）。控制台加 ollama 选项。live：Ollama qwen2.5:0.5b 端到端出答案；deepseek-v4-pro「推理过程」入轨迹 | 对齐生产 Qwen |
-| P1 | 高可用：网关/调度多副本 + NATS 集群 + 自愈 | 解单点 |
+| ~~P1~~ 🟡 | 高可用（代码层就绪）：调度/工具本就队列组可多副本（实测 2 副本 8 任务 4/4 分摊）；**网关改为队列组订阅**（eval/usage/status/config），多副本不再重复落库/重复计费（单测+实测去重）。剩 NATS 集群 + 网关 LB + PG/Redis HA 属部署期 | 解单点 |
 | P1 | 备份/灾备演练（PG/Milvus/Neo4j） | 数据安全 |
 | ~~P1~~ ✅ | ~~优雅停机 drain~~：三 Go 服务全覆盖（gateway HTTP Shutdown / dispatcher 在途任务跑完 / mcp-go 在途工具回完），SIGTERM 后等在途至 SHUTDOWN_DRAIN_TIMEOUT(默认30s)再退；在途任务 ctx 脱离信号 ctx 不被掐断。剩 exec 轨迹 Redis 回放 | 可靠性 |
 | P1 | 核心链路集成测试 + 前端 E2E | 测试纵深 |

sundynix-shared/bus/bus.go

@@ -333,8 +333,9 @@ func (b *Bus) PublishTaskStatus(ev *contract.TaskStatusEvent) error {
 }
 
 // SubscribeTaskStatus 订阅任务状态流转（网关调用，落 PG + 推 UI）。
+// 队列组订阅：多网关副本下每条状态只由一个副本落库，避免重复写（HA）。
 func (b *Bus) SubscribeTaskStatus(onEvent func(*contract.TaskStatusEvent)) (unsub func() error, err error) {
-	sub, err := b.nc.Subscribe(contract.SubjectTaskStatus, func(m *nats.Msg) {
+	sub, err := b.nc.QueueSubscribe(contract.SubjectTaskStatus, contract.QueueGateway, func(m *nats.Msg) {
 		var ev contract.TaskStatusEvent
 		if json.Unmarshal(m.Data, &ev) == nil {
 			onEvent(&ev)
@@ -357,9 +358,9 @@ func (b *Bus) PublishEval(ev *contract.EvalEvent) error {
 	return b.nc.Publish(contract.SubjectEval, data)
 }
 
-// SubscribeEval 订阅评测结果（网关调用，落 PG）。
+// SubscribeEval 订阅评测结果（网关调用，落 PG）。队列组：多副本下每条只落一次（HA）。
 func (b *Bus) SubscribeEval(onEvent func(*contract.EvalEvent)) (unsub func() error, err error) {
-	sub, err := b.nc.Subscribe(contract.SubjectEval, func(m *nats.Msg) {
+	sub, err := b.nc.QueueSubscribe(contract.SubjectEval, contract.QueueGateway, func(m *nats.Msg) {
 		var ev contract.EvalEvent
 		if json.Unmarshal(m.Data, &ev) == nil {
 			onEvent(&ev)
@@ -381,8 +382,9 @@ func (b *Bus) PublishUsage(ev *contract.UsageEvent) error {
 }
 
 // SubscribeUsage 订阅 token 用量（网关调用，累计到用户日预算 / 计费）。
+// 队列组：多副本下每条用量只累加一次，避免日预算被重复计（HA）。
 func (b *Bus) SubscribeUsage(onEvent func(*contract.UsageEvent)) (unsub func() error, err error) {
-	sub, err := b.nc.Subscribe(contract.SubjectUsage, func(m *nats.Msg) {
+	sub, err := b.nc.QueueSubscribe(contract.SubjectUsage, contract.QueueGateway, func(m *nats.Msg) {
 		var ev contract.UsageEvent
 		if json.Unmarshal(m.Data, &ev) == nil {
 			onEvent(&ev)
@@ -480,8 +482,9 @@ func (b *Bus) RequestConfigWithRetry(ctx context.Context, kind string, apply fun
 }
 
 // ServeConfig 让控制面响应某 kind 的配置请求；provide 返回当前激活配置（可为 nil）。
+// 队列组：多网关副本下每个配置请求只由一个副本应答，避免请求方收到多份重复应答（HA）。
 func (b *Bus) ServeConfig(kind string, provide func() *contract.ModelConfig) (unsub func() error, err error) {
-	sub, err := b.nc.Subscribe(contract.ConfigGetSubject(kind), func(m *nats.Msg) {
+	sub, err := b.nc.QueueSubscribe(contract.ConfigGetSubject(kind), contract.QueueGateway, func(m *nats.Msg) {
 		var data []byte
 		if cfg := provide(); cfg != nil {
 			data, _ = json.Marshal(cfg)

sundynix-shared/bus/bus_e2e_test.go

@@ -3,6 +3,7 @@ package bus_test
 import (
 	"context"
 	"encoding/json"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -360,3 +361,48 @@ func TestGracefulDrain(t *testing.T) {
 	}
 	t.Logf("✓ drain 等待在途任务跑完：耗时 %v", elapsed)
 }
+
+// TestGatewayQueueDedup 验证多网关副本下事件不重复处理：两个网关实例都订阅评测/用量/状态，
+// 经队列组应「每条只被一个副本处理」（旧的广播订阅会被两副本各处理一遍 → 重复落库/重复计费）。
+func TestGatewayQueueDedup(t *testing.T) {
+	url := startEmbeddedNATS(t)
+	pub, err := bus.Connect(url)
+	if err != nil {
+		t.Fatalf("pub connect: %v", err)
+	}
+	defer pub.Close()
+	gwA, err := bus.Connect(url)
+	if err != nil {
+		t.Fatalf("gwA connect: %v", err)
+	}
+	defer gwA.Close()
+	gwB, err := bus.Connect(url)
+	if err != nil {
+		t.Fatalf("gwB connect: %v", err)
+	}
+	defer gwB.Close()
+
+	var total int64
+	count := func(_ *contract.EvalEvent) { atomic.AddInt64(&total, 1) }
+	if _, err := gwA.SubscribeEval(count); err != nil {
+		t.Fatalf("gwA sub: %v", err)
+	}
+	if _, err := gwB.SubscribeEval(count); err != nil {
+		t.Fatalf("gwB sub: %v", err)
+	}
+	time.Sleep(100 * time.Millisecond) // 等订阅就绪
+
+	const n = 50
+	for i := 0; i < n; i++ {
+		if err := pub.PublishEval(&contract.EvalEvent{TaskID: "t", Overall: 1}); err != nil {
+			t.Fatalf("publish: %v", err)
+		}
+	}
+	time.Sleep(400 * time.Millisecond) // 等投递
+
+	got := atomic.LoadInt64(&total)
+	if got != n {
+		t.Fatalf("两副本共处理 %d 条，期望 %d（广播会得 %d=重复处理）", got, n, 2*n)
+	}
+	t.Logf("✓ 队列组去重：%d 条事件被两网关副本合计处理 %d 次（无重复）", n, got)
+}

sundynix-shared/contract/task.go

@@ -25,6 +25,10 @@ const (
 	QueueToolsGo      = "mcp-go-workers"      // mcp-go 队列组（多副本负载均衡）
 	QueueToolsPy      = "mcp-py-workers"      // mcp-py 队列组
 
+	// QueueGateway 是网关侧事件订阅/配置应答的队列组：多网关副本下，每条
+	// eval/usage/status 事件与每个 config 请求只由组内一个副本处理，避免重复落库/重复应答（HA）。
+	QueueGateway = "gateway-workers"
+
 	// 服务探活：dispatcher 既无 HTTP 端点也不挂工具，单独用一个 core NATS
 	// request-reply 心跳主题让控制面（管理端「服务状态」）能判定它在不在线。
 	SubjectHealthDispatcher = "sundynix.health.dispatcher"