diff --git a/internal/workbenchruntime/cache_test.go b/internal/workbenchruntime/cache_test.go index 62c6597a..77b0bf5b 100644 --- a/internal/workbenchruntime/cache_test.go +++ b/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) { diff --git a/internal/workbenchruntime/service.go b/internal/workbenchruntime/service.go index 24ce1ce5..47c2e3ff 100644 --- a/internal/workbenchruntime/service.go +++ b/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) {