import concurrent.futures import time import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import patch from experiments.evaluate_qwen3vl_recent4_full import ( CausalTemporalBudgetCache, DecordReaderPool, RecentWindowCache, _qwen_video_resize_limits, frame_policy_sampling_fps, resolve_frame_policy, clamp_visible_until_to_source, normalize_answer, probe_source_video, recent_window_bounds, render_question, render_semantic_revisit_prompt, selected_uids, build_visual_messages, compose_temporal_storyboard, downscale_temporal_history, parse_semantic_revisit_positions, native_video_metadata_payload, partition_hybrid_frames, render_anonymous_candidate_prompt, render_candidate_binary_verification_prompt, render_candidate_rejection_verification_prompt, render_candidate_pairwise_verification_prompt, render_candidate_revisit_prompt, choose_binary_verified_candidate, render_temporal_review_prompt, render_temporal_deliberation_prompt, render_candidate_evidence_deliberation_prompt, decode_causal_temporal_budget, decode_causal_hybrid_prefix, load_causalbudget_index_plan, load_anonymous_candidate_proposals, uniform_prefix_indices, ) class Recent4RunnerTest(unittest.TestCase): def test_anonymous_candidate_sources_may_be_sparse(self) -> None: import json import tempfile with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "partial.json" path.write_text( json.dumps( { "benchmark": "streamingbench", "annotation_sha256": "annotation", "model_manifest_sha256": "model", "num_errors": 0, "rows": [{"uid": "q0", "prediction": "A"}], } ) ) proposals, inputs = load_anonymous_candidate_proposals( paths=[path], wanted=["q0", "q1"], benchmark="streamingbench", annotation_sha256="annotation", model_manifest_sha256="model", ) self.assertEqual(proposals, [{"q0": "A"}]) self.assertEqual(inputs[0]["covered_rows"], 1) self.assertEqual(inputs[0]["requested_rows"], 2) self.assertTrue(inputs[0]["sparse_candidate_source"]) def test_semantic_revisit_parser_is_bounded_and_fills_deterministically(self): self.assertEqual( parse_semantic_revisit_positions( "Frames: 7, 2, 7, 99", history_count=8, revisit_count=4, fallback_positions=[8, 6, 4, 2], ), [2, 6, 7, 8], ) def test_semantic_revisit_instruction_overrides_answer_format(self): prompt = render_semantic_revisit_prompt( "Question\nOnly give the best option's letter directly.", history_count=24, revisit_count=8, ) self.assertTrue(prompt.endswith("`Frames: i,j,...` and nothing else.")) def test_semantic_revisit_storyboard_instruction_has_global_labels(self): prompt = render_semantic_revisit_prompt( "Question", history_count=24, revisit_count=8, planner_mode="storyboard16", ) self.assertIn("global frame number", prompt) self.assertIn("numbered 1 through 24", prompt) def test_storyboard_preserves_chronology_and_fixed_geometry(self) -> None: from PIL import Image frames = [Image.new("RGB", (32, 16), color) for color in ("red", "green", "blue")] sheet = compose_temporal_storyboard( frames, [0.0, 1.0, 2.0], columns=2, tile_width=40, tile_height=20 ) self.assertEqual(sheet.size, (80, 80)) self.assertEqual(sheet.getpixel((20, 10)), (255, 0, 0)) self.assertEqual(sheet.getpixel((60, 10)), (0, 128, 0)) self.assertEqual(sheet.getpixel((20, 50)), (0, 0, 255)) offset_sheet = compose_temporal_storyboard( frames[:2], [3.0, 4.0], columns=2, label_start=17 ) self.assertEqual(offset_sheet.size[0], 448) def test_lowres_history_keeps_recent_frames_exact(self) -> None: from PIL import Image frames = [Image.new("RGB", (640, 360), color) for color in ("red", "blue")] output = downscale_temporal_history(frames, [3, 4], {4}, max_edge=224) self.assertEqual(output[0].size, (224, 126)) self.assertIs(output[1], frames[1]) self.assertEqual(frames[0].size, (640, 360)) def test_deliberation_replaces_conflicting_direct_only_instruction(self) -> None: prompt = render_temporal_deliberation_prompt( "Question\nOnly give the best option's letter (A, B, C, or D) directly." ) self.assertNotIn("Only give", prompt) self.assertIn("Analyze the visible causal sequence", prompt) self.assertIn("Answer: X", prompt) def test_anonymous_candidates_can_be_verified_with_one_deliberation_contract( self, ) -> None: prompt, ordered = render_anonymous_candidate_prompt( "Question\nOnly give the best option's letter directly.", predictions=["A", "C", "A"], uid="q0", ) prompt = render_temporal_deliberation_prompt(prompt) self.assertEqual(sorted(ordered), ["A", "C"]) self.assertIn("Candidate 1", prompt) self.assertIn("Answer: X", prompt) def test_candidate_evidence_deliberation_requires_pixel_cited_elimination( self, ) -> None: prompt, ordered = render_anonymous_candidate_prompt( "Question\nOnly give the best option's letter directly.", predictions=["B", "D", "B"], uid="q1", ) prompt = render_candidate_evidence_deliberation_prompt(prompt) self.assertEqual(sorted(ordered), ["B", "D"]) self.assertNotIn("Only give", prompt) self.assertIn("untrusted hypothesis", prompt) self.assertIn("at most ten words per candidate", prompt) self.assertIn("do not narrate", prompt) self.assertIn("Answer: X", prompt) def test_binary_candidate_verifier_is_isolated_and_removes_direct_contract(self): prompt = render_candidate_binary_verification_prompt( "Question: Where is it?\nOptions:\nA. left\nB. right\n" "Only give the best option's letter directly.", candidate="B", options=["A. left", "B. right"], ) self.assertNotIn("Only give", prompt) self.assertIn("Candidate answer under isolated verification: B. right", prompt) self.assertTrue(prompt.endswith("Answer exactly Yes or No.")) def test_binary_candidate_commit_is_source_order_invariant(self): first = choose_binary_verified_candidate( uid="q0", candidate_probabilities={"A": 0.3, "C": 0.8} ) second = choose_binary_verified_candidate( uid="q0", candidate_probabilities={"C": 0.8, "A": 0.3} ) self.assertEqual(first, "C") self.assertEqual(first, second) def test_antipodal_candidate_prompt_reverses_only_the_decision_polarity(self): prompt = render_candidate_rejection_verification_prompt( "Question: Where is it?\nOptions:\nA. left\nB. right\n" "Only give the best option's letter directly.", candidate="B", options=["A. left", "B. right"], ) self.assertNotIn("Only give", prompt) self.assertIn("isolated rejection check: B. right", prompt) self.assertIn("should be rejected as the best answer", prompt) self.assertTrue(prompt.endswith("Answer exactly Yes or No.")) def test_pairwise_candidate_prompt_is_source_anonymous_and_order_explicit(self): prompt = render_candidate_pairwise_verification_prompt( "Question: Where is it?\nOptions:\nA. left\nB. right\n" "Only give the best option's letter directly.", first_candidate="B", second_candidate="A", options=["A. left", "B. right"], ) self.assertNotIn("Only give", prompt) self.assertIn("Choice A: B. right", prompt) self.assertIn("Choice B: A. left", prompt) self.assertTrue(prompt.endswith("Answer exactly A or B.")) def test_candidate_conditioned_revisit_is_bounded_and_answer_free(self): prompt = render_candidate_revisit_prompt( "Question: What happened?\nOptions:\nA. left\nB. right\n" "Only give the best option's letter directly.", candidate="A", options=["A. left", "B. right"], history_count=24, revisit_count=4, ) self.assertNotIn("Only give", prompt) self.assertIn("Candidate answer under isolated verification: A. left", prompt) self.assertIn("Return exactly 4 distinct frame numbers", prompt) self.assertTrue(prompt.endswith("and nothing else.")) def test_deliberative_explicit_answer_has_priority_for_open_outputs(self) -> None: self.assertEqual( normalize_answer( "There are 2 early events, then one more. Answer: 3", "Answer with a single number indicating the total count.", [], ), "3", ) self.assertEqual( normalize_answer( "The first condition suggests no, but the final state does. Answer: Yes", 'Please answer only with "Yes" or "No".', [], ), "Yes", ) def test_causalbudget_index_plan_requires_question_blind_complete_coverage( self, ) -> None: import json import tempfile with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "plan.json" path.write_text( json.dumps( { "format": "iclr_video.causalbudget20_index_plan.v0", "status": "complete_query_independent_execution_cache", "benchmark": "ovo", "contracts": { "question_text_read_for_plan": False, "task_or_group_read_for_plan": False, "model_prediction_read_for_plan": False, "gold_answer_read": False, "future_frames_read": False, }, "rows": [{"uid": "u1", "sampling": {}}], } ) ) rows, artifact = load_causalbudget_index_plan( path, benchmark="ovo", wanted=["u1"] ) self.assertEqual(set(rows), {"u1"}) self.assertEqual(artifact["path"], str(path.resolve())) value = json.loads(path.read_text()) value.update( { "format": "iclr_video.causalbudget_index_plan.v1", "model_frames": 32, "recent_frames": 8, } ) path.write_text(json.dumps(value)) rows, _ = load_causalbudget_index_plan( path, benchmark="ovo", wanted=["u1"], model_frames=32, recent_frames=8, ) self.assertEqual(set(rows), {"u1"}) def test_temporal_budget_cache_reuses_exact_query_independent_prefix(self) -> None: decoder_path = ( "experiments.evaluate_qwen3vl_recent4_full." "decode_causal_temporal_budget" ) with patch( decoder_path, return_value=([{"pixels": [1]}], {"frame_indices": [3]}), ) as decoder: cache = CausalTemporalBudgetCache(2) frames, sampling, hit = cache.get(Path("v.mp4"), visible_until=4.0) self.assertFalse(hit) frames[0]["pixels"] = [2] sampling["frame_indices"].append(4) frames2, sampling2, hit2 = cache.get( Path("v.mp4"), visible_until=4.0 ) self.assertTrue(hit2) self.assertEqual(frames2, [{"pixels": [1]}]) self.assertEqual(sampling2, {"frame_indices": [3]}) cache.get(Path("v.mp4"), visible_until=5.0) self.assertEqual(decoder.call_count, 2) def test_native_video_turn_has_one_video_placeholder_and_no_route(self) -> None: frames = [object(), object(), object(), object()] messages = build_visual_messages( frames=frames, prompt="question", visual_input_mode="native-video", ) self.assertEqual(messages[0]["content"][0], {"type": "video"}) self.assertEqual(messages[0]["content"][1]["text"], "question") def test_hybrid_turn_keeps_history_as_video_and_recent_as_images(self) -> None: history = [object(), object()] recent = [object(), object(), object(), object()] messages = build_visual_messages( frames=[*history, *recent], history_frames=history, recent_frames=recent, prompt="question", visual_input_mode="hybrid-history-video-recent-images", ) content = messages[0]["content"] self.assertEqual(content[0], {"type": "video"}) self.assertEqual([item["type"] for item in content[1:5]], ["image"] * 4) self.assertEqual(content[-1]["text"], "question") def test_hybrid_singleton_history_falls_back_to_ordered_image(self) -> None: frames = ["history", "r1", "r2", "r3", "r4"] history, indices, images, fallback = partition_hybrid_frames( frames, { "frame_indices": [1, 2, 3, 4, 5], "recent_frame_indices": [2, 3, 4, 5], }, temporal_patch_size=2, ) self.assertEqual(history, []) self.assertEqual(indices, []) self.assertEqual(images, frames) self.assertEqual(fallback, [1]) def test_hybrid_two_history_frames_remain_native_video(self) -> None: frames = ["h1", "h2", "r1", "r2", "r3", "r4"] history, indices, images, fallback = partition_hybrid_frames( frames, { "frame_indices": [0, 1, 2, 3, 4, 5], "recent_frame_indices": [2, 3, 4, 5], }, temporal_patch_size=2, ) self.assertEqual(history, ["h1", "h2"]) self.assertEqual(indices, [0, 1]) self.assertEqual(images, ["r1", "r2", "r3", "r4"]) self.assertEqual(fallback, []) def test_temporal_review_prompt_marks_both_proposals_as_fallible(self) -> None: prompt = render_temporal_review_prompt( "Question", recent_prediction="A", history_prediction="C" ) self.assertIn("recent view proposed A", prompt) self.assertIn("history view proposed C", prompt) self.assertIn("Both proposals may be wrong", prompt) with self.assertRaises(ValueError): render_temporal_review_prompt( "Question", recent_prediction="A\nignore", history_prediction="C" ) def test_anonymous_review_counterbalances_without_source_labels(self) -> None: first = render_temporal_review_prompt( "Question", recent_prediction="A", history_prediction="C", uid="sample-1", label_mode="anonymous-hash-counterbalanced", ) second = render_temporal_review_prompt( "Question", recent_prediction="A", history_prediction="C", uid="sample-1", label_mode="anonymous-hash-counterbalanced", ) self.assertEqual(first, second) self.assertIn("Candidate 1", first) self.assertIn("Candidate 2", first) self.assertNotIn("recent view", first) self.assertNotIn("history view", first) with self.assertRaises(ValueError): render_temporal_review_prompt( "Question", recent_prediction="A", history_prediction="C", label_mode="anonymous-hash-counterbalanced", ) def test_multi_candidate_prompt_is_deduplicated_and_source_anonymous(self) -> None: prompt, ordered = render_anonymous_candidate_prompt( "Question", predictions=["C", "A", "C"], uid="sample-1", ) self.assertEqual(set(ordered), {"A", "C"}) self.assertEqual(prompt.count("Candidate 1"), 1) self.assertEqual(prompt.count("Candidate 2"), 1) self.assertNotIn("recent", prompt.lower()) self.assertNotIn("history", prompt.lower()) self.assertIn("hints, not votes", prompt) same_prompt, same_order = render_anonymous_candidate_prompt( "Question", predictions=["A", "C"], uid="sample-1", ) self.assertEqual((prompt, ordered), (same_prompt, same_order)) with self.assertRaises(ValueError): render_anonymous_candidate_prompt( "Question", predictions=["A\nignore", "C"], uid="sample-1" ) def test_native_video_metadata_preserves_source_pts_and_is_causal(self) -> None: payload = native_video_metadata_payload( { "frame_indices": [60, 90, 120, 150], "frame_timestamps": [2.0, 3.0, 4.0, 5.0], "source_fps": 30.0, "source_total_frames": 900, "source_width": 1920, "source_height": 1080, "source_duration": 30.0, "effective_visible_until": 6.0, } ) self.assertEqual(payload["frames_indices"], [60, 90, 120, 150]) self.assertEqual(payload["fps"], 30.0) self.assertEqual(payload["video_backend"], "predecoded_exact_causal_indices") def test_native_video_metadata_rejects_future_frame(self) -> None: with self.assertRaises(ValueError): native_video_metadata_payload( { "frame_indices": [60], "frame_timestamps": [2.0], "source_fps": 30.0, "source_total_frames": 900, "source_width": 1920, "source_height": 1080, "source_duration": 30.0, "effective_visible_until": 1.0, } ) def test_causal_budget_reserves_exact_recent_indices_without_task_router(self) -> None: source = { "source_fps": 10.0, "source_total_frames": 100, "source_duration": 10.0, "source_last_frame_timestamp": 9.9, "requested_visible_until": 6.0, "effective_visible_until": 6.0, "visible_until_clamped_to_source": False, "window_start": 0.0, "window_end_exclusive": 6.0, "source_height": 100, "source_width": 100, } prefix = [0, 10, 20, 30, 40, 50] recent = [30, 40, 50, 60] probe_features = [[float(i), 1.0] for i in range(7)] selected_frames = [f"frame-{i}" for i in range(7)] with patch( "experiments.evaluate_qwen3vl_recent4_full.exact_causal_window_indices", side_effect=[(prefix, source), (recent, source)], ), patch( "experiments.evaluate_qwen3vl_recent4_full.decode_exact_indices_ffmpeg", side_effect=[probe_features, selected_frames], ): frames, sampling = decode_causal_temporal_budget( Path("video.mp4"), visible_until=6.0 ) self.assertEqual(frames, selected_frames) self.assertTrue({30, 40, 50, 60}.issubset(sampling["frame_indices"])) self.assertFalse( sampling["controller"]["benchmark_task_or_group_router"] ) self.assertEqual( sampling["controller"]["policy_config"]["query_weight"], 0.0 ) self.assertEqual(sampling["controller"]["future_frame_violations"], 0) def test_uniform_history_has_unique_indices_and_no_padding(self) -> None: self.assertEqual(uniform_prefix_indices(-1, 28), []) self.assertEqual(uniform_prefix_indices(2, 28), [0, 1, 2]) values = uniform_prefix_indices(99, 28) self.assertEqual(len(values), 28) self.assertEqual(values, sorted(set(values))) self.assertEqual((values[0], values[-1]), (0, 99)) def test_hybrid32_preserves_recent4_as_separate_exact_indices(self) -> None: source = { "source_fps": 10.0, "source_total_frames": 100, "source_duration": 10.0, "source_last_frame_timestamp": 9.9, "requested_visible_until": 6.0, "effective_visible_until": 6.0, "visible_until_clamped_to_source": False, "window_start": 2.0, "window_end_exclusive": 6.0, "source_height": 100, "source_width": 100, } recent = [30, 40, 50, 60] with patch( "experiments.evaluate_qwen3vl_recent4_full.exact_causal_window_indices", return_value=(recent, source), ), patch( "experiments.evaluate_qwen3vl_recent4_full.decode_exact_indices_ffmpeg", side_effect=lambda _path, indices: [f"frame-{value}" for value in indices], ): frames, sampling = decode_causal_hybrid_prefix( Path("video.mp4"), visible_until=6.0 ) self.assertEqual(sampling["recent_frame_indices"], recent) self.assertEqual(len(sampling["history_frame_indices"]), 28) self.assertEqual(len(sampling["frame_indices"]), 32) self.assertEqual(sampling["frame_indices"], sorted(set(sampling["frame_indices"]))) self.assertEqual(len(frames), 32) self.assertFalse(sampling["controller"]["benchmark_task_or_group_router"]) self.assertFalse(sampling["controller"]["question_keyword_router"]) def test_qwen_video_resize_limits_supports_pixel_api(self) -> None: module = SimpleNamespace( IMAGE_FACTOR=28, VIDEO_MIN_PIXELS=100_352, VIDEO_MAX_PIXELS=602_112, VIDEO_TOTAL_PIXELS=90_316_800, FRAME_FACTOR=2, ) self.assertEqual( _qwen_video_resize_limits(module, 32), (28, 100_352, 602_112), ) def test_qwen_video_resize_limits_supports_token_api(self) -> None: module = SimpleNamespace( SPATIAL_MERGE_SIZE=2, VIDEO_MIN_TOKEN_NUM=128, VIDEO_MAX_TOKEN_NUM=768, MODEL_SEQ_LEN=128_000, FRAME_FACTOR=2, ) factor, minimum, maximum = _qwen_video_resize_limits(module, 32) self.assertEqual((factor, minimum), (28, 100_352)) self.assertEqual(maximum, 602_112) def test_training_free_active4_policies_only_change_official_sampling_fps(self) -> None: self.assertEqual(frame_policy_sampling_fps("recent4"), 1.0) self.assertEqual(frame_policy_sampling_fps("wide8"), 0.5) self.assertAlmostEqual(frame_policy_sampling_fps("wide12"), 1.0 / 3.0) self.assertEqual(frame_policy_sampling_fps("wide16"), 0.25) def test_modelonly_frame_policy_is_question_independent(self) -> None: self.assertEqual(resolve_frame_policy("recent4"), "recent4") self.assertEqual(resolve_frame_policy("wide12"), "wide12") self.assertEqual(resolve_frame_policy("prefix8"), "prefix8") self.assertEqual(resolve_frame_policy("prefix32"), "prefix32") self.assertEqual(resolve_frame_policy("causalbudget20"), "causalbudget20") self.assertEqual( resolve_frame_policy("causalbudget32r8"), "causalbudget32r8" ) self.assertEqual(resolve_frame_policy("causalhybrid32"), "causalhybrid32") self.assertEqual( resolve_frame_policy("causalhybrid32r8"), "causalhybrid32r8" ) with self.assertRaises(ValueError): resolve_frame_policy("adaptive_query_budget32") def test_source_probe_has_small_bounded_cache(self) -> None: self.assertEqual(probe_source_video.cache_parameters()["maxsize"], 8) def test_decord_reader_pool_bounds_native_decoder_threads(self) -> None: calls = [] class Reader: pass def video_reader(path, *, num_threads): calls.append((path, num_threads)) return Reader() pool = DecordReaderPool(2, num_threads=3) with patch.dict( "sys.modules", {"decord": SimpleNamespace(VideoReader=video_reader)} ): first, first_hit = pool.reader("a.mp4") second, second_hit = pool.reader("a.mp4") self.assertIs(first, second) self.assertFalse(first_hit) self.assertTrue(second_hit) self.assertEqual(calls, [("a.mp4", 3)]) def test_decord_reader_pool_serializes_same_video_prefetch(self) -> None: state_lock = __import__("threading").Lock() active = 0 peak = 0 instances = 0 class Frame: shape = (16, 32, 3) class Reader: def get_avg_fps(self): nonlocal active, peak with state_lock: active += 1 peak = max(peak, active) time.sleep(0.01) with state_lock: active -= 1 return 30.0 def __len__(self): return 10 def __getitem__(self, index): return Frame() def video_reader(path, *, num_threads): nonlocal instances instances += 1 return Reader() pool = DecordReaderPool(2, num_threads=3) with patch.dict( "sys.modules", {"decord": SimpleNamespace(VideoReader=video_reader)} ): with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: values = list(executor.map(pool.metadata, ["a.mp4"] * 8)) self.assertEqual(values, [(30.0, 10, 16, 32)] * 8) self.assertEqual(instances, 1) self.assertEqual(peak, 1) def test_exact_decode_cache_reuses_only_matching_sampling_request(self) -> None: from pathlib import Path class Frame: def __init__(self, value: int) -> None: self.value = value def copy(self): return Frame(self.value) calls = [] def decoder(path, **kwargs): calls.append((path, kwargs)) return [Frame(len(calls))], {"frame_indices": [len(calls)]} cache = RecentWindowCache(2, decoder=decoder) first_frames, first_sampling, first_hit = cache.get( Path("video.mp4"), visible_until=10.0, recent_frames=4, sampling_fps=1.0, ) first_frames[0].value = 99 first_sampling["frame_indices"][0] = 99 second_frames, second_sampling, second_hit = cache.get( Path("video.mp4"), visible_until=10.0, recent_frames=4, sampling_fps=1.0, ) _, _, third_hit = cache.get( Path("video.mp4"), visible_until=11.0, recent_frames=4, sampling_fps=1.0, ) self.assertFalse(first_hit) self.assertTrue(second_hit) self.assertFalse(third_hit) self.assertEqual(len(calls), 2) self.assertEqual(second_frames[0].value, 1) self.assertEqual(second_sampling["frame_indices"], [1]) self.assertEqual(second_sampling["frame_policy"], "recent4") self.assertEqual(second_sampling["policy_sampling_fps"], 1.0) self.assertEqual((cache.hits, cache.misses), (1, 2)) def test_prefix_policy_uses_dedicated_sequential_decoder(self) -> None: cache = RecentWindowCache(0) decoded = ([object()], {"frame_indices": [0]}) with patch( "experiments.evaluate_qwen3vl_recent4_full.decode_prefix_window_ffmpeg", return_value=decoded, ) as decoder: frames, sampling, hit = cache.get( Path("video.mp4"), visible_until=10.0, recent_frames=4, sampling_fps=1.0, frame_policy="prefix32", ) self.assertIs(frames, decoded[0]) self.assertFalse(hit) self.assertEqual(sampling["frame_policy"], "prefix32") decoder.assert_called_once_with( Path("video.mp4"), visible_until=10.0, recent_frames=32, sampling_fps=3.2, ) def test_exact_decode_cache_is_safe_for_parallel_prefix_misses(self) -> None: class Frame: def __init__(self, value: str) -> None: self.value = value def copy(self): return Frame(self.value) def decoder(path, **kwargs): time.sleep(0.01) return [Frame(str(path))], {"frame_indices": [str(path)]} cache = RecentWindowCache(8, decoder=decoder) paths = [Path(f"video-{index}.mp4") for index in range(8)] with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: values = list( executor.map( lambda path: cache.get( path, visible_until=10.0, recent_frames=32, sampling_fps=3.2, frame_policy="prefix32", ), paths, ) ) self.assertEqual([value[0][0].value for value in values], list(map(str, paths))) self.assertEqual(cache.misses, 8) def test_exact_decode_cache_singleflights_matching_parallel_requests(self) -> None: class Frame: def copy(self): return Frame() calls = [] def decoder(path, **kwargs): calls.append((path, kwargs)) time.sleep(0.05) return [Frame()], {"frame_indices": [1]} cache = RecentWindowCache(8, decoder=decoder) def request(_): return cache.get( Path("same-video.mp4"), visible_until=10.0, recent_frames=4, sampling_fps=1.0, ) with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: values = list(executor.map(request, range(16))) self.assertEqual(len(calls), 1) self.assertEqual(cache.misses, 1) self.assertEqual(cache.hits, 15) self.assertGreaterEqual(cache.singleflight_hits, 1) self.assertEqual(sum(bool(value[2]) for value in values), 15) def test_visible_prefix_is_clamped_to_last_real_frame(self) -> None: effective, clamped = clamp_visible_until_to_source(10.0, 2.0, 9) self.assertEqual(effective, 4.0) self.assertTrue(clamped) effective, clamped = clamp_visible_until_to_source(3.0, 2.0, 9) self.assertEqual(effective, 3.0) self.assertFalse(clamped) def test_explicit_uid_subset_keeps_export_order(self) -> None: import tempfile from pathlib import Path with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "samples.jsonl" path.write_text( '{"uid":"a"}\n{"uid":"b"}\n{"uid":"c"}\n', encoding="utf-8", ) self.assertEqual( selected_uids(path, 0, -1, ["c", "a"]), ["a", "c"], ) def test_batched_runner_keeps_auditable_resume_controls(self) -> None: source = Path("experiments/evaluate_qwen3vl_recent4_full.py").read_text( encoding="utf-8" ) launcher = Path("scripts/run_recent4_single_benchmark.sh").read_text( encoding="utf-8" ) self.assertIn('processor.tokenizer.padding_side = "left"', source) self.assertIn("validate_processor_contract(processor_path, protocol)", source) self.assertIn('"inference_batch_size": args.batch_size', source) self.assertIn('"execution_optimization",', source) self.assertIn('--uids-file "$uid_file"', launcher) self.assertIn('uids[index::count]', launcher) self.assertIn('--batch-size "$attempt_batch"', launcher) self.assertIn('attempt_batch=$EVAL_BATCH_SIZE', launcher) self.assertIn('|| smoke_rc=$?', launcher) self.assertIn('|| runner_rc=$?', launcher) self.assertIn( 'after writing a complete validated result', launcher, ) def test_recent_window_is_half_open_and_causal(self) -> None: self.assertEqual(recent_window_bounds(10.0, 4, 1.0), (6.0, 10.0)) self.assertEqual(recent_window_bounds(1.0, 4, 1.0), (0.0, 1.0)) def test_answer_normalization(self) -> None: options = ["A. first", "B. second", "C. third", "D. fourth"] self.assertEqual(normalize_answer("(C) third", "question", options), "C") self.assertEqual(normalize_answer("second", "question", options), "B") self.assertEqual(normalize_answer("A map", "question", ["A. compass", "B. A map", "C. sky", "D. radar"]), "B") self.assertEqual( normalize_answer( "Yes", "Answer only with “Yes” or “No”.", [], ), "Yes", ) def test_streaming_prompt_matches_release_contract(self) -> None: prompt = render_question( "streamingbench", "What is visible?", ["A. first", "B. second", "C. third", "D. fourth"], ) self.assertIn("advanced video question-answering AI assistant", prompt) self.assertTrue(prompt.endswith( "Only give the best option's letter (A, B, C, or D) directly." )) def test_ovo_mcq_prompt_matches_release_contract(self) -> None: prompt = render_question( "ovo", "What happened?", ["one", "two", "three", "four"], ) self.assertEqual( prompt, "What happened?\nOptions: A. one; B. two; C. three; D. four;" "\nOnly give the best option's letter directly.", ) def test_ovo_requires_full_runtime_asset_attestation(self) -> None: launcher = Path("scripts/run_recent4_single_benchmark.sh").read_text( encoding="utf-8" ) self.assertIn("OVO_RUNTIME_REPORT", launcher) self.assertIn(".expanded_queries == 3035", launcher) self.assertIn(".covered_queries == 3035", launcher) self.assertIn(".missing_videos == 0", launcher) self.assertIn(".annotation_sha256 == $annotation_sha256", launcher) self.assertIn(".frozen_export_sha256 == $frozen_export_sha256", launcher) if __name__ == "__main__": unittest.main()