[AZ-238] [AZ-239] Add resource restart tests

Co-authored-by: Cursor <cursoragent@cursor.com>

e2e/replay/harness.py

@@ -84,6 +84,16 @@ class ReplayEstimate:
     capture_to_output_latency_ms: float
 
 
+@dataclass(frozen=True)
+class ResourceSample:
+    timestamp_s: float
+    process_rss_bytes: int
+    shared_memory_used_bytes: int
+    cuda_allocated_bytes: int
+    throttle_active: bool
+    temperature_c: float
+
+
 @dataclass(frozen=True)
 class ScenarioReport:
     scenario_id: str
@@ -487,6 +497,56 @@ def validate_derkachi_alignment(
     }
 
 
+def relocalization_required(
+    visual_overlap_fraction: float,
+    disconnected_duration_s: float,
+    max_disconnected_duration_s: float = 3.0,
+) -> bool:
+    if not 0.0 <= visual_overlap_fraction <= 1.0:
+        raise ValueError("visual overlap fraction must be within [0, 1]")
+    return visual_overlap_fraction < 0.05 or disconnected_duration_s > max_disconnected_duration_s
+
+
+def summarize_cold_start_trials(
+    first_fix_latencies_s: Sequence[float],
+    peak_memory_bytes: Sequence[int],
+    first_fix_budget_s: float = 30.0,
+    memory_budget_bytes: int = 8 * 1024 * 1024 * 1024,
+) -> Mapping[str, float | str | bool]:
+    if len(first_fix_latencies_s) != len(peak_memory_bytes):
+        raise ValueError("cold-start latency and memory trial counts must match")
+    if not first_fix_latencies_s:
+        raise ValueError("cold-start trials are empty")
+
+    p95_first_fix_s = percentile(first_fix_latencies_s, 95)
+    peak_memory = max(peak_memory_bytes)
+    return {
+        "trial_count": float(len(first_fix_latencies_s)),
+        "p95_first_fix_s": p95_first_fix_s,
+        "peak_memory_bytes": float(peak_memory),
+        "first_fix_passed": p95_first_fix_s < first_fix_budget_s,
+        "memory_passed": peak_memory < memory_budget_bytes,
+    }
+
+
+def summarize_resource_samples(samples: Sequence[ResourceSample]) -> Mapping[str, float | str | bool]:
+    if not samples:
+        raise ValueError("resource samples are empty")
+    duration_s = samples[-1].timestamp_s - samples[0].timestamp_s
+    if duration_s < 0.0:
+        raise ValueError("resource sample timestamps must be monotonic")
+    return {
+        "duration_s": duration_s,
+        "peak_process_rss_bytes": float(max(sample.process_rss_bytes for sample in samples)),
+        "peak_shared_memory_used_bytes": float(
+            max(sample.shared_memory_used_bytes for sample in samples)
+        ),
+        "peak_cuda_allocated_bytes": float(max(sample.cuda_allocated_bytes for sample in samples)),
+        "throttle_observed": any(sample.throttle_active for sample in samples),
+        "max_temperature_c": max(sample.temperature_c for sample in samples),
+    }
+
+
 def percentile(values: Sequence[float], percentile_value: int) -> float:
     if not values:
         raise ValueError("cannot compute percentile for empty values")