Merge pull request #1895 from pikasTech/fix/1870-p6-cache-consistency

fix: 补齐 Workbench Redis cache freshness 诊断

internal/workbenchruntime/cache_test.go

@@ -158,6 +158,38 @@ func TestSessionsSummaryCacheKeyActorIsolation(t *testing.T) {
 	}
 }
 
+func TestSessionsSummaryRevisionMetadataExplainsFreshness(t *testing.T) {
+	payload := map[string]any{"sessions": []any{
+		map[string]any{"sessionId": "ses_1", "projectedSeq": int64(12), "updatedAt": "2026-06-22T06:00:00Z", "projection": map[string]any{"lastProjectedSeq": int64(30)}},
+		map[string]any{"sessionId": "ses_2", "projectedSeq": int64(44), "updatedAt": "2026-06-22T06:01:00Z", "turnSummary": map[string]any{"eventCount": int64(41)}},
+	}}
+	revision := sessionsSummaryRevisionForPayload(payload)
+	if revision.SessionCount != 2 || revision.ProjectionSeqMax != 44 || revision.SessionUpdatedAtMax != "2026-06-22T06:01:00Z" {
+		t.Fatalf("unexpected revision: %#v", revision)
+	}
+	if !strings.HasPrefix(revision.ProjectionRevision, "digest:") {
+		t.Fatalf("revision should use a bounded digest, got %q", revision.ProjectionRevision)
+	}
+
+	meta := sessionsCacheMetadata("miss")
+	applySessionsCacheRevisionMetadata(meta, revision)
+	if meta["invalidationMode"] != "ttl_slo" || meta["revisionSource"] != "response_payload" {
+		t.Fatalf("unexpected freshness metadata: %#v", meta)
+	}
+	if meta["projectionRevision"] == nil || meta["projectionSeq"] != int64(44) || meta["sessionCount"] != 2 {
+		t.Fatalf("missing revision metadata: %#v", meta)
+	}
+}
+
+func TestProjectionRevisionTokenUsesProjectedSeq(t *testing.T) {
+	if got := projectionRevisionToken(42); got != "seq:42" {
+		t.Fatalf("projection revision token=%q", got)
+	}
+	if got := projectionRevisionToken(0); got != "" {
+		t.Fatalf("zero projection revision token=%q", got)
+	}
+}
+
 func TestFactsCacheClassAndTTL(t *testing.T) {
 	tracePage := factQuery{TraceID: "trc_1", Families: []string{"traceEvents"}, AfterProjectedSeq: 10, Limit: 51}
 	if got := factsCacheClass(tracePage); got != terminalTracePageCacheClass {

internal/workbenchruntime/service.go

@@ -30,11 +30,15 @@ const (
 )
 
 type sessionsCacheEntry struct {
-	CachedAt       string         `json:"cachedAt"`
+	CachedAt            string         `json:"cachedAt"`
-	Payload        map[string]any `json:"payload"`
+	Payload             map[string]any `json:"payload"`
-	PayloadBytes   int            `json:"payloadBytes"`
+	PayloadBytes        int            `json:"payloadBytes"`
-	TTLMillis      int64          `json:"ttlMs"`
+	TTLMillis           int64          `json:"ttlMs"`
-	ValuesRedacted bool           `json:"valuesRedacted"`
+	ProjectionRevision  string         `json:"projectionRevision,omitempty"`
+	ProjectionSeqMax    int64          `json:"projectionSeqMax,omitempty"`
+	SessionUpdatedAtMax string         `json:"sessionUpdatedAtMax,omitempty"`
+	SessionCount        int            `json:"sessionCount,omitempty"`
+	ValuesRedacted      bool           `json:"valuesRedacted"`
 }
 
 type factsCacheEntry struct {
@@ -56,6 +60,14 @@ type factsCachePlan struct {
 	Status           string
 }
 
+type sessionsSummaryRevision struct {
+	ProjectionRevision  string
+	ProjectionSeqMax    int64
+	SessionUpdatedAtMax string
+	SessionCount        int
+	RevisionSource      string
+}
+
 type Config struct {
 	Addr           string
 	DatabaseURL    string
@@ -283,11 +295,13 @@ func (s *Server) queryFactsResponse(ctx context.Context, query factQuery) (map[s
 				plan.Meta["dbQueryDurationMs"] = int64(0)
 				plan.Meta["payloadBytes"] = entry.PayloadBytes
 				plan.Meta["ttlMs"] = entry.TTLMillis
+				plan.Meta["freshnessSloMs"] = entry.TTLMillis
 				attachFactsCacheMetadata(payload, plan.Meta)
 				return payload, nil
 			}
 			setFactsCacheStatus(plan.Meta, "stale")
 			plan.Meta["cacheAgeMs"] = ageMs
+			plan.Meta["freshnessSloMs"] = entry.TTLMillis
 		}
 	}
 
@@ -308,6 +322,7 @@ func (s *Server) queryFactsResponse(ctx context.Context, query factQuery) (map[s
 	ttl, unstable, ttlReason := factsCacheTTL(plan, s.config.Cache)
 	plan.Meta["unstable"] = unstable
 	plan.Meta["ttlMs"] = ttl.Milliseconds()
+	plan.Meta["freshnessSloMs"] = ttl.Milliseconds()
 	plan.Meta["ttlReason"] = ttlReason
 	storedPayload := factsCacheStorablePayload(payload)
 	payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes)
@@ -377,9 +392,12 @@ func (s *Server) factsCachePlan(ctx context.Context, query factQuery) factsCache
 	plan.ProjectionHealth = state.ProjectionHealth
 	plan.Status = state.Status
 	meta["projectionSeq"] = state.ProjectedSeq
+	meta["projectionRevision"] = projectionRevisionToken(state.ProjectedSeq)
 	meta["projectionStatus"] = state.ProjectionStatus
 	meta["projectionHealth"] = state.ProjectionHealth
 	meta["traceStatus"] = state.Status
+	meta["revisionSource"] = "workbench_projection_checkpoints"
+	meta["invalidationMode"] = "projection_revision_key"
 	if state.ProjectedSeq <= 0 {
 		setFactsCacheStatus(meta, "skipped")
 		meta["skipReason"] = "missing_projection_revision"
@@ -513,6 +531,10 @@ func factsCacheMetadata(class string, status string) map[string]any {
 		"dbQueryDurationMs":    nil,
 		"payloadBytes":         nil,
 		"ttlMs":                nil,
+		"freshnessSloMs":       nil,
+		"projectionRevision":   nil,
+		"invalidationMode":     nil,
+		"revisionSource":       nil,
 		"valuesRedacted":       true,
 		"secretMaterialStored": false,
 	}
@@ -707,11 +729,15 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[
 				cacheMeta["dbQueryDurationMs"] = int64(0)
 				cacheMeta["payloadBytes"] = entry.PayloadBytes
 				cacheMeta["ttlMs"] = entry.TTLMillis
+				cacheMeta["freshnessSloMs"] = entry.TTLMillis
+				applySessionsCacheRevisionMetadata(cacheMeta, sessionsSummaryRevision{ProjectionRevision: entry.ProjectionRevision, ProjectionSeqMax: entry.ProjectionSeqMax, SessionUpdatedAtMax: entry.SessionUpdatedAtMax, SessionCount: entry.SessionCount, RevisionSource: "cached_payload"})
 				attachSessionsCacheMetadata(payload, cacheMeta)
 				return payload, nil
 			}
 			setSessionsCacheStatus(cacheMeta, "stale")
 			cacheMeta["cacheAgeMs"] = ageMs
+			cacheMeta["freshnessSloMs"] = entry.TTLMillis
+			applySessionsCacheRevisionMetadata(cacheMeta, sessionsSummaryRevision{ProjectionRevision: entry.ProjectionRevision, ProjectionSeqMax: entry.ProjectionSeqMax, SessionUpdatedAtMax: entry.SessionUpdatedAtMax, SessionCount: entry.SessionCount, RevisionSource: "cached_payload"})
 		}
 	}
 
@@ -731,7 +757,10 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[
 	ttl, unstable, ttlReason := sessionsSummaryCacheTTL(payload, s.config.Cache)
 	cacheMeta["unstable"] = unstable
 	cacheMeta["ttlMs"] = ttl.Milliseconds()
+	cacheMeta["freshnessSloMs"] = ttl.Milliseconds()
 	cacheMeta["ttlReason"] = ttlReason
+	revision := sessionsSummaryRevisionForPayload(payload)
+	applySessionsCacheRevisionMetadata(cacheMeta, revision)
 	storedPayload := sessionsCacheStorablePayload(payload)
 	payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes)
 	if payloadErr != nil {
@@ -747,7 +776,7 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[
 		attachSessionsCacheMetadata(payload, cacheMeta)
 		return payload, nil
 	}
-	entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: storedPayload, PayloadBytes: len(payloadBytes), TTLMillis: ttl.Milliseconds(), ValuesRedacted: true}
+	entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: storedPayload, PayloadBytes: len(payloadBytes), TTLMillis: ttl.Milliseconds(), ProjectionRevision: revision.ProjectionRevision, ProjectionSeqMax: revision.ProjectionSeqMax, SessionUpdatedAtMax: revision.SessionUpdatedAtMax, SessionCount: revision.SessionCount, ValuesRedacted: true}
 	if err := s.cache.Set(ctx, cacheKey, entry, ttl); err != nil {
 		setSessionsCacheStatus(cacheMeta, "set_failed")
 		cacheMeta["errorKind"] = classifyCacheError(err)
@@ -863,6 +892,13 @@ func sessionsCacheMetadata(status string) map[string]any {
 		"dbQueryDurationMs":    nil,
 		"payloadBytes":         nil,
 		"ttlMs":                nil,
+		"freshnessSloMs":       nil,
+		"projectionRevision":   nil,
+		"projectionSeq":        nil,
+		"sessionUpdatedAtMax":  nil,
+		"sessionCount":         nil,
+		"revisionSource":       nil,
+		"invalidationMode":     "ttl_slo",
 		"valuesRedacted":       true,
 		"secretMaterialStored": false,
 	}
@@ -901,6 +937,68 @@ func sessionsSummaryCacheTTL(payload map[string]any, config cacheConfig) (time.D
 	return config.SessionsSummaryTTL, false, "stable_page"
 }
 
+func sessionsSummaryRevisionForPayload(payload map[string]any) sessionsSummaryRevision {
+	rows := sessionsPayloadRows(payload["sessions"])
+	revision := sessionsSummaryRevision{RevisionSource: "response_payload"}
+	maxUpdatedAt := ""
+	var maxSeq int64
+	for _, row := range rows {
+		item := objectMap(row)
+		if len(item) == 0 {
+			continue
+		}
+		revision.SessionCount++
+		if updatedAt := text(item["updatedAt"]); updatedAt > maxUpdatedAt {
+			maxUpdatedAt = updatedAt
+		}
+		maxSeq = maxInt64(maxSeq, int64FromAny(item["projectedSeq"]))
+		projection := objectMap(item["projection"])
+		maxSeq = maxInt64(maxSeq, int64FromAny(projection["lastProjectedSeq"]))
+		turnSummary := objectMap(item["turnSummary"])
+		maxSeq = maxInt64(maxSeq, int64FromAny(turnSummary["eventCount"]))
+	}
+	revision.ProjectionSeqMax = maxSeq
+	revision.SessionUpdatedAtMax = maxUpdatedAt
+	raw := fmt.Sprintf("contract=workbench-sessions-v1|count=%d|maxProjectionSeq=%d|maxUpdatedAt=%s", revision.SessionCount, revision.ProjectionSeqMax, revision.SessionUpdatedAtMax)
+	revision.ProjectionRevision = "digest:" + cacheHash(raw)
+	return revision
+}
+
+func applySessionsCacheRevisionMetadata(meta map[string]any, revision sessionsSummaryRevision) {
+	if meta == nil {
+		return
+	}
+	if revision.ProjectionRevision != "" {
+		meta["projectionRevision"] = revision.ProjectionRevision
+	}
+	meta["projectionSeq"] = revision.ProjectionSeqMax
+	meta["sessionUpdatedAtMax"] = nilIfEmpty(revision.SessionUpdatedAtMax)
+	meta["sessionCount"] = revision.SessionCount
+	meta["revisionSource"] = first(revision.RevisionSource, "response_payload")
+	meta["invalidationMode"] = "ttl_slo"
+}
+
+func projectionRevisionToken(projectedSeq int64) string {
+	if projectedSeq <= 0 {
+		return ""
+	}
+	return "seq:" + strconv.FormatInt(projectedSeq, 10)
+}
+
+func maxInt64(a int64, b int64) int64 {
+	if b > a {
+		return b
+	}
+	return a
+}
+
+func nilIfEmpty(value string) any {
+	if strings.TrimSpace(value) == "" {
+		return nil
+	}
+	return value
+}
+
 func sessionsSummaryPayloadUnstable(payload map[string]any) bool {
 	for _, row := range sessionsPayloadRows(payload["sessions"]) {
 		item := objectMap(row)
@@ -1221,8 +1319,9 @@ func (s *Server) querySessionSummariesAttempt(ctx context.Context, stage string,
 		defer rows.Close()
 		for rows.Next() {
 			var sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID sql.NullString
+			var projectedSeq sql.NullInt64
 			var createdAt, updatedAt sql.NullString
-			if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &createdAt, &updatedAt); err != nil {
+			if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &createdAt, &updatedAt); err != nil {
 				return err
 			}
 			if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" {
@@ -1237,6 +1336,7 @@ func (s *Server) querySessionSummariesAttempt(ctx context.Context, stage string,
 				"threadId":       nullableString(threadID),
 				"status":         first(nullableStringValue(status), "unknown"),
 				"lastTraceId":    nullableString(lastTraceID),
+				"projectedSeq":   nullableInt(projectedSeq),
 				"createdAt":      nullableTimeText(createdAt),
 				"updatedAt":      first(nullableTimeText(updatedAt), nullableTimeText(createdAt)),
 				"valuesPrinted":  false,
@@ -1601,6 +1701,7 @@ func sessionSummarySelectClause() string {
 		"thread_id",
 		"status",
 		"last_trace_id",
+		"projected_seq",
 		"created_at",
 		"updated_at",
 	}, ", ")