pikasTech-HWLAB/internal/workbenchruntime/cache_test.go

package workbenchruntime

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/alicebob/miniredis/v2"
	"github.com/redis/go-redis/v9"
)

func TestBuildCacheKeyActorIsolationAndAuthority(t *testing.T) {
	base := cacheKeyParts{Class: "sessions.summary", ActorID: "actor-a", SessionID: "ses-1", TraceID: "trace-1", Cursor: "idx:0", ProjectionSeq: 42}
	keyA, err := buildCacheKey(512, base)
	if err != nil {
		t.Fatalf("build key actor a: %v", err)
	}
	base.ActorID = "actor-b"
	keyB, err := buildCacheKey(512, base)
	if err != nil {
		t.Fatalf("build key actor b: %v", err)
	}
	if keyA == keyB {
		t.Fatalf("expected actor-isolated keys, got identical key %q", keyA)
	}
	if !strings.Contains(keyA, cacheKeySchemaVersion) || !strings.Contains(keyA, "class=sessions.summary") {
		t.Fatalf("key missing schema/class: %s", keyA)
	}
	if strings.Contains(keyA, "actor-a") || strings.Contains(keyB, "actor-b") {
		t.Fatalf("cache key should hash raw actor ids: %s / %s", keyA, keyB)
	}

	_, err = buildCacheKey(512, cacheKeyParts{Class: "sessions.summary", ActorID: "actor-a"})
	if cacheErrKind(err) != "cache_key_invalid" {
		t.Fatalf("expected missing authority to be rejected, got %v", err)
	}
	_, err = buildCacheKey(512, cacheKeyParts{Class: "sessions.summary", ProjectionSeq: 1})
	if cacheErrKind(err) != "cache_key_invalid" {
		t.Fatalf("expected missing actor to be rejected, got %v", err)
	}
}

func TestCachePayloadBoundary(t *testing.T) {
	_, err := cachePayloadBytes(map[string]any{"sessionId": "ses-1", "summary": map[string]any{"messageCount": 2}}, 1024)
	if err != nil {
		t.Fatalf("safe payload rejected: %v", err)
	}

	for _, payload := range []map[string]any{
		{"authorization": "bearer secret"},
		{"session": map[string]any{"prompt": "full prompt text"}},
		{"providerPayload": map[string]any{"raw": true}},
		{"stdout": strings.Repeat("x", 16)},
	} {
		_, err := cachePayloadBytes(payload, 1024)
		if cacheErrKind(err) != "cache_payload_rejected" {
			t.Fatalf("expected payload rejection for %#v, got %v", payload, err)
		}
	}

	_, err = cachePayloadBytes(map[string]any{"sessionId": strings.Repeat("x", 64)}, 16)
	if cacheErrKind(err) != "cache_payload_too_large" {
		t.Fatalf("expected payload too large, got %v", err)
	}
}

func TestCacheErrorClassification(t *testing.T) {
	cases := []struct {
		err  error
		kind string
	}{
		{redis.Nil, "cache_miss"},
		{context.DeadlineExceeded, "cache_timeout"},
		{context.Canceled, "cache_context_canceled"},
		{errors.New("dial tcp 127.0.0.1:6379: connect: connection refused"), "cache_unavailable"},
	}
	for _, tc := range cases {
		if got := classifyCacheError(tc.err); got != tc.kind {
			t.Fatalf("classifyCacheError(%v)=%s, want %s", tc.err, got, tc.kind)
		}
	}
}

func TestDisabledCacheDoesNotFailReadPath(t *testing.T) {
	cache := newDerivedCache(cacheConfig{Enabled: false})
	hit, err := cache.Get(context.Background(), "ignored", &map[string]any{})
	if err != nil || hit {
		t.Fatalf("disabled cache should be transparent, hit=%v err=%v", hit, err)
	}
	diag := cache.Diagnostic(context.Background())
	if diag["enabled"] != false || diag["available"] != false {
		t.Fatalf("unexpected disabled diagnostic: %#v", diag)
	}
}

func TestDerivedCacheHitMissAndMetricsWithRedis(t *testing.T) {
	redisServer := miniredis.RunT(t)
	cache := newDerivedCache(cacheConfig{Enabled: true, RedisURL: "redis://" + redisServer.Addr() + "/0", ConnectTimeout: 100 * time.Millisecond, OperationTimeout: 100 * time.Millisecond, MaxKeyBytes: 512, MaxPayloadBytes: 4096})
	defer cache.Close()
	key, err := cache.BuildKey(cacheKeyParts{Class: "sessions.summary", ActorID: "actor-a", Cursor: "idx:0|limit:20", ProjectionRevision: "workbench-sessions-v1", Authority: "contract=workbench-sessions-v1|limit=20"})
	if err != nil {
		t.Fatalf("build key: %v", err)
	}

	var missTarget map[string]any
	hit, err := cache.Get(context.Background(), key, &missTarget)
	if err != nil || hit {
		t.Fatalf("expected first read miss without error, hit=%v err=%v", hit, err)
	}
	payload := map[string]any{"contractVersion": "workbench-sessions-v1", "sessions": []any{map[string]any{"sessionId": "ses_cache_test", "terminal": true, "valuesRedacted": true}}, "valuesRedacted": true}
	if err := cache.Set(context.Background(), key, payload, time.Minute); err != nil {
		t.Fatalf("set cache payload: %v", err)
	}
	var got map[string]any
	hit, err = cache.Get(context.Background(), key, &got)
	if err != nil || !hit {
		t.Fatalf("expected second read hit, hit=%v err=%v", hit, err)
	}
	if got["contractVersion"] != "workbench-sessions-v1" || got["valuesRedacted"] != true {
		t.Fatalf("unexpected cached payload: %#v", got)
	}

	diagnostic := cache.Diagnostic(context.Background())
	metrics := objectMap(diagnostic["metrics"])
	if metrics["hits"] != int64(1) || metrics["misses"] != int64(1) || metrics["sets"] != int64(1) || diagnostic["available"] != true {
		t.Fatalf("unexpected cache diagnostic: %#v", diagnostic)
	}
	text := cache.PrometheusText(serviceID)
	for _, want := range []string{`cache_key_class="sessions.summary"`, `cache_status="miss"`, `cache_status="hit"`, `cache_status="stored"`, "workbench_cache_payload_bytes"} {
		if !strings.Contains(text, want) {
			t.Fatalf("metrics missing %s in:\n%s", want, text)
		}
	}
	if strings.Contains(text, "actor-a") || strings.Contains(text, key) || strings.Contains(text, "ses_cache_test") {
		t.Fatalf("metrics leaked actor, full key, or payload identity:\n%s", text)
	}
}

func TestDerivedCacheUnavailableIsObservableWithoutFallbackAuthority(t *testing.T) {
	redisServer := miniredis.RunT(t)
	addr := redisServer.Addr()
	redisServer.Close()
	cache := newDerivedCache(cacheConfig{Enabled: true, RedisURL: "redis://" + addr + "/0", ConnectTimeout: 50 * time.Millisecond, OperationTimeout: 50 * time.Millisecond, MaxKeyBytes: 512, MaxPayloadBytes: 4096})
	defer cache.Close()
	key, err := cache.BuildKey(cacheKeyParts{Class: "sessions.summary", ActorID: "actor-a", Cursor: "idx:0|limit:20", ProjectionRevision: "workbench-sessions-v1", Authority: "contract=workbench-sessions-v1|limit=20"})
	if err != nil {
		t.Fatalf("build key: %v", err)
	}

	var target map[string]any
	hit, err := cache.Get(context.Background(), key, &target)
	errKind := cacheErrKind(err)
	if hit || (errKind != "cache_unavailable" && errKind != "cache_timeout") {
		t.Fatalf("expected unavailable/timeout miss, hit=%v err=%v kind=%s", hit, err, errKind)
	}
	diagnostic := cache.Diagnostic(context.Background())
	if diagnostic["available"] != false || diagnostic["degraded"] != true || diagnostic["role"] != "derived-read-cache" {
		t.Fatalf("unexpected unavailable diagnostic: %#v", diagnostic)
	}
	text := cache.PrometheusText(serviceID)
	if !strings.Contains(text, `cache_status="unavailable"`) || !strings.Contains(text, "workbench_cache_unavailable_total") {
		t.Fatalf("unavailable metrics missing:\n%s", text)
	}
}

func TestStaleCacheMetadataDoesNotClaimDBAvoided(t *testing.T) {
	now := time.Now().UTC()
	entry := sessionsCacheEntry{CachedAt: now.Add(-2 * time.Minute).Format(time.RFC3339Nano), Payload: map[string]any{"sessions": []any{}}, PayloadBytes: 64, TTLMillis: int64((30 * time.Second).Milliseconds()), ProjectionRevision: "digest:old", ProjectionSeqMax: 7, ValuesRedacted: true}
	ageMs, stale := sessionsCacheEntryAgeMs(entry, now)
	if !stale || ageMs <= entry.TTLMillis {
		t.Fatalf("expected stale cache entry, ageMs=%d ttlMs=%d stale=%v", ageMs, entry.TTLMillis, stale)
	}
	meta := sessionsCacheMetadata("miss")
	setSessionsCacheStatus(meta, "stale")
	meta["cacheAgeMs"] = ageMs
	meta["freshnessSloMs"] = entry.TTLMillis
	applySessionsCacheRevisionMetadata(meta, sessionsSummaryRevision{ProjectionRevision: entry.ProjectionRevision, ProjectionSeqMax: entry.ProjectionSeqMax, RevisionSource: "cached_payload"})
	if meta["hit"] == true || meta["dbQueryAvoided"] == true || meta["status"] != "stale" {
		t.Fatalf("stale cache must remain diagnostic and not claim DB avoidance: %#v", meta)
	}
}

func TestCachePrometheusMetricsUseLowCardinalityLabels(t *testing.T) {
	var metrics cacheMetrics
	metrics.recordCacheObservation("turn.terminal.snapshot", "get", "hit", 12*time.Millisecond, 4096)
	metrics.recordCacheObservation("sessions.summary", "get", "miss", 25*time.Millisecond, 0)
	metrics.recordStale("sessions.summary")
	metrics.recordDBQueryAvoided("turn.terminal.snapshot")

	text := metrics.prometheusText(serviceID)
	for _, name := range []string{"workbench_cache_requests_total", "workbench_cache_hit_ratio", "workbench_cache_operation_duration_seconds", "workbench_cache_payload_bytes", "workbench_cache_stale_total", "workbench_db_query_avoided_total"} {
		if !strings.Contains(text, name) {
			t.Fatalf("metrics missing %s in:\n%s", name, text)
		}
	}
	if !strings.Contains(text, `cache_key_class="turn.terminal.snapshot"`) || !strings.Contains(text, `cache_status="hit"`) {
		t.Fatalf("metrics missing cache labels:\n%s", text)
	}
	if strings.Contains(text, "trc_") || strings.Contains(text, "ses_") || strings.Contains(text, "actor=") || strings.Contains(text, "cacheKey") {
		t.Fatalf("metrics leaked high-cardinality or key material:\n%s", text)
	}
}

func TestSessionsSummaryCacheTTLUsesShortTTLForUnstablePages(t *testing.T) {
	config := cacheConfig{SessionsSummaryTTL: 30 * time.Second, RunningPageTTL: 5 * time.Second}
	stable := map[string]any{"sessions": []any{map[string]any{"terminal": true, "running": false, "projectionStatus": "caught-up"}}}
	ttl, unstable, reason := sessionsSummaryCacheTTL(stable, config)
	if ttl != 30*time.Second || unstable || reason != "stable_page" {
		t.Fatalf("stable ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
	}

	running := map[string]any{"sessions": []any{map[string]any{"terminal": false, "running": true, "projectionStatus": "projecting"}}}
	ttl, unstable, reason = sessionsSummaryCacheTTL(running, config)
	if ttl != 5*time.Second || !unstable || reason != "running_or_projecting_page" {
		t.Fatalf("running ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
	}

	config.RunningPageTTL = 0
	ttl, unstable, reason = sessionsSummaryCacheTTL(running, config)
	if ttl != 0 || !unstable || reason != "running_page_ttl_disabled" {
		t.Fatalf("disabled running ttl=%s unstable=%v reason=%s", ttl, unstable, reason)
	}
}

func TestSessionsCacheStorablePayloadRemovesRuntimeOnlySecretKeys(t *testing.T) {
	payload := map[string]any{
		"ok":                   true,
		"contractVersion":      "workbench-sessions-v1",
		"secretMaterialStored": false,
		"persistence": map[string]any{
			"cache": map[string]any{"secretMaterialStored": false, "valuesRedacted": true},
		},
		"sessions": []any{map[string]any{"sessionId": "ses_1", "terminal": true, "projectionStatus": "caught-up", "valuesRedacted": true}},
	}
	stored := sessionsCacheStorablePayload(payload)
	if _, ok := stored["secretMaterialStored"]; ok {
		t.Fatalf("storable payload must not keep secretMaterialStored: %#v", stored)
	}
	if _, ok := stored["persistence"]; ok {
		t.Fatalf("storable payload must not keep runtime persistence summary: %#v", stored)
	}
	entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: stored, PayloadBytes: 128, TTLMillis: 30000, ValuesRedacted: true}
	if _, err := cachePayloadBytes(entry, 4096); err != nil {
		t.Fatalf("storable sessions payload rejected: %v", err)
	}
}

func TestSessionsSummaryCacheKeyActorIsolation(t *testing.T) {
	config := cacheConfig{Enabled: true, RedisURL: "redis://localhost:6379/0", MaxKeyBytes: 512}
	server := &Server{config: Config{Cache: config}, cache: newDerivedCache(config)}
	query := sessionListQuery{Limit: 20, Actor: sessionActor{ID: "actor-a", Role: "user"}}
	keyA, meta := server.sessionsSummaryCacheKey(query, 20, 0)
	if keyA == "" || meta["status"] != "miss" || meta["enabled"] != true {
		t.Fatalf("unexpected key/meta: key=%q meta=%#v", keyA, meta)
	}
	query.Actor.ID = "actor-b"
	keyB, _ := server.sessionsSummaryCacheKey(query, 20, 0)
	if keyA == keyB {
		t.Fatalf("expected actor-isolated session cache keys, got %q", keyA)
	}
	query.Actor.ID = ""
	keyMissingActor, meta := server.sessionsSummaryCacheKey(query, 20, 0)
	if keyMissingActor != "" || meta["status"] != "skipped" || meta["skipReason"] != "missing_actor" {
		t.Fatalf("expected missing actor skip, key=%q meta=%#v", keyMissingActor, meta)
	}
}

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 TestFactsCacheKeyChangesWithProjectionRevision(t *testing.T) {
	config := cacheConfig{Enabled: true, RedisURL: "redis://localhost:6379/0", MaxKeyBytes: 512}
	cache := newDerivedCache(config)
	defer cache.Close()
	base := cacheKeyParts{Class: terminalTurnCacheClass, ActorID: "workbench-runtime", TraceID: "trc-cache-key", Cursor: "after:0|limit:100|families:sessions,messages,turns,checkpoints", Authority: "contract=workbench-runtime-facts-v1|class=turn.terminal.snapshot|projectionStatus=caught-up|projectionHealth=healthy"}
	base.ProjectionSeq = 41
	keyA, err := cache.BuildKey(base)
	if err != nil {
		t.Fatalf("build key A: %v", err)
	}
	base.ProjectionSeq = 42
	keyB, err := cache.BuildKey(base)
	if err != nil {
		t.Fatalf("build key B: %v", err)
	}
	if keyA == keyB {
		t.Fatalf("projection revision must invalidate facts cache key: %q", keyA)
	}
	if strings.Contains(keyA, "trc-cache-key") || strings.Contains(keyB, "trc-cache-key") {
		t.Fatalf("facts cache keys must hash raw trace ids: %s / %s", keyA, keyB)
	}
}

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, "providerPayload": map[string]any{"raw": true}, "stdout": "internal", "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)
	}
	traceEvents := objectMap(stored["facts"])["traceEvents"].([]any)
	traceEvent := objectMap(traceEvents[0])
	if _, ok := traceEvent["providerPayload"]; ok {
		t.Fatalf("facts storable payload kept providerPayload: %#v", traceEvent)
	}
	if _, ok := traceEvent["stdout"]; ok {
		t.Fatalf("facts storable payload kept stdout: %#v", traceEvent)
	}
	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) {
		return cacheErr.Kind
	}
	return classifyCacheError(err)
}