diff --git a/internal/chat/session.go b/internal/chat/session.go index b6b38140..85b99276 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -7,6 +7,7 @@ import ( "sync" "time" + "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/coordinator" "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/intent" @@ -50,6 +51,7 @@ type Session struct { Recognizer intent.Recognizer `json:"-"` Coordinator *coordinator.CoordinatorEngine `json:"-"` CtxMgr *ctxmgr.ContextManager `json:"-"` + ChatLog chatflow.ChatLog `json:"-"` Actions *ActionServices `json:"-"` Messages []Message `json:"messages"` CreatedAt time.Time `json:"created_at"` @@ -211,7 +213,19 @@ func (s *Session) syncRecalledMemories(ctx context.Context, userInput string, re TopK: 3, MinScore: 0.20, } - items, err := s.CtxMgr.SearchChatMemory(ctx, opts) + var items []ctxmgr.RecalledMemoryItem + var err error + if s.ChatLog != nil { + items, err = s.ChatLog.SearchMemory(ctx, chatflow.Principal{UserID: s.UserID, Action: chatflow.ActionAssemble}, chatflow.MemoryQuery{ + Query: userInput, + ProjectID: s.ProjectID, + ExcludeSessionID: s.ID, + TopK: 3, + MinScore: 0.20, + }) + } else { + items, err = s.CtxMgr.SearchChatMemory(ctx, opts) + } if err == nil && len(items) > 0 { s.UserCtx.RecalledMemories = items s.UserCtx.RecalledMemorySnippet = ctxmgr.FormatRecalledMemorySnippet(items, s.ID) diff --git a/internal/server/chat_digest_api.go b/internal/server/chat_digest_api.go index 5ef7e1ed..9be10a46 100644 --- a/internal/server/chat_digest_api.go +++ b/internal/server/chat_digest_api.go @@ -5,6 +5,7 @@ import ( "net/http" "strings" + "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/ctxmgr" ) @@ -16,7 +17,7 @@ func (s *Server) handleListDailyDigests(w http.ResponseWriter, r *http.Request) return } - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } @@ -32,16 +33,12 @@ func (s *Server) handleListDailyDigests(w http.ResponseWriter, r *http.Request) return } - if _, ok := s.authorizeSessionRead(w, r, sessionID, user); !ok { - return - } - from := strings.TrimSpace(r.URL.Query().Get("from")) to := strings.TrimSpace(r.URL.Query().Get("to")) - digests, err := s.svc.ContextManager().ListDailyDigests(r.Context(), sessionID, from, to) + digests, err := s.chatLog.ListDigests(r.Context(), s.chatPrincipal(user, chatflow.ActionDigest), sessionID, from, to) if err != nil { - http.Error(w, "ListDailyDigests error: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } @@ -67,7 +64,7 @@ func (s *Server) handleGenerateDailyDigest(w http.ResponseWriter, r *http.Reques return } - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } @@ -91,13 +88,9 @@ func (s *Server) handleGenerateDailyDigest(w http.ResponseWriter, r *http.Reques return } - if _, ok := s.authorizeSessionRead(w, r, req.SessionID, user); !ok { - return - } - - digest, err := s.svc.ContextManager().GenerateDailyDigest(r.Context(), req.SessionID, req.DayKey) + digest, err := s.chatLog.GenerateDigest(r.Context(), s.chatPrincipal(user, chatflow.ActionDigest), req.SessionID, req.DayKey) if err != nil { - http.Error(w, "GenerateDailyDigest error: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } diff --git a/internal/server/chat_export_api.go b/internal/server/chat_export_api.go index 7508553f..aee37b97 100644 --- a/internal/server/chat_export_api.go +++ b/internal/server/chat_export_api.go @@ -10,6 +10,7 @@ import ( "net/http" "strings" + "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/replyexport" ) @@ -31,43 +32,24 @@ func (s *Server) handleChatExportMessages(w http.ResponseWriter, r *http.Request return } - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } - if !s.assertSessionOwnedByUser(w, r, sessionID, user.ID) { - return - } - dateFrom := strings.TrimSpace(r.URL.Query().Get("date_from")) - dateTo := strings.TrimSpace(r.URL.Query().Get("date_to")) - category := strings.TrimSpace(r.URL.Query().Get("category")) - - // 查询该会话的历史消息（上限 5000 条） - msgs, err := s.svc.ContextManager().GetChatHistory(r.Context(), sessionID, 5000, 0) + msgs, err := s.chatLog.Export(r.Context(), s.chatPrincipal(user, chatflow.ActionExport), chatflow.ExportQuery{ + SessionID: sessionID, + DateFrom: strings.TrimSpace(r.URL.Query().Get("date_from")), + DateTo: strings.TrimSpace(r.URL.Query().Get("date_to")), + Category: strings.TrimSpace(r.URL.Query().Get("category")), + }) if err != nil { - http.Error(w, "query chat history failed: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } - // 过滤 + 导出策略脱敏 filtered := make([]ctxmgr.ChatMessageRecord, 0, len(msgs)) for _, m := range msgs { - dk := m.DayKey - if dk == "" && !m.CreatedAt.IsZero() { - dk = m.CreatedAt.Format("2006-01-02") - } - - if dateFrom != "" && dk < dateFrom { - continue - } - if dateTo != "" && dk > dateTo { - continue - } - if category != "" && category != "all" && m.Category != category { - continue - } - hasCard := false if m.Metadata != nil { if _, exists := m.Metadata["interactive_card"]; exists { @@ -84,7 +66,6 @@ func (s *Server) handleChatExportMessages(w http.ResponseWriter, r *http.Request if cap.Policy == replyexport.PolicyNone { continue } - filtered = append(filtered, m) } diff --git a/internal/server/chat_history_api_test.go b/internal/server/chat_history_api_test.go index fed6ef2d..e7d5d6b4 100644 --- a/internal/server/chat_history_api_test.go +++ b/internal/server/chat_history_api_test.go @@ -6,6 +6,7 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "strconv" "testing" "time" @@ -25,17 +26,24 @@ func TestServer_ChatHistoryAndCaseRecommendationAPIs(t *testing.T) { svc := bizdev.MustNewService(t, cfg) srv := server.NewServer(cfg, svc) handler := srv.Routes() + token := loginTestAdmin(t, handler, cfg) + adminID := fetchAuthUserID(t, handler, token) ctx := context.Background() sessID := "sess_api_history_001" - userID := int64(108) + userID := adminID + + if _, err := svc.ContextManager().EnsureChatSession(ctx, sessID, userID, 0, "历史会话"); err != nil { + t.Fatalf("ensure session: %v", err) + } // 1. 模拟写入对话消息 _ = svc.ContextManager().RecordUserMessage(ctx, sessID, userID, "user", "我们准备立项一款便携无线掌上超声，预算大约150万", "SUPPLEMENT_INFO", "EXPECTATION_HIT", nil) _ = svc.ContextManager().RecordUserMessage(ctx, sessID, 0, "assistant", "已为您建立超声设备立项基准，预估研发周期6个月", "ADVANCE_STAGE", "EXPECTATION_HIT", nil) // 2. 测试 GET /api/v1/chat/sessions - reqSessions := httptest.NewRequest("GET", "/api/v1/chat/sessions?user_id=108", nil) + reqSessions := httptest.NewRequest("GET", "/api/v1/chat/sessions?user_id="+strconv.FormatInt(userID, 10), nil) + withBearerAuth(reqSessions, token) rrSessions := httptest.NewRecorder() handler.ServeHTTP(rrSessions, reqSessions) @@ -61,6 +69,7 @@ func TestServer_ChatHistoryAndCaseRecommendationAPIs(t *testing.T) { }) reqUpsert := httptest.NewRequest("POST", "/api/v1/chat/sessions", bytes.NewReader(upsertBody)) reqUpsert.Header.Set("Content-Type", "application/json") + withBearerAuth(reqUpsert, token) rrUpsert := httptest.NewRecorder() handler.ServeHTTP(rrUpsert, reqUpsert) if rrUpsert.Code != http.StatusOK { @@ -76,6 +85,7 @@ func TestServer_ChatHistoryAndCaseRecommendationAPIs(t *testing.T) { // 3. 测试 GET /api/v1/chat/messages reqMsgs := httptest.NewRequest("GET", "/api/v1/chat/messages?session_id="+sessID, nil) + withBearerAuth(reqMsgs, token) rrMsgs := httptest.NewRecorder() handler.ServeHTTP(rrMsgs, reqMsgs) diff --git a/internal/server/chat_search_api.go b/internal/server/chat_search_api.go index a8a4fdaa..5523995b 100644 --- a/internal/server/chat_search_api.go +++ b/internal/server/chat_search_api.go @@ -6,6 +6,7 @@ import ( "strconv" "strings" + "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/ctxmgr" ) @@ -22,7 +23,7 @@ func (s *Server) handleChatSearch(w http.ResponseWriter, r *http.Request) { return } - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } @@ -33,11 +34,6 @@ func (s *Server) handleChatSearch(w http.ResponseWriter, r *http.Request) { return } - authCtx, ok := s.authorizeSessionRead(w, r, sessionID, user) - if !ok { - return - } - query := strings.TrimSpace(r.URL.Query().Get("q")) if query == "" { http.Error(w, "missing or empty query param 'q'", http.StatusBadRequest) @@ -56,25 +52,20 @@ func (s *Server) handleChatSearch(w http.ResponseWriter, r *http.Request) { } } - searchQuery := ctxmgr.ChatSearchQuery{ + hits, err := s.chatLog.Search(r.Context(), s.chatPrincipal(user, chatflow.ActionSearch), ctxmgr.ChatSearchQuery{ SessionID: sessionID, - UserID: user.ID, Query: query, Mode: mode, Category: category, DateFrom: dateFrom, DateTo: dateTo, Limit: limit, - } - - hits, err := s.svc.ContextManager().SearchChatMessages(r.Context(), searchQuery) + }) if err != nil { - http.Error(w, "SearchChatMessages error: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } - hits = s.filterChatSearchHitsByAuth(authCtx, hits) - if hits == nil { hits = []ctxmgr.ChatSearchHit{} } diff --git a/internal/server/chat_unread_and_export_api_test.go b/internal/server/chat_unread_and_export_api_test.go index 00fa0e7b..9ee042a0 100644 --- a/internal/server/chat_unread_and_export_api_test.go +++ b/internal/server/chat_unread_and_export_api_test.go @@ -43,27 +43,27 @@ func TestServer_ChatUnreadAndExportAPIs(t *testing.T) { }) // 消息 2: assistant, day 1, category=casual id2, _ := ctxMgr.RecordUserMessageWithID(ctx, sessID, 0, "assistant", "消息2：收到需求，开始规划。以下为初步里程碑拆解与责任分工建议，供项目组评审。", "", "", map[string]any{ - "day_key": dayKey1, + "day_key": dayKey1, "category": "casual", }) // 消息 3: user, day 2, category=review (DOCGOV) _, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, 1, "user", "消息3：关于型检机构认证进度", "DOCGOV_REVIEW", "", map[string]any{ - "day_key": dayKey2, + "day_key": dayKey2, "category": "review", }) // 消息 4: assistant, day 2, category=review — 可导出正文 _, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, 0, "assistant", "消息4：关于型检机构认证进度，已整理成以下要点：1) 受理号已确认；2) 补正材料清单如下；3) 预计复核周期 15 个工作日。", "DOCGOV_REVIEW", "", map[string]any{ - "day_key": dayKey2, + "day_key": dayKey2, "category": "review", }) // 消息 5: user, day 2 _, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, 1, "user", "消息5：另外请补充EMC整改报告", "DOCGOV_AUDIT", "", map[string]any{ - "day_key": dayKey2, + "day_key": dayKey2, "category": "review", }) t.Run("GET /api/v1/chat/unread 统计 since_message_id 之后的未读", func(t *testing.T) { - // since_message_id=id2, 应统计 id3 与 id4 + // since_message_id=id2：当前登录用户自己的 user 消息不计入未读，仅 assistant 消息 4 未读 req := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/chat/unread?session_id=%s&since_message_id=%d", sessID, id2), nil) withBearerAuth(req, token) w := httptest.NewRecorder() @@ -78,8 +78,8 @@ func TestServer_ChatUnreadAndExportAPIs(t *testing.T) { t.Fatalf("decode unread summary failed: %v", err) } - if summary.Total != 3 { - t.Errorf("expected total 3, got %d", summary.Total) + if summary.Total != 1 { + t.Errorf("expected total 1, got %d", summary.Total) } if len(summary.Items) != 1 { t.Fatalf("expected 1 unread day bucket, got %d", len(summary.Items)) @@ -87,8 +87,8 @@ func TestServer_ChatUnreadAndExportAPIs(t *testing.T) { if summary.Items[0].Date != dayKey2 { t.Errorf("expected date %s, got %s", dayKey2, summary.Items[0].Date) } - if summary.Items[0].Count != 3 { - t.Errorf("expected count 3, got %d", summary.Items[0].Count) + if summary.Items[0].Count != 1 { + t.Errorf("expected count 1, got %d", summary.Items[0].Count) } if summary.NextMessageID == 0 { t.Errorf("expected NextMessageID > 0, got %d", summary.NextMessageID) @@ -114,8 +114,8 @@ func TestServer_ChatUnreadAndExportAPIs(t *testing.T) { var summary2 ctxmgr.ChatUnreadSummary _ = json.NewDecoder(w2.Body).Decode(&summary2) - if summary2.Total != 2 { - t.Errorf("expected total 2 after consuming one unread, got %d", summary2.Total) + if summary2.Total != 0 { + t.Errorf("expected total 0 after consuming the assistant unread, got %d", summary2.Total) } }) diff --git a/internal/server/chat_unread_api.go b/internal/server/chat_unread_api.go index e3f289cc..a7b4696a 100644 --- a/internal/server/chat_unread_api.go +++ b/internal/server/chat_unread_api.go @@ -2,14 +2,17 @@ package server // chat_unread_api.go — P3 / Step 2 双轨制未读消息与位点推进 API // -// GET /api/v1/chat/unread?session_id=...&user_id=...&since_message_id=... +// GET /api/v1/chat/unread?session_id=...&since_message_id=... // POST /api/v1/chat/read +// +// user_id 只取自登录 Principal，忽略 query/body 中的身份字段。 import ( "encoding/json" "net/http" "strconv" + "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/ctxmgr" ) @@ -25,18 +28,6 @@ func (s *Server) handleChatUnread(w http.ResponseWriter, r *http.Request) { return } - authCtx, ok := s.authorizeSessionRead(w, r, sessionID, user) - if !ok { - return - } - - var userID int64 - if uStr := r.URL.Query().Get("user_id"); uStr != "" { - if uid, err := strconv.ParseInt(uStr, 10, 64); err == nil { - userID = uid - } - } - var sinceID int64 if sinceStr := r.URL.Query().Get("since_message_id"); sinceStr != "" { if id, err := strconv.ParseInt(sinceStr, 10, 64); err == nil { @@ -44,7 +35,7 @@ func (s *Server) handleChatUnread(w http.ResponseWriter, r *http.Request) { } } - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { writeJSONResponse(w, http.StatusOK, ctxmgr.ChatUnreadSummary{ Total: 0, NextMessageID: 0, @@ -54,32 +45,18 @@ func (s *Server) handleChatUnread(w http.ResponseWriter, r *http.Request) { return } - summary, err := s.svc.ContextManager().GetChatUnreadSummary(r.Context(), sessionID, userID, sinceID) + summary, err := s.chatLog.Unread(r.Context(), s.chatPrincipal(user, chatflow.ActionUnread), chatflow.UnreadQuery{ + SessionID: sessionID, + SinceMessageID: sinceID, + }) if err != nil { - writeJSONResponse(w, http.StatusOK, ctxmgr.ChatUnreadSummary{ - Total: 0, - NextMessageID: 0, - UnreadMessageIDs: []int64{}, - Items: []ctxmgr.UnreadDayItem{}, - }) + s.writeChatLogError(w, err) return } - - if authCtx.StageAccess != nil && authCtx.StageAccess.Restricted { - visible, err := s.listVisibleSessionMessages(r.Context(), authCtx) - if err == nil { - allowed := make(map[int64]bool, len(visible)) - for _, msg := range visible { - allowed[msg.ID] = true - } - filteredIDs := make([]int64, 0, len(summary.UnreadMessageIDs)) - for _, id := range summary.UnreadMessageIDs { - if allowed[id] { - filteredIDs = append(filteredIDs, id) - } - } - summary.UnreadMessageIDs = filteredIDs - summary.Total = len(filteredIDs) + if summary == nil { + summary = &ctxmgr.ChatUnreadSummary{ + UnreadMessageIDs: []int64{}, + Items: []ctxmgr.UnreadDayItem{}, } } @@ -107,18 +84,14 @@ func (s *Server) handleChatRead(w http.ResponseWriter, r *http.Request) { return } - if _, ok := s.authorizeSessionRead(w, r, req.SessionID, user); !ok { - return - } - - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusInternalServerError) return } - pos, err := s.svc.ContextManager().MarkChatRead(r.Context(), req) + pos, err := s.chatLog.MarkRead(r.Context(), s.chatPrincipal(user, chatflow.ActionUnread), req) if err != nil { - http.Error(w, "mark chat read failed: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } diff --git a/internal/server/chat_window_api.go b/internal/server/chat_window_api.go index fc94bda3..fb3af997 100644 --- a/internal/server/chat_window_api.go +++ b/internal/server/chat_window_api.go @@ -6,13 +6,14 @@ import ( "os" "strconv" + "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/ctxmgr" ) // handleGetChatWindow 获取指定会话最近 N 日的消息窗口 // GET /api/v1/chat/messages/window?session_id=&days=&before_day= func (s *Server) handleGetChatWindow(w http.ResponseWriter, r *http.Request) { - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } @@ -28,37 +29,29 @@ func (s *Server) handleGetChatWindow(w http.ResponseWriter, r *http.Request) { return } - authCtx, ok := s.authorizeSessionRead(w, r, sessionID, user) - if !ok { - return - } - daysStr := r.URL.Query().Get("days") days, _ := strconv.Atoi(daysStr) if days <= 0 { days = ctxmgr.DefaultWindowDays } - beforeDay := r.URL.Query().Get("before_day") - res, err := s.svc.ContextManager().GetChatWindow(r.Context(), ctxmgr.ChatWindowQuery{ + res, err := s.chatLog.GetWindow(r.Context(), s.chatPrincipal(user, chatflow.ActionReadWindow), ctxmgr.ChatWindowQuery{ SessionID: sessionID, Days: days, - BeforeDay: beforeDay, + BeforeDay: r.URL.Query().Get("before_day"), }) if err != nil { - http.Error(w, "get chat window failed: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } - s.applyChatWindowAuthFilter(authCtx, res) - writeJSONResponse(w, http.StatusOK, res) } // handleGetChatDayMessages 获取指定日消息 (支持 anchor_id 上下文补齐或游标分页) // GET /api/v1/chat/messages/day?session_id=&date=&anchor_id=&before_id=&after_id=&limit= func (s *Server) handleGetChatDayMessages(w http.ResponseWriter, r *http.Request) { - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } @@ -74,11 +67,6 @@ func (s *Server) handleGetChatDayMessages(w http.ResponseWriter, r *http.Request return } - authCtx, ok := s.authorizeSessionRead(w, r, sessionID, user) - if !ok { - return - } - date := r.URL.Query().Get("date") if date == "" { date = r.URL.Query().Get("day_key") @@ -92,35 +80,33 @@ func (s *Server) handleGetChatDayMessages(w http.ResponseWriter, r *http.Request return } - beforeIDStr := r.URL.Query().Get("before_id") - beforeID, _ := strconv.ParseInt(beforeIDStr, 10, 64) - - afterIDStr := r.URL.Query().Get("after_id") - afterID, _ := strconv.ParseInt(afterIDStr, 10, 64) - - limitStr := r.URL.Query().Get("limit") - limit, _ := strconv.Atoi(limitStr) + beforeID, _ := strconv.ParseInt(r.URL.Query().Get("before_id"), 10, 64) + afterID, _ := strconv.ParseInt(r.URL.Query().Get("after_id"), 10, 64) + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) if limit <= 0 { limit = 100 } - res, err := s.svc.ContextManager().GetChatDayMessages(r.Context(), sessionID, date, anchorID, beforeID, afterID, limit) + res, err := s.chatLog.GetDay(r.Context(), s.chatPrincipal(user, chatflow.ActionReadDay), chatflow.DayQuery{ + SessionID: sessionID, + Date: date, + AnchorID: anchorID, + BeforeID: beforeID, + AfterID: afterID, + Limit: limit, + }) if err != nil { - http.Error(w, "get chat day messages failed: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } - res.Messages = s.filterChatMessagesByAuth(authCtx, res.Messages) - res.LoadedCount = len(res.Messages) - res.TotalCount = len(res.Messages) - writeJSONResponse(w, http.StatusOK, res) } // handleListChatDates 列出会话有消息的全部日期 (用于日历打点) // GET /api/v1/chat/dates?session_id= func (s *Server) handleListChatDates(w http.ResponseWriter, r *http.Request) { - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } @@ -136,24 +122,19 @@ func (s *Server) handleListChatDates(w http.ResponseWriter, r *http.Request) { return } - authCtx, ok := s.authorizeSessionRead(w, r, sessionID, user) - if !ok { - return - } - - visible, err := s.listVisibleSessionMessages(r.Context(), authCtx) + res, err := s.chatLog.DateStats(r.Context(), s.chatPrincipal(user, chatflow.ActionReadStats), sessionID) if err != nil { - http.Error(w, "list chat dates failed: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } - writeJSONResponse(w, http.StatusOK, aggregateDateStats(visible)) + writeJSONResponse(w, http.StatusOK, res) } // handleListChatCategories 列出会话各分类消息统计 (用于分类筛选条) // GET /api/v1/chat/categories?session_id= func (s *Server) handleListChatCategories(w http.ResponseWriter, r *http.Request) { - if s.svc == nil || s.svc.ContextManager() == nil { + if s.chatLog == nil { http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } @@ -169,18 +150,13 @@ func (s *Server) handleListChatCategories(w http.ResponseWriter, r *http.Request return } - authCtx, ok := s.authorizeSessionRead(w, r, sessionID, user) - if !ok { - return - } - - visible, err := s.listVisibleSessionMessages(r.Context(), authCtx) + res, err := s.chatLog.CategoryStats(r.Context(), s.chatPrincipal(user, chatflow.ActionReadStats), sessionID) if err != nil { - http.Error(w, "list chat categories failed: "+err.Error(), http.StatusInternalServerError) + s.writeChatLogError(w, err) return } - writeJSONResponse(w, http.StatusOK, aggregateCategoryStats(visible)) + writeJSONResponse(w, http.StatusOK, res) } // handleImportChatMessages 一次性批量回填历史消息 (仅允许在 dev/demo 环境下调用) diff --git a/internal/server/chatflow_http.go b/internal/server/chatflow_http.go new file mode 100644 index 00000000..d01cead5 --- /dev/null +++ b/internal/server/chatflow_http.go @@ -0,0 +1,37 @@ +package server + +import ( + "errors" + "net/http" + + "chat-agent-system/internal/app/chatflow" + "chat-agent-system/internal/auth" +) + +func (s *Server) chatPrincipal(user *auth.UserAccount, action chatflow.Action) chatflow.Principal { + if user == nil { + return chatflow.Principal{Action: action} + } + return chatflow.Principal{UserID: user.ID, Action: action} +} + +func (s *Server) writeChatLogError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, chatflow.ErrUnauthenticated): + http.Error(w, `{"error":"ERR_UNAUTHENTICATED","message":"请先登录"}`, http.StatusUnauthorized) + case errors.Is(err, chatflow.ErrForbidden): + writeJSONResponse(w, http.StatusForbidden, map[string]string{ + "error_code": "SESSION_FORBIDDEN", + "message": "无权访问该会话", + }) + case errors.Is(err, chatflow.ErrNotFound): + writeJSONResponse(w, http.StatusNotFound, map[string]string{ + "error_code": "SESSION_NOT_FOUND", + "message": "指定会话不存在或已被删除", + }) + case errors.Is(err, chatflow.ErrClosed): + http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 7f9873e8..4de4ff31 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/app/docflow" "chat-agent-system/internal/app/uow" "chat-agent-system/internal/bizdev" @@ -173,6 +174,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { req.UserRole, ) sess.UserID = user.ID + s.hydrateVisibleHistory(r.Context(), sess, user.ID) if req.ProjectID > 0 { sess.ProjectID = req.ProjectID } @@ -946,14 +948,19 @@ func (s *Server) handleGetChatMessages(w http.ResponseWriter, r *http.Request) { return } - msgs, err := s.svc.ContextManager().GetChatHistory(r.Context(), sessionID, limit, offset) - if err != nil { - http.Error(w, "get chat messages failed: "+err.Error(), http.StatusInternalServerError) + if s.chatLog == nil { + http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) return } - if projectID, pidErr := s.svc.ContextManager().GetSessionProjectID(r.Context(), sessionID); pidErr == nil && projectID > 0 { - msgs = s.filterChatMessagesByStageAccess(r.Context(), projectID, user.ID, msgs) + msgs, err := s.chatLog.History(r.Context(), s.chatPrincipal(user, chatflow.ActionReadWindow), chatflow.HistoryQuery{ + SessionID: sessionID, + Limit: limit, + Offset: offset, + }) + if err != nil { + s.writeChatLogError(w, err) + return } writeJSONResponse(w, http.StatusOK, msgs) diff --git a/internal/server/server.go b/internal/server/server.go index f0e703d4..666199d0 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -11,6 +11,7 @@ import ( "time" "chat-agent-system/internal/agent" + "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/app/uow" "chat-agent-system/internal/artifactvault" "chat-agent-system/internal/auth" @@ -49,6 +50,8 @@ type Server struct { resolver *workspace.Resolver wsCap *workspace.Capability cmds *taskrun.Commands + chatLog chatflow.ChatLog + assembler chatflow.ContextAssembler httpServer *http.Server } @@ -128,6 +131,8 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { wsCap: wsCap, cmds: cmds, } + s.chatLog = chatflow.NewAdapter(svc.ContextManager(), s.resolveStageAccess, s.failClosedAuthz()) + s.assembler = chatflow.NewAssembler(coordinator.NewContextAssembler(), s.chatLog) if instSvc != nil && instSvc.Available() { svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc)) coordEngine.SetInstanceService(instSvc) @@ -562,6 +567,7 @@ func (s *Server) getOrCreateSession( } sess := chat.NewSession(sessionID, userCtx, s.rec, s.engine) + sess.ChatLog = s.chatLog if s.svc != nil { sess.CtxMgr = s.svc.ContextManager() sess.Actions = &chat.ActionServices{ @@ -571,7 +577,8 @@ func (s *Server) getOrCreateSession( } } - // 若从持久层恢复了会话历史，回填消息列表、事实黑板与近期对话 (G3) + // 若从持久层恢复了会话历史，回填消息列表、事实黑板与近期对话 (G3)。 + // 正文注入走 ChatLog.Restore，避免把未授权消息写进 Prompt 窗口。 if restored != nil { if restored.UserID > 0 { sess.UserID = restored.UserID @@ -584,17 +591,19 @@ func (s *Server) getOrCreateSession( userCtx.BlackboardFacts[k] = v } } - for _, rm := range restored.Messages { - sess.Messages = append(sess.Messages, chat.Message{ - ID: fmt.Sprintf("msg_%d", rm.ID), - Role: rm.Role, - Content: rm.Content, - Intent: intent.IntentType(rm.Intent), - Branch: intent.ResolutionBranch(rm.Branch), - Timestamp: rm.CreatedAt, - }) + if s.chatLog == nil { + for _, rm := range restored.Messages { + sess.Messages = append(sess.Messages, chat.Message{ + ID: fmt.Sprintf("msg_%d", rm.ID), + Role: rm.Role, + Content: rm.Content, + Intent: intent.IntentType(rm.Intent), + Branch: intent.ResolutionBranch(rm.Branch), + Timestamp: rm.CreatedAt, + }) + } + sess.SyncRecentDialogue() } - sess.SyncRecentDialogue() if sess.UserID > 0 && sess.CtxMgr != nil { if guidance, err := sess.CtxMgr.CompilePromptGuidance(context.Background(), sess.UserID); err == nil && guidance != "" { @@ -608,6 +617,30 @@ func (s *Server) getOrCreateSession( return sess } +func (s *Server) hydrateVisibleHistory(ctx context.Context, sess *chat.Session, viewerUserID int64) { + if s.chatLog == nil || sess == nil || viewerUserID <= 0 { + return + } + if len(sess.Messages) > 0 { + return + } + restored, err := s.chatLog.Restore(ctx, chatflow.Principal{UserID: viewerUserID, Action: chatflow.ActionAssemble}, sess.ID) + if err != nil || restored == nil { + return + } + for _, rm := range restored.Messages { + sess.Messages = append(sess.Messages, chat.Message{ + ID: fmt.Sprintf("msg_%d", rm.ID), + Role: rm.Role, + Content: rm.Content, + Intent: intent.IntentType(rm.Intent), + Branch: intent.ResolutionBranch(rm.Branch), + Timestamp: rm.CreatedAt, + }) + } + sess.SyncRecentDialogue() +} + func (s *Server) corsMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { s.writeCORS(w, r) diff --git a/internal/server/sse.go b/internal/server/sse.go index ffad0f19..921fcad3 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -188,6 +188,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { req.UserRole, ) sess.UserID = req.UserID + s.hydrateVisibleHistory(r.Context(), sess, user.ID) if req.ProjectID > 0 { sess.ProjectID = req.ProjectID }