feat(chatflow): add ChatLog and ContextAssembler ports

Keep ctxmgr as table owner and filter the authorized visible set in-process so window, stats, restore, digest, and prompt assembly share one Principal-scoped collection without a package move.

Co-authored-by: Cursor <cursoragent@cursor.com>

.doc/system-design/module-boundaries.yaml

@@ -27,6 +27,7 @@ imports:
     - stream
     - workflowdef
   chat:
+    - app/chatflow
     - bizdev
     - coordinator
     - ctxmgr
@@ -56,6 +57,11 @@ imports:
   db: []
   docduty: []
   app: []
+  app/chatflow:
+    - bizdev
+    - coordinator
+    - ctxmgr
+    - projectinstance
   app/docflow:
     - app/uow
     - docgov
@@ -99,6 +105,7 @@ imports:
   replyexport: []
   server:
     - agent
+    - app/chatflow
     - app/docflow
     - app/uow
     - auth

internal/app/chatflow/assembler.go

@@ -0,0 +1,106 @@
+package chatflow
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/coordinator"
+	"chat-agent-system/internal/ctxmgr"
+)
+
+// Assembler adapts coordinator.ContextAssembler and ChatLog so prompt context
+// only contains the authorized message set.
+type Assembler struct {
+	Inner *coordinator.ContextAssembler
+	Log   ChatLog
+}
+
+func NewAssembler(inner *coordinator.ContextAssembler, log ChatLog) *Assembler {
+	if inner == nil {
+		inner = coordinator.NewContextAssembler()
+	}
+	return &Assembler{Inner: inner, Log: log}
+}
+
+func (a *Assembler) Assemble(ctx context.Context, p Principal, req AssembleRequest) (*coordinator.UserProjectContext, error) {
+	if p.UserID <= 0 {
+		return nil, ErrUnauthenticated
+	}
+	p.Action = ActionAssemble
+
+	var userCtx *coordinator.UserProjectContext
+	if a.Inner != nil && strings.TrimSpace(req.WorkDir) != "" {
+		userCtx, _ = a.Inner.AssembleFromWorkspace(req.WorkDir, req.User)
+	}
+	if userCtx == nil {
+		userCtx = &coordinator.UserProjectContext{
+			SnapshotID:       fmt.Sprintf("ctx_%d", time.Now().UnixNano()),
+			AssembledAt:      time.Now(),
+			User:             req.User,
+			WorkDir:          req.WorkDir,
+			Project:          req.Fallback,
+			BlackboardFacts:  make(map[string]string),
+			AvailableActions: coordinator.ComputeAvailableActions(bizdev.StageV01_Triage, req.User.ProjectRole),
+		}
+		if userCtx.Project.CurrentStage == "" {
+			userCtx.Project.CurrentStage = bizdev.StageV01_Triage
+		}
+		if userCtx.Project.CurrentWorkflow == "" {
+			userCtx.Project.CurrentWorkflow = "intake"
+		}
+	}
+
+	if a.Log == nil || strings.TrimSpace(req.SessionID) == "" {
+		return userCtx, nil
+	}
+	restored, err := a.Log.Restore(ctx, p, req.SessionID)
+	if err != nil {
+		if err == ErrForbidden || err == ErrNotFound || err == ErrUnauthenticated {
+			userCtx.RecentDialogue = nil
+			userCtx.DialogueSummary = ""
+			userCtx.RecalledMemories = nil
+			userCtx.RecalledMemorySnippet = ""
+			return userCtx, nil
+		}
+		return nil, err
+	}
+	if restored == nil {
+		return userCtx, nil
+	}
+	if restored.ProjectID > 0 && userCtx.Project.ID == "" {
+		userCtx.Project.ID = fmt.Sprintf("%d", restored.ProjectID)
+	}
+	if userCtx.BlackboardFacts == nil {
+		userCtx.BlackboardFacts = make(map[string]string)
+	}
+	for k, v := range restored.Facts {
+		if _, exists := userCtx.BlackboardFacts[k]; !exists {
+			userCtx.BlackboardFacts[k] = v
+		}
+	}
+	turns := make([]coordinator.DialogueTurn, 0, len(restored.Messages))
+	records := make([]ctxmgr.ChatMessageRecord, 0, len(restored.Messages))
+	for _, m := range restored.Messages {
+		if strings.TrimSpace(m.Content) == "" {
+			continue
+		}
+		turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content})
+		records = append(records, m)
+	}
+	cfg := ctxmgr.DefaultWindowConfig()
+	window, _, _ := ctxmgr.AssembleWindow(records, cfg)
+	if len(window) > 0 {
+		turns = turns[:0]
+		for _, m := range window {
+			if strings.TrimSpace(m.Content) == "" {
+				continue
+			}
+			turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content})
+		}
+	}
+	userCtx.RecentDialogue = turns
+	return userCtx, nil
+}

internal/app/chatflow/chatlog.go

@@ -0,0 +1,443 @@
+package chatflow
+
+import (
+	"context"
+	"sort"
+	"strings"
+
+	"chat-agent-system/internal/ctxmgr"
+	"chat-agent-system/internal/projectinstance"
+)
+
+type sessionAuth struct {
+	SessionID      string
+	SessionOwnerID int64
+	ProjectID      int64
+	ViewerUserID   int64
+	StageAccess    *projectinstance.StageAccess
+}
+
+// Adapter implements ChatLog on top of ctxmgr without moving table ownership.
+type Adapter struct {
+	Mgr        *ctxmgr.ContextManager
+	Resolve    StageAccessResolver
+	FailClosed bool
+}
+
+func NewAdapter(mgr *ctxmgr.ContextManager, resolve StageAccessResolver, failClosed bool) *Adapter {
+	return &Adapter{Mgr: mgr, Resolve: resolve, FailClosed: failClosed}
+}
+
+func (a *Adapter) authorize(ctx context.Context, p Principal, sessionID string) (*sessionAuth, error) {
+	if a == nil || a.Mgr == nil {
+		return nil, ErrClosed
+	}
+	if p.UserID <= 0 {
+		return nil, ErrUnauthenticated
+	}
+	if strings.TrimSpace(sessionID) == "" {
+		return nil, ErrNotFound
+	}
+	ownerID, err := a.Mgr.GetSessionOwnerID(ctx, sessionID)
+	if err != nil || ownerID <= 0 {
+		return nil, ErrNotFound
+	}
+	if ownerID != p.UserID {
+		return nil, ErrForbidden
+	}
+	auth := &sessionAuth{
+		SessionID:      sessionID,
+		SessionOwnerID: ownerID,
+		ViewerUserID:   p.UserID,
+	}
+	if projectID, err := a.Mgr.GetSessionProjectID(ctx, sessionID); err == nil {
+		auth.ProjectID = projectID
+	}
+	if auth.ProjectID > 0 && a.Resolve != nil {
+		access, err := a.Resolve(ctx, auth.ProjectID, p.UserID)
+		if err != nil {
+			if a.FailClosed {
+				return nil, ErrForbidden
+			}
+		} else {
+			auth.StageAccess = access
+		}
+	}
+	return auth, nil
+}
+
+func (a *Adapter) visible(auth *sessionAuth, msg ctxmgr.ChatMessageRecord) bool {
+	if auth == nil {
+		return false
+	}
+	return projectinstance.MessageRecordVisible(
+		msg.ScopeKind,
+		msg.StageKey,
+		msg.Metadata,
+		auth.StageAccess,
+		auth.SessionOwnerID,
+		auth.ViewerUserID,
+	)
+}
+
+func (a *Adapter) filterMessages(auth *sessionAuth, msgs []ctxmgr.ChatMessageRecord) []ctxmgr.ChatMessageRecord {
+	if len(msgs) == 0 {
+		return msgs
+	}
+	out := make([]ctxmgr.ChatMessageRecord, 0, len(msgs))
+	for _, msg := range msgs {
+		if a.visible(auth, msg) {
+			out = append(out, msg)
+		}
+	}
+	return out
+}
+
+func (a *Adapter) GetWindow(ctx context.Context, p Principal, q ctxmgr.ChatWindowQuery) (*ctxmgr.ChatWindowResult, error) {
+	auth, err := a.authorize(ctx, p, q.SessionID)
+	if err != nil {
+		return nil, err
+	}
+	res, err := a.Mgr.GetChatWindow(ctx, q)
+	if err != nil {
+		return nil, err
+	}
+	applyWindowFilter(auth, a, res)
+	return res, nil
+}
+
+func applyWindowFilter(auth *sessionAuth, a *Adapter, res *ctxmgr.ChatWindowResult) {
+	if res == nil || auth == nil {
+		return
+	}
+	filtered := make([]ctxmgr.ChatDayBucket, 0, len(res.Days))
+	for i := range res.Days {
+		msgs := a.filterMessages(auth, 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 (a *Adapter) GetDay(ctx context.Context, p Principal, q DayQuery) (*ctxmgr.ChatDayMessagesResult, error) {
+	auth, err := a.authorize(ctx, p, q.SessionID)
+	if err != nil {
+		return nil, err
+	}
+	res, err := a.Mgr.GetChatDayMessages(ctx, q.SessionID, q.Date, q.AnchorID, q.BeforeID, q.AfterID, q.Limit)
+	if err != nil {
+		return nil, err
+	}
+	res.Messages = a.filterMessages(auth, res.Messages)
+	res.LoadedCount = len(res.Messages)
+	res.TotalCount = len(res.Messages)
+	return res, nil
+}
+
+func (a *Adapter) History(ctx context.Context, p Principal, q HistoryQuery) ([]ctxmgr.ChatMessageRecord, error) {
+	auth, err := a.authorize(ctx, p, q.SessionID)
+	if err != nil {
+		return nil, err
+	}
+	if q.Limit <= 0 {
+		q.Limit = 50
+	}
+	if q.Offset < 0 {
+		q.Offset = 0
+	}
+	msgs, err := a.Mgr.GetChatHistory(ctx, q.SessionID, q.Limit, q.Offset)
+	if err != nil {
+		return nil, err
+	}
+	return a.filterMessages(auth, msgs), nil
+}
+
+func (a *Adapter) ListVisible(ctx context.Context, p Principal, sessionID string) ([]ctxmgr.ChatMessageRecord, error) {
+	auth, err := a.authorize(ctx, p, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	msgs, err := a.Mgr.ListSessionMessagesForAuth(ctx, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	return a.filterMessages(auth, msgs), nil
+}
+
+func (a *Adapter) CategoryStats(ctx context.Context, p Principal, sessionID string) ([]ctxmgr.ChatCategoryStat, error) {
+	visible, err := a.ListVisible(ctx, p, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	return aggregateCategoryStats(visible), nil
+}
+
+func (a *Adapter) DateStats(ctx context.Context, p Principal, sessionID string) ([]ctxmgr.ChatDateStat, error) {
+	visible, err := a.ListVisible(ctx, p, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	return aggregateDateStats(visible), nil
+}
+
+func (a *Adapter) Search(ctx context.Context, p Principal, q ctxmgr.ChatSearchQuery) ([]ctxmgr.ChatSearchHit, error) {
+	auth, err := a.authorize(ctx, p, q.SessionID)
+	if err != nil {
+		return nil, err
+	}
+	q.UserID = p.UserID
+	hits, err := a.Mgr.SearchChatMessages(ctx, q)
+	if err != nil {
+		return nil, err
+	}
+	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, auth.StageAccess, auth.SessionOwnerID, auth.ViewerUserID) {
+			out = append(out, hit)
+		}
+	}
+	return out, nil
+}
+
+func (a *Adapter) Unread(ctx context.Context, p Principal, q UnreadQuery) (*ctxmgr.ChatUnreadSummary, error) {
+	auth, err := a.authorize(ctx, p, q.SessionID)
+	if err != nil {
+		return nil, err
+	}
+	summary, err := a.Mgr.GetChatUnreadSummary(ctx, q.SessionID, p.UserID, q.SinceMessageID)
+	if err != nil {
+		return &ctxmgr.ChatUnreadSummary{UnreadMessageIDs: []int64{}, Items: []ctxmgr.UnreadDayItem{}}, nil
+	}
+	if summary == nil {
+		return &ctxmgr.ChatUnreadSummary{UnreadMessageIDs: []int64{}, Items: []ctxmgr.UnreadDayItem{}}, nil
+	}
+	visible, err := a.Mgr.ListSessionMessagesForAuth(ctx, q.SessionID)
+	if err != nil {
+		return summary, nil
+	}
+	allowed := make(map[int64]bool, len(visible))
+	for _, msg := range a.filterMessages(auth, 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)
+	return summary, nil
+}
+
+func (a *Adapter) MarkRead(ctx context.Context, p Principal, req ctxmgr.MarkChatReadRequest) (*ctxmgr.ChatReadPosition, error) {
+	if _, err := a.authorize(ctx, p, req.SessionID); err != nil {
+		return nil, err
+	}
+	req.UserID = p.UserID
+	return a.Mgr.MarkChatRead(ctx, req)
+}
+
+func (a *Adapter) Export(ctx context.Context, p Principal, q ExportQuery) ([]ctxmgr.ChatMessageRecord, error) {
+	auth, err := a.authorize(ctx, p, q.SessionID)
+	if err != nil {
+		return nil, err
+	}
+	msgs, err := a.Mgr.GetChatHistory(ctx, q.SessionID, 5000, 0)
+	if err != nil {
+		return nil, err
+	}
+	visible := a.filterMessages(auth, msgs)
+	out := make([]ctxmgr.ChatMessageRecord, 0, len(visible))
+	for _, m := range visible {
+		dk := m.DayKey
+		if dk == "" && !m.CreatedAt.IsZero() {
+			dk = m.CreatedAt.Format("2006-01-02")
+		}
+		if q.DateFrom != "" && dk < q.DateFrom {
+			continue
+		}
+		if q.DateTo != "" && dk > q.DateTo {
+			continue
+		}
+		if q.Category != "" && q.Category != "all" && m.Category != q.Category {
+			continue
+		}
+		out = append(out, m)
+	}
+	return out, nil
+}
+
+func (a *Adapter) ListDigests(ctx context.Context, p Principal, sessionID, from, to string) ([]ctxmgr.DailyDigest, error) {
+	visible, err := a.ListVisible(ctx, p, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	days := make(map[string]struct{}, len(visible))
+	for _, msg := range visible {
+		if msg.DayKey != "" {
+			days[msg.DayKey] = struct{}{}
+		}
+	}
+	digests, err := a.Mgr.ListDailyDigests(ctx, sessionID, from, to)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]ctxmgr.DailyDigest, 0, len(digests))
+	for _, d := range digests {
+		if _, ok := days[d.DayKey]; ok {
+			out = append(out, d)
+		}
+	}
+	return out, nil
+}
+
+func (a *Adapter) GenerateDigest(ctx context.Context, p Principal, sessionID, dayKey string) (*ctxmgr.DailyDigest, error) {
+	visible, err := a.ListVisible(ctx, p, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	dayMsgs := make([]ctxmgr.ChatMessageRecord, 0, len(visible))
+	for _, msg := range visible {
+		if msg.DayKey == dayKey {
+			dayMsgs = append(dayMsgs, msg)
+		}
+	}
+	return a.Mgr.GenerateDailyDigestFromMessages(ctx, sessionID, dayKey, dayMsgs)
+}
+
+func (a *Adapter) Restore(ctx context.Context, p Principal, sessionID string) (*ctxmgr.RestoredSessionContext, error) {
+	auth, err := a.authorize(ctx, p, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	restored, err := a.Mgr.RestoreSessionContext(ctx, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	if restored == nil {
+		return nil, nil
+	}
+	restored.Messages = a.filterMessages(auth, restored.Messages)
+	return restored, nil
+}
+
+func (a *Adapter) SearchMemory(ctx context.Context, p Principal, q MemoryQuery) ([]ctxmgr.RecalledMemoryItem, error) {
+	if a == nil || a.Mgr == nil {
+		return nil, ErrClosed
+	}
+	if p.UserID <= 0 {
+		return nil, ErrUnauthenticated
+	}
+	items, err := a.Mgr.SearchChatMemory(ctx, ctxmgr.ChatMemorySearchOptions{
+		Query:            q.Query,
+		UserID:           p.UserID,
+		ProjectID:        q.ProjectID,
+		SessionID:        q.SessionID,
+		ExcludeSessionID: q.ExcludeSessionID,
+		TopK:             q.TopK,
+		MinScore:         q.MinScore,
+	})
+	if err != nil {
+		return nil, err
+	}
+	out := make([]ctxmgr.RecalledMemoryItem, 0, len(items))
+	for _, item := range items {
+		ownerID := item.UserID
+		if item.SessionID != "" {
+			if sidOwner, err := a.Mgr.GetSessionOwnerID(ctx, item.SessionID); err == nil && sidOwner > 0 {
+				ownerID = sidOwner
+			}
+		}
+		var access *projectinstance.StageAccess
+		if item.ProjectID > 0 && a.Resolve != nil {
+			if resolved, err := a.Resolve(ctx, item.ProjectID, p.UserID); err == nil {
+				access = resolved
+			} else if a.FailClosed {
+				continue
+			}
+		}
+		if projectinstance.MessageRecordVisible("", "", item.Metadata, access, ownerID, p.UserID) {
+			out = append(out, item)
+		}
+	}
+	return out, nil
+}
+
+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
+}

internal/app/chatflow/chatlog_test.go

@@ -0,0 +1,145 @@
+package chatflow_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"chat-agent-system/internal/app/chatflow"
+	"chat-agent-system/internal/coordinator"
+	"chat-agent-system/internal/ctxmgr"
+	"chat-agent-system/internal/projectinstance"
+)
+
+func setupLog(t *testing.T) (*ctxmgr.ContextManager, *chatflow.Adapter, int64, string) {
+	t.Helper()
+	mgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{})
+	ctx := context.Background()
+	const ownerID int64 = 9
+	sess := "sess-u10-owner"
+	if _, err := mgr.EnsureChatSession(ctx, sess, ownerID, 42, "U10"); err != nil {
+		t.Fatal(err)
+	}
+	restricted := &projectinstance.StageAccess{
+		Restricted:       true,
+		ActiveStageKey:   "V0.1_triage",
+		AllowedStageKeys: map[string]bool{"V0.1_triage": true},
+	}
+	log := chatflow.NewAdapter(mgr, func(context.Context, int64, int64) (*projectinstance.StageAccess, error) {
+		return restricted, nil
+	}, true)
+	return mgr, log, ownerID, sess
+}
+
+func record(t *testing.T, mgr *ctxmgr.ContextManager, sess string, userID int64, content, scopeKind, stageKey string) {
+	t.Helper()
+	meta := map[string]any{"scope_kind": scopeKind, "stage_key": stageKey}
+	projectinstance.ApplyScopeToMetadata(meta, &projectinstance.MessageScope{ScopeKind: scopeKind, StageKey: stageKey})
+	if _, err := mgr.RecordUserMessageWithID(context.Background(), sess, userID, "user", content, "CASUAL_CHAT", "", meta); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestChatLog_WindowAndStatsSameSet(t *testing.T) {
+	mgr, log, ownerID, sess := setupLog(t)
+	ctx := context.Background()
+	p := chatflow.Principal{UserID: ownerID, Action: chatflow.ActionReadWindow}
+
+	record(t, mgr, sess, ownerID, "激活前沟通", projectinstance.ScopePreActivation, "")
+	record(t, mgr, sess, ownerID, "当前阶段任务", projectinstance.ScopeWorkflowStage, "V0.1_triage")
+	record(t, mgr, sess, ownerID, "报价阶段秘密", projectinstance.ScopeWorkflowStage, "V1.0_quoting")
+
+	win, err := log.GetWindow(ctx, p, ctxmgr.ChatWindowQuery{SessionID: sess, Days: 7})
+	if err != nil {
+		t.Fatal(err)
+	}
+	n := 0
+	for _, d := range win.Days {
+		n += len(d.Messages)
+	}
+	if n != 2 {
+		t.Fatalf("window visible=%d want 2", n)
+	}
+	cats, err := log.CategoryStats(ctx, p, sess)
+	if err != nil {
+		t.Fatal(err)
+	}
+	catTotal := 0
+	for _, c := range cats {
+		catTotal += c.Count
+	}
+	if catTotal != n {
+		t.Fatalf("category total %d != window %d", catTotal, n)
+	}
+	dates, err := log.DateStats(ctx, p, sess)
+	if err != nil {
+		t.Fatal(err)
+	}
+	dateTotal := 0
+	for _, d := range dates {
+		dateTotal += d.Count
+	}
+	if dateTotal != n {
+		t.Fatalf("date total %d != window %d", dateTotal, n)
+	}
+}
+
+func TestChatLog_RejectsNonOwnerAndQueryIdentity(t *testing.T) {
+	_, log, _, sess := setupLog(t)
+	ctx := context.Background()
+	_, err := log.GetWindow(ctx, chatflow.Principal{UserID: 99, Action: chatflow.ActionReadWindow}, ctxmgr.ChatWindowQuery{SessionID: sess, Days: 7})
+	if !errors.Is(err, chatflow.ErrForbidden) {
+		t.Fatalf("guest want forbidden, got %v", err)
+	}
+	_, err = log.GetWindow(ctx, chatflow.Principal{}, ctxmgr.ChatWindowQuery{SessionID: sess, Days: 7})
+	if !errors.Is(err, chatflow.ErrUnauthenticated) {
+		t.Fatalf("empty principal want unauthenticated, got %v", err)
+	}
+}
+
+func TestChatLog_RestoreAndDigestHideInvisible(t *testing.T) {
+	mgr, log, ownerID, sess := setupLog(t)
+	ctx := context.Background()
+	p := chatflow.Principal{UserID: ownerID, Action: chatflow.ActionAssemble}
+	record(t, mgr, sess, ownerID, "可见", projectinstance.ScopeWorkflowStage, "V0.1_triage")
+	record(t, mgr, sess, ownerID, "不可见报价", projectinstance.ScopeWorkflowStage, "V1.0_quoting")
+
+	restored, err := log.Restore(ctx, p, sess)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(restored.Messages) != 1 || restored.Messages[0].Content != "可见" {
+		t.Fatalf("restore leaked invisible: %+v", restored.Messages)
+	}
+
+	digest, err := log.GenerateDigest(ctx, p, sess, restored.Messages[0].DayKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if digest != nil && digest.MessageCount > 1 {
+		t.Fatalf("digest counted invisible messages: %+v", digest)
+	}
+}
+
+func TestAssembler_OnlyAuthorizedDialogue(t *testing.T) {
+	mgr, log, ownerID, sess := setupLog(t)
+	record(t, mgr, sess, ownerID, "可见对话", projectinstance.ScopePreActivation, "")
+	record(t, mgr, sess, ownerID, "隐藏阶段", projectinstance.ScopeWorkflowStage, "V1.0_quoting")
+	asm := chatflow.NewAssembler(nil, log)
+	ctx := context.Background()
+	userCtx, err := asm.Assemble(ctx, chatflow.Principal{UserID: ownerID}, chatflow.AssembleRequest{
+		SessionID: sess,
+		User:      coordinator.UserInfo{ID: "u9", Name: "张工", ProjectRole: "composer"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, turn := range userCtx.RecentDialogue {
+		if turn.Content == "隐藏阶段" {
+			t.Fatal("assembler injected invisible message into RecentDialogue")
+		}
+	}
+	if len(userCtx.RecentDialogue) == 0 {
+		t.Fatal("expected visible dialogue in assembled context")
+	}
+}

internal/app/chatflow/ports.go

@@ -0,0 +1,106 @@
+package chatflow
+
+import (
+	"context"
+	"errors"
+
+	"chat-agent-system/internal/coordinator"
+	"chat-agent-system/internal/ctxmgr"
+	"chat-agent-system/internal/projectinstance"
+)
+
+var (
+	ErrUnauthenticated = errors.New("chatflow: unauthenticated")
+	ErrForbidden       = errors.New("chatflow: forbidden")
+	ErrNotFound        = errors.New("chatflow: not found")
+	ErrClosed          = errors.New("chatflow: closed")
+)
+
+type Action string
+
+const (
+	ActionReadWindow Action = "chat.read_window"
+	ActionReadDay    Action = "chat.read_day"
+	ActionReadStats  Action = "chat.read_stats"
+	ActionSearch     Action = "chat.search"
+	ActionUnread     Action = "chat.unread"
+	ActionExport     Action = "chat.export"
+	ActionDigest     Action = "chat.digest"
+	ActionAssemble   Action = "context.assemble"
+)
+
+// Principal is taken only from the authenticated session. Callers must not
+// copy user/project identifiers from the request body or query into UserID.
+type Principal struct {
+	UserID int64
+	Action Action
+}
+
+type DayQuery struct {
+	SessionID string
+	Date      string
+	AnchorID  int64
+	BeforeID  int64
+	AfterID   int64
+	Limit     int
+}
+
+type HistoryQuery struct {
+	SessionID string
+	Limit     int
+	Offset    int
+}
+
+type UnreadQuery struct {
+	SessionID      string
+	SinceMessageID int64
+}
+
+type ExportQuery struct {
+	SessionID string
+	DateFrom  string
+	DateTo    string
+	Category  string
+}
+
+type MemoryQuery struct {
+	Query            string
+	ProjectID        int64
+	SessionID        string
+	ExcludeSessionID string
+	TopK             int
+	MinScore         float64
+}
+
+type AssembleRequest struct {
+	SessionID string
+	WorkDir   string
+	User      coordinator.UserInfo
+	Fallback  coordinator.ProjectInfo
+}
+
+type StageAccessResolver func(ctx context.Context, projectID, viewerUserID int64) (*projectinstance.StageAccess, error)
+
+// ChatLog is the consumer-side port for authorized chat_* reads.
+// Table ownership stays in ctxmgr; this port never accepts request-body identity as auth fact.
+type ChatLog interface {
+	GetWindow(ctx context.Context, p Principal, q ctxmgr.ChatWindowQuery) (*ctxmgr.ChatWindowResult, error)
+	GetDay(ctx context.Context, p Principal, q DayQuery) (*ctxmgr.ChatDayMessagesResult, error)
+	History(ctx context.Context, p Principal, q HistoryQuery) ([]ctxmgr.ChatMessageRecord, error)
+	ListVisible(ctx context.Context, p Principal, sessionID string) ([]ctxmgr.ChatMessageRecord, error)
+	CategoryStats(ctx context.Context, p Principal, sessionID string) ([]ctxmgr.ChatCategoryStat, error)
+	DateStats(ctx context.Context, p Principal, sessionID string) ([]ctxmgr.ChatDateStat, error)
+	Search(ctx context.Context, p Principal, q ctxmgr.ChatSearchQuery) ([]ctxmgr.ChatSearchHit, error)
+	Unread(ctx context.Context, p Principal, q UnreadQuery) (*ctxmgr.ChatUnreadSummary, error)
+	MarkRead(ctx context.Context, p Principal, req ctxmgr.MarkChatReadRequest) (*ctxmgr.ChatReadPosition, error)
+	Export(ctx context.Context, p Principal, q ExportQuery) ([]ctxmgr.ChatMessageRecord, error)
+	ListDigests(ctx context.Context, p Principal, sessionID, from, to string) ([]ctxmgr.DailyDigest, error)
+	GenerateDigest(ctx context.Context, p Principal, sessionID, dayKey string) (*ctxmgr.DailyDigest, error)
+	Restore(ctx context.Context, p Principal, sessionID string) (*ctxmgr.RestoredSessionContext, error)
+	SearchMemory(ctx context.Context, p Principal, q MemoryQuery) ([]ctxmgr.RecalledMemoryItem, error)
+}
+
+// ContextAssembler is the consumer-side port for prompt context assembly.
+type ContextAssembler interface {
+	Assemble(ctx context.Context, p Principal, req AssembleRequest) (*coordinator.UserProjectContext, error)
+}

internal/ctxmgr/chat_digest.go

@@ -17,10 +17,7 @@ func (m *ContextManager) GenerateDailyDigest(ctx context.Context, sessionID, day
 		return nil, fmt.Errorf("sessionID and dayKey are required")
 	}
 
-	// 1. 获取当日全部消息
 	var msgs []ChatMessageRecord
-	var projectID int64
-
 	if m.pool != nil {
 		rows, err := m.pool.Query(ctx, `
 			SELECT id, session_id, user_id, role, content, intent, branch, metadata, day_key::text, category, created_at
@@ -53,21 +50,33 @@ func (m *ContextManager) GenerateDailyDigest(ctx context.Context, sessionID, day
 			rec.CreatedAt = createdAt
 			msgs = append(msgs, rec)
 		}
-
-		// 获取 project_id
-		var pid *int64
-		_ = m.pool.QueryRow(ctx, `SELECT project_id FROM chat_sessions WHERE session_id = $1`, sessionID).Scan(&pid)
-		if pid != nil {
-			projectID = *pid
-		}
 	} else {
-		// 内存模式
 		m.mu.RLock()
 		for _, r := range m.messages[sessionID] {
 			if r.DayKey == dayKey {
 				msgs = append(msgs, r)
 			}
 		}
+		m.mu.RUnlock()
+	}
+
+	return m.GenerateDailyDigestFromMessages(ctx, sessionID, dayKey, msgs)
+}
+
+// GenerateDailyDigestFromMessages 仅使用调用方已授权的消息集合生成摘要，避免把不可见正文写入 Prompt/缓存。
+func (m *ContextManager) GenerateDailyDigestFromMessages(ctx context.Context, sessionID, dayKey string, msgs []ChatMessageRecord) (*DailyDigest, error) {
+	if sessionID == "" || dayKey == "" {
+		return nil, fmt.Errorf("sessionID and dayKey are required")
+	}
+	var projectID int64
+	if m.pool != nil {
+		var pid *int64
+		_ = m.pool.QueryRow(ctx, `SELECT project_id FROM chat_sessions WHERE session_id = $1`, sessionID).Scan(&pid)
+		if pid != nil {
+			projectID = *pid
+		}
+	} else {
+		m.mu.RLock()
 		if s, ok := m.sessions[sessionID]; ok {
 			projectID = s.ProjectID
 		}
@@ -89,10 +98,8 @@ func (m *ContextManager) GenerateDailyDigest(ctx context.Context, sessionID, day
 		}, nil
 	}
 
-	// 2. 规则/启发式提取摘要 (作为可靠保障)
 	digest := m.extractRuleDigest(sessionID, projectID, dayKey, msgs)
 
-	// 3. 持久化存储 (Postgres 或内存)
 	if m.pool != nil {
 		keyPointsJSON, _ := json.Marshal(digest.KeyPoints)
 		deliverablesJSON, _ := json.Marshal(digest.Deliverables)