Merge pull request #1893 from pikasTech/feat/hwlab-1870-p5-terminal-trace-cache

feat: cache terminal workbench trace reads

internal/workbenchruntime/cache_test.go

@@ -158,6 +158,58 @@ func TestSessionsSummaryCacheKeyActorIsolation(t *testing.T) {
 	}
 }
 
+func TestFactsCacheClassAndTTL(t *testing.T) {
+	tracePage := factQuery{TraceID: "trc_1", Families: []string{"traceEvents"}, AfterProjectedSeq: 10, Limit: 51}
+	if got := factsCacheClass(tracePage); got != terminalTracePageCacheClass {
+		t.Fatalf("trace page class=%q", got)
+	}
+	turnSnapshot := factQuery{TraceID: "trc_1", Families: []string{"sessions", "messages", "turns", "checkpoints"}, Limit: 100}
+	if got := factsCacheClass(turnSnapshot); got != terminalTurnCacheClass {
+		t.Fatalf("turn snapshot class=%q", got)
+	}
+	uncacheable := factQuery{TraceID: "trc_1", Families: []string{"sessions", "messages"}}
+	if got := factsCacheClass(uncacheable); got != "" {
+		t.Fatalf("unexpected cache class=%q", got)
+	}
+
+	config := cacheConfig{TerminalTurnTTL: 5 * time.Minute, TerminalTracePageTTL: 10 * time.Minute, RunningPageTTL: 5 * time.Second}
+	ttl, unstable, reason := factsCacheTTL(factsCachePlan{Class: terminalTracePageCacheClass, ProjectionStatus: "caught-up", Status: "completed"}, config)
+	if ttl != 10*time.Minute || unstable || reason != "terminal_trace_page" {
+		t.Fatalf("terminal trace ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
+	}
+	ttl, unstable, reason = factsCacheTTL(factsCachePlan{Class: terminalTurnCacheClass, ProjectionStatus: "projecting", Status: "running"}, config)
+	if ttl != 5*time.Second || !unstable || reason != "running_or_projecting_trace" {
+		t.Fatalf("running turn ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
+	}
+	ttl, unstable, reason = factsCacheTTL(factsCachePlan{Class: terminalTracePageCacheClass, ProjectionStatus: "blocked", Status: "blocked"}, config)
+	if ttl != 10*time.Minute || unstable || reason != "terminal_trace_page" {
+		t.Fatalf("blocked trace ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
+	}
+}
+
+func TestFactsCacheStorablePayloadRemovesRuntimeOnlySecretKeys(t *testing.T) {
+	payload := map[string]any{
+		"contractVersion":      "workbench-runtime-facts-v1",
+		"secretMaterialStored": false,
+		"persistence":          map[string]any{"cache": map[string]any{"secretMaterialStored": false}},
+		"facts": map[string]any{
+			"traceEvents": []any{map[string]any{"traceId": "trc_1", "projectedSeq": 1, "valuesRedacted": true}},
+		},
+		"valuesRedacted": true,
+	}
+	stored := factsCacheStorablePayload(payload)
+	if _, ok := stored["secretMaterialStored"]; ok {
+		t.Fatalf("facts storable payload kept secretMaterialStored: %#v", stored)
+	}
+	if _, ok := stored["persistence"]; ok {
+		t.Fatalf("facts storable payload kept persistence: %#v", stored)
+	}
+	entry := factsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: stored, PayloadBytes: 128, TTLMillis: 300000, ValuesRedacted: true}
+	if _, err := cachePayloadBytes(entry, 4096); err != nil {
+		t.Fatalf("storable facts payload rejected: %v", err)
+	}
+}
+
 func cacheErrKind(err error) string {
 	var cacheErr cacheError
 	if errors.As(err, &cacheErr) {

internal/workbenchruntime/service.go

@@ -25,6 +25,8 @@ const (
 	sessionSummaryQueryAttemptTimeout = 2 * time.Second
 	sessionSummaryQueryRetryBackoff   = 150 * time.Millisecond
 	sessionsSummaryCacheClass         = "sessions.summary"
+	terminalTurnCacheClass            = "turn.terminal.snapshot"
+	terminalTracePageCacheClass       = "trace.terminal.events"
 )
 
 type sessionsCacheEntry struct {
@@ -35,6 +37,25 @@ type sessionsCacheEntry struct {
 	ValuesRedacted bool           `json:"valuesRedacted"`
 }
 
+type factsCacheEntry struct {
+	CachedAt       string         `json:"cachedAt"`
+	Payload        map[string]any `json:"payload"`
+	PayloadBytes   int            `json:"payloadBytes"`
+	TTLMillis      int64          `json:"ttlMs"`
+	ValuesRedacted bool           `json:"valuesRedacted"`
+}
+
+type factsCachePlan struct {
+	Class            string
+	Key              string
+	Meta             map[string]any
+	TraceID          string
+	ProjectionSeq    int64
+	ProjectionStatus string
+	ProjectionHealth string
+	Status           string
+}
+
 type Config struct {
 	Addr           string
 	DatabaseURL    string
@@ -236,22 +257,338 @@ func (s *Server) handleFacts(w http.ResponseWriter, r *http.Request) {
 	}
 	ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout)
 	defer cancel()
-	facts, err := s.queryFacts(ctx, query)
+	payload, err := s.queryFactsResponse(ctx, query)
 	if err != nil {
 		writeQueryError(w, err)
 		return
 	}
+	writeJSON(w, http.StatusOK, payload)
+}
+
+func (s *Server) queryFactsResponse(ctx context.Context, query factQuery) (map[string]any, error) {
+	plan := s.factsCachePlan(ctx, query)
+	if plan.Key != "" {
+		var entry factsCacheEntry
+		hit, err := s.cache.Get(ctx, plan.Key, &entry)
+		if err != nil {
+			setFactsCacheStatus(plan.Meta, "unavailable")
+			plan.Meta["errorKind"] = classifyCacheError(err)
+		} else if hit && entry.Payload != nil {
+			ageMs, stale := factsCacheEntryAgeMs(entry, time.Now().UTC())
+			if !stale {
+				payload := factsCacheResponsePayload(entry.Payload, s.persistenceSummary())
+				setFactsCacheStatus(plan.Meta, "hit")
+				plan.Meta["cacheAgeMs"] = ageMs
+				plan.Meta["dbQueryAvoided"] = true
+				plan.Meta["dbQueryDurationMs"] = int64(0)
+				plan.Meta["payloadBytes"] = entry.PayloadBytes
+				plan.Meta["ttlMs"] = entry.TTLMillis
+				attachFactsCacheMetadata(payload, plan.Meta)
+				return payload, nil
+			}
+			setFactsCacheStatus(plan.Meta, "stale")
+			plan.Meta["cacheAgeMs"] = ageMs
+		}
+	}
+
+	dbStart := time.Now()
+	facts, err := s.queryFacts(ctx, query)
+	dbDuration := time.Since(dbStart)
+	if err != nil {
+		return nil, err
+	}
+	payload := factsResponsePayload(facts, s.persistenceSummary())
+	plan.Meta["dbQueryAvoided"] = false
+	plan.Meta["dbQueryDurationMs"] = dbDuration.Milliseconds()
+	if plan.Key == "" {
+		attachFactsCacheMetadata(payload, plan.Meta)
+		return payload, nil
+	}
+
+	ttl, unstable, ttlReason := factsCacheTTL(plan, s.config.Cache)
+	plan.Meta["unstable"] = unstable
+	plan.Meta["ttlMs"] = ttl.Milliseconds()
+	plan.Meta["ttlReason"] = ttlReason
+	storedPayload := factsCacheStorablePayload(payload)
+	payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes)
+	if payloadErr != nil {
+		setFactsCacheStatus(plan.Meta, "skipped")
+		plan.Meta["skipReason"] = classifyCacheError(payloadErr)
+		attachFactsCacheMetadata(payload, plan.Meta)
+		return payload, nil
+	}
+	plan.Meta["payloadBytes"] = len(payloadBytes)
+	if ttl <= 0 {
+		setFactsCacheStatus(plan.Meta, "skipped")
+		plan.Meta["skipReason"] = ttlReason
+		attachFactsCacheMetadata(payload, plan.Meta)
+		return payload, nil
+	}
+	entry := factsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: storedPayload, PayloadBytes: len(payloadBytes), TTLMillis: ttl.Milliseconds(), ValuesRedacted: true}
+	if err := s.cache.Set(ctx, plan.Key, entry, ttl); err != nil {
+		setFactsCacheStatus(plan.Meta, "set_failed")
+		plan.Meta["errorKind"] = classifyCacheError(err)
+		attachFactsCacheMetadata(payload, plan.Meta)
+		return payload, nil
+	}
+	if plan.Meta["status"] != "stale" {
+		setFactsCacheStatus(plan.Meta, "miss")
+	}
+	plan.Meta["stored"] = true
+	attachFactsCacheMetadata(payload, plan.Meta)
+	return payload, nil
+}
+
+func factsResponsePayload(facts map[string][]any, persistence map[string]any) map[string]any {
 	count := 0
 	for _, rows := range facts {
 		count += len(rows)
 	}
-	writeJSON(w, http.StatusOK, map[string]any{
+	return map[string]any{
 		"contractVersion": "workbench-runtime-facts-v1",
 		"facts":           facts,
 		"count":           count,
-		"persistence":     s.persistenceSummary(),
+		"persistence":     persistence,
 		"valuesRedacted":  true,
-	})
+	}
+}
+
+func (s *Server) factsCachePlan(ctx context.Context, query factQuery) factsCachePlan {
+	class := factsCacheClass(query)
+	meta := factsCacheMetadata(class, "disabled")
+	plan := factsCachePlan{Class: class, Meta: meta, TraceID: strings.TrimSpace(query.TraceID)}
+	if s == nil || s.cache == nil || !s.config.Cache.Enabled || class == "" {
+		return plan
+	}
+	meta["enabled"] = true
+	if plan.TraceID == "" {
+		setFactsCacheStatus(meta, "skipped")
+		meta["skipReason"] = "missing_trace"
+		return plan
+	}
+	state, err := s.traceProjectionState(ctx, plan.TraceID)
+	if err != nil {
+		setFactsCacheStatus(meta, "skipped")
+		meta["skipReason"] = "projection_revision_unavailable"
+		return plan
+	}
+	plan.ProjectionSeq = state.ProjectedSeq
+	plan.ProjectionStatus = state.ProjectionStatus
+	plan.ProjectionHealth = state.ProjectionHealth
+	plan.Status = state.Status
+	meta["projectionSeq"] = state.ProjectedSeq
+	meta["projectionStatus"] = state.ProjectionStatus
+	meta["projectionHealth"] = state.ProjectionHealth
+	meta["traceStatus"] = state.Status
+	if state.ProjectedSeq <= 0 {
+		setFactsCacheStatus(meta, "skipped")
+		meta["skipReason"] = "missing_projection_revision"
+		return plan
+	}
+	cursor := fmt.Sprintf("after:%d|limit:%d|families:%s", query.AfterProjectedSeq, boundedLimit(query.Limit, 0, 1000), strings.Join(requestedFactFamilies(query), ","))
+	authority := fmt.Sprintf("contract=workbench-runtime-facts-v1|class=%s|projectionStatus=%s|projectionHealth=%s", class, state.ProjectionStatus, state.ProjectionHealth)
+	key, err := s.cache.BuildKey(cacheKeyParts{Class: class, ActorID: "workbench-runtime", TraceID: plan.TraceID, Cursor: cursor, ProjectionSeq: state.ProjectedSeq, Authority: authority})
+	if err != nil {
+		setFactsCacheStatus(meta, "skipped")
+		meta["skipReason"] = classifyCacheError(err)
+		return plan
+	}
+	plan.Key = key
+	setFactsCacheStatus(meta, "miss")
+	return plan
+}
+
+type traceProjectionState struct {
+	ProjectedSeq     int64
+	ProjectionStatus string
+	ProjectionHealth string
+	Status           string
+}
+
+func (s *Server) traceProjectionState(ctx context.Context, traceID string) (traceProjectionState, error) {
+	facts, err := s.queryFacts(ctx, factQuery{TraceID: traceID, Families: []string{"checkpoints", "turns"}, Limit: 1, TurnsOrder: "updated_desc", CheckpointsOrder: "updated_desc"})
+	if err != nil {
+		return traceProjectionState{}, err
+	}
+	checkpoint := firstObject(facts["checkpoints"])
+	turn := firstObject(facts["turns"])
+	state := traceProjectionState{
+		ProjectedSeq:     int64FromAny(seq(checkpoint)),
+		ProjectionStatus: normalizeProjectionStatus(text(checkpoint["projectionStatus"])),
+		ProjectionHealth: normalizeProjectionHealth(text(checkpoint["projectionHealth"]), normalizeProjectionStatus(text(checkpoint["projectionStatus"]))),
+		Status:           normalizeStatus(first(text(checkpoint["status"]), text(turn["status"]))),
+	}
+	if state.ProjectionStatus == "" {
+		state.ProjectionStatus = "unknown"
+	}
+	if state.ProjectionHealth == "" {
+		state.ProjectionHealth = "unknown"
+	}
+	return state, nil
+}
+
+func factsCacheClass(query factQuery) string {
+	if strings.TrimSpace(query.TraceID) == "" {
+		return ""
+	}
+	families := requestedFactFamilies(query)
+	if stringSetEqual(families, []string{"traceEvents"}) {
+		return terminalTracePageCacheClass
+	}
+	if query.AfterProjectedSeq == 0 && stringSetEqual(families, []string{"sessions", "messages", "turns", "checkpoints"}) {
+		return terminalTurnCacheClass
+	}
+	return ""
+}
+
+func requestedFactFamilies(query factQuery) []string {
+	selected := query.Families
+	if len(selected) == 0 {
+		selected = query.FactFamilies
+	}
+	result := []string{}
+	seen := map[string]bool{}
+	for _, family := range selected {
+		family = strings.TrimSpace(family)
+		if family != "" && !seen[family] {
+			result = append(result, family)
+			seen[family] = true
+		}
+	}
+	return result
+}
+
+func stringSetEqual(values []string, expected []string) bool {
+	if len(values) != len(expected) {
+		return false
+	}
+	set := map[string]bool{}
+	for _, value := range values {
+		set[value] = true
+	}
+	for _, value := range expected {
+		if !set[value] {
+			return false
+		}
+	}
+	return true
+}
+
+func factsCacheTTL(plan factsCachePlan, config cacheConfig) (time.Duration, bool, string) {
+	stableProjection := plan.ProjectionStatus == "caught-up" || plan.ProjectionStatus == "blocked" || plan.ProjectionStatus == "stalled"
+	terminal := isTerminal(plan.Status) || stableProjection
+	if !terminal {
+		if config.RunningPageTTL <= 0 {
+			return 0, true, "running_page_ttl_disabled"
+		}
+		return config.RunningPageTTL, true, "running_or_projecting_trace"
+	}
+	if plan.Class == terminalTracePageCacheClass {
+		if config.TerminalTracePageTTL <= 0 {
+			return 0, false, "terminal_trace_page_ttl_disabled"
+		}
+		return config.TerminalTracePageTTL, false, "terminal_trace_page"
+	}
+	if config.TerminalTurnTTL <= 0 {
+		return 0, false, "terminal_turn_ttl_disabled"
+	}
+	return config.TerminalTurnTTL, false, "terminal_turn_snapshot"
+}
+
+func factsCacheMetadata(class string, status string) map[string]any {
+	if class == "" {
+		class = "unknown"
+	}
+	meta := map[string]any{
+		"class":                class,
+		"schemaVersion":        cacheKeySchemaVersion,
+		"enabled":              false,
+		"status":               status,
+		"hit":                  false,
+		"miss":                 false,
+		"stale":                false,
+		"unavailable":          false,
+		"dbQueryAvoided":       false,
+		"cacheAgeMs":           nil,
+		"dbQueryDurationMs":    nil,
+		"payloadBytes":         nil,
+		"ttlMs":                nil,
+		"valuesRedacted":       true,
+		"secretMaterialStored": false,
+	}
+	setFactsCacheStatus(meta, status)
+	return meta
+}
+
+func setFactsCacheStatus(meta map[string]any, status string) {
+	if meta == nil {
+		return
+	}
+	meta["status"] = status
+	meta["hit"] = status == "hit"
+	meta["miss"] = status == "miss"
+	meta["stale"] = status == "stale"
+	meta["unavailable"] = status == "unavailable" || status == "set_failed"
+}
+
+func attachFactsCacheMetadata(payload map[string]any, meta map[string]any) {
+	if payload == nil || meta == nil {
+		return
+	}
+	payload["cache"] = meta
+}
+
+func factsCacheStorablePayload(payload map[string]any) map[string]any {
+	clone := cloneJSONMap(payload)
+	delete(clone, "cache")
+	delete(clone, "persistence")
+	delete(clone, "secretMaterialStored")
+	return clone
+}
+
+func factsCacheResponsePayload(payload map[string]any, persistence map[string]any) map[string]any {
+	clone := cloneJSONMap(payload)
+	clone["persistence"] = persistence
+	clone["valuesRedacted"] = true
+	return clone
+}
+
+func factsCacheEntryAgeMs(entry factsCacheEntry, now time.Time) (int64, bool) {
+	cachedAt, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(entry.CachedAt))
+	if err != nil {
+		return 0, true
+	}
+	age := now.Sub(cachedAt)
+	if age < 0 {
+		age = 0
+	}
+	if entry.TTLMillis > 0 && age.Milliseconds() > entry.TTLMillis {
+		return age.Milliseconds(), true
+	}
+	return age.Milliseconds(), false
+}
+
+func firstObject(rows []any) map[string]any {
+	if len(rows) == 0 {
+		return map[string]any{}
+	}
+	return objectMap(rows[0])
+}
+
+func int64FromAny(value any) int64 {
+	switch typed := value.(type) {
+	case int64:
+		return typed
+	case int:
+		return int64(typed)
+	case float64:
+		return int64(typed)
+	case json.Number:
+		parsed, _ := typed.Int64()
+		return parsed
+	default:
+		return 0
+	}
 }
 
 func (s *Server) handleTraceEvents(w http.ResponseWriter, r *http.Request) {