weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): route chat history reads through ChatLog

Browse Source 
Ignore request user_id as identity, authorize from the session Principal, and hydrate restore/memory from the same visible set so HTTP, SSE, and prompt assembly cannot amplify hidden messages.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-15 04:11:27 +00:00
co-authored by Cursor
parent 18d40fa8a9
commit 65bdda8aae
12 changed files with 196 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/chat/session.go
+15 -1
View File
@@ -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)
internal/server/chat_digest_api.go
+7 -14
View File
@@ -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
}
internal/server/chat_export_api.go
+9 -28
View File
@@ -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)
}
internal/server/chat_history_api_test.go
+12 -2
View File
@@ -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)
internal/server/chat_search_api.go
+5 -14
View File
@@ -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{}
}
internal/server/chat_unread_and_export_api_test.go
+11 -11
View File
@@ -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)
}
})
internal/server/chat_unread_api.go
+17 -44
View File
@@ -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
}
internal/server/chat_window_api.go
+26 -50
View File
@@ -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 环境下调用)
internal/server/chatflow_http.go
+37
View File
@@ -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)
}
}
internal/server/handlers.go
+12 -5
View File
@@ -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)
internal/server/server.go
+44 -11
View File
@@ -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)
internal/server/sse.go
+1
View File
@@ -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
}