feat(ops): 优雅停机 drain —— 三 Go 服务 SIGTERM 后排空在途，不硬切

滚动更新/重启时旧的"信号到→进程退"会硬切在途工作：dispatcher 在途任务被掐断、
mcp-go 在途工具调用让 dispatcher 干等超时、gateway 在途 HTTP 请求被截断。

共享 bus 加在途追踪 + drain：
- ConsumeTasks/ServeTool 各加 sync.WaitGroup 跟踪在途 goroutine，返回 drain(ctx)：
  先停止接新活（cc.Stop / Unsubscribe），再等在途跑完至 drain 超时。
- 关键修复：任务 handler 的 ctx 改为派生自 context.Background()（而非信号 ctx），
  否则 SIGTERM 会立即取消在途任务的 ctx，drain 形同虚设。超时未跑完才由 JetStream
  AckWait 重投兜底（不丢任务）。
- DrainTimeout()：SHUTDOWN_DRAIN_TIMEOUT 秒，默认 30s。

各服务收尾：
- gateway：r.Run → http.Server + signal.NotifyContext + srv.Shutdown(drain 在途请求)，
  随后 defer 关 db/redis/bus（HTTP 排空后才断后端连接）。
- dispatcher：收到信号 → drain 在途任务跑完再退。
- mcp-go：收到信号 → drain 在途工具调用回完再退（dispatcher 拿到结果而非超时）。

bus 加 TestGracefulDrain（drain 等满在途任务 + 验证在途 ctx 不被取消），e2e 测试适配
新签名，四模块全绿。live：在途任务执行中 kill -TERM dispatcher → 日志「drain 在途任务」
→ 11s 后 task done(511字完整生成) → 「drain 完成，退出」，任务终态 done 非 failed；
gateway/mcp-go 同样优雅退出。

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

project_analysis.md

@@ -147,7 +147,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
 | **备份 / 灾备** | PG/Milvus/Neo4j 无备份恢复方案 | 数据丢失风险 |
 | **安全审计** | 全自述，未渗透/未审计 | "纵深防御"未被验证 |
 | **配额 / 多租户** | 仅按 IP 全局限流；owner 基础隔离 | 无团队/组织/按用户配额 |
-| **可靠性细节** | exec 轨迹流丢事件；停机不 drain | 观测缺口 + 滚更硬切在途任务 |
+| **可靠性细节** | exec 轨迹流丢事件；~~停机不 drain~~（已优雅停机 drain） | 观测缺口 |
 | **计量计费** | 计价配置有，计量×单价+配额未落地 | 商业化未闭环 |
 | **本地模型** | 仅 OpenAI 兼容在线 API | vLLM/Ollama、模型路由/fallback 未做 |
 
@@ -207,7 +207,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
 | P0 | 本地模型(vLLM/Ollama) + 推理模型 reasoning_content 适配 | 对齐生产 Qwen |
 | P1 | 高可用：网关/调度多副本 + NATS 集群 + 自愈 | 解单点 |
 | P1 | 备份/灾备演练（PG/Milvus/Neo4j） | 数据安全 |
-| P1 | exec 轨迹 Redis 回放 + 优雅停机 drain | 可靠性 |
+| ~~P1~~ ✅ | ~~优雅停机 drain~~：三 Go 服务全覆盖（gateway HTTP Shutdown / dispatcher 在途任务跑完 / mcp-go 在途工具回完），SIGTERM 后等在途至 SHUTDOWN_DRAIN_TIMEOUT(默认30s)再退；在途任务 ctx 脱离信号 ctx 不被掐断。剩 exec 轨迹 Redis 回放 | 可靠性 |
 | P1 | 核心链路集成测试 + 前端 E2E | 测试纵深 |
 | P2 | 计量计费闭环 + 按用户配额；OCR 完善；KbView 拆分；循环节点 | |
 | P2 | 安全审计 / 渗透（让"纵深防御"从自述变已验证） | 外部 |

sundynix-dispatcher/internal/nats/subscriber.go

@@ -33,15 +33,21 @@ func MustConnect(url string) *Subscriber {
 
 // ConsumeTasks 从 sundynix.tasks.* 持续消费任务（队列组负载均衡），阻塞至 ctx 取消。
 func (s *Subscriber) ConsumeTasks(ctx context.Context, h TaskHandler) error {
-	stop, err := s.inner.ConsumeTasks(ctx, func(c context.Context, t *contract.Task) error {
+	drain, err := s.inner.ConsumeTasks(ctx, func(c context.Context, t *contract.Task) error {
 		return h(c, t)
 	})
 	if err != nil {
 		return err
 	}
-	defer stop()
-	<-ctx.Done()
-	return ctx.Err()
+	<-ctx.Done() // 收到停机信号
+	// 优雅停机：停止消费新任务 + 等在途任务跑完（至多 DrainTimeout），再退出。
+	to := sharedbus.DrainTimeout()
+	log.Printf("[dispatcher] 收到停机信号，drain 在途任务（≤%s）…", to)
+	dctx, cancel := context.WithTimeout(context.Background(), to)
+	defer cancel()
+	drain(dctx)
+	log.Printf("[dispatcher] drain 完成，退出")
+	return nil
 }
 
 // PublishToken / CompleteStream 让 Subscriber 满足 eino.TokenSink，

sundynix-gateway/cmd/server/main.go

@@ -5,13 +5,17 @@ import (
 	"context"
 	"encoding/json"
 	"log"
+	"net/http"
 	"os"
+	"os/signal"
+	"syscall"
 	"time"
 
 	"github.com/sundynix/sundynix-gateway/internal/blob"
 	"github.com/sundynix/sundynix-gateway/internal/nats"
 	"github.com/sundynix/sundynix-gateway/internal/router"
 	"github.com/sundynix/sundynix-gateway/internal/store"
+	sharedbus "github.com/sundynix/sundynix-shared/bus"
 	"github.com/sundynix/sundynix-shared/contract"
 	"github.com/sundynix/sundynix-shared/otelx"
 	"github.com/sundynix/sundynix-shared/secrets"
@@ -94,10 +98,29 @@ func main() {
 
 	r := router.New(db, cache, bus, blobStore)
 	addr := envOr("GATEWAY_ADDR", ":8080")
-	log.Printf("[gateway] listening on %s", addr)
-	if err := r.Run(addr); err != nil {
-		log.Fatalf("[gateway] exit: %v", err)
+	srv := &http.Server{Addr: addr, Handler: r}
+
+	// 后台监听；ListenAndServe 在 Shutdown 后返回 ErrServerClosed（正常退出）。
+	go func() {
+		log.Printf("[gateway] listening on %s", addr)
+		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			log.Fatalf("[gateway] listen: %v", err)
+		}
+	}()
+
+	// 优雅停机：收到信号 → 停止收新连接、drain 在途请求（至多 DrainTimeout），再退出
+	// （defer 的 db/cache/bus/blob Close 随后执行，确保 HTTP 排空后才断后端连接）。
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+	<-ctx.Done()
+	to := sharedbus.DrainTimeout()
+	log.Printf("[gateway] 收到停机信号，优雅关闭 HTTP（drain 在途请求 ≤%s）…", to)
+	sctx, cancel := context.WithTimeout(context.Background(), to)
+	defer cancel()
+	if err := srv.Shutdown(sctx); err != nil {
+		log.Printf("[gateway] HTTP 优雅关闭超时/出错: %v", err)
 	}
+	log.Println("[gateway] 已优雅停机")
 }
 
 func envOr(key, def string) string {

sundynix-mcp-go/internal/mcp/gateway.go

@@ -68,15 +68,21 @@ func NewGateway(b *sharedbus.Bus, s *search.Hybrid, m *memory.Store, h *history.
 
 // Serve 以队列组通配订阅 sundynix.tools.go.>，按工具名分发并阻塞。
 func (g *Gateway) Serve(ctx context.Context) error {
-	unsub, err := g.bus.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo, g.dispatch)
+	drain, err := g.bus.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo, g.dispatch)
 	if err != nil {
 		return err
 	}
-	defer func() { _ = unsub() }()
 	log.Printf("[mcp_go] tools ready on %s (queue=%s): wiki_search, kb_ingest, kb_search, kb_graph, report_render, memory_*, history_*, echo",
 		contract.SubjectToolsGoAll, contract.QueueToolsGo)
-	<-ctx.Done()
-	return ctx.Err()
+	<-ctx.Done() // 收到停机信号
+	// 优雅停机：停止接新工具调用 + 等在途调用回完（至多 DrainTimeout），让 dispatcher 拿到结果而非干等超时。
+	to := sharedbus.DrainTimeout()
+	log.Printf("[mcp_go] 收到停机信号，drain 在途工具调用（≤%s）…", to)
+	dctx, cancel := context.WithTimeout(context.Background(), to)
+	defer cancel()
+	drain(dctx)
+	log.Printf("[mcp_go] drain 完成，退出")
+	return nil
 }
 
 // buildRegistry 注册 mcp-go 全部工具：名称 → (中文名, 作用, 处理函数)。

sundynix-shared/bus/bus.go

@@ -9,6 +9,7 @@ import (
 	"log"
 	"os"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/nats-io/nats.go"
@@ -246,14 +247,17 @@ func toolConcurrency() int {
 // ServeTool 以队列组订阅工具主题（可用通配 sundynix.tools.go.>），队列组内多副本自动负载均衡（水平扩）。
 // 单实例内每个请求分发到独立 goroutine 并发处理（垂直扩，上限 MCP_TOOL_CONCURRENCY）——
 // 回调立即返回不阻塞 NATS 投递；信号量约束同时执行的 handler 数。返回的 unsub 用于退订。
-func (b *Bus) ServeTool(subject, queue string, h ToolHandler) (unsub func() error, err error) {
+func (b *Bus) ServeTool(subject, queue string, h ToolHandler) (drain func(context.Context), err error) {
 	sem := make(chan struct{}, toolConcurrency())
+	var wg sync.WaitGroup // 跟踪在途工具调用，供优雅停机 drain 等待回完
 	sub, err := b.nc.QueueSubscribe(subject, queue, func(m *nats.Msg) {
 		// 立即起 goroutine 并返回，让 NATS 继续投递下一条（核心 NATS 单订阅回调本是串行）。
+		wg.Add(1)
 		go func() {
 			sem <- struct{}{} // 限并发：超额则在此短暂排队
 			defer func() {
 				<-sem
+				wg.Done()
 				if r := recover(); r != nil { // handler panic → 回错误结果，调用方不必干等超时
 					respond(m, &contract.ToolResult{OK: false, Error: fmt.Sprintf("tool panic: %v", r)})
 				}
@@ -279,7 +283,11 @@ func (b *Bus) ServeTool(subject, queue string, h ToolHandler) (unsub func() erro
 	if err != nil {
 		return nil, fmt.Errorf("serve tool %s: %w", subject, err)
 	}
-	return sub.Unsubscribe, nil
+	// drain：退订（停止接新请求），等在途工具调用回完（至多 dctx）。
+	return func(dctx context.Context) {
+		_ = sub.Unsubscribe()
+		drainWait(&wg, dctx)
+	}, nil
 }
 
 func respond(m *nats.Msg, res *contract.ToolResult) {
@@ -523,10 +531,31 @@ func taskConcurrency() int {
 	return 8
 }
 
+// DrainTimeout 是优雅停机时等待在途工作（任务/工具调用/HTTP 请求）跑完的上限。
+// 经 SHUTDOWN_DRAIN_TIMEOUT 秒配置，缺省 30s。超时即放弃等待退出（在途未 ack 由 JetStream 重投兜底）。
+func DrainTimeout() time.Duration {
+	if v := os.Getenv("SHUTDOWN_DRAIN_TIMEOUT"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return time.Duration(n) * time.Second
+		}
+	}
+	return 30 * time.Second
+}
+
+// drainWait 停止接新活后，等待在途 WaitGroup 跑完，或 dctx 到期放弃。
+func drainWait(wg *sync.WaitGroup, dctx context.Context) {
+	done := make(chan struct{})
+	go func() { wg.Wait(); close(done) }()
+	select {
+	case <-done:
+	case <-dctx.Done():
+	}
+}
+
 // ConsumeTasks 在持久消费者上消费任务，队列组内负载均衡。
 // 每个任务分发到独立 worker goroutine 并发执行——一个慢任务/HITL 待审不再阻塞后续任务。
 // 并发上限由信号量 + 消费者 MaxAckPending 双重约束（背压）。返回的 stop 用于优雅停止消费。
-func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (stop func(), err error) {
+func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (drain func(context.Context), err error) {
 	concurrency := taskConcurrency()
 	cons, err := b.js.CreateOrUpdateConsumer(ctx, contract.StreamTasks, jetstream.ConsumerConfig{
 		Durable:       contract.ConsumerDurable,
@@ -542,6 +571,7 @@ func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (stop func(), err
 		return nil, fmt.Errorf("create consumer: %w", err)
 	}
 	sem := make(chan struct{}, concurrency) // 限并发：最多 N 个任务同时执行
+	var wg sync.WaitGroup                   // 跟踪在途任务，供优雅停机 drain 等待跑完
 	cc, err := cons.Consume(func(msg jetstream.Msg) {
 		t, err := contract.Unmarshal(msg.Data())
 		if err != nil {
@@ -554,17 +584,20 @@ func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (stop func(), err
 		case <-ctx.Done():
 			return
 		}
+		wg.Add(1)
 		go func() {
 			defer func() {
 				<-sem // 释放并发额度
+				wg.Done()
 				if r := recover(); r != nil {
 					// 任务处理 panic：丢弃不重投（避免崩溃循环），记录后继续。
 					log.Printf("[bus] task %s handler panic: %v", t.ID, r)
 					_ = msg.Term()
 				}
 			}()
-			// 从消息头还原上游链路，开消费 span（成为 gateway 发布 span 的子节点）。
-			mctx := extractTrace(ctx, nats.Header(msg.Headers()))
+			// 关键：handler ctx 派生自 Background（而非信号 ctx），使「停止消费」不会立刻掐断在途任务——
+			// 在途任务由 drain 等待至完成；drain 超时未跑完才由 JetStream AckWait 重投兜底。
+			mctx := extractTrace(context.Background(), nats.Header(msg.Headers()))
 			mctx, span := tracer().Start(mctx, "nats.consume task",
 				trace.WithSpanKind(trace.SpanKindConsumer),
 				trace.WithAttributes(attribute.String("sundynix.task_id", t.ID)))
@@ -580,5 +613,9 @@ func (b *Bus) ConsumeTasks(ctx context.Context, h TaskHandler) (stop func(), err
 	if err != nil {
 		return nil, fmt.Errorf("consume: %w", err)
 	}
-	return cc.Stop, nil
+	// drain：停止新投递，等在途任务跑完（至多 dctx）。
+	return func(dctx context.Context) {
+		cc.Stop()
+		drainWait(&wg, dctx)
+	}, nil
 }

sundynix-shared/bus/bus_e2e_test.go

@@ -55,14 +55,14 @@ func TestTaskRoundTrip(t *testing.T) {
 	defer dp.Close()
 
 	got := make(chan *contract.Task, 1)
-	stop, err := dp.ConsumeTasks(ctx, func(_ context.Context, task *contract.Task) error {
+	drain, err := dp.ConsumeTasks(ctx, func(_ context.Context, task *contract.Task) error {
 		got <- task
 		return nil
 	})
 	if err != nil {
 		t.Fatalf("consume: %v", err)
 	}
-	defer stop()
+	defer drain(context.Background())
 
 	// --- Gateway 发布一个任务 ---
 	want := &contract.Task{
@@ -104,7 +104,7 @@ func TestToolCallRoundTrip(t *testing.T) {
 	}
 	defer srv.Close()
 
-	unsub, err := srv.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo,
+	drain, err := srv.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo,
 		func(_ context.Context, call *contract.ToolCall) *contract.ToolResult {
 			if call.Tool != "wiki_search" {
 				return &contract.ToolResult{OK: false, Error: "unknown tool"}
@@ -114,7 +114,7 @@ func TestToolCallRoundTrip(t *testing.T) {
 	if err != nil {
 		t.Fatalf("serve tool: %v", err)
 	}
-	defer func() { _ = unsub() }()
+	defer drain(context.Background())
 
 	// --- Dispatcher 侧：同步调用工具 ---
 	dp, err := bus.Connect(url)
@@ -220,7 +220,7 @@ func TestConcurrentConsume(t *testing.T) {
 
 	blockA := make(chan struct{})
 	doneB := make(chan string, 1)
-	stop, err := dp.ConsumeTasks(ctx, func(_ context.Context, task *contract.Task) error {
+	drain, err := dp.ConsumeTasks(ctx, func(_ context.Context, task *contract.Task) error {
 		switch task.ID {
 		case "A":
 			<-blockA // 模拟审批长阻塞
@@ -232,7 +232,7 @@ func TestConcurrentConsume(t *testing.T) {
 	if err != nil {
 		t.Fatalf("consume: %v", err)
 	}
-	defer stop()
+	defer drain(context.Background())
 	defer close(blockA)
 
 	// 先发 A 并等它被消费、卡在 handler 里；再发 B。
@@ -267,7 +267,7 @@ func TestConcurrentToolServe(t *testing.T) {
 	defer srv.Close()
 
 	blockSlow := make(chan struct{})
-	unsub, err := srv.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo,
+	drain, err := srv.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo,
 		func(_ context.Context, call *contract.ToolCall) *contract.ToolResult {
 			if call.Tool == "slow" {
 				<-blockSlow // 模拟慢工具（如 RAG 嵌入）
@@ -277,7 +277,7 @@ func TestConcurrentToolServe(t *testing.T) {
 	if err != nil {
 		t.Fatalf("serve tool: %v", err)
 	}
-	defer unsub()
+	defer drain(context.Background())
 	defer close(blockSlow)
 
 	cli, err := bus.Connect(url)
@@ -306,3 +306,57 @@ func TestConcurrentToolServe(t *testing.T) {
 	}
 	t.Log("✓ 单实例工具并发生效：slow 仍阻塞时 fast 已返回")
 }
+
+// TestGracefulDrain 验证优雅停机：drain 会等在途任务跑完（不被信号掐断），
+// 且在途任务的 ctx 不随消费停止而取消。
+func TestGracefulDrain(t *testing.T) {
+	url := startEmbeddedNATS(t)
+	gw, err := bus.Connect(url)
+	if err != nil {
+		t.Fatalf("gateway connect: %v", err)
+	}
+	defer gw.Close()
+	if err := gw.EnsureTaskStream(context.Background()); err != nil {
+		t.Fatalf("ensure stream: %v", err)
+	}
+	dp, err := bus.Connect(url)
+	if err != nil {
+		t.Fatalf("dispatcher connect: %v", err)
+	}
+	defer dp.Close()
+
+	started := make(chan struct{})
+	finished := make(chan struct{})
+	drain, err := dp.ConsumeTasks(context.Background(), func(c context.Context, _ *contract.Task) error {
+		close(started)
+		time.Sleep(500 * time.Millisecond) // 模拟在途长任务
+		if c.Err() != nil {                // 关键：消费停止不应取消在途任务的 ctx
+			t.Errorf("在途任务 ctx 不应被取消: %v", c.Err())
+		}
+		close(finished)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("consume: %v", err)
+	}
+	if _, err := gw.PublishTask(context.Background(), &contract.Task{ID: "drain1", Graph: json.RawMessage(`{}`)}); err != nil {
+		t.Fatalf("publish: %v", err)
+	}
+	<-started // 等任务进入 handler
+
+	t0 := time.Now()
+	dctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	drain(dctx) // 应阻塞至在途任务跑完
+	elapsed := time.Since(t0)
+
+	select {
+	case <-finished:
+	default:
+		t.Fatal("drain 返回时在途任务竟未完成（被掐断或未等待）")
+	}
+	if elapsed < 300*time.Millisecond {
+		t.Fatalf("drain 过早返回（%v），未真正等待在途任务", elapsed)
+	}
+	t.Logf("✓ drain 等待在途任务跑完：耗时 %v", elapsed)
+}