feat(prompts): v2 DB 控制面热切换 —— 版本留存 + 不重启即生效

在 v1(注册表+文件覆盖)上加 DB 管理层与热切换，镜像 model-config 控制面： - store: sundynix_prompt 表(key/version/content/active) + ActivePrompts/ListPrompts/ CreateVersion/Activate/Deactivate - 控制面: ServePrompts/RequestActivePrompts(+Retry)/PublishPromptsUpdated/SubscribePromptsUpdated； prompts.ApplyOverrides 整体替换覆盖集(DB 激活集为权威) - gateway API: GET/POST /api/v1/prompts、version/activate/deactivate；激活/撤销即广播 - dispatcher/mcp-go: 启动拉激活集 + 订阅热更新(不重启) - live: 建版本→激活→mcp-go 图谱抽取 2→0→回滚 2(全程不重启)；deactivate 回退代码默认；版本可回溯 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-30 14:59:52 +08:00
parent aa61590b6c
commit 5736ad145e
12 changed files with 306 additions and 1 deletions
@@ -53,6 +53,12 @@ func main() {
 	}
 	go sub.FetchModelConfigWithRetry(context.Background(), pool.SetConfig)

+	// Prompt 控制面：拉激活集覆盖内置默认 + 订阅热更新（管理端激活某版即生效，不重启）。
+	go sub.FetchPromptsWithRetry(context.Background(), prompts.ApplyOverrides)
+	if _, err := sub.SubscribePromptsUpdated(prompts.ApplyOverrides); err != nil {
+		log.Printf("[dispatcher] subscribe prompts: %v", err)
+	}
+
 	// sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)、
 	// 任务状态回写(StatusSink)、HITL 审批等待(ApprovalWaiter)与评测落库(EvalSink)出口。
 	orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub, sub, sub)
@@ -133,4 +133,14 @@ func (s *Subscriber) FetchModelConfigWithRetry(ctx context.Context, apply func(*
 	s.inner.RequestConfigWithRetry(ctx, contract.ConfigKindChat, apply)
 }

+// FetchPromptsWithRetry 后台重试拉取初始激活 prompt 集（覆盖内置默认）。
+func (s *Subscriber) FetchPromptsWithRetry(ctx context.Context, apply func(map[string]string)) {
+	s.inner.RequestPromptsWithRetry(ctx, apply)
+}
+
+// SubscribePromptsUpdated 订阅 prompt 激活集热更新。
+func (s *Subscriber) SubscribePromptsUpdated(onUpdate func(map[string]string)) (func() error, error) {
+	return s.inner.SubscribePromptsUpdated(onUpdate)
+}
+
 func (s *Subscriber) Close() { s.inner.Close() }
@@ -60,6 +60,13 @@ func main() {
 		}
 	}

+	// Prompt 控制面：响应各服务「取激活 prompt」请求（DB 激活集；空则服务用内置默认）。
+	if _, err := bus.ServePrompts(func() map[string]string {
+		return db.ActivePrompts(context.Background())
+	}); err != nil {
+		log.Printf("[gateway] serve prompts: %v", err)
+	}
+
 	// 任务生命周期：订阅 dispatcher 回写的状态流转（running/done/failed/timeout），落 PG 供 UI 查询。
 	if _, err := bus.SubscribeTaskStatus(func(ev *contract.TaskStatusEvent) {
 		if err := db.UpdateTaskStatus(context.Background(), ev.TaskID, ev.Status, ev.Detail); err != nil {
@@ -0,0 +1,81 @@
+package handler
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/sundynix/sundynix-shared/prompts"
+)
+
+// PromptList: GET /api/v1/prompts —— 列出受管 prompt 的全部版本 + 可配键（管理端浏览/对比/回滚）。
+func (h *Handler) PromptList(c *gin.Context) {
+	rows, err := h.db.ListPrompts(c.Request.Context())
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	out := make([]gin.H, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, gin.H{"key": r.Key, "version": r.Version, "active": r.Active, "note": r.Note, "content": r.Content})
+	}
+	c.JSON(http.StatusOK, gin.H{"keys": prompts.Known, "versions": out})
+}
+
+// PromptCreateVersion: POST /api/v1/prompts/version {key, content, note} —— 为某 key 建新版本（不自动激活）。
+func (h *Handler) PromptCreateVersion(c *gin.Context) {
+	var body struct {
+		Key     string `json:"key"`
+		Content string `json:"content"`
+		Note    string `json:"note"`
+	}
+	if err := c.ShouldBindJSON(&body); err != nil || body.Key == "" || body.Content == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "key/content required"})
+		return
+	}
+	v, err := h.db.CreatePromptVersion(c.Request.Context(), body.Key, body.Content, body.Note)
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"key": body.Key, "version": v})
+}
+
+// PromptActivate: POST /api/v1/prompts/activate {key, version} —— 激活某版本 → 经控制面热下发各服务（不重启即生效）。
+func (h *Handler) PromptActivate(c *gin.Context) {
+	var body struct {
+		Key     string `json:"key"`
+		Version int    `json:"version"`
+	}
+	if err := c.ShouldBindJSON(&body); err != nil || body.Key == "" || body.Version <= 0 {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "key/version required"})
+		return
+	}
+	if err := h.db.ActivatePrompt(c.Request.Context(), body.Key, body.Version); err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	// 广播最新激活集 → dispatcher/mcp-go 热更新覆盖。
+	if err := h.bus.PublishPromptsUpdated(h.db.ActivePrompts(c.Request.Context())); err != nil {
+		c.JSON(http.StatusOK, gin.H{"key": body.Key, "version": body.Version, "warn": "已激活但广播失败: " + err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"key": body.Key, "version": body.Version, "activated": true})
+}
+
+// PromptDeactivate: POST /api/v1/prompts/deactivate {key} —— 撤销某 key 激活，回退代码内置默认（热下发）。
+func (h *Handler) PromptDeactivate(c *gin.Context) {
+	var body struct {
+		Key string `json:"key"`
+	}
+	if err := c.ShouldBindJSON(&body); err != nil || body.Key == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "key required"})
+		return
+	}
+	if err := h.db.DeactivatePrompt(c.Request.Context(), body.Key); err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	_ = h.bus.PublishPromptsUpdated(h.db.ActivePrompts(c.Request.Context()))
+	c.JSON(http.StatusOK, gin.H{"key": body.Key, "deactivated": true})
+}
@@ -119,4 +119,16 @@ func (b *Bus) ConsumeIngestJobs(ctx context.Context, h sharedbus.IngestHandler)
 	return b.inner.ConsumeIngestJobs(ctx, h)
 }

+// ---- Prompt 控制面 ----
+
+// ServePrompts 让网关响应「取全部激活 prompt」请求。
+func (b *Bus) ServePrompts(provide func() map[string]string) (func() error, error) {
+	return b.inner.ServePrompts(provide)
+}
+
+// PublishPromptsUpdated 广播最新激活集 → 各服务热更新。
+func (b *Bus) PublishPromptsUpdated(m map[string]string) error {
+	return b.inner.PublishPromptsUpdated(m)
+}
+
 func (b *Bus) Close() { b.inner.Close() }
@@ -65,6 +65,12 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
 			p.GET("/kb/vault", h.KbVault)               // 文库列表
 			p.GET("/kb/doc", h.KbDoc)                   // 取单篇文档
 			p.DELETE("/kb/doc", h.KbDeleteDoc)          // 级联删文档（三库+MinIO+PG）
+
+			// Prompt 控制面（平台级配置：建版本 → 激活 → 控制面热下发各服务）
+			p.GET("/prompts", h.PromptList)                 // 列出全部版本 + 可配键
+			p.POST("/prompts/version", h.PromptCreateVersion) // 建新版本（不自动激活）
+			p.POST("/prompts/activate", h.PromptActivate)     // 激活某版本 → 广播热更新
+			p.POST("/prompts/deactivate", h.PromptDeactivate) // 撤销激活 → 回退代码默认（热）
 			p.GET("/kb/links", h.KbLinks)               // 某库双链
 			p.POST("/kb/note", h.KbSaveNote)            // 新建/编辑笔记
 			p.GET("/kb/graph", h.KbGraph)               // 知识图谱三元组
@@ -66,7 +66,7 @@ func OpenPostgres(dsn string) *Postgres {
 	migrateLegacyIntIDs(db)
 	migrateDocLinkToID(db)

-	if err := db.AutoMigrate(&User{}, &Task{}, &Eval{}, &LLMModel{}, &KB{}, &Doc{}, &Agent{}, &DocLink{}, &Pricing{}); err != nil {
+	if err := db.AutoMigrate(&User{}, &Task{}, &Eval{}, &LLMModel{}, &KB{}, &Doc{}, &Agent{}, &DocLink{}, &Pricing{}, &Prompt{}); err != nil {
 		log.Printf("[store] postgres AutoMigrate 失败，降级运行: %v", err)
 		return &Postgres{}
 	}
@@ -0,0 +1,82 @@
+package store
+
+import (
+	"context"
+
+	"gorm.io/gorm"
+)
+
+// Prompt 是一条受管系统提示词的某个版本。表名 sundynix_prompt，(key,version) 唯一。
+// 平台级配置（非按 owner），由管理端维护：建版本 → 激活某版 → 控制面热下发各服务。
+type Prompt struct {
+	BaseModel
+	Key     string `gorm:"size:64;uniqueIndex:idx_prompt_kv;index"` // 受管 prompt 键，如 graph.extract
+	Version int    `gorm:"uniqueIndex:idx_prompt_kv"`               // 该 key 下递增版本号
+	Content string `gorm:"type:text"`                               // 提示词正文
+	Active  bool   `gorm:"index"`                                   // 是否为该 key 当前激活版（每 key 至多一条）
+	Note    string `gorm:"size:200"`                                // 版本说明
+}
+
+func (Prompt) TableName() string { return "sundynix_prompt" }
+
+// ActivePrompts 返回全部激活版的 key→content（控制面下发用）。
+func (p *Postgres) ActivePrompts(ctx context.Context) map[string]string {
+	out := map[string]string{}
+	if p.db == nil {
+		return out
+	}
+	var rows []Prompt
+	if err := p.db.WithContext(ctx).Where("active = ?", true).Find(&rows).Error; err != nil {
+		return out
+	}
+	for _, r := range rows {
+		out[r.Key] = r.Content
+	}
+	return out
+}
+
+// ListPrompts 返回全部 prompt 版本（按 key、version 排序，供管理端浏览/对比/回滚）。
+func (p *Postgres) ListPrompts(ctx context.Context) ([]Prompt, error) {
+	if p.db == nil {
+		return nil, nil
+	}
+	var rows []Prompt
+	err := p.db.WithContext(ctx).Order("key asc, version asc").Find(&rows).Error
+	return rows, err
+}
+
+// CreatePromptVersion 为某 key 新建一个版本（version=该 key 现有最大+1，默认不激活），返回新版本号。
+func (p *Postgres) CreatePromptVersion(ctx context.Context, key, content, note string) (int, error) {
+	if p.db == nil {
+		return 0, nil
+	}
+	var maxV struct{ V int }
+	p.db.WithContext(ctx).Model(&Prompt{}).Select("coalesce(max(version),0) as v").Where("key = ?", key).Scan(&maxV)
+	v := maxV.V + 1
+	row := Prompt{Key: key, Version: v, Content: content, Note: note, Active: false}
+	if err := p.db.WithContext(ctx).Create(&row).Error; err != nil {
+		return 0, err
+	}
+	return v, nil
+}
+
+// DeactivatePrompt 撤销某 key 的全部激活（控制面下发后该 key 回退到代码内置默认）。
+func (p *Postgres) DeactivatePrompt(ctx context.Context, key string) error {
+	if p.db == nil {
+		return nil
+	}
+	return p.db.WithContext(ctx).Model(&Prompt{}).Where("key = ?", key).Update("active", false).Error
+}
+
+// ActivatePrompt 激活某 key 的指定版本（同 key 其它版本置非激活）。事务保证每 key 至多一个激活版。
+func (p *Postgres) ActivatePrompt(ctx context.Context, key string, version int) error {
+	if p.db == nil {
+		return nil
+	}
+	return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if err := tx.Model(&Prompt{}).Where("key = ?", key).Update("active", false).Error; err != nil {
+			return err
+		}
+		return tx.Model(&Prompt{}).Where("key = ? AND version = ?", key, version).Update("active", true).Error
+	})
+}
@@ -98,6 +98,12 @@ func main() {
 	go b.RequestConfigWithRetry(ctx, contract.ConfigKindEmbedding, applyEmbed)
 	go b.RequestConfigWithRetry(ctx, contract.ConfigKindChat, applyChat)

+	// Prompt 控制面：拉取激活集覆盖内置默认 + 订阅热更新（管理端激活某版即生效，不重启）。
+	go b.RequestPromptsWithRetry(ctx, prompts.ApplyOverrides)
+	if _, err := b.SubscribePromptsUpdated(prompts.ApplyOverrides); err != nil {
+		log.Printf("[mcp_go] subscribe prompts: %v", err)
+	}
+
 	gw := mcp.NewGateway(b, engine, mem, hist, ragEngine, pgDSN)

 	log.Println("[mcp_go] serving MCP over sundynix.tools.go.* (Ctrl-C to quit)")
@@ -588,6 +588,75 @@ func (b *Bus) SubscribeConfigUpdated(kind string, onUpdate func(*contract.ModelC
 	return sub.Unsubscribe, nil
 }

+// ---- Prompt 控制面（core NATS request-reply + broadcast，镜像 config）----
+
+// ServePrompts 让控制面响应「取全部激活 prompt」请求；provide 返回 key→content 映射（可空）。
+// 队列组：多网关副本下每个请求只由一个副本应答。
+func (b *Bus) ServePrompts(provide func() map[string]string) (unsub func() error, err error) {
+	sub, err := b.nc.QueueSubscribe(contract.SubjectPromptsGet, contract.QueueGateway, func(m *nats.Msg) {
+		data, _ := json.Marshal(provide())
+		_ = m.Respond(data)
+	})
+	if err != nil {
+		return nil, fmt.Errorf("serve prompts: %w", err)
+	}
+	return sub.Unsubscribe, nil
+}
+
+// RequestActivePrompts 向控制面请求当前全部激活 prompt（key→content）。无人应答/空集返回 (nil,nil)，调用方用内置默认。
+func (b *Bus) RequestActivePrompts(ctx context.Context) (map[string]string, error) {
+	msg, err := b.nc.RequestWithContext(ctx, contract.SubjectPromptsGet, nil)
+	if err != nil || len(msg.Data) == 0 {
+		return nil, nil
+	}
+	var m map[string]string
+	if err := json.Unmarshal(msg.Data, &m); err != nil {
+		return nil, fmt.Errorf("unmarshal prompts: %w", err)
+	}
+	return m, nil
+}
+
+// RequestPromptsWithRetry 后台重试拉取初始激活 prompt 集（容忍消费方早于网关启动），拿到非空即 apply。
+func (b *Bus) RequestPromptsWithRetry(ctx context.Context, apply func(map[string]string)) {
+	for i := 0; i < 60; i++ {
+		cctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+		m, _ := b.RequestActivePrompts(cctx)
+		cancel()
+		if len(m) > 0 {
+			apply(m)
+			return
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(2 * time.Second):
+		}
+	}
+}
+
+// PublishPromptsUpdated 广播激活集变更（携带全量 key→content，消费方据此整体热更新）。
+func (b *Bus) PublishPromptsUpdated(m map[string]string) error {
+	data, err := json.Marshal(m)
+	if err != nil {
+		return err
+	}
+	return b.nc.Publish(contract.SubjectPromptsUpdated, data)
+}
+
+// SubscribePromptsUpdated 订阅激活集变更，回调拿到全量 key→content。
+func (b *Bus) SubscribePromptsUpdated(onUpdate func(map[string]string)) (unsub func() error, err error) {
+	sub, err := b.nc.Subscribe(contract.SubjectPromptsUpdated, func(m *nats.Msg) {
+		var mp map[string]string
+		if json.Unmarshal(m.Data, &mp) == nil {
+			onUpdate(mp)
+		}
+	})
+	if err != nil {
+		return nil, fmt.Errorf("subscribe prompts updated: %w", err)
+	}
+	return sub.Unsubscribe, nil
+}
+
 // TaskHandler 处理一个消费到的任务。
 type TaskHandler func(ctx context.Context, t *contract.Task) error

@@ -0,0 +1,8 @@
+package contract
+
+// Prompt 控制面：DB 存各 prompt 的多版本，激活某版 → 经控制面热下发给各服务，
+// 覆盖内置默认（不重启即生效）。镜像 model-config 控制面(ConfigGetSubject/ConfigUpdatedSubject)。
+const (
+	SubjectPromptsGet     = "sundynix.prompts.get"     // request-reply：取全部激活 prompt（map key→content）
+	SubjectPromptsUpdated = "sundynix.prompts.updated" // 广播：激活集变更，消费方据此热更新
+)
@@ -20,6 +20,9 @@ const (
 	MemoryExtract   = "memory.extract"    // dispatcher：长期记忆对账
 )

+// Known 是平台受管的全部 prompt key（供管理端列出可配项；网关进程不登记默认，靠它枚举）。
+var Known = []string{GraphExtract, EvalQuality, EvalRefine, GuardJailbreak, CoordinatorLead, MemoryExtract}
+
 // Registry 持有默认与覆盖；Get 取覆盖优先、否则默认。并发安全。
 type Registry struct {
 	mu        sync.RWMutex
@@ -53,6 +56,21 @@ func (r *Registry) SetOverride(key, text string) {
 	r.overrides[key] = text
 }

+// ApplyOverrides 用给定集合整体替换当前覆盖集（控制面热下发：DB 激活集为权威）。
+// 空值条目忽略；传空 map 即清空所有覆盖、全回退内置默认。
+func ApplyOverrides(m map[string]string) { global.ApplyOverrides(m) }
+func (r *Registry) ApplyOverrides(m map[string]string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	next := make(map[string]string, len(m))
+	for k, v := range m {
+		if v != "" {
+			next[k] = v
+		}
+	}
+	r.overrides = next
+}
+
 // Get 取生效内容：有覆盖用覆盖，否则用默认（都没有则空串）。
 func Get(key string) string { return global.Get(key) }
 func (r *Registry) Get(key string) string {