Blame for code/tests/test_recent4_runner.py

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()