fix(chat): stop treating failed summary writes as durable

新增 chat_summaries 迁移。数据库写失败会返回错误并标明仅本进程可见。test_manager 血缘改用文档版本，不再写死 1。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-20 04:02:41 +00:00
parent 033462a4e6
commit 848a0457a0
9 changed files with 337 additions and 47 deletions
@@ -3,25 +3,40 @@ package ctxmgr
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"log/slog"
 	"strings"
 	"time"
+
+	"github.com/jackc/pgx/v5"
 )

 // ChatSummary 会话滚动摘要数据模型
 type ChatSummary struct {
-	SessionID     string                `json:"session_id"`
-	Narrative     string                `json:"narrative"`      // 叙事摘要（≤300 字）；PU6 shadow 下始终为 deterministic
-	Facts         map[string]string     `json:"facts"`          // 硬事实槽位字典（永不被压缩稀释）
-	OpenQuestions []string              `json:"open_questions"` // 未决技术或商务待办
-	CoveredCount  int                   `json:"covered_count"`  // 累计已覆盖并压缩的历史消息条数
-	CoveredUntil  time.Time             `json:"covered_until"`  // 摘要覆盖消息的截止时间戳
-	UpdatedAt     time.Time             `json:"updated_at"`
-	Source        string                `json:"source,omitempty"`         // deterministic（served）
-	FreshnessNote string                `json:"freshness_note,omitempty"` // shadow 失败时的可解释回退说明
-	Shadow        *SemanticShadowReport `json:"shadow,omitempty"`         // advisory only；REPLACE 未接线
+	SessionID        string                `json:"session_id"`
+	Narrative        string                `json:"narrative"`      // 叙事摘要（≤300 字）；PU6 shadow 下始终为 deterministic
+	Facts            map[string]string     `json:"facts"`          // 硬事实槽位字典（永不被压缩稀释）
+	OpenQuestions    []string              `json:"open_questions"` // 未决技术或商务待办
+	CoveredCount     int                   `json:"covered_count"`  // 累计已覆盖并压缩的历史消息条数
+	CoveredUntil     time.Time             `json:"covered_until"`  // 摘要覆盖消息的截止时间戳
+	UpdatedAt        time.Time             `json:"updated_at"`
+	Source           string                `json:"source,omitempty"`            // deterministic（served）
+	FreshnessNote    string                `json:"freshness_note,omitempty"`    // shadow 失败或未持久化时的可解释回退说明
+	Shadow           *SemanticShadowReport `json:"shadow,omitempty"`            // advisory only；REPLACE 未接线
+	VisibilityDigest string                `json:"visibility_digest,omitempty"` // 授权可见集；未计算时为空
+	FromMessageID    int64                 `json:"from_message_id,omitempty"`
+	ToMessageID      int64                 `json:"to_message_id,omitempty"`
+	PreviousRevision int                   `json:"previous_revision,omitempty"`
+	Revision         int                   `json:"revision,omitempty"`
+	ProtocolVersion  string                `json:"protocol_version,omitempty"`
+	ModelVersion     string                `json:"model_version,omitempty"`
+	// Persisted 为 true 表示配置的存储接受了这次写入。有数据库时只有 INSERT 成功才为 true。
+	Persisted bool `json:"persisted,omitempty"`
 }

+const summaryNotDurableNote = "摘要未写入数据库，仅本进程可见"
+
 // RenderPromptSnippet 渲染为 Prompt 中的【早期对话摘要】段
 func (s *ChatSummary) RenderPromptSnippet() string {
 	if s == nil || (s.Narrative == "" && len(s.Facts) == 0) {
@@ -61,18 +76,13 @@ func (m *ContextManager) GetChatSummary(ctx context.Context, sessionID string) (
 	}

 	if m.pool != nil {
-		var sum ChatSummary
-		var factsJSON, openQuestionsJSON []byte
-		err := m.pool.QueryRow(ctx, `
-			SELECT session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at
-			FROM chat_summaries
-			WHERE session_id = $1
-		`, sessionID).Scan(&sum.SessionID, &sum.Narrative, &factsJSON, &openQuestionsJSON, &sum.CoveredCount, &sum.CoveredUntil, &sum.UpdatedAt)
-		if err == nil {
-			_ = json.Unmarshal(factsJSON, &sum.Facts)
-			_ = json.Unmarshal(openQuestionsJSON, &sum.OpenQuestions)
-			m.Cache().Set(LayerC5Summary, sessionID, &sum, 15)
-			return &sum, nil
+		sum, err := m.loadChatSummary(ctx, sessionID)
+		if err == nil && sum != nil {
+			m.Cache().Set(LayerC5Summary, sessionID, sum, 15)
+			return sum, nil
+		}
+		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+			slog.Warn("chat summary load failed", "session_id", sessionID, "err", err)
 		}
 	}

@@ -100,26 +110,87 @@ func (m *ContextManager) SaveChatSummary(ctx context.Context, sum *ChatSummary)
 	m.Cache().Invalidate(LayerC5Summary, sum.SessionID)
 	m.Cache().InvalidateLayer(LayerC4Context)

-	if m.pool != nil {
-		factsJSON, _ := json.Marshal(sum.Facts)
-		openQuestionsJSON, _ := json.Marshal(sum.OpenQuestions)
-		_, err := m.pool.Exec(ctx, `
-			INSERT INTO chat_summaries (session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at)
-			VALUES ($1, $2, $3, $4, $5, $6, $7)
-			ON CONFLICT (session_id) DO UPDATE SET
-				narrative = EXCLUDED.narrative,
-				facts = EXCLUDED.facts,
-				open_questions = EXCLUDED.open_questions,
-				covered_count = EXCLUDED.covered_count,
-				covered_until = EXCLUDED.covered_until,
-				updated_at = EXCLUDED.updated_at
-		`, sum.SessionID, sum.Narrative, factsJSON, openQuestionsJSON, sum.CoveredCount, sum.CoveredUntil, sum.UpdatedAt)
-		if err != nil {
-			// DB 失败时仍降级写入内存
-			_ = err
-		}
+	if m.pool == nil {
+		sum.Persisted = true
+		m.rememberSummary(sum)
+		return nil
 	}

+	factsJSON, err := json.Marshal(sum.Facts)
+	if err != nil {
+		return fmt.Errorf("chat summary facts: %w", err)
+	}
+	openQuestionsJSON, err := json.Marshal(sum.OpenQuestions)
+	if err != nil {
+		return fmt.Errorf("chat summary questions: %w", err)
+	}
+	_, err = m.pool.Exec(ctx, `
+		INSERT INTO chat_summaries (
+			session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at,
+			source, visibility_digest, from_message_id, to_message_id, previous_revision, revision,
+			protocol_version, model_version
+		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
+		ON CONFLICT (session_id) DO UPDATE SET
+			narrative = EXCLUDED.narrative,
+			facts = EXCLUDED.facts,
+			open_questions = EXCLUDED.open_questions,
+			covered_count = EXCLUDED.covered_count,
+			covered_until = EXCLUDED.covered_until,
+			updated_at = EXCLUDED.updated_at,
+			source = EXCLUDED.source,
+			visibility_digest = EXCLUDED.visibility_digest,
+			from_message_id = EXCLUDED.from_message_id,
+			to_message_id = EXCLUDED.to_message_id,
+			previous_revision = EXCLUDED.previous_revision,
+			revision = EXCLUDED.revision,
+			protocol_version = EXCLUDED.protocol_version,
+			model_version = EXCLUDED.model_version
+	`, sum.SessionID, sum.Narrative, factsJSON, openQuestionsJSON, sum.CoveredCount, sum.CoveredUntil, sum.UpdatedAt,
+		sum.Source, sum.VisibilityDigest, nullSummaryID(sum.FromMessageID), nullSummaryID(sum.ToMessageID),
+		sum.PreviousRevision, sum.Revision, sum.ProtocolVersion, sum.ModelVersion)
+	if err != nil {
+		sum.Persisted = false
+		noteSummaryNotDurable(sum)
+		m.rememberSummary(sum)
+		slog.Warn("chat summary persist failed", "session_id", sum.SessionID, "err", err)
+		return fmt.Errorf("chat summary persist failed: %w", err)
+	}
+	sum.Persisted = true
+	m.rememberSummary(sum)
+	return nil
+}
+
+func (m *ContextManager) loadChatSummary(ctx context.Context, sessionID string) (*ChatSummary, error) {
+	var sum ChatSummary
+	var factsJSON, openQuestionsJSON []byte
+	var fromID, toID *int64
+	err := m.pool.QueryRow(ctx, `
+		SELECT session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at,
+		       source, visibility_digest, from_message_id, to_message_id, previous_revision, revision,
+		       protocol_version, model_version
+		FROM chat_summaries
+		WHERE session_id = $1
+	`, sessionID).Scan(
+		&sum.SessionID, &sum.Narrative, &factsJSON, &openQuestionsJSON, &sum.CoveredCount, &sum.CoveredUntil, &sum.UpdatedAt,
+		&sum.Source, &sum.VisibilityDigest, &fromID, &toID, &sum.PreviousRevision, &sum.Revision,
+		&sum.ProtocolVersion, &sum.ModelVersion,
+	)
+	if err != nil {
+		return nil, err
+	}
+	_ = json.Unmarshal(factsJSON, &sum.Facts)
+	_ = json.Unmarshal(openQuestionsJSON, &sum.OpenQuestions)
+	if fromID != nil {
+		sum.FromMessageID = *fromID
+	}
+	if toID != nil {
+		sum.ToMessageID = *toID
+	}
+	sum.Persisted = true
+	return &sum, nil
+}
+
+func (m *ContextManager) rememberSummary(sum *ChatSummary) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if m.summaries == nil {
@@ -127,7 +198,24 @@ func (m *ContextManager) SaveChatSummary(ctx context.Context, sum *ChatSummary)
 	}
 	copySum := *sum
 	m.summaries[sum.SessionID] = &copySum
-	return nil
+}
+
+func nullSummaryID(id int64) any {
+	if id <= 0 {
+		return nil
+	}
+	return id
+}
+
+func noteSummaryNotDurable(sum *ChatSummary) {
+	if sum == nil || strings.Contains(sum.FreshnessNote, summaryNotDurableNote) {
+		return
+	}
+	if sum.FreshnessNote == "" {
+		sum.FreshnessNote = summaryNotDurableNote
+		return
+	}
+	sum.FreshnessNote += "；" + summaryNotDurableNote
 }

 // GenerateRollingSummary 结合已覆盖旧摘要与本批被淘汰出窗口的早期消息，滚动增量压缩生成新摘要
@@ -192,12 +280,27 @@ func (m *ContextManager) GenerateRollingSummary(ctx context.Context, sessionID s
 	}
 	newSum.Narrative = fullNarrative
 	newSum.Source = "deterministic"
+	newSum.ProtocolVersion = "deterministic-v1"
+	if len(earlyMsgs) > 0 {
+		newSum.FromMessageID = earlyMsgs[0].ID
+		newSum.ToMessageID = earlyMsgs[len(earlyMsgs)-1].ID
+	}
+	if oldSum != nil {
+		prev := oldSum.Revision
+		if prev < 1 {
+			prev = 1
+		}
+		newSum.PreviousRevision = prev
+		newSum.Revision = prev + 1
+	} else {
+		newSum.Revision = 1
+	}

 	// SEMANTIC_SUMMARY_SHADOW：仅在授权可见集合上评估语义摘要；失败保留 deterministic + freshness。
 	// SEMANTIC_SUMMARY_REPLACE 故意不接线，served Narrative 永不被语义结果替换。
 	applySemanticShadow(ctx, newSum, earlyMsgs, m.semanticSummarizerOrDefault())

-	// 保存并返回
+	// 写库失败仍返回确定性摘要，供本轮使用；错误留在 Persisted/FreshnessNote，不伪装成已持久化。
 	_ = m.SaveChatSummary(ctx, newSum)
 	return newSum, nil
 }
@@ -0,0 +1,79 @@
+package ctxmgr
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func TestSaveChatSummary_MemoryStoreCountsAsPersisted(t *testing.T) {
+	mgr := NewContextManager(nil, nil, DefaultTimeDecayConfig())
+	sum := &ChatSummary{SessionID: "sess_mem", Narrative: "仅内存", Facts: map[string]string{}, OpenQuestions: []string{}}
+	if err := mgr.SaveChatSummary(context.Background(), sum); err != nil {
+		t.Fatal(err)
+	}
+	if !sum.Persisted {
+		t.Fatal("nil pool is the configured store and must report persisted")
+	}
+}
+
+func TestSaveChatSummary_DatabaseFailureIsNotSuccess(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	pool, err := pgxpool.New(ctx, "postgres://summary:summary@127.0.0.1:1/none?sslmode=disable&connect_timeout=1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(pool.Close)
+
+	mgr := NewContextManager(pool, nil, DefaultTimeDecayConfig())
+	sum := &ChatSummary{
+		SessionID: "sess_db_down", Narrative: "不应伪装成功",
+		Facts: map[string]string{}, OpenQuestions: []string{},
+	}
+	err = mgr.SaveChatSummary(ctx, sum)
+	if err == nil {
+		t.Fatal("database write failure must be returned")
+	}
+	if sum.Persisted {
+		t.Fatal("failed write must not be marked persisted")
+	}
+	if !strings.Contains(sum.FreshnessNote, summaryNotDurableNote) {
+		t.Fatalf("freshness note = %q", sum.FreshnessNote)
+	}
+	got, getErr := mgr.GetChatSummary(context.Background(), sum.SessionID)
+	if getErr != nil || got == nil {
+		t.Fatalf("current process should still serve the summary, err=%v", getErr)
+	}
+	if got.Persisted || !strings.Contains(got.FreshnessNote, summaryNotDurableNote) {
+		t.Fatalf("served fallback must stay non-durable: %+v", got)
+	}
+}
+
+func TestGenerateRollingSummary_RecordsRevisionAndMessageSpan(t *testing.T) {
+	mgr := NewContextManager(nil, nil, DefaultTimeDecayConfig())
+	ctx := context.Background()
+	first := []ChatMessageRecord{{
+		ID: 10, Role: "user", Content: "临床评价路径该怎么走？", CreatedAt: time.Now(),
+	}}
+	sum, err := mgr.GenerateRollingSummary(ctx, "sess_rev", nil, first)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !sum.Persisted || sum.Revision != 1 || sum.FromMessageID != 10 || sum.ToMessageID != 10 || sum.ProtocolVersion != "deterministic-v1" {
+		t.Fatalf("first summary = %+v", sum)
+	}
+	second := []ChatMessageRecord{{
+		ID: 11, Role: "user", Content: "预算大约在200万左右", CreatedAt: time.Now(),
+	}}
+	next, err := mgr.GenerateRollingSummary(ctx, "sess_rev", sum, second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if next.Revision != 2 || next.PreviousRevision != 1 || next.FromMessageID != 11 || next.Facts["budget"] != "200万元" {
+		t.Fatalf("rolled summary = %+v", next)
+	}
+}