mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: cache terminal workbench trace reads

Browse Source
This commit is contained in:
lyon
2026-06-22 13:46:12 +08:00
parent 33c8c18da0
commit 6378541c5b
2 changed files with 393 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/workbenchruntime/cache_test.go
+52
View File
@@ -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
+341 -4
View File
@@ -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) {