feat(benchmark): UUID cache-coherency validation on the router-replay path

benchmark/auto.go

@@ -91,6 +91,51 @@ type AutoBenchmarkConfig struct {
 	// remains the normal budget C (--concurrency).
 	ReplayStopAtLowConcurrency bool
 
+	// UUID-based cache-coherency validation (--replay-inject-uuids). ROUTER
+	// PATH ONLY (cfg.RouterReplayFile != ""); the CLI rejects this combined
+	// with --from-dataset (see cli/benchmark_commands.go). One deterministic
+	// UUID is injected per user turn, spread through the conversation
+	// (rather than clustered at a session boundary) — see the package doc
+	// in replay_router_uuid.go for the full design. Each request recites a
+	// bounded WINDOW (first turn + up to 3 most-recent turns, excluding the
+	// current turn, capped at 4), keeping the recite cost/response budget
+	// constant regardless of session length.
+	ReplayInjectUUIDs bool
+	// ReplayUUIDSeed seeds the UUID generator (see newUUIDGenerator); 0 = crypto/rand
+	// (non-deterministic across runs).
+	ReplayUUIDSeed int64
+	// ReplayReciteEveryRequest: ask the model to recite the first-line UUID
+	// window on EVERY request (default true), not just each instance's
+	// final request.
+	ReplayReciteEveryRequest bool
+	// replaySessionTurnHashes is the precomputed per-session ordered list of
+	// qualifying user-turn hashes, populated once by RunAutoBenchmark before
+	// any per-model goroutine spawns — see computeSessionTurnHashes.
+	// replaySessionTurnHashes[i][t] = session i's turn-t message hash.
+	// Shared, read-only, across every model in a multi-model run (same
+	// sharing rationale as replayConversations below).
+	replaySessionTurnHashes [][]string
+	// replayUUIDSets is the precomputed per-session-per-turn UUID
+	// assignment, populated once alongside replaySessionTurnHashes — see
+	// buildSessionTurnUUIDs. Index i = session i's ordered turn-UUID list
+	// (index t = turn t's UUID); i corresponds to seriesNum-1, the order
+	// sessions are dispatched in (see the sizing note at the router-replay
+	// precompute call site). Shared, read-only, across every model in a
+	// multi-model run so every model sees the identical assignment.
+	replayUUIDSets [][]string
+	// replayBlockSessionCounts maps a replay-v3 block hash to the number of
+	// DISTINCT SESSIONS that reference it (see computeBlockSessionCounts) —
+	// a hash referenced by more than one session is never eligible for
+	// turn-stamping (isQualifyingUserTurn requires count==1). Populated once
+	// alongside replayUUIDSets.
+	replayBlockSessionCounts map[string]int
+	// replayUUIDOwner is the precomputed reverse uuid -> owning-session-index
+	// map, populated once alongside replayUUIDSets — see
+	// buildSessionTurnUUIDs. Read-only after precompute; used by
+	// findLeakedUUIDsByOwner (replay_uuid.go) to flag cross-session
+	// contamination in O(response) time.
+	replayUUIDOwner map[string]int
+
 	// RunID is populated internally by RunAutoBenchmark at the start of each
 	// run. It's the UUID injected into every conversation's system prompt
 	// (when ReplayNoStamp is false). Per-run scope — conversations that share
@@ -203,6 +248,18 @@ type requestDataRecord struct {
 	Question        string `json:"question,omitempty"`
 	ResponseText    string `json:"response_text,omitempty"`
 	RawResponseTail string `json:"raw_response_tail,omitempty"`
+
+	// UUID validation (router-replay --replay-inject-uuids only). The three
+	// counts are always populated (0 when the feature is off); the raw detail
+	// lists are populated ONLY on a miss or a leak (mirrors the
+	// failed-request-only policy above — avoid bloating every row).
+	UUIDExpected     int      `json:"uuid_expected"`
+	UUIDFound        int      `json:"uuid_found"`
+	UUIDLeaked       int      `json:"uuid_leaked"`
+	UUIDExactMatch   bool     `json:"uuid_exact_match"`
+	ExpectedUUIDsRaw []string `json:"expected_uuids_raw,omitempty"`
+	FoundMask        []bool   `json:"found_mask,omitempty"`
+	LeakedUUIDsRaw   []string `json:"leaked_uuids_raw,omitempty"`
 }
 
 // requestDataWriter writes requestDataRecord entries as JSONL, safe for concurrent use.
@@ -896,6 +953,18 @@ type autoState struct {
 	coldStartTTFTCount atomic.Int64 // count of cold-start samples (used for series-scaling gate)
 	ttftDegradedCount  atomic.Int64 // requests disqualified from cache-hit by TTFT degradation
 
+	// UUID validation (replay --replay-inject-uuids only). All zero when the
+	// feature is off — recordReplayRequest only touches these when
+	// metrics.ExpectedUUIDs is non-empty.
+	valReqs              atomic.Int64 // requests that carried >=1 expected UUID (i.e. validation ran)
+	valUUIDChecks        atomic.Int64 // total per-UUID presence checks made
+	valUUIDFound         atomic.Int64 // per-UUID presence checks that found the UUID
+	valExactMatchReqs    atomic.Int64 // requests whose first line was the exact ordered UUID list (output conformity)
+	valPresenceMissUUIDs atomic.Int64 // per-UUID PRESENCE_MISS count (expected UUID absent)
+	valCrossContamUUIDs  atomic.Int64 // per-UUID CROSS_CONTAMINATION count (other-conversation UUID present)
+	valPresenceMissReqs  atomic.Int64 // requests with >=1 PRESENCE_MISS
+	valCrossContamReqs   atomic.Int64 // requests with >=1 CROSS_CONTAMINATION
+
 	// Persistent early-sample buffers — never trimmed, survive stream eviction.
 	printMu sync.Mutex // serialises --print-responses output across concurrent series
 
@@ -989,6 +1058,17 @@ type autoBenchmarkResult struct {
 	totalInputWarm    int64 // input tokens from warm requests (prefix was cached)
 	totalOutput       int64 // output tokens across all requests
 	totalCachedTokens int64 // server-reported cached prompt tokens
+
+	// UUID validation (replay --replay-inject-uuids only); all zero when the
+	// feature is off. See autoState's val* atomics for field meanings.
+	valReqs              int64
+	valUUIDChecks        int64
+	valUUIDFound         int64
+	valExactMatchReqs    int64
+	valPresenceMissUUIDs int64
+	valCrossContamUUIDs  int64
+	valPresenceMissReqs  int64
+	valCrossContamReqs   int64
 }
 
 // displaySnapshot is an atomic snapshot of state for the display goroutine.
@@ -1140,6 +1220,19 @@ func printAutoSummary(res autoBenchmarkResult, cfg AutoBenchmarkConfig) {
 	fmt.Println(strings.Repeat("-", 62))
 	fmt.Printf(" Total completed    : %d\n", res.totalCompleted)
 	fmt.Printf(" Total errors       : %d\n", res.totalErrors)
+	if cfg.ReplayInjectUUIDs {
+		fmt.Println(strings.Repeat("-", 62))
+		fmt.Println(" UUID validation (replay)")
+		fmt.Printf("   Requests validated                  : %d\n", res.valReqs)
+		// Two tests, mirroring the cache-coherency eval CLI's layout: UUID
+		// correctness (per-stamp presence, Contains anywhere in the response)
+		// and output conformity (first line is exactly the ordered,
+		// comma-joined UUID list — see firstLineConformity).
+		fmt.Printf("   UUID correctness (presence)          : %d/%d\n", res.valUUIDFound, res.valUUIDChecks)
+		fmt.Printf("   Output conformity (first-line exact) : %d/%d\n", res.valExactMatchReqs, res.valReqs)
+		fmt.Printf("   PRESENCE_MISS (expected UUID absent) : %d across %d requests\n", res.valPresenceMissUUIDs, res.valPresenceMissReqs)
+		fmt.Printf("   CROSS_CONTAMINATION (other-conv)     : %d across %d requests\n", res.valCrossContamUUIDs, res.valCrossContamReqs)
+	}
 	fmt.Println(strings.Repeat("=", 62))
 }
 
@@ -1560,6 +1653,23 @@ func runSingleModelBenchmark(
 			}
 			pp.outputRatio = cfg.ReplayOutputRatio
 			pp.forceOutput = cfg.ReplayForceOutput
+			// UUID cache-coherency injection (--replay-inject-uuids). Same
+			// global/read-only refs set on the per-instance poster in
+			// replay_router.go — every poster in the run, per-instance or
+			// pooled-per-endpoint, shares these identical slices/maps, so
+			// this poster (potentially shared across many concurrent
+			// sessions once picked by the router) can safely serve any
+			// session: do()/dryDo() derive the per-request sessionIdx from
+			// seriesNum and buildInjection looks up that session's data from
+			// these globals, with no mutable per-session state on the
+			// poster itself.
+			if cfg.ReplayInjectUUIDs {
+				pp.uuidEnabled = true
+				pp.allUUIDSets = cfg.replayUUIDSets
+				pp.sessionTurnHashes = cfg.replaySessionTurnHashes
+				pp.owner = cfg.replayUUIDOwner
+				pp.reciteEveryRequest = cfg.ReplayReciteEveryRequest
+			}
 			posters[i] = pp
 		}
 		if ok {
@@ -2323,6 +2433,14 @@ func runSingleModelBenchmark(
 	res.totalInputWarm = tt.inputWarm
 	res.totalOutput = tt.output
 	res.totalCachedTokens = tt.cached
+	res.valReqs = st.valReqs.Load()
+	res.valUUIDChecks = st.valUUIDChecks.Load()
+	res.valUUIDFound = st.valUUIDFound.Load()
+	res.valExactMatchReqs = st.valExactMatchReqs.Load()
+	res.valPresenceMissUUIDs = st.valPresenceMissUUIDs.Load()
+	res.valCrossContamUUIDs = st.valCrossContamUUIDs.Load()
+	res.valPresenceMissReqs = st.valPresenceMissReqs.Load()
+	res.valCrossContamReqs = st.valCrossContamReqs.Load()
 
 	// Send a final snapshot with termReason set so multi-model display shows DONE.
 	{
@@ -2515,6 +2633,82 @@ func RunAutoBenchmark(ctx context.Context, cfg AutoBenchmarkConfig) error {
 		fmt.Printf("Header: %d sessions / %d instances / %d requests / %d fan-out turns / max fan-out %d\n\n",
 			hdr.Summary.Sessions, hdr.Summary.Instances, hdr.Summary.Requests,
 			hdr.Summary.FanOutTurns, hdr.Summary.MaxFanOutInOneTurn)
+
+		// Precompute UUID cache-coherency injection ONCE here, before any
+		// per-model goroutine spawns below, so every model's
+		// runSingleModelBenchmark sees the identical per-session assignment
+		// (same sharing rationale as replayConversations above).
+		if cfg.ReplayInjectUUIDs {
+			// Effective session count: the SAME formula
+			// openRouterReplayStream uses to compute its own `total` (see
+			// replay_router.go), so the array is sized to exactly the
+			// number of sessions that will ever be dispatched (dispatch
+			// order == array index, since both walk the file in the same
+			// filtered, sequential order). This is what makes true
+			// lazy-growth unnecessary: the count is fully determined by
+			// the header + --replay-series + --replay-series-indices/
+			// --replay-series-range before any session is ever pulled.
+			effectiveSessions := hdr.Summary.Sessions
+			if cfg.ReplaySeries > 0 && cfg.ReplaySeries < effectiveSessions {
+				effectiveSessions = cfg.ReplaySeries
+			}
+			if len(cfg.RouterReplaySeriesIndices) > 0 {
+				effectiveSessions = len(cfg.RouterReplaySeriesIndices)
+				if cfg.ReplaySeries > 0 && cfg.ReplaySeries < effectiveSessions {
+					effectiveSessions = cfg.ReplaySeries
+				}
+			}
+			if effectiveSessions <= 0 {
+				fmt.Fprintf(os.Stderr,
+					"[router-replay] WARNING: --replay-inject-uuids could not determine an effective session count (header sessions=%d) — UUID injection disabled for this run\n",
+					hdr.Summary.Sessions)
+			} else {
+				counts, cerr := computeBlockSessionCounts(cfg.RouterReplayFile, cfg.RouterReplaySeriesIndices, cfg.ReplaySeries)
+				if cerr != nil {
+					return fmt.Errorf("compute block session counts for --replay-inject-uuids: %w", cerr)
+				}
+				cfg.replayBlockSessionCounts = counts
+
+				// Per-session ordered list of qualifying user-turn hashes
+				// (role=="user", has a text block, hash referenced by
+				// exactly this one session — see isQualifyingUserTurn),
+				// each turn getting its own UUID (buildSessionTurnUUIDs)
+				// rather than N stamps clustered at a session boundary.
+				turnHashes, terr := computeSessionTurnHashes(cfg.RouterReplayFile, cfg.RouterReplaySeriesIndices, cfg.ReplaySeries, counts)
+				if terr != nil {
+					return fmt.Errorf("compute session turn hashes for --replay-inject-uuids: %w", terr)
+				}
+				cfg.replaySessionTurnHashes = turnHashes
+
+				turnCounts := make([]int, len(turnHashes))
+				for i, h := range turnHashes {
+					turnCounts[i] = len(h)
+				}
+				cfg.replayUUIDSets, cfg.replayUUIDOwner = buildSessionTurnUUIDs(turnCounts, cfg.ReplayUUIDSeed)
+
+				sharedHashes := 0
+				for _, n := range counts {
+					if n > 1 {
+						sharedHashes++
+					}
+				}
+				totalTurns, minTurns, maxTurns := 0, 0, 0
+				if len(turnCounts) > 0 {
+					minTurns, maxTurns = turnCounts[0], turnCounts[0]
+					for _, n := range turnCounts {
+						totalTurns += n
+						if n < minTurns {
+							minTurns = n
+						}
+						if n > maxTurns {
+							maxTurns = n
+						}
+					}
+				}
+				fmt.Printf("UUID validation enabled: %d session(s) prepared, %d turn(s) total, %d-%d turns/session, %d cross-session-shared block hash(es), seed=%d, recite-every-request=%v\n",
+					effectiveSessions, totalTurns, minTurns, maxTurns, sharedHashes, cfg.ReplayUUIDSeed, cfg.ReplayReciteEveryRequest)
+			}
+		}
 	}
 
 	// Create per-run subdirectory for request data if configured.

benchmark/replay.go

@@ -101,7 +101,7 @@ func runReplaySeriesLoop(
 		seriesNum := convIdx + 1
 		seriesGUID := conv.ID
 
-		fullyWalked := runReplayConversation(benchCtx, cfg, st, rdw, conv, seriesNum, seriesGUID, endpointOverride, reqTimeout, gate)
+		fullyWalked := runReplayConversation(benchCtx, cfg, st, rdw, conv, convIdx, seriesNum, seriesGUID, endpointOverride, reqTimeout, gate)
 		// The conversation's slot is retired — its context no longer counts
 		// toward the active dataset.
 		st.datasetTracker.Reset(seriesNum)
@@ -116,6 +116,9 @@ func runReplaySeriesLoop(
 // benchmark request per gpt turn. Errors on individual requests are recorded
 // but don't abort the series — the next turn still runs.
 //
+// convIdx is this conversation's index into cfg.replayConversations
+// (== seriesNum-1, passed explicitly rather than re-derived).
+//
 // Returns true if the whole conversation was walked; false if a stop signal
 // (--total reached, context cancel) cut it short mid-walk.
 func runReplayConversation(
@@ -124,6 +127,7 @@ func runReplayConversation(
 	st *autoState,
 	rdw *requestDataWriter,
 	conv Conversation,
+	convIdx int,
 	seriesNum int,
 	seriesGUID string,
 	endpointOverride string,
@@ -200,7 +204,7 @@ func runReplayConversation(
 	var pending strings.Builder
 	turnNum := 0
 
-	flush := func() bool {
+	flush := func(gptIdx int) bool {
 		userContent := strings.TrimSpace(pending.String())
 		pending.Reset()
 		if userContent == "" {
@@ -232,7 +236,11 @@ func runReplayConversation(
 		requestNum := int(st.totalCompleted.Load()) + 1
 
 		// Observe the accumulated history (including this turn's user message)
-		// with the content-level estimator BEFORE the server responds.
+		// with the content-level estimator BEFORE the server responds. The
+		// recite-instruction tail (below) is deliberately NOT part of what the
+		// estimator/history see — it's per-request boilerplate, not
+		// conversation content, and would otherwise skew the cache-ratio
+		// estimate every single turn (recite-every-turn is the default).
 		history.WriteString(userContent)
 		ratio := st.estimator.Observe(history.String())
 
@@ -281,6 +289,7 @@ func runReplayConversation(
 		}
 
 		metrics.LocalCacheRatio = ratio
+
 		recordReplayRequest(cfg, st, rdw, metrics, isFirstRequest, &coldStartTTFT)
 		isFirstRequest = false
 		return true
@@ -289,7 +298,7 @@ func runReplayConversation(
 	for i := firstIdx; i < len(conv.Turns); i++ {
 		t := conv.Turns[i]
 		if t.From == "gpt" {
-			if !flush() {
+			if !flush(i) {
 				return false
 			}
 			continue
@@ -347,6 +356,34 @@ func recordReplayRequest(
 	isErr := metrics.Error != nil
 	explicitCache := metrics.UsageData.CachedTokens.Count > 0
 
+	// UUID validation tallies (--replay-inject-uuids only). metrics.ExpectedUUIDs
+	// is nil/empty for every request when the feature is off (default), so this
+	// block — and the val* counters it touches — is fully inert then.
+	uuidExpectedCount := len(metrics.ExpectedUUIDs)
+	uuidFoundCount := 0
+	for _, found := range metrics.UUIDFound {
+		if found {
+			uuidFoundCount++
+		}
+	}
+	uuidLeakedCount := len(metrics.LeakedUUIDs)
+	if uuidExpectedCount > 0 {
+		st.valReqs.Add(1)
+		st.valUUIDChecks.Add(int64(uuidExpectedCount))
+		st.valUUIDFound.Add(int64(uuidFoundCount))
+		if metrics.ExactMatch {
+			st.valExactMatchReqs.Add(1)
+		}
+		if missCount := uuidExpectedCount - uuidFoundCount; missCount > 0 {
+			st.valPresenceMissUUIDs.Add(int64(missCount))
+			st.valPresenceMissReqs.Add(1)
+		}
+		if uuidLeakedCount > 0 {
+			st.valCrossContamUUIDs.Add(int64(uuidLeakedCount))
+			st.valCrossContamReqs.Add(1)
+		}
+	}
+
 	earlyColdBaseline := st.earlyColdStartTTFT()
 
 	st.mu.Lock()
@@ -390,7 +427,7 @@ func recordReplayRequest(
 		if metrics.Error != nil {
 			errMsg = metrics.Error.Error()
 		}
-		if writeErr := rdw.write(requestDataRecord{
+		rec := requestDataRecord{
 			StartTime:            reqStart,
 			EndTime:              reqEnd,
 			TTFT:                 float64(metrics.TimeToFirstToken.Milliseconds()),
@@ -409,7 +446,21 @@ func recordReplayRequest(
 			ErrorMessage:         errMsg,
 			IsEmpty:              metrics.IsEmpty,
 			LocalCacheRatio:      metrics.LocalCacheRatio,
-		}); writeErr != nil {
+			UUIDExpected:         uuidExpectedCount,
+			UUIDFound:            uuidFoundCount,
+			UUIDLeaked:           uuidLeakedCount,
+			UUIDExactMatch:       metrics.ExactMatch,
+		}
+		// Raw detail lists only on a miss or a leak — mirrors the
+		// failed-request-only policy on PromptText/ResponseText/RawResponseTail
+		// above (avoid bloating every row with data that matters only when
+		// something's wrong).
+		if uuidFoundCount < uuidExpectedCount || uuidLeakedCount > 0 {
+			rec.ExpectedUUIDsRaw = metrics.ExpectedUUIDs
+			rec.FoundMask = metrics.UUIDFound
+			rec.LeakedUUIDsRaw = metrics.LeakedUUIDs
+		}
+		if writeErr := rdw.write(rec); writeErr != nil {
 			fmt.Fprintf(os.Stderr, "warning: failed to write request data: %v\n", writeErr)
 		}
 	}