删除多余文件-改菜单

internal/opencode/direct_client.go

@@ -202,7 +202,7 @@ func (c *DirectClient) SendPromptStream(ctx context.Context, sessionID string, p
 	fmt.Printf("🔍 [opencode.DirectClient] 请求体: %s\n", string(reqBody))
 	fmt.Printf("🔍 [opencode.DirectClient] 端口: %d\n", c.port)
 
-	// 测试异步端点
+	// 发送异步请求到 opencode（触发AI处理）
 	url := fmt.Sprintf("%s/session/%s/prompt_async", c.baseURL, sessionID)
 	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(reqBody))
 	if err != nil {
@@ -214,82 +214,35 @@ func (c *DirectClient) SendPromptStream(ctx context.Context, sessionID string, p
 	fmt.Printf("🔍 [opencode.DirectClient] 请求 URL: %s\n", url)
 	fmt.Printf("🔍 [opencode.DirectClient] 请求头: %v\n", req.Header)
 
-	// 为SSE流创建独立的httpClient，不设置超时限制
+	// 发送请求，不等待SSE响应（事件通过EventDispatcher分发）
 	sseClient := &http.Client{
-		// 不设置Timeout，允许长连接
-		// Timeout: 0 表示无超时限制
+		Timeout: 30 * time.Second, // 设置超时，避免长时间阻塞
 	}
 	resp, err := sseClient.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("发送请求失败: %w", err)
 	}
+	defer resp.Body.Close()
 
 	fmt.Printf("🔍 [opencode.DirectClient] 响应状态: %d\n", resp.StatusCode)
 	fmt.Printf("🔍 [opencode.DirectClient] 响应头: %v\n", resp.Header)
 
-	if resp.StatusCode != http.StatusOK {
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
 		body, _ := io.ReadAll(resp.Body)
-		resp.Body.Close()
 		fmt.Printf("🔍 [opencode.DirectClient] 错误响应体: %s\n", string(body))
-
-		// 如果返回 204，尝试其他流式端点
-		if resp.StatusCode == http.StatusNoContent {
-			fmt.Printf("🔍 [opencode.DirectClient] 异步端点返回 204，尝试 /global/event SSE 流\n")
-			return c.subscribeGlobalEvents(ctx)
-		}
-
 		return nil, fmt.Errorf("请求失败，状态码: %d", resp.StatusCode)
 	}
 
-	ch := make(chan string, 100)
+	// 异步读取响应体（避免资源泄漏）
 	go func() {
-		defer resp.Body.Close()
-		defer close(ch)
-
-		reader := bufio.NewReader(resp.Body)
-		eventCount := 0
-
-		for {
-			line, err := reader.ReadString('\n')
-			if err != nil {
-				if err == io.EOF {
-					fmt.Printf("🔍 [opencode.DirectClient] SSE流结束，共收到 %d 个事件\n", eventCount)
-				} else {
-					// 区分正常取消和错误
-					if ctx.Err() != nil {
-						fmt.Printf("🔍 [opencode.DirectClient] SSE流正常结束（上下文取消）\n")
-					} else {
-						fmt.Printf("🔍 [opencode.DirectClient] 读取错误: %v\n", err)
-					}
-				}
-				return
-			}
-
-			line = strings.TrimSpace(line)
-			if line == "" {
-				continue
-			}
-
-			if strings.HasPrefix(line, "data: ") {
-				data := strings.TrimPrefix(line, "data: ")
-				eventCount++
-				fmt.Printf("🔍 [opencode.DirectClient] 收到SSE数据[%d]: %s\n", eventCount, data)
-
-				// 写入日志文件用于分析
-				writeStreamLog(sessionID, data)
-
-				select {
-				case ch <- data:
-				case <-ctx.Done():
-					fmt.Printf("🔍 [opencode.DirectClient] 上下文取消\n")
-					return
-				}
-			} else {
-				fmt.Printf("🔍 [opencode.DirectClient] 忽略非数据行: %s\n", line)
-			}
-		}
+		io.Copy(io.Discard, resp.Body)
 	}()
 
+	fmt.Printf("🔍 [opencode.DirectClient] 异步请求发送成功，事件将通过EventDispatcher分发\n")
+
+	// 返回一个立即关闭的空通道（保持接口兼容，实际事件通过EventDispatcher分发）
+	ch := make(chan string)
+	close(ch)
 	return ch, nil
 }
 

internal/routes/menu_routes.go

@@ -0,0 +1,59 @@
+package routes
+
+import (
+	"context"
+
+	"git.x2erp.com/qdy/go-base/authbase"
+	"git.x2erp.com/qdy/go-base/ctx"
+	"git.x2erp.com/qdy/go-base/logger"
+	"git.x2erp.com/qdy/go-base/model/response"
+	"git.x2erp.com/qdy/go-base/webx/router"
+	"git.x2erp.com/qdy/go-svc-code/internal/service/menu"
+)
+
+// RegisterMenuRoutes 注册菜单路由
+func RegisterMenuRoutes(ws *router.RouterService, mappingService *menu.MappingService) {
+	// 获取顶部菜单（需要Token认证）
+	ws.GET("/api/menu/top",
+		func(ctx context.Context, reqCtx *ctx.RequestContext) (*response.QueryResult[[]menu.MenuItem], error) {
+			return menu.GetTopMenu(ctx, reqCtx), nil
+		},
+	).Use(authbase.TokenAuth).Desc("获取顶部菜单").Register()
+
+	// 获取菜单项的所有会话ID（需要Token认证）
+	ws.GET("/api/menu/sessions",
+		func(menuItemID string, ctx context.Context, reqCtx *ctx.RequestContext) (*response.QueryResult[[]string], error) {
+			if menuItemID == "" {
+				return &response.QueryResult[[]string]{
+					Success: false,
+					Message: "参数 menu_item_id 不能为空",
+				}, nil
+			}
+
+			// 获取用户ID
+			userID := reqCtx.UserID
+			if userID == "" {
+				logger.Warn("无法从请求上下文获取用户ID")
+				return &response.QueryResult[[]string]{
+					Success: false,
+					Message: "用户认证信息不完整",
+				}, nil
+			}
+
+			// 获取菜单项的所有会话ID
+			sessionIDs, err := mappingService.GetSessionIDsByMenuItemID(ctx, menuItemID, userID)
+			if err != nil {
+				logger.Error("获取菜单项会话ID失败", "menu_item_id", menuItemID, "user_id", userID, "error", err)
+				return &response.QueryResult[[]string]{
+					Success: false,
+					Message: "获取菜单项会话ID失败: " + err.Error(),
+				}, nil
+			}
+
+			return &response.QueryResult[[]string]{
+				Success: true,
+				Data:    sessionIDs,
+			}, nil
+		},
+	).Use(authbase.TokenAuth).Desc("获取菜单项的所有会话ID").Register()
+}

internal/routes/prompt_stream_routes.go

@@ -12,6 +12,7 @@ import (
 	"git.x2erp.com/qdy/go-base/webx"
 	"git.x2erp.com/qdy/go-base/webx/router"
 	"git.x2erp.com/qdy/go-svc-code/internal/opencode"
+	"git.x2erp.com/qdy/go-svc-code/internal/service/event"
 )
 
 // PromptStreamRequest 流式对话请求
@@ -52,7 +53,7 @@ func StreamPromptHandler(client opencode.OpenCodeClient) http.HandlerFunc {
 		// 创建 prompt 请求
 		prompt := &opencode.PromptRequest{
 			Parts: req.Parts,
-			Agent: req.Agent,
+			Agent: "code-sql", //req.Agent,
 			Model: req.Model,
 		}
 
@@ -66,15 +67,47 @@ func StreamPromptHandler(client opencode.OpenCodeClient) http.HandlerFunc {
 		ctx, cancel := context.WithTimeout(r.Context(), 15*time.Minute)
 		defer cancel()
 
-		fmt.Printf("🔍 [StreamPromptHandler] 调用 SendPromptStream, sessionID=%s\n", req.SessionID)
-		// 获取流式响应通道
-		ch, err := client.SendPromptStream(ctx, req.SessionID, prompt)
+		// 获取事件分发器实例
+		dispatcher := event.GetEventDispatcher(client.GetBaseURL(), client.GetPort())
+
+		// 从认证上下文中获取用户ID（用于缓存，按sessionID分发事件）
+		userID := "unknown-user"
+
+		// TokenAuth中间件通常将用户信息存储在context中
+		if userVal := r.Context().Value("user"); userVal != nil {
+			if userMap, ok := userVal.(map[string]interface{}); ok {
+				if id, ok := userMap["id"].(string); ok {
+					userID = id
+				} else if id, ok := userMap["user_id"].(string); ok {
+					userID = id
+				}
+			}
+		}
+
+		fmt.Printf("🔍 [StreamPromptHandler] 用户ID: %s, 会话ID: %s\n", userID, req.SessionID)
+
+		// 注册会话到缓存
+		dispatcher.RegisterSession(req.SessionID, userID)
+
+		// 订阅该会话的事件
+		ch, err := dispatcher.Subscribe(req.SessionID, userID)
+		if err != nil {
+			fmt.Printf("🔍 [StreamPromptHandler] 订阅事件失败: %v\n", err)
+			http.Error(w, fmt.Sprintf("订阅事件失败: %v", err), http.StatusInternalServerError)
+			return
+		}
+		defer dispatcher.Unsubscribe(req.SessionID, ch)
+
+		// 发送异步请求到 opencode（触发AI处理）
+		fmt.Printf("🔍 [StreamPromptHandler] 发送异步请求到 opencode, sessionID=%s\n", req.SessionID)
+		// 忽略返回的通道，事件通过EventDispatcher分发
+		_, err = client.SendPromptStream(ctx, req.SessionID, prompt)
 		if err != nil {
-			fmt.Printf("🔍 [StreamPromptHandler] 发送流式请求失败: %v\n", err)
-			http.Error(w, fmt.Sprintf("发送流式请求失败: %v", err), http.StatusInternalServerError)
+			fmt.Printf("🔍 [StreamPromptHandler] 发送请求失败: %v\n", err)
+			http.Error(w, fmt.Sprintf("发送请求失败: %v", err), http.StatusInternalServerError)
 			return
 		}
-		fmt.Printf("🔍 [StreamPromptHandler] 成功获取流式响应通道\n")
+		fmt.Printf("🔍 [StreamPromptHandler] 异步请求发送成功，等待事件流\n")
 
 		// 发送流式响应
 		flusher, ok := w.(http.Flusher)

internal/routes/session_routes.go

@@ -5,14 +5,17 @@ import (
 
 	"git.x2erp.com/qdy/go-base/authbase"
 	"git.x2erp.com/qdy/go-base/ctx"
+	"git.x2erp.com/qdy/go-base/logger"
 	"git.x2erp.com/qdy/go-base/model/response"
 	"git.x2erp.com/qdy/go-base/webx/router"
 	"git.x2erp.com/qdy/go-svc-code/internal/opencode"
+	"git.x2erp.com/qdy/go-svc-code/internal/service/menu"
 )
 
 // SessionCreateRequest 创建会话请求
 type SessionCreateRequest struct {
-	Title string `json:"title" binding:"required"`
+	Title      string `json:"title" binding:"required"`
+	MenuItemID string `json:"menu_item_id" binding:"required"` // 菜单项ID，必填
 }
 
 // SessionResponse 会话响应
@@ -24,7 +27,7 @@ type SessionResponse struct {
 }
 
 // RegisterSessionRoutes 注册会话管理路由
-func RegisterSessionRoutes(ws *router.RouterService, client opencode.OpenCodeClient) {
+func RegisterSessionRoutes(ws *router.RouterService, client opencode.OpenCodeClient, mappingService *menu.MappingService) {
 	// 创建会话（需要Token认证）
 	ws.POST("/api/session/create",
 		func(req *SessionCreateRequest, ctx context.Context, reqCtx *ctx.RequestContext) (*response.QueryResult[SessionResponse], error) {
@@ -36,6 +39,29 @@ func RegisterSessionRoutes(ws *router.RouterService, client opencode.OpenCodeCli
 				}, nil
 			}
 
+			// 从认证信息中获取用户ID和租户ID
+			// TokenAuth中间件已经在reqCtx中设置了UserID和TenantID
+			userID := reqCtx.UserID
+			tenantID := reqCtx.TenantID
+
+			// 如果无法获取，使用占位符（实际生产环境应该确保认证信息正确设置）
+			if userID == "" {
+				userID = "unknown_user"
+				logger.Warn("无法从请求上下文获取用户ID，使用默认值", "session_id", session.ID)
+			}
+			if tenantID == "" {
+				tenantID = "default"
+			}
+			if tenantID == "" {
+				tenantID = "default"
+			}
+
+			// 保存会话-菜单映射
+			if err := mappingService.CreateMapping(ctx, session.ID, req.MenuItemID, userID, tenantID); err != nil {
+				logger.Warn("创建会话-菜单映射失败", "session_id", session.ID, "menu_item_id", req.MenuItemID, "error", err)
+				// 注意：映射创建失败不影响会话创建，但会记录警告
+			}
+
 			return &response.QueryResult[SessionResponse]{
 				Success: true,
 				Data: SessionResponse{
@@ -66,6 +92,33 @@ func RegisterSessionRoutes(ws *router.RouterService, client opencode.OpenCodeCli
 		},
 	).Use(authbase.TokenAuth).Desc("获取 opencode 会话列表").Register()
 
+	// 获取会话的菜单项ID（需要Token认证）
+	ws.GET("/api/session/menu",
+		func(sessionID string, ctx context.Context, reqCtx *ctx.RequestContext) (*response.QueryResult[string], error) {
+			if sessionID == "" {
+				return &response.QueryResult[string]{
+					Success: false,
+					Message: "参数 session_id 不能为空",
+				}, nil
+			}
+
+			// 获取会话的菜单项ID
+			menuItemID, err := mappingService.GetMenuItemBySessionID(ctx, sessionID)
+			if err != nil {
+				logger.Error("获取会话菜单项失败", "session_id", sessionID, "error", err)
+				return &response.QueryResult[string]{
+					Success: false,
+					Message: "获取会话菜单项失败: " + err.Error(),
+				}, nil
+			}
+
+			return &response.QueryResult[string]{
+				Success: true,
+				Data:    menuItemID,
+			}, nil
+		},
+	).Use(authbase.TokenAuth).Desc("获取会话的菜单项ID").Register()
+
 	// 获取单个会话（暂不实现，因为 opencode API 可能不支持）
 	// ws.GET("/api/session/get",
 	// 	func(ctx context.Context, reqCtx *ctx.RequestContext) (*response.QueryResult[opencode.Session], error) {

internal/service/event/cache.go

@@ -0,0 +1,118 @@
+package event
+
+import (
+	"sync"
+	"time"
+)
+
+// SessionUserCache 会话-用户映射缓存，支持TTL过期
+type SessionUserCache struct {
+	mu       sync.RWMutex
+	cache    map[string]cacheEntry
+	ttl      time.Duration
+	stopChan chan struct{}
+}
+
+// cacheEntry 缓存条目
+type cacheEntry struct {
+	userID    string
+	expiresAt time.Time
+}
+
+// NewSessionUserCache 创建新的会话-用户缓存
+func NewSessionUserCache(ttl time.Duration) *SessionUserCache {
+	cache := &SessionUserCache{
+		cache:    make(map[string]cacheEntry),
+		ttl:      ttl,
+		stopChan: make(chan struct{}),
+	}
+
+	// 启动清理协程
+	go cache.cleanupWorker()
+
+	return cache
+}
+
+// Set 设置会话-用户映射
+func (c *SessionUserCache) Set(sessionID, userID string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.cache[sessionID] = cacheEntry{
+		userID:    userID,
+		expiresAt: time.Now().Add(c.ttl),
+	}
+}
+
+// Get 获取用户ID，如果不存在或已过期返回空字符串
+func (c *SessionUserCache) Get(sessionID string) string {
+	c.mu.RLock()
+	entry, exists := c.cache[sessionID]
+	c.mu.RUnlock()
+
+	if !exists {
+		return ""
+	}
+
+	// 检查是否过期
+	if time.Now().After(entry.expiresAt) {
+		// 异步删除过期条目
+		go c.deleteIfExpired(sessionID)
+		return ""
+	}
+
+	return entry.userID
+}
+
+// Delete 删除指定会话的缓存
+func (c *SessionUserCache) Delete(sessionID string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	delete(c.cache, sessionID)
+}
+
+// deleteIfExpired 检查并删除过期条目
+func (c *SessionUserCache) deleteIfExpired(sessionID string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if entry, exists := c.cache[sessionID]; exists {
+		if time.Now().After(entry.expiresAt) {
+			delete(c.cache, sessionID)
+		}
+	}
+}
+
+// cleanupWorker 定期清理过期条目的工作协程
+func (c *SessionUserCache) cleanupWorker() {
+	ticker := time.NewTicker(c.ttl / 2)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			c.cleanupExpired()
+		case <-c.stopChan:
+			return
+		}
+	}
+}
+
+// cleanupExpired 清理所有过期条目
+func (c *SessionUserCache) cleanupExpired() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	now := time.Now()
+	for sessionID, entry := range c.cache {
+		if now.After(entry.expiresAt) {
+			delete(c.cache, sessionID)
+		}
+	}
+}
+
+// Stop 停止缓存清理协程
+func (c *SessionUserCache) Stop() {
+	close(c.stopChan)
+}

internal/service/event/dispatcher.go

@@ -0,0 +1,490 @@
+package event
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"git.x2erp.com/qdy/go-base/logger"
+)
+
+// EventDispatcher 事件分发器 - 单例模式
+type EventDispatcher struct {
+	mu               sync.RWMutex
+	baseURL          string
+	port             int
+	subscriptions    map[string]map[chan string]struct{} // sessionID -> set of channels
+	sessionUserCache *SessionUserCache                   // sessionID -> userID 映射缓存（用于用户验证）
+	client           *http.Client
+	cancelFunc       context.CancelFunc
+	running          bool
+}
+
+// EventData opencode事件数据结构 - 匹配实际事件格式
+type EventData struct {
+	Directory string                 `json:"directory,omitempty"`
+	Payload   map[string]interface{} `json:"payload"`
+}
+
+// PayloadData payload内部结构（辅助类型）
+type PayloadData struct {
+	Type       string                 `json:"type"`
+	Properties map[string]interface{} `json:"properties,omitempty"`
+}
+
+// NewEventDispatcher 创建新的事件分发器
+func NewEventDispatcher(baseURL string, port int) *EventDispatcher {
+	return &EventDispatcher{
+		baseURL:          baseURL,
+		port:             port,
+		subscriptions:    make(map[string]map[chan string]struct{}),
+		sessionUserCache: NewSessionUserCache(20 * time.Minute),
+		client: &http.Client{
+			Timeout: 0, // 无超时限制，用于长连接
+		},
+		running: false,
+	}
+}
+
+// Start 启动事件分发器，连接到opencode全局事件流
+func (ed *EventDispatcher) Start(ctx context.Context) error {
+	ed.mu.Lock()
+	if ed.running {
+		ed.mu.Unlock()
+		return fmt.Errorf("event dispatcher already running")
+	}
+
+	// 创建子上下文用于控制SSE连接
+	sseCtx, cancel := context.WithCancel(ctx)
+	ed.cancelFunc = cancel
+	ed.running = true
+	ed.mu.Unlock()
+
+	// 启动SSE连接协程
+	go ed.runSSEConnection(sseCtx)
+
+	logger.Info(fmt.Sprintf("事件分发器已启动 baseURL=%s port=%d",
+		ed.baseURL, ed.port))
+	return nil
+}
+
+// Stop 停止事件分发器
+func (ed *EventDispatcher) Stop() {
+	ed.mu.Lock()
+	if !ed.running {
+		ed.mu.Unlock()
+		return
+	}
+
+	if ed.cancelFunc != nil {
+		ed.cancelFunc()
+	}
+
+	// 清理所有订阅通道
+	for sessionID, channels := range ed.subscriptions {
+		for ch := range channels {
+			close(ch)
+		}
+		delete(ed.subscriptions, sessionID)
+	}
+
+	ed.running = false
+	ed.mu.Unlock()
+
+	logger.Info("事件分发器已停止")
+}
+
+// Subscribe 订阅指定会话的事件
+func (ed *EventDispatcher) Subscribe(sessionID, userID string) (<-chan string, error) {
+	ed.mu.Lock()
+	defer ed.mu.Unlock()
+
+	// 缓存会话-用户映射（用于未来需要用户验证时）
+	ed.sessionUserCache.Set(sessionID, userID)
+
+	// 创建缓冲通道
+	ch := make(chan string, 100)
+
+	// 添加到订阅列表
+	if _, exists := ed.subscriptions[sessionID]; !exists {
+		ed.subscriptions[sessionID] = make(map[chan string]struct{})
+	}
+	ed.subscriptions[sessionID][ch] = struct{}{}
+
+	logger.Debug(fmt.Sprintf("新订阅添加 sessionID=%s userID=%s totalSubscriptions=%d",
+		sessionID, userID, len(ed.subscriptions[sessionID])))
+
+	return ch, nil
+}
+
+// Unsubscribe 取消订阅指定会话的事件
+func (ed *EventDispatcher) Unsubscribe(sessionID string, ch <-chan string) {
+	ed.mu.Lock()
+	defer ed.mu.Unlock()
+
+	if channels, exists := ed.subscriptions[sessionID]; exists {
+		// 遍历查找对应的通道（因为ch是只读通道，无法直接作为key）
+		var foundChan chan string
+		for candidate := range channels {
+			// 比较通道值
+			if candidate == ch {
+				foundChan = candidate
+				break
+			}
+		}
+
+		if foundChan != nil {
+			close(foundChan)
+			delete(channels, foundChan)
+			logger.Debug(fmt.Sprintf("订阅已移除 sessionID=%s remainingSubscriptions=%d",
+				sessionID, len(channels)))
+		}
+
+		// 如果没有订阅者了，清理该会话的映射
+		if len(channels) == 0 {
+			delete(ed.subscriptions, sessionID)
+			ed.sessionUserCache.Delete(sessionID)
+		}
+	}
+}
+
+// RegisterSession 注册会话（前端调用SendPromptStream时调用）
+func (ed *EventDispatcher) RegisterSession(sessionID, userID string) {
+	ed.sessionUserCache.Set(sessionID, userID)
+	logger.Debug(fmt.Sprintf("会话已注册 sessionID=%s userID=%s",
+		sessionID, userID))
+}
+
+// buildSSEURL 构建SSE URL，避免端口重复
+func (ed *EventDispatcher) buildSSEURL() string {
+	// 检查baseURL是否已包含端口
+	base := ed.baseURL
+	// 简单检查：如果baseURL已经包含端口号模式（冒号后跟数字），就不再加端口
+	// 查找最后一个冒号的位置
+	lastColon := -1
+	for i := len(base) - 1; i >= 0; i-- {
+		if base[i] == ':' {
+			lastColon = i
+			break
+		}
+	}
+
+	if lastColon != -1 {
+		// 检查冒号后是否都是数字（端口号）
+		hasPort := true
+		for i := lastColon + 1; i < len(base); i++ {
+			if base[i] < '0' || base[i] > '9' {
+				hasPort = false
+				break
+			}
+		}
+		if hasPort {
+			// baseURL已有端口，直接拼接路径
+			if strings.HasSuffix(base, "/") {
+				return base + "global/event"
+			}
+			return base + "/global/event"
+		}
+	}
+
+	// baseURL没有端口或端口格式不正确，添加端口
+	if strings.HasSuffix(base, "/") {
+		return fmt.Sprintf("%s:%d/global/event", strings.TrimSuffix(base, "/"), ed.port)
+	}
+	return fmt.Sprintf("%s:%d/global/event", base, ed.port)
+}
+
+// runSSEConnection 运行SSE连接，读取全局事件并分发
+func (ed *EventDispatcher) runSSEConnection(ctx context.Context) {
+	// 构建SSE URL，避免重复端口
+	url := ed.buildSSEURL()
+
+	for {
+		select {
+		case <-ctx.Done():
+			logger.Info("SSE连接停止（上下文取消）")
+			return
+		default:
+			// 建立SSE连接
+			logger.Info(fmt.Sprintf("正在连接SSE流 url=%s",
+				url))
+			if err := ed.connectAndProcessSSE(ctx, url); err != nil {
+				logger.Error(fmt.Sprintf("SSE连接失败，5秒后重试 error=%s url=%s",
+					err.Error(), url))
+
+				select {
+				case <-ctx.Done():
+					return
+				case <-time.After(5 * time.Second):
+					continue
+				}
+			}
+		}
+	}
+}
+
+// connectAndProcessSSE 连接并处理SSE流
+func (ed *EventDispatcher) connectAndProcessSSE(ctx context.Context, url string) error {
+	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
+	if err != nil {
+		return fmt.Errorf("创建请求失败: %w", err)
+	}
+	req.Header.Set("Accept", "text/event-stream")
+
+	resp, err := ed.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("发送请求失败: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("SSE请求失败，状态码: %d, 响应: %s", resp.StatusCode, string(body))
+	}
+
+	logger.Info(fmt.Sprintf("SSE连接已建立 url=%s",
+		url))
+
+	reader := bufio.NewReader(resp.Body)
+	eventCount := 0
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				if err == io.EOF {
+					logger.Info(fmt.Sprintf("SSE流正常结束 totalEvents=%d",
+						eventCount))
+				} else if ctx.Err() != nil {
+					logger.Info("SSE流上下文取消")
+				} else {
+					logger.Error(fmt.Sprintf("读取SSE流错误 error=%s",
+						err.Error()))
+				}
+				return err
+			}
+
+			line = strings.TrimSpace(line)
+			if line == "" {
+				continue
+			}
+
+			if strings.HasPrefix(line, "data: ") {
+				data := strings.TrimPrefix(line, "data: ")
+				eventCount++
+
+				// 分发事件
+				ed.dispatchEvent(data)
+
+				if eventCount%100 == 0 {
+					logger.Debug(fmt.Sprintf("事件处理统计 totalEvents=%d activeSessions=%d",
+						eventCount, len(ed.subscriptions)))
+				}
+			}
+		}
+	}
+}
+
+// dispatchEvent 分发事件到相关订阅者
+func (ed *EventDispatcher) dispatchEvent(data string) {
+	// 解析事件数据获取sessionID
+	sessionID := extractSessionIDFromEvent(data)
+	if sessionID == "" {
+		// 没有sessionID的事件（如全局心跳）分发给所有订阅者
+		logger.Debug(fmt.Sprintf("广播全局事件 dataPreview=%s", safeSubstring(data, 0, 100)))
+		ed.broadcastToAll(data)
+		return
+	}
+
+	logger.Debug(fmt.Sprintf("路由事件到会话 sessionID=%s dataPreview=%s",
+		sessionID, safeSubstring(data, 0, 100)))
+
+	// 只分发给订阅该会话的通道
+	ed.mu.RLock()
+	channels, exists := ed.subscriptions[sessionID]
+	ed.mu.RUnlock()
+
+	if !exists {
+		// 没有该会话的订阅者，忽略事件
+		logger.Debug(fmt.Sprintf("忽略事件，无订阅者 sessionID=%s", sessionID))
+		return
+	}
+
+	// 发送事件到所有订阅该会话的通道
+	ed.mu.RLock()
+	for ch := range channels {
+		select {
+		case ch <- data:
+			// 成功发送
+		default:
+			// 通道已满，丢弃事件并记录警告
+			logger.Warn(fmt.Sprintf("事件通道已满，丢弃事件 sessionID=%s",
+				sessionID))
+		}
+	}
+	ed.mu.RUnlock()
+}
+
+// broadcastToAll 广播事件给所有订阅者（用于全局事件如心跳）
+func (ed *EventDispatcher) broadcastToAll(data string) {
+	logger.Debug(fmt.Sprintf("广播事件给所有订阅者 dataPreview=%s", safeSubstring(data, 0, 100)))
+	ed.mu.RLock()
+	defer ed.mu.RUnlock()
+
+	for sessionID, channels := range ed.subscriptions {
+		for ch := range channels {
+			select {
+			case ch <- data:
+				// 成功发送
+			default:
+				// 通道已满，丢弃事件
+				logger.Warn(fmt.Sprintf("事件通道已满，丢弃全局事件 sessionID=%s",
+					sessionID))
+			}
+		}
+	}
+}
+
+// extractSessionIDFromEvent 从事件数据中提取sessionID
+func extractSessionIDFromEvent(data string) string {
+	// 尝试解析为JSON
+	var eventMap map[string]interface{}
+	if err := json.Unmarshal([]byte(data), &eventMap); err != nil {
+		logger.Debug(fmt.Sprintf("无法解析事件JSON error=%s dataPreview=%s",
+			err.Error(), safeSubstring(data, 0, 200)))
+		return ""
+	}
+
+	// 添加调试日志，显示完整事件结构（仅调试时启用）
+	debugMode := false
+	if debugMode {
+		eventJSON, _ := json.MarshalIndent(eventMap, "", "  ")
+		logger.Debug(fmt.Sprintf("事件数据结构 eventStructure=%s",
+			string(eventJSON)))
+	}
+
+	// 递归查找sessionID字段
+	sessionID := findSessionIDRecursive(eventMap)
+
+	if sessionID == "" {
+		logger.Debug(fmt.Sprintf("未找到sessionID字段 eventType=%s dataPreview=%s",
+			getEventType(eventMap), safeSubstring(data, 0, 100)))
+	} else {
+		logger.Debug(fmt.Sprintf("成功提取sessionID sessionID=%s eventType=%s",
+			sessionID, getEventType(eventMap)))
+	}
+
+	return sessionID
+}
+
+// findSessionIDRecursive 递归查找sessionID字段
+func findSessionIDRecursive(data interface{}) string {
+	switch v := data.(type) {
+	case map[string]interface{}:
+		// 检查当前层级的sessionID字段（支持多种命名变体）
+		for _, key := range []string{"sessionID", "session_id", "sessionId"} {
+			if val, ok := v[key]; ok {
+				if str, ok := val.(string); ok && str != "" {
+					return str
+				}
+			}
+		}
+
+		// 检查常见嵌套路径
+		// 1. payload.properties.sessionID (session.status事件)
+		if payload, ok := v["payload"].(map[string]interface{}); ok {
+			if props, ok := payload["properties"].(map[string]interface{}); ok {
+				if sessionID, ok := props["sessionID"].(string); ok && sessionID != "" {
+					return sessionID
+				}
+			}
+		}
+
+		// 2. payload.properties.part.sessionID (message.part.updated事件)
+		if payload, ok := v["payload"].(map[string]interface{}); ok {
+			if props, ok := payload["properties"].(map[string]interface{}); ok {
+				if part, ok := props["part"].(map[string]interface{}); ok {
+					if sessionID, ok := part["sessionID"].(string); ok && sessionID != "" {
+						return sessionID
+					}
+				}
+			}
+		}
+
+		// 3. payload.properties.info.sessionID (message.updated事件)
+		if payload, ok := v["payload"].(map[string]interface{}); ok {
+			if props, ok := payload["properties"].(map[string]interface{}); ok {
+				if info, ok := props["info"].(map[string]interface{}); ok {
+					if sessionID, ok := info["sessionID"].(string); ok && sessionID != "" {
+						return sessionID
+					}
+				}
+			}
+		}
+
+		// 递归遍历所有值
+		for _, value := range v {
+			if result := findSessionIDRecursive(value); result != "" {
+				return result
+			}
+		}
+
+	case []interface{}:
+		// 遍历数组
+		for _, item := range v {
+			if result := findSessionIDRecursive(item); result != "" {
+				return result
+			}
+		}
+	}
+
+	return ""
+}
+
+// getEventType 获取事件类型
+func getEventType(eventMap map[string]interface{}) string {
+	if payload, ok := eventMap["payload"].(map[string]interface{}); ok {
+		if eventType, ok := payload["type"].(string); ok {
+			return eventType
+		}
+	}
+	return "unknown"
+}
+
+// safeSubstring 安全的子字符串函数
+func safeSubstring(s string, start, length int) string {
+	if start < 0 {
+		start = 0
+	}
+	if start >= len(s) {
+		return ""
+	}
+	end := start + length
+	if end > len(s) {
+		end = len(s)
+	}
+	return s[start:end]
+}
+
+// GetInstance 获取单例实例（线程安全）
+var (
+	instance     *EventDispatcher
+	instanceOnce sync.Once
+)
+
+// GetEventDispatcher 获取事件分发器单例
+func GetEventDispatcher(baseURL string, port int) *EventDispatcher {
+	instanceOnce.Do(func() {
+		instance = NewEventDispatcher(baseURL, port)
+	})
+	return instance
+}

internal/service/menu/get_menu.go

@@ -0,0 +1,59 @@
+package menu
+
+import (
+	"context"
+
+	"git.x2erp.com/qdy/go-base/ctx"
+	"git.x2erp.com/qdy/go-base/logger"
+	"git.x2erp.com/qdy/go-base/model/response"
+	"git.x2erp.com/qdy/go-base/util"
+)
+
+// MenuItem 顶部菜单项结构
+type MenuItem struct {
+	ID    string `json:"id"`
+	Name  string `json:"name"`
+	Icon  string `json:"icon,omitempty"`
+	Type  string `json:"type"` // top-menu
+	Route string `json:"route,omitempty"`
+}
+
+// GetTopMenu 获取顶部菜单
+func GetTopMenu(ctx context.Context, reqCtx *ctx.RequestContext) *response.QueryResult[[]MenuItem] {
+	logger.Debug("GetTopMenu-开始获取顶部菜单")
+
+	// 构建顶部菜单项
+	menu := []MenuItem{
+		{
+			ID:    "replenish",
+			Name:  "补货",
+			Icon:  "inventory_2", // 库存补充图标
+			Type:  "top-menu",
+			Route: "/replenish",
+		},
+		{
+			ID:    "transfer",
+			Name:  "调拨",
+			Icon:  "swap_horiz", // 横向交换图标
+			Type:  "top-menu",
+			Route: "/transfer",
+		},
+		{
+			ID:    "allocation",
+			Name:  "配货",
+			Icon:  "local_shipping", // 本地货运图标
+			Type:  "top-menu",
+			Route: "/allocation",
+		},
+		{
+			ID:    "report",
+			Name:  "报表",
+			Icon:  "assessment", // 评估报表图标
+			Type:  "top-menu",
+			Route: "/report",
+		},
+	}
+
+	logger.Debug("返回顶部菜单结构")
+	return util.CreateSuccessResultData(menu, reqCtx)
+}

internal/service/menu/mapping.go

@@ -0,0 +1,150 @@
+package menu
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"git.x2erp.com/qdy/go-base/logger"
+	"git.x2erp.com/qdy/go-db/factory/mongodb"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+// SessionMenuMapping 会话-菜单映射结构
+type SessionMenuMapping struct {
+	ID         primitive.ObjectID `bson:"_id,omitempty" json:"id,omitempty"`
+	SessionID  string             `bson:"session_id" json:"session_id"`     // 会话ID
+	MenuItemID string             `bson:"menu_item_id" json:"menu_item_id"` // 菜单项ID
+	UserID     string             `bson:"user_id" json:"user_id"`           // 用户ID
+	TenantID   string             `bson:"tenant_id" json:"tenant_id"`       // 租户ID
+	CreatedAt  time.Time          `bson:"created_at" json:"created_at"`     // 创建时间
+}
+
+// CollectionName 返回MongoDB集合名称
+func (SessionMenuMapping) CollectionName() string {
+	return "session_menu_mappings"
+}
+
+// MappingService 会话-菜单映射服务
+type MappingService struct {
+	mongoFactory *mongodb.MongoDBFactory
+}
+
+// NewMappingService 创建新的映射服务
+func NewMappingService(factory *mongodb.MongoDBFactory) *MappingService {
+	return &MappingService{mongoFactory: factory}
+}
+
+// CreateMapping 创建会话-菜单映射
+func (s *MappingService) CreateMapping(ctx context.Context, sessionID, menuItemID, userID, tenantID string) error {
+	mapping := SessionMenuMapping{
+		SessionID:  sessionID,
+		MenuItemID: menuItemID,
+		UserID:     userID,
+		TenantID:   tenantID,
+		CreatedAt:  time.Now(),
+	}
+
+	_, success := s.mongoFactory.InsertOneWithResult(SessionMenuMapping{}.CollectionName(), mapping)
+	if !success {
+		logger.Error("创建会话-菜单映射失败", "session_id", sessionID, "menu_item_id", menuItemID)
+		return errors.New("failed to create session-menu mapping")
+	}
+
+	logger.Debug("创建会话-菜单映射成功", "session_id", sessionID, "menu_item_id", menuItemID)
+	return nil
+}
+
+// GetMenuItemBySessionID 根据会话ID获取菜单项ID
+func (s *MappingService) GetMenuItemBySessionID(ctx context.Context, sessionID string) (string, error) {
+	filter := bson.M{"session_id": sessionID}
+	var mapping SessionMenuMapping
+
+	err := s.mongoFactory.FindOne(SessionMenuMapping{}.CollectionName(), filter, &mapping)
+	if err != nil {
+		if err.Error() == "mongo: no documents in result" {
+			return "", nil // 未找到映射，返回空字符串
+		}
+		logger.Error("根据会话ID查询菜单项失败", "session_id", sessionID, "error", err)
+		return "", err
+	}
+
+	return mapping.MenuItemID, nil
+}
+
+// GetSessionIDsByMenuItemID 根据菜单项ID获取会话ID列表
+func (s *MappingService) GetSessionIDsByMenuItemID(ctx context.Context, menuItemID, userID string) ([]string, error) {
+	filter := bson.M{"menu_item_id": menuItemID, "user_id": userID}
+	var mappings []SessionMenuMapping
+
+	err := s.mongoFactory.Find(SessionMenuMapping{}.CollectionName(), filter, &mappings)
+	if err != nil {
+		logger.Error("根据菜单项ID查询会话ID失败", "menu_item_id", menuItemID, "error", err)
+		return nil, err
+	}
+
+	var sessionIDs []string
+	for _, mapping := range mappings {
+		sessionIDs = append(sessionIDs, mapping.SessionID)
+	}
+
+	return sessionIDs, nil
+}
+
+// GetMappingsByUser 获取用户的所有会话-菜单映射
+func (s *MappingService) GetMappingsByUser(ctx context.Context, userID string) (map[string]string, error) {
+	filter := bson.M{"user_id": userID}
+	var mappings []SessionMenuMapping
+
+	err := s.mongoFactory.Find(SessionMenuMapping{}.CollectionName(), filter, &mappings)
+	if err != nil {
+		logger.Error("查询用户会话-菜单映射失败", "user_id", userID, "error", err)
+		return nil, err
+	}
+
+	result := make(map[string]string)
+	for _, mapping := range mappings {
+		result[mapping.SessionID] = mapping.MenuItemID
+	}
+
+	return result, nil
+}
+
+// DeleteMappingBySessionID 根据会话ID删除映射
+func (s *MappingService) DeleteMappingBySessionID(ctx context.Context, sessionID string) error {
+	filter := bson.M{"session_id": sessionID}
+	success, deletedCount := s.mongoFactory.DeleteOne(SessionMenuMapping{}.CollectionName(), filter)
+	if !success {
+		logger.Error("根据会话ID删除映射失败", "session_id", sessionID)
+		return errors.New("failed to delete session-menu mapping")
+	}
+
+	logger.Debug("删除会话-菜单映射成功", "session_id", sessionID, "deleted_count", deletedCount)
+	return nil
+}
+
+// EnsureIndexes 确保集合索引
+func (s *MappingService) EnsureIndexes(ctx context.Context) error {
+	// 创建会话ID唯一索引
+	sessionIDIndexKeys := bson.D{{Key: "session_id", Value: 1}}
+	sessionIDSuccess := s.mongoFactory.CreateIndex(SessionMenuMapping{}.CollectionName(), sessionIDIndexKeys)
+	if !sessionIDSuccess {
+		logger.Error("创建会话ID索引失败")
+		return errors.New("failed to create session_id index")
+	}
+
+	// 创建复合索引：菜单项ID + 用户ID
+	menuItemUserIndexKeys := bson.D{
+		{Key: "menu_item_id", Value: 1},
+		{Key: "user_id", Value: 1},
+	}
+	menuItemUserSuccess := s.mongoFactory.CreateIndex(SessionMenuMapping{}.CollectionName(), menuItemUserIndexKeys)
+	if !menuItemUserSuccess {
+		logger.Error("创建菜单项用户复合索引失败")
+		return errors.New("failed to create menu_item_id-user_id index")
+	}
+
+	logger.Debug("会话-菜单映射索引创建成功")
+	return nil
+}

main.go

@@ -1,8 +1,8 @@
 package main
 
 import (
+	"context"
 	"log"
-	"sync"
 
 	"git.x2erp.com/qdy/go-base/config"
 	"git.x2erp.com/qdy/go-base/container"
@@ -12,9 +12,12 @@ import (
 	"git.x2erp.com/qdy/go-base/webx"
 	"git.x2erp.com/qdy/go-base/webx/health"
 	"git.x2erp.com/qdy/go-base/webx/router"
+	"git.x2erp.com/qdy/go-db/factory/mongodb"
 
 	"git.x2erp.com/qdy/go-svc-code/internal/opencode"
 	"git.x2erp.com/qdy/go-svc-code/internal/routes"
+	"git.x2erp.com/qdy/go-svc-code/internal/service/event"
+	"git.x2erp.com/qdy/go-svc-code/internal/service/menu"
 )
 
 var (
@@ -24,9 +27,9 @@ var (
 
 // 全局变量存储 opencode 进程信息
 var (
-	opencodeProcess *opencode.Process
-	opencodePort    int
-	opencodeMutex   sync.Mutex
+	//opencodeProcess *opencode.Process
+	opencodePort int
+	//opencodeMutex   sync.Mutex
 )
 
 func main() {
@@ -47,6 +50,20 @@ func main() {
 	// 3. 启用运行日志（需要在启动 opencode 服务之前）
 	container.Create(ctr, logger.InitRuntimeLogger)
 
+	// 创建mongodb
+	mongoDBFactory := container.Create(ctr, mongodb.CreateFactory)
+	mongoDBFactory.TestConnection()
+
+	// 创建会话-菜单映射服务
+	mappingService := menu.NewMappingService(mongoDBFactory)
+	// 确保索引存在
+	ctx := context.Background()
+	if err := mappingService.EnsureIndexes(ctx); err != nil {
+		log.Printf("警告：创建会话-菜单映射索引失败: %v", err)
+	} else {
+		log.Printf("会话-菜单映射索引确保成功")
+	}
+
 	// 4. 创建 configure 客户端
 	configClient, err := configure.NewClient()
 	if err != nil {
@@ -61,6 +78,15 @@ func main() {
 	}
 	log.Printf("opencode 客户端已创建，连接端口: %d", opencodePort)
 
+	// 启动事件分发器（用于多用户流式对话隔离）
+	dispatcher := event.GetEventDispatcher(client.GetBaseURL(), client.GetPort())
+	if err := dispatcher.Start(context.Background()); err != nil {
+		log.Printf("警告：事件分发器启动失败（流式功能可能受影响）: %v", err)
+	} else {
+		log.Printf("事件分发器已启动")
+		defer dispatcher.Stop()
+	}
+
 	// 6. 得到 webservice 服务工厂
 	webxFactory := webx.GetWebServiceFactory()
 
@@ -77,7 +103,7 @@ func main() {
 	health.RegisterConsulHealthCheck(routerService)
 
 	// 9. 注册路由--api
-	registerRoutes(routerService, webService, configClient, client, nil, opencodePort)
+	registerRoutes(routerService, webService, configClient, client, nil, opencodePort, mappingService)
 
 	// 9. 注册前端静态文件服务
 	//registerStaticFiles(webService)
@@ -93,12 +119,12 @@ func main() {
 }
 
 // registerRoutes 注册所有 API 路由
-func registerRoutes(ws *router.RouterService, webService *webx.WebService, configClient *configure.Client, client opencode.OpenCodeClient, opencodeProcess *opencode.Process, opencodePort int) {
+func registerRoutes(ws *router.RouterService, webService *webx.WebService, configClient *configure.Client, client opencode.OpenCodeClient, opencodeProcess *opencode.Process, opencodePort int, mappingService *menu.MappingService) {
 	// 认证路由（公开登录接口）
 	routes.RegisterAuthRoutes(ws, configClient)
 
 	// 会话管理路由
-	routes.RegisterSessionRoutes(ws, client)
+	routes.RegisterSessionRoutes(ws, client, mappingService)
 
 	// 同步对话路由
 	routes.RegisterPromptSyncRoutes(ws, client)
@@ -108,4 +134,7 @@ func registerRoutes(ws *router.RouterService, webService *webx.WebService, confi
 
 	// 日志流路由（需要直接 HTTP 处理器）
 	routes.RegisterLogStreamRoutes(ws, webService, opencodeProcess, opencodePort)
+
+	// 菜单路由
+	routes.RegisterMenuRoutes(ws, mappingService)
 }