From 4b8d8bf6d6a4f7913bef60af70e0a7e488939ff1 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 13 Sep 2026 03:06:11 +0000 Subject: [PATCH] fix(chat): unify history read authorization and stats filtering Apply session ownership plus scope_kind visibility across window, day, search, dates, categories, unread, and digest endpoints; add PG refresh regression test for pre-activation and task-scoped messages. Co-authored-by: Cursor --- internal/server/chat_digest_api.go | 18 ++ internal/server/chat_history_auth.go | 245 ++++++++++++++++++ .../server/chat_history_refresh_pg_test.go | 160 ++++++++++++ internal/server/chat_search_api.go | 9 +- internal/server/chat_unread_api.go | 37 +++ internal/server/chat_window_api.go | 49 +++- internal/server/chat_window_api_test.go | 2 +- internal/server/stage_access_helpers.go | 100 ++++--- internal/server/task_run_handlers.go | 33 ++- 9 files changed, 576 insertions(+), 77 deletions(-) create mode 100644 internal/server/chat_history_auth.go create mode 100644 internal/server/chat_history_refresh_pg_test.go diff --git a/internal/server/chat_digest_api.go b/internal/server/chat_digest_api.go index e84b4e45..5ef7e1ed 100644 --- a/internal/server/chat_digest_api.go +++ b/internal/server/chat_digest_api.go @@ -21,12 +21,21 @@ func (s *Server) handleListDailyDigests(w http.ResponseWriter, r *http.Request) return } + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return + } + sessionID := strings.TrimSpace(r.URL.Query().Get("session_id")) if sessionID == "" { http.Error(w, "missing required query param 'session_id'", http.StatusBadRequest) 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")) @@ -63,6 +72,11 @@ func (s *Server) handleGenerateDailyDigest(w http.ResponseWriter, r *http.Reques return } + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return + } + var req GenerateDigestRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "invalid request json body: "+err.Error(), http.StatusBadRequest) @@ -77,6 +91,10 @@ 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) if err != nil { http.Error(w, "GenerateDailyDigest error: "+err.Error(), http.StatusInternalServerError) diff --git a/internal/server/chat_history_auth.go b/internal/server/chat_history_auth.go new file mode 100644 index 00000000..0413cdb3 --- /dev/null +++ b/internal/server/chat_history_auth.go @@ -0,0 +1,245 @@ +package server + +import ( + "context" + "net/http" + "sort" + + "chat-agent-system/internal/auth" + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/projectinstance" +) + +// chatHistoryAuth 会话历史读取授权上下文。 +type chatHistoryAuth struct { + SessionID string + SessionOwnerID int64 + ProjectID int64 + ViewerUserID int64 + StageAccess *projectinstance.StageAccess +} + +func (s *Server) authorizeSessionRead( + w http.ResponseWriter, + r *http.Request, + sessionID string, + user *auth.UserAccount, +) (*chatHistoryAuth, bool) { + if user == nil { + http.Error(w, `{"error":"ERR_UNAUTHENTICATED","message":"请先登录"}`, http.StatusUnauthorized) + return nil, false + } + if !s.assertSessionOwnedByUser(w, r, sessionID, user.ID) { + return nil, false + } + authCtx := &chatHistoryAuth{ + SessionID: sessionID, + SessionOwnerID: user.ID, + ViewerUserID: user.ID, + } + if s.svc != nil && s.svc.ContextManager() != nil { + if projectID, err := s.svc.ContextManager().GetSessionProjectID(r.Context(), sessionID); err == nil { + authCtx.ProjectID = projectID + } + } + if authCtx.ProjectID > 0 { + access, err := s.resolveStageAccess(r.Context(), authCtx.ProjectID, user.ID) + if err != nil && s.failClosedAuthz() { + writeJSONResponse(w, http.StatusForbidden, map[string]string{ + "error_code": "WORKFLOW_ACCESS_DENIED", + "message": "无权访问该阶段内容", + }) + return nil, false + } + authCtx.StageAccess = access + } + return authCtx, true +} + +func (a *chatHistoryAuth) messageVisible(msg ctxmgr.ChatMessageRecord) bool { + if a == nil { + return false + } + return projectinstance.MessageRecordVisible( + msg.ScopeKind, + msg.StageKey, + msg.Metadata, + a.StageAccess, + a.SessionOwnerID, + a.ViewerUserID, + ) +} + +func (s *Server) filterChatMessagesByAuth( + authCtx *chatHistoryAuth, + msgs []ctxmgr.ChatMessageRecord, +) []ctxmgr.ChatMessageRecord { + if authCtx == nil || len(msgs) == 0 { + return msgs + } + out := make([]ctxmgr.ChatMessageRecord, 0, len(msgs)) + for _, msg := range msgs { + if authCtx.messageVisible(msg) { + out = append(out, msg) + } + } + return out +} + +func (s *Server) filterChatSearchHitsByAuth( + authCtx *chatHistoryAuth, + hits []ctxmgr.ChatSearchHit, +) []ctxmgr.ChatSearchHit { + if authCtx == nil || len(hits) == 0 { + return hits + } + out := make([]ctxmgr.ChatSearchHit, 0, len(hits)) + for _, hit := range hits { + meta := map[string]any{"stage_key": hit.StageKey} + if projectinstance.MessageRecordVisible("", hit.StageKey, meta, authCtx.StageAccess, authCtx.SessionOwnerID, authCtx.ViewerUserID) { + out = append(out, hit) + } + } + return out +} + +func (s *Server) applyChatWindowAuthFilter(authCtx *chatHistoryAuth, res *ctxmgr.ChatWindowResult) { + if res == nil || authCtx == nil { + return + } + filtered := make([]ctxmgr.ChatDayBucket, 0, len(res.Days)) + for i := range res.Days { + msgs := s.filterChatMessagesByAuth(authCtx, res.Days[i].Messages) + if len(msgs) == 0 { + continue + } + res.Days[i].Messages = msgs + res.Days[i].LoadedCount = len(msgs) + res.Days[i].Count = len(msgs) + res.Days[i].FirstMessageID = msgs[0].ID + res.Days[i].LastMessageID = msgs[len(msgs)-1].ID + filtered = append(filtered, res.Days[i]) + } + res.Days = filtered + if len(filtered) > 0 { + res.OldestLoadedDay = filtered[0].DayKey + } else { + res.OldestLoadedDay = "" + } +} + +func aggregateDateStats(msgs []ctxmgr.ChatMessageRecord) []ctxmgr.ChatDateStat { + statMap := make(map[string]*ctxmgr.ChatDateStat) + var orderedKeys []string + for _, msg := range msgs { + dKey := msg.DayKey + if dKey == "" { + continue + } + st, ok := statMap[dKey] + if !ok { + st = &ctxmgr.ChatDateStat{ + DayKey: dKey, + Categories: make(map[string]int), + } + statMap[dKey] = st + orderedKeys = append(orderedKeys, dKey) + } + st.Count++ + cat := msg.Category + if cat == "" { + cat = ctxmgr.CategoryUncategorized + } + st.Categories[cat]++ + } + sort.Strings(orderedKeys) + res := make([]ctxmgr.ChatDateStat, 0, len(orderedKeys)) + for _, k := range orderedKeys { + res = append(res, *statMap[k]) + } + return res +} + +func aggregateCategoryStats(msgs []ctxmgr.ChatMessageRecord) []ctxmgr.ChatCategoryStat { + statMap := make(map[string]*ctxmgr.ChatCategoryStat) + for _, msg := range msgs { + cat := msg.Category + if cat == "" { + cat = ctxmgr.CategoryUncategorized + } + st, ok := statMap[cat] + if !ok { + st = &ctxmgr.ChatCategoryStat{ + Category: cat, + Label: ctxmgr.GetCategoryLabel(cat), + } + statMap[cat] = st + } + st.Count++ + if st.FirstMessageID == 0 || msg.ID < st.FirstMessageID { + st.FirstMessageID = msg.ID + } + if msg.ID > st.LastMessageID { + st.LastMessageID = msg.ID + } + } + res := make([]ctxmgr.ChatCategoryStat, 0, len(statMap)) + for _, st := range statMap { + res = append(res, *st) + } + sort.Slice(res, func(i, j int) bool { + return res[i].Count > res[j].Count + }) + return res +} + +func (s *Server) listVisibleSessionMessages(ctx context.Context, authCtx *chatHistoryAuth) ([]ctxmgr.ChatMessageRecord, error) { + if s.svc == nil || s.svc.ContextManager() == nil || authCtx == nil { + return nil, nil + } + msgs, err := s.svc.ContextManager().ListSessionMessagesForAuth(ctx, authCtx.SessionID) + if err != nil { + return nil, err + } + return s.filterChatMessagesByAuth(authCtx, msgs), nil +} + +func (s *Server) resolveTaskMessageScope(ctx context.Context, projectID int64, taskKey string) (*projectinstance.MessageScope, error) { + if s.instSvc == nil || !s.instSvc.Available() { + if s.cfg != nil && s.cfg.AllowInMemory { + return &projectinstance.MessageScope{ScopeKind: projectinstance.ScopePreActivation}, nil + } + return nil, projectinstance.ErrTaskScopeNotResolved + } + return s.instSvc.ResolveMessageScopeForTask(ctx, projectID, taskKey) +} + +func (s *Server) resolveChatMessageScope(ctx context.Context, projectID int64, activeStageKey string) (*projectinstance.MessageScope, error) { + if s.instSvc == nil || !s.instSvc.Available() { + if projectID <= 0 { + return &projectinstance.MessageScope{ScopeKind: projectinstance.ScopeNonProjectPrivate}, nil + } + return &projectinstance.MessageScope{ScopeKind: projectinstance.ScopePreActivation}, nil + } + return s.instSvc.ResolveMessageScopeForChat(ctx, projectID, activeStageKey) +} + +func (s *Server) applyScopeToMeta(ctx context.Context, meta map[string]any, projectID int64, activeStageKey, taskKey string) error { + if meta == nil { + return nil + } + var scope *projectinstance.MessageScope + var err error + if taskKey != "" { + scope, err = s.resolveTaskMessageScope(ctx, projectID, taskKey) + } else { + scope, err = s.resolveChatMessageScope(ctx, projectID, activeStageKey) + } + if err != nil { + return err + } + if scope != nil { + projectinstance.ApplyScopeToMetadata(meta, scope) + } + return nil +} diff --git a/internal/server/chat_history_refresh_pg_test.go b/internal/server/chat_history_refresh_pg_test.go new file mode 100644 index 00000000..6bbfa75f --- /dev/null +++ b/internal/server/chat_history_refresh_pg_test.go @@ -0,0 +1,160 @@ +package server_test + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/workflowdef" +) + +func TestChatHistoryRefresh_PG_PreActivationAndTaskScope(t *testing.T) { + pool := openPGPoolWithMigrations(t) + if pool == nil { + return + } + defer pool.Close() + + ctx := context.Background() + handler, srv := setupPGGovernanceServer(t, pool) + ctxMgr := srv.BizdevService().ContextManager() + if ctxMgr == nil { + t.Fatal("context manager required") + } + + wfKey := workflowdef.OfficialKeyMain + seedOfficialWorkflowManifest(t, pool, wfKey) + + creatorToken := loginTestUser(t, handler, "zhanggong", "123456") + creatorID := fetchAuthUserID(t, handler, creatorToken) + + var projectID int64 + code := "pg-refresh-" + time.Now().Format("150405.000") + if err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision) + VALUES ($1, 'PG Refresh Test', 'active', $2, 'planning', '', 1, 0) RETURNING id + `, code, creatorID).Scan(&projectID); err != nil { + t.Fatalf("insert project: %v", err) + } + _, _ = pool.Exec(ctx, ` + INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at) + VALUES ($1, $2, 'owner', 'active', now(), now()) ON CONFLICT DO NOTHING + `, projectID, creatorID) + t.Cleanup(func() { cleanupProject(t, pool, projectID) }) + + sessID := fmt.Sprintf("sess-p%d-u%d", projectID, creatorID) + _, _ = ctxMgr.EnsureChatSession(ctx, sessID, creatorID, projectID, "Refresh Test") + + preMeta := map[string]any{"project_id": projectID} + projectinstance.ApplyScopeToMetadata(preMeta, &projectinstance.MessageScope{ScopeKind: projectinstance.ScopePreActivation}) + if _, err := ctxMgr.RecordUserMessageWithID(ctx, sessID, creatorID, "user", "激活前需求沟通", "SUPPLEMENT_INFO", "", preMeta); err != nil { + t.Fatalf("record pre-activation: %v", err) + } + + reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=0", projectID, wfKey), nil) + withBearerAuth(reqDefault, creatorToken) + recDefault := httptest.NewRecorder() + handler.ServeHTTP(recDefault, reqDefault) + if recDefault.Code != http.StatusOK { + t.Fatalf("default plan: %d %s", recDefault.Code, recDefault.Body.String()) + } + var plan projectinstance.WorkflowPlan + _ = json.Unmarshal(recDefault.Body.Bytes(), &plan) + reviewerID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "lizhuguan", "123456")) + approverID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "wangzong", "123456")) + for i := range plan.Stages { + for j := range plan.Stages[i].DocTasks { + plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{ + AuthorUserID: creatorID, ReviewerUserID: reviewerID, ApproverUserID: approverID, + } + } + } + saveBody, _ := json.Marshal(projectinstance.SavePlanInput{Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1}) + reqSave := httptest.NewRequest(http.MethodPut, fmt.Sprintf("/api/v1/projects/%d/workflow-plan", projectID), bytes.NewReader(saveBody)) + withBearerAuth(reqSave, creatorToken) + recSave := httptest.NewRecorder() + handler.ServeHTTP(recSave, reqSave) + if recSave.Code != http.StatusOK { + t.Fatalf("save plan: %d %s", recSave.Code, recSave.Body.String()) + } + + reqPreview := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/activation-preview", projectID), nil) + withBearerAuth(reqPreview, creatorToken) + recPreview := httptest.NewRecorder() + handler.ServeHTTP(recPreview, reqPreview) + if recPreview.Code != http.StatusOK { + t.Fatalf("activation preview: %d %s", recPreview.Code, recPreview.Body.String()) + } + var preview projectinstance.ActivationPreview + _ = json.Unmarshal(recPreview.Body.Bytes(), &preview) + activateBody, _ := json.Marshal(projectinstance.ActivateInput{ + ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion, + ExpectedPlanRevision: preview.ExpectedPlanRevision, + }) + reqActivate := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", projectID), bytes.NewReader(activateBody)) + withBearerAuth(reqActivate, creatorToken) + recActivate := httptest.NewRecorder() + handler.ServeHTTP(recActivate, reqActivate) + if recActivate.Code != http.StatusOK { + t.Fatalf("activate: %d %s", recActivate.Code, recActivate.Body.String()) + } + + instSvc := projectinstance.NewService(projectinstance.NewStore(pool), nil, nil, nil) + taskScope, err := instSvc.ResolveMessageScopeForTask(ctx, projectID, "intake") + if err != nil { + t.Fatalf("resolve task scope: %v", err) + } + taskUserMeta := map[string]any{"project_id": projectID, "task_key": "intake", "client_turn_id": "turn-doc-1"} + projectinstance.ApplyScopeToMetadata(taskUserMeta, taskScope) + _, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, creatorID, "user", "请生成 intake 文档", "TRIGGER_TASK", "", taskUserMeta) + taskAsstMeta := map[string]any{"project_id": projectID, "task_key": "intake", "task_status": "completed"} + projectinstance.ApplyScopeToMetadata(taskAsstMeta, taskScope) + _, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, 0, "assistant", "文档已生成", "TRIGGER_TASK", "", taskAsstMeta) + + reqWindow := httptest.NewRequest(http.MethodGet, "/api/v1/chat/messages/window?session_id="+sessID+"&days=7", nil) + withBearerAuth(reqWindow, creatorToken) + recWindow := httptest.NewRecorder() + handler.ServeHTTP(recWindow, reqWindow) + if recWindow.Code != http.StatusOK { + t.Fatalf("window: %d %s", recWindow.Code, recWindow.Body.String()) + } + var win ctxmgr.ChatWindowResult + _ = json.Unmarshal(recWindow.Body.Bytes(), &win) + totalMsgs := 0 + for _, d := range win.Days { + totalMsgs += len(d.Messages) + } + if totalMsgs != 3 { + t.Fatalf("expected 3 visible messages after refresh, got %d", totalMsgs) + } + + reqCat := httptest.NewRequest(http.MethodGet, "/api/v1/chat/categories?session_id="+sessID, nil) + withBearerAuth(reqCat, creatorToken) + recCat := httptest.NewRecorder() + handler.ServeHTTP(recCat, reqCat) + var cats []ctxmgr.ChatCategoryStat + _ = json.Unmarshal(recCat.Body.Bytes(), &cats) + catTotal := 0 + for _, c := range cats { + catTotal += c.Count + } + if catTotal != 3 { + t.Fatalf("expected category count 3, got %d", catTotal) + } + + guestToken := loginTestUser(t, handler, "guest", "123456") + reqForbidden := httptest.NewRequest(http.MethodGet, "/api/v1/chat/messages/window?session_id="+sessID+"&days=7", nil) + withBearerAuth(reqForbidden, guestToken) + recForbidden := httptest.NewRecorder() + handler.ServeHTTP(recForbidden, reqForbidden) + if recForbidden.Code != http.StatusForbidden { + t.Fatalf("expected 403 for non-owner, got %d %s", recForbidden.Code, recForbidden.Body.String()) + } +} diff --git a/internal/server/chat_search_api.go b/internal/server/chat_search_api.go index ef89bf36..a8a4fdaa 100644 --- a/internal/server/chat_search_api.go +++ b/internal/server/chat_search_api.go @@ -33,6 +33,11 @@ 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) @@ -68,9 +73,7 @@ func (s *Server) handleChatSearch(w http.ResponseWriter, r *http.Request) { return } - if projectID, pidErr := s.svc.ContextManager().GetSessionProjectID(r.Context(), sessionID); pidErr == nil && projectID > 0 { - hits = s.filterChatSearchHitsByStageAccess(r.Context(), projectID, user.ID, hits) - } + hits = s.filterChatSearchHitsByAuth(authCtx, hits) if hits == nil { hits = []ctxmgr.ChatSearchHit{} diff --git a/internal/server/chat_unread_api.go b/internal/server/chat_unread_api.go index b0e1a3bc..e3f289cc 100644 --- a/internal/server/chat_unread_api.go +++ b/internal/server/chat_unread_api.go @@ -14,12 +14,22 @@ import ( ) func (s *Server) handleChatUnread(w http.ResponseWriter, r *http.Request) { + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return + } + sessionID := r.URL.Query().Get("session_id") if sessionID == "" { http.Error(w, "missing session_id", http.StatusBadRequest) 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 { @@ -55,6 +65,24 @@ func (s *Server) handleChatUnread(w http.ResponseWriter, r *http.Request) { 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) + } + } + writeJSONResponse(w, http.StatusOK, summary) } @@ -64,6 +92,11 @@ func (s *Server) handleChatRead(w http.ResponseWriter, r *http.Request) { return } + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return + } + var req ctxmgr.MarkChatReadRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest) @@ -74,6 +107,10 @@ 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 { http.Error(w, "context manager not initialized", http.StatusInternalServerError) return diff --git a/internal/server/chat_window_api.go b/internal/server/chat_window_api.go index 62593ac1..fc94bda3 100644 --- a/internal/server/chat_window_api.go +++ b/internal/server/chat_window_api.go @@ -28,6 +28,11 @@ 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 { @@ -45,9 +50,7 @@ func (s *Server) handleGetChatWindow(w http.ResponseWriter, r *http.Request) { return } - if projectID, pidErr := s.svc.ContextManager().GetSessionProjectID(r.Context(), sessionID); pidErr == nil && projectID > 0 { - s.applyChatWindowStageFilter(r.Context(), projectID, user.ID, res) - } + s.applyChatWindowAuthFilter(authCtx, res) writeJSONResponse(w, http.StatusOK, res) } @@ -70,6 +73,12 @@ func (s *Server) handleGetChatDayMessages(w http.ResponseWriter, r *http.Request http.Error(w, "missing session_id query param", http.StatusBadRequest) 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") @@ -101,9 +110,9 @@ func (s *Server) handleGetChatDayMessages(w http.ResponseWriter, r *http.Request return } - if projectID, pidErr := s.svc.ContextManager().GetSessionProjectID(r.Context(), sessionID); pidErr == nil && projectID > 0 { - res.Messages = s.filterChatMessagesByStageAccess(r.Context(), projectID, user.ID, res.Messages) - } + res.Messages = s.filterChatMessagesByAuth(authCtx, res.Messages) + res.LoadedCount = len(res.Messages) + res.TotalCount = len(res.Messages) writeJSONResponse(w, http.StatusOK, res) } @@ -116,19 +125,29 @@ func (s *Server) handleListChatDates(w http.ResponseWriter, r *http.Request) { return } + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return + } + sessionID := r.URL.Query().Get("session_id") if sessionID == "" { http.Error(w, "missing session_id query param", http.StatusBadRequest) return } - stats, err := s.svc.ContextManager().ListChatDates(r.Context(), sessionID) + authCtx, ok := s.authorizeSessionRead(w, r, sessionID, user) + if !ok { + return + } + + visible, err := s.listVisibleSessionMessages(r.Context(), authCtx) if err != nil { http.Error(w, "list chat dates failed: "+err.Error(), http.StatusInternalServerError) return } - writeJSONResponse(w, http.StatusOK, stats) + writeJSONResponse(w, http.StatusOK, aggregateDateStats(visible)) } // handleListChatCategories 列出会话各分类消息统计 (用于分类筛选条) @@ -139,19 +158,29 @@ func (s *Server) handleListChatCategories(w http.ResponseWriter, r *http.Request return } + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return + } + sessionID := r.URL.Query().Get("session_id") if sessionID == "" { http.Error(w, "missing session_id query param", http.StatusBadRequest) return } - stats, err := s.svc.ContextManager().ListChatCategories(r.Context(), sessionID) + authCtx, ok := s.authorizeSessionRead(w, r, sessionID, user) + if !ok { + return + } + + visible, err := s.listVisibleSessionMessages(r.Context(), authCtx) if err != nil { http.Error(w, "list chat categories failed: "+err.Error(), http.StatusInternalServerError) return } - writeJSONResponse(w, http.StatusOK, stats) + writeJSONResponse(w, http.StatusOK, aggregateCategoryStats(visible)) } // handleImportChatMessages 一次性批量回填历史消息 (仅允许在 dev/demo 环境下调用) diff --git a/internal/server/chat_window_api_test.go b/internal/server/chat_window_api_test.go index 0917f413..311b8e8f 100644 --- a/internal/server/chat_window_api_test.go +++ b/internal/server/chat_window_api_test.go @@ -25,7 +25,7 @@ func TestServer_ChatWindowAndDayAPIs(t *testing.T) { handler := srv.Routes() sessID := "sess_api_window_001" - userID := int64(201) + userID := int64(1) // 与 AllowInMemory legacy principal 一致 // 1. 测试 POST /api/v1/chat/messages/import (批量历史回填) importBody := []map[string]any{ diff --git a/internal/server/stage_access_helpers.go b/internal/server/stage_access_helpers.go index 5cf45916..7a680103 100644 --- a/internal/server/stage_access_helpers.go +++ b/internal/server/stage_access_helpers.go @@ -95,14 +95,19 @@ func (s *Server) allowActiveStageChatAccess(ctx context.Context, projectID, user } func (s *Server) enrichSessionWorkflowStage(ctx context.Context, sess *chat.Session, userID int64) { - if sess == nil || sess.ProjectID <= 0 || s.instSvc == nil || !s.instSvc.Available() { + if sess == nil { return } - settings, err := s.instSvc.GetSettings(ctx, sess.ProjectID, userID, false) - if err != nil || settings == nil { - return + if sess.ProjectID > 0 && s.instSvc != nil && s.instSvc.Available() { + settings, err := s.instSvc.GetSettings(ctx, sess.ProjectID, userID, false) + if err == nil && settings != nil { + sess.ActiveStageKey = settings.ActiveStageKey + } + } + scope, err := s.resolveChatMessageScope(ctx, sess.ProjectID, sess.ActiveStageKey) + if err == nil && scope != nil { + sess.MessageScope = scope } - sess.ActiveStageKey = settings.ActiveStageKey } func (s *Server) filterWorkflowDocsByStageAccess( @@ -137,30 +142,22 @@ func (s *Server) filterChatMessagesByStageAccess( projectID, userID int64, msgs []ctxmgr.ChatMessageRecord, ) []ctxmgr.ChatMessageRecord { - if len(msgs) == 0 || projectID <= 0 { + if len(msgs) == 0 { return msgs } - access, err := s.resolveStageAccess(ctx, projectID, userID) - if err != nil { - if s.failClosedAuthz() { + authCtx := &chatHistoryAuth{ + SessionOwnerID: userID, + ProjectID: projectID, + ViewerUserID: userID, + } + if projectID > 0 { + access, err := s.resolveStageAccess(ctx, projectID, userID) + if err != nil && s.failClosedAuthz() { return nil } - return msgs + authCtx.StageAccess = access } - if access == nil || !access.Restricted { - return msgs - } - out := make([]ctxmgr.ChatMessageRecord, 0, len(msgs)) - for _, msg := range msgs { - stageKey := projectinstance.MessageStageKey(msg.Metadata) - if stageKey == "" { - continue - } - if access.StageKeyAllowed(stageKey) { - out = append(out, msg) - } - } - return out + return s.filterChatMessagesByAuth(authCtx, msgs) } func (s *Server) filterChatSearchHitsByStageAccess( @@ -168,47 +165,42 @@ func (s *Server) filterChatSearchHitsByStageAccess( projectID, userID int64, hits []ctxmgr.ChatSearchHit, ) []ctxmgr.ChatSearchHit { - if len(hits) == 0 || projectID <= 0 { + if len(hits) == 0 { return hits } - access, err := s.resolveStageAccess(ctx, projectID, userID) - if err != nil { - if s.failClosedAuthz() { + authCtx := &chatHistoryAuth{ + SessionOwnerID: userID, + ProjectID: projectID, + ViewerUserID: userID, + } + if projectID > 0 { + access, err := s.resolveStageAccess(ctx, projectID, userID) + if err != nil && s.failClosedAuthz() { return nil } - return hits + authCtx.StageAccess = access } - if access == nil || !access.Restricted { - return hits - } - out := make([]ctxmgr.ChatSearchHit, 0, len(hits)) - for _, hit := range hits { - if hit.StageKey == "" { - continue - } - if access.StageKeyAllowed(hit.StageKey) { - out = append(out, hit) - } - } - return out + return s.filterChatSearchHitsByAuth(authCtx, hits) } func (s *Server) applyChatWindowStageFilter(ctx context.Context, projectID, userID int64, res *ctxmgr.ChatWindowResult) { - if res == nil || projectID <= 0 { + if res == nil { return } - for i := range res.Days { - res.Days[i].Messages = s.filterChatMessagesByStageAccess(ctx, projectID, userID, res.Days[i].Messages) - res.Days[i].LoadedCount = len(res.Days[i].Messages) - res.Days[i].Count = len(res.Days[i].Messages) - if len(res.Days[i].Messages) > 0 { - res.Days[i].FirstMessageID = res.Days[i].Messages[0].ID - res.Days[i].LastMessageID = res.Days[i].Messages[len(res.Days[i].Messages)-1].ID - } else { - res.Days[i].FirstMessageID = 0 - res.Days[i].LastMessageID = 0 - } + authCtx := &chatHistoryAuth{ + SessionOwnerID: userID, + ProjectID: projectID, + ViewerUserID: userID, } + if projectID > 0 { + access, err := s.resolveStageAccess(ctx, projectID, userID) + if err != nil && s.failClosedAuthz() { + res.Days = nil + return + } + authCtx.StageAccess = access + } + s.applyChatWindowAuthFilter(authCtx, res) } func workflowAccessDeniedJSON() string { diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index e4c5ad1d..1947e5a3 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -13,6 +13,7 @@ import ( "chat-agent-system/internal/agent" "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/squad" "chat-agent-system/internal/stream" "chat-agent-system/internal/taskrun" @@ -78,6 +79,15 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { _, _ = ctxMgr.EnsureChatSession(r.Context(), req.SessionID, req.UserID, req.ProjectID, req.ProjectName) } + taskScope, scopeErr := s.resolveTaskMessageScope(r.Context(), req.ProjectID, req.TaskKey) + if scopeErr != nil { + writeJSONResponse(w, http.StatusConflict, map[string]string{ + "error_code": "TASK_SCOPE_UNRESOLVED", + "message": "无法解析文档任务的工作流阶段作用域", + }) + return + } + var assistantMsgID int64 if ctxMgr != nil && req.UserMessage != "" { userMeta := map[string]any{ @@ -85,6 +95,7 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { "client_turn_id": req.ClientTurnID, "task_key": req.TaskKey, } + projectinstance.ApplyScopeToMetadata(userMeta, taskScope) _, _ = ctxMgr.RecordUserMessageWithID(r.Context(), req.SessionID, req.UserID, "user", req.UserMessage, "TRIGGER_TASK", "", userMeta) asstMeta := map[string]any{ "project_id": req.ProjectID, @@ -92,20 +103,24 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { "task_key": req.TaskKey, "task_status": "running", } + projectinstance.ApplyScopeToMetadata(asstMeta, taskScope) assistantMsgID, _ = ctxMgr.RecordUserMessageWithID(r.Context(), req.SessionID, 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", asstMeta) } runID := fmt.Sprintf("run_%d", time.Now().UnixNano()) record := &taskrun.RunRecord{ - RunID: runID, - SessionID: req.SessionID, - ProjectID: req.ProjectID, - UserID: req.UserID, - ClientTurnID: req.ClientTurnID, - TaskKey: req.TaskKey, - Stage: req.Stage, - AssistantMsgID: assistantMsgID, - Status: "running", + RunID: runID, + SessionID: req.SessionID, + ProjectID: req.ProjectID, + UserID: req.UserID, + ClientTurnID: req.ClientTurnID, + TaskKey: req.TaskKey, + Stage: req.Stage, + WorkflowInstanceID: taskScope.WorkflowInstanceID, + StageInstanceID: taskScope.StageInstanceID, + WorkflowStageKey: taskScope.StageKey, + AssistantMsgID: assistantMsgID, + Status: "running", } taskReq := s.buildBizdevTaskRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, runID)