From 18d40fa8a92d7e009fb65e5c5756bfca9cba7f0b Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 04:11:16 +0000 Subject: [PATCH] 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 --- ...-09-13-chat-history-refresh-remediation.md | 0 ...-09-13-chat-history-refresh-remediation.md | 0 .doc/system-design/module-boundaries.yaml | 7 + internal/app/chatflow/assembler.go | 106 +++++ internal/app/chatflow/chatlog.go | 443 ++++++++++++++++++ internal/app/chatflow/chatlog_test.go | 145 ++++++ internal/app/chatflow/ports.go | 106 +++++ internal/ctxmgr/chat_digest.go | 33 +- 8 files changed, 827 insertions(+), 13 deletions(-) rename .doc/{ops/short-term => archive/ops}/2026-09-13-chat-history-refresh-remediation.md (100%) rename .doc/{ => archive}/ops/checklists/2026-09-13-chat-history-refresh-remediation.md (100%) create mode 100644 internal/app/chatflow/assembler.go create mode 100644 internal/app/chatflow/chatlog.go create mode 100644 internal/app/chatflow/chatlog_test.go create mode 100644 internal/app/chatflow/ports.go diff --git a/.doc/ops/short-term/2026-09-13-chat-history-refresh-remediation.md b/.doc/archive/ops/2026-09-13-chat-history-refresh-remediation.md similarity index 100% rename from .doc/ops/short-term/2026-09-13-chat-history-refresh-remediation.md rename to .doc/archive/ops/2026-09-13-chat-history-refresh-remediation.md diff --git a/.doc/ops/checklists/2026-09-13-chat-history-refresh-remediation.md b/.doc/archive/ops/checklists/2026-09-13-chat-history-refresh-remediation.md similarity index 100% rename from .doc/ops/checklists/2026-09-13-chat-history-refresh-remediation.md rename to .doc/archive/ops/checklists/2026-09-13-chat-history-refresh-remediation.md diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 2793bef0..14748570 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.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 diff --git a/internal/app/chatflow/assembler.go b/internal/app/chatflow/assembler.go new file mode 100644 index 00000000..e310bb7c --- /dev/null +++ b/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 +} diff --git a/internal/app/chatflow/chatlog.go b/internal/app/chatflow/chatlog.go new file mode 100644 index 00000000..ab2a7930 --- /dev/null +++ b/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 +} diff --git a/internal/app/chatflow/chatlog_test.go b/internal/app/chatflow/chatlog_test.go new file mode 100644 index 00000000..556cb32a --- /dev/null +++ b/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") + } +} diff --git a/internal/app/chatflow/ports.go b/internal/app/chatflow/ports.go new file mode 100644 index 00000000..30c1067a --- /dev/null +++ b/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) +} diff --git a/internal/ctxmgr/chat_digest.go b/internal/ctxmgr/chat_digest.go index dec18f77..7222f4ed 100644 --- a/internal/ctxmgr/chat_digest.go +++ b/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)