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 <cursoragent@cursor.com>

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)

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
+}

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())
+	}
+}

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{}

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

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 环境下调用)

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{

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 {

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)