mbx/mb-project-info.git

			@@ -1,8 +1,11 @@
			from __future__ import annotations

			import csv
			from datetime import datetime, timedelta, timezone
			import hashlib
			import importlib.util
			import io
			import json
			import os
			from pathlib import Path
			import re
			@@ -10,6 +13,7 @@
			import subprocess
			import sys
			import tempfile
			import textwrap
			import unittest
			from unittest import mock
			import zlib
			@@ -26,21 +30,63 @@
			SPEC.loader.exec_module(slides)


			def feature(index: int, array: np.ndarray, score: float = 0.2, sharpness: float \| None = None):
			def palette(marker_bin: int \| None = None, coverage: float = 0.20):
			histogram = np.full(slides.PALETTE_BIN_COUNT, 1 / slides.PALETTE_BIN_COUNT, np.float32)
			if marker_bin is not None:
			histogram.fill((1.0 - coverage) / (slides.PALETTE_BIN_COUNT - 1))
			histogram[marker_bin] = coverage
			histogram /= float(np.sum(histogram, dtype=np.float64))
			return histogram


			def feature(
			index: int,
			array: np.ndarray,
			score: float = 0.2,
			sharpness: float \| None = None,
			page_palette: np.ndarray \| None = None,
			):
			value = int(np.mean(array))
			phash = int.from_bytes(hashlib.sha256(array.tobytes()).digest()[:8], "big")
			return slides.FrameFeature(
			sample_index=index,
			relative_seconds=float(index),
			gray=array.astype(np.uint8).copy(),
			palette_histogram=(palette() if page_palette is None else page_palette.copy()),
			perceptual_hash=phash,
			content_score=score,
			sharpness=float(value if sharpness is None else sharpness),
			)


			def candidate(timestamp: float, array: np.ndarray, score: float, phash: int = 0):
			return slides.PageCandidate(timestamp, array.astype(np.uint8).copy(), phash, score, 1.0)
			def candidate(
			timestamp: float,
			array: np.ndarray,
			score: float,
			phash: int = 0,
			page_palette: np.ndarray \| None = None,
			):
			return slides.PageCandidate(
			timestamp,
			array.astype(np.uint8).copy(),
			palette() if page_palette is None else page_palette.copy(),
			phash,
			score,
			1.0,
			)


			def logical_page(order: int, page_palette: np.ndarray \| None = None):
			page_feature = np.full((8, 8), order % 256, np.uint8)
			return slides.LogicalPage(
			logical_anchor_feature=page_feature.copy(),
			current_feature=page_feature,
			current_palette=palette() if page_palette is None else page_palette.copy(),
			current_hash=order,
			current_timestamp=float(order),
			current_score=0.2,
			first_seen_order=order,
			)


			def sha256(path: Path) -> str:
			@@ -148,20 +194,915 @@
			recent = np.full((8, 8), 4, np.uint8)
			classifier.consume(candidate(1, historical, 0.20, 0))
			classifier.pages.append(
			slides.LogicalPage(recent.copy(), recent.copy(), 0, 2.0, 0.20, 1)
			slides.LogicalPage(recent.copy(), recent.copy(), palette(), 0, 2.0, 0.20, 1)
			)
			action = classifier.consume(candidate(3, historical, 0.22, 0))
			self.assertEqual(action, "animation_updated")
			self.assertEqual(len(classifier.pages), 2)


			class MainSequenceTests(unittest.TestCase):
			def test_palette_histogram_distinguishes_dark_blue_from_black(self):
			image = np.array(
			[
			[[16, 16, 48], [16, 16, 48]],
			[[16, 16, 16], [16, 16, 16]],
			],
			dtype=np.uint8,
			)
			histogram = slides._palette_histogram(image)
			self.assertEqual(histogram.shape, (512,))
			self.assertTrue(np.all(np.isfinite(histogram)))
			self.assertAlmostEqual(float(np.sum(histogram)), 1.0, places=6)
			self.assertEqual(float(histogram[1]), 0.5)
			self.assertEqual(float(histogram[0]), 0.5)

			def test_selects_old_pages_002_through_038_and_keeps_two_gaps(self):
			marker_pages = {2, range(4, 9), range(10, 39)}
			pages = [
			logical_page(index - 1, palette(1) if index in marker_pages else palette())
			for index in range(1, 54)
			]
			selected = slides.select_main_sequence(pages)
			self.assertEqual(selected, pages[1:38])
			self.assertIs(selected[1], pages[2]) # old slide_003: first one-page gap
			self.assertIs(selected[7], pages[8]) # old slide_009: second one-page gap
			self.assertEqual([page.first_seen_order for page in selected], list(range(1, 38)))

			def test_conservative_gates_keep_all_candidates(self):
			cases = {
			"markers": [palette(1) if index < 7 else palette() for index in range(10)],
			"share": [palette(1) if 3 <= index <= 10 else palette() for index in range(20)],
			"density": [
			palette(1) if index in {0, 2, 4, 6, 7, 8, 9, 10} else palette()
			for index in range(11)
			],
			}
			for name, histograms in cases.items():
			with self.subTest(name=name):
			pages = [logical_page(index, value) for index, value in enumerate(histograms)]
			selected = slides.select_main_sequence(pages)
			self.assertEqual(selected, pages)
			self.assertTrue(all(left is right for left, right in zip(selected, pages)))

			def test_all_theme_pages_are_preserved(self):
			pages = [logical_page(index, palette(7)) for index in range(12)]
			selected = slides.select_main_sequence(pages)
			self.assertEqual(selected, pages)
			self.assertTrue(all(left is right for left, right in zip(selected, pages)))

			def test_invalid_palette_is_explicit_failure(self):
			invalid = [
			np.zeros(5, np.float32),
			np.full(512, np.nan, np.float32),
			np.zeros(512, np.float32),
			]
			for value in invalid:
			with self.subTest(shape=value.shape, finite=bool(np.all(np.isfinite(value)))):
			with self.assertRaises(slides.SlideExtractionError):
			slides.select_main_sequence([logical_page(0, value)])


			class ProcessAndCommandTests(unittest.TestCase):
			def test_command_freezes_time_mapping(self):
			scan = slides._scan_command("ffmpeg", Path("x.mp4"), 320, 180)
			self.assertIn("setpts=PTS-STARTPTS,fps=fps=1:start_time=0,scale=320:180:flags=bilinear", scan)
			input_index = scan.index("-i")
			self.assertLess(scan.index("-threads"), input_index)
			self.assertEqual(scan[scan.index("-threads") + 1], str(slides.MEDIA_DECODE_THREADS))
			self.assertEqual(scan[scan.index("-filter_threads") + 1], str(slides.MEDIA_FILTER_THREADS))
			extract = slides._extract_command("ffmpeg", Path("x.mp4"), 12.0, Path("x.png"))
			self.assertEqual(extract[extract.index("-ss") + 1], "12.000")
			self.assertLess(extract.index("-threads"), extract.index("-i"))
			self.assertEqual(feature(9, np.zeros((2, 2), np.uint8)).relative_seconds, 9.0)

			def test_hardware_commands_place_decoder_and_threads_before_input(self):
			for codec, decoder in (("h264", "h264_cuvid"), ("hevc", "hevc_cuvid")):
			with self.subTest(codec=codec):
			self.assertEqual(slides._decoder_for_codec(codec, "yuv420p"), decoder)
			command = slides._scan_command(
			"ffmpeg", Path(f"{codec}.mp4"), 320, 180, decoder
			)
			input_index = command.index("-i")
			for option in ("-hwaccel", "-hwaccel_output_format", "-c:v", "-threads"):
			self.assertLess(command.index(option), input_index)
			self.assertEqual(command[command.index("-hwaccel") + 1], "cuda")
			self.assertEqual(command[command.index("-c:v") + 1], decoder)
			self.assertEqual(
			command[command.index("-threads") + 1],
			str(slides.MEDIA_DECODE_THREADS),
			)
			self.assertIn("hwdownload,format=nv12", command[command.index("-vf") + 1])
			self.assertEqual(slides._decoder_for_codec("h264", "gbrp"), "h264_cuvid")
			self.assertEqual(slides._decoder_for_codec("hevc", "p010le"), "hevc_cuvid")
			self.assertIsNone(slides._decoder_for_codec("vp9", "yuv420p"))

			def test_hardware_external_failures_have_typed_boundary(self):
			class FakeStdout:
			def __init__(self, payloads):
			self.payloads = list(payloads)

			def read(self, size):
			return self.payloads.pop(0) if self.payloads else b""

			def close(self):
			return None

			class FakeProcess:
			def __init__(self, payloads, code):
			self.stdout = FakeStdout(payloads)
			self.code = code

			def wait(self, timeout=None):
			return self.code

			def poll(self):
			return self.code

			def terminate(self):
			return None

			def kill(self):
			return None

			info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
			cases = (
			("nonzero", FakeProcess([], 7), "exit=7"),
			("zero", FakeProcess([], 0), "未输出任何完整"),
			("partial", FakeProcess([b"x", b""], 0), "输出半帧"),
			)
			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_which_or_error", return_value="ffmpeg"
			):
			root = Path(raw)
			with mock.patch.object(slides.subprocess, "Popen", side_effect=OSError("start denied")):
			with self.assertRaisesRegex(slides.HardwareDecodePathError, "启动失败"):
			slides._scan_video_once(Path("x.mp4"), root, info, decoder="h264_cuvid")
			for name, process, pattern in cases:
			with self.subTest(name=name), mock.patch.object(
			slides.subprocess, "Popen", return_value=process
			):
			with self.assertRaisesRegex(slides.HardwareDecodePathError, pattern):
			slides._scan_video_once(
			Path("x.mp4"), root, info, decoder="h264_cuvid"
			)

			def test_hardware_cleanup_failure_does_not_trigger_cpu_fallback(self):
			info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
			hardware_error = slides.HardwareDecodePathError("hardware-root")
			slides._attach_cleanup_note(hardware_error, "cleanup-failed")
			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_scan_video_once", side_effect=hardware_error
			) as scan:
			with self.assertRaises(slides.HardwareDecodePathError) as caught:
			slides.scan_video(Path("x.mp4"), Path(raw), info)
			self.assertIs(caught.exception, hardware_error)
			self.assertEqual(scan.call_count, 1)
			self.assertIn("cleanup-failed", "\n".join(caught.exception.__notes__))

			def test_active_errors_keep_identity_when_process_cleanup_fails(self):
			frame_bytes = 320 * 180 * 3

			class Stream:
			def __init__(self, error=None, partial=False):
			self.error = error
			self.partial = partial
			self.calls = 0

			def read(self, size):
			self.calls += 1
			if self.error is not None:
			raise self.error
			if self.partial:
			return b"x" if self.calls == 1 else b""
			return b"\0" * frame_bytes if self.calls == 1 else b""

			def close(self):
			return None

			class FakeProcess:
			def __init__(self, stream):
			self.stdout = stream

			def wait(self, timeout=None):
			return 0

			def poll(self):
			return None

			def terminate(self):
			return None

			cleanup = OSError("cleanup-failed")
			info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
			cases = (
			("business", None, ValueError("business-root")),
			("cpu", None, slides.SlideExtractionError("cpu-root")),
			("base", KeyboardInterrupt(), None),
			)
			for name, read_error, feature_error in cases:
			root_error = read_error or feature_error
			stream = Stream(error=read_error)
			patches = [
			mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"),
			mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess(stream)),
			mock.patch.object(slides, "_attempt_reap", return_value=cleanup),
			]
			if feature_error is not None:
			patches.append(mock.patch.object(slides, "_gray_feature", side_effect=feature_error))
			with self.subTest(name=name), tempfile.TemporaryDirectory() as raw:
			for item in patches:
			item.start()
			try:
			with self.assertRaises(type(root_error)) as caught:
			slides._scan_video_once(
			Path("x.mp4"), Path(raw), info, decoder=None
			)
			self.assertIs(caught.exception, root_error)
			self.assertIn("cleanup-failed", "\n".join(caught.exception.__notes__))
			finally:
			for item in reversed(patches):
			item.stop()

			partial_process = FakeProcess(Stream(partial=True))
			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_which_or_error", return_value="ffmpeg"
			), mock.patch.object(slides.subprocess, "Popen", return_value=partial_process), mock.patch.object(
			slides, "_attempt_reap", return_value=cleanup
			):
			with self.assertRaises(slides.HardwareDecodePathError) as caught:
			slides._scan_video_once(
			Path("x.mp4"), Path(raw), info, decoder="h264_cuvid"
			)
			hardware_error = caught.exception
			self.assertTrue(getattr(hardware_error, "_media_cleanup_failed", False))
			self.assertIn("cleanup-failed", "\n".join(hardware_error.__notes__))
			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_scan_video_once", side_effect=hardware_error
			) as scan:
			with self.assertRaises(slides.HardwareDecodePathError) as caught_again:
			slides.scan_video(Path("x.mp4"), Path(raw), info)
			self.assertIs(caught_again.exception, hardware_error)
			self.assertEqual(scan.call_count, 1)

			def test_product_popen_uses_below_normal_creation_flags(self):
			class FakeProcess:
			stdout = None

			def wait(self, timeout=None):
			return 0

			def poll(self):
			return 0

			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_subprocess_creation_flags", return_value=12345
			), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess()) as popen:
			slides._run_file_command(["tool"], Path(raw) / "stderr.log", "stage")
			self.assertEqual(popen.call_args.kwargs["creationflags"], 12345)

			def test_fallback_only_handles_hardware_error_and_never_retries_cpu(self):
			info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
			page = logical_page(0)
			success = slides.ScanResult((page,), 10, 2, (("new_page", 1),), None)
			hardware_error = slides.HardwareDecodePathError("decoder failed")
			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_scan_video_once", side_effect=[hardware_error, success]
			) as scan, mock.patch("sys.stderr", new_callable=io.StringIO) as stderr:
			self.assertEqual(slides.scan_video(Path("x.mp4"), Path(raw), info), [page])
			self.assertEqual([call.kwargs["decoder"] for call in scan.call_args_list], [
			"h264_cuvid", None
			])
			self.assertIn("decoder failed", stderr.getvalue())

			cpu_error = slides.SlideExtractionError("cpu root cause")
			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_scan_video_once", side_effect=[hardware_error, cpu_error]
			) as scan:
			with self.assertRaises(slides.SlideExtractionError) as caught:
			slides.scan_video(Path("x.mp4"), Path(raw), info)
			self.assertIs(caught.exception, cpu_error)
			self.assertEqual(scan.call_count, 2)

			def test_algorithm_empty_and_baseexception_never_fall_back(self):
			info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
			empty = slides.ScanResult((), 10, 0, (), "h264_cuvid")
			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_scan_video_once", return_value=empty
			) as scan:
			with self.assertRaisesRegex(slides.SlideExtractionError, "未发现"):
			slides.scan_video(Path("x.mp4"), Path(raw), info)
			scan.assert_called_once()
			for error in (RuntimeError("classifier failed"), KeyboardInterrupt(), SystemExit(9)):
			with self.subTest(error=type(error).__name__), tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_scan_video_once", side_effect=error
			) as scan:
			with self.assertRaises(type(error)) as caught:
			slides.scan_video(Path("x.mp4"), Path(raw), info)
			self.assertIs(caught.exception, error)
			scan.assert_called_once()

			def test_algorithm_exception_inside_hardware_scan_preserves_original_type(self):
			frame_bytes = 320 * 180 * 3

			class FullFrameStream:
			def __init__(self):
			self.done = False

			def read(self, size):
			if self.done:
			return b""
			self.done = True
			return b"\0" * frame_bytes

			def close(self):
			return None

			class FakeProcess:
			stdout = FullFrameStream()

			def wait(self, timeout=None):
			return 0

			def poll(self):
			return 0

			root_error = ValueError("feature root cause")
			info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
			with tempfile.TemporaryDirectory() as raw, mock.patch.object(
			slides, "_which_or_error", return_value="ffmpeg"
			), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess()), mock.patch.object(
			slides, "_gray_feature", side_effect=root_error
			):
			with self.assertRaises(ValueError) as caught:
			slides._scan_video_once(
			Path("x.mp4"), Path(raw), info, decoder="h264_cuvid"
			)
			self.assertIs(caught.exception, root_error)

			@unittest.skipUnless(shutil.which("powershell"), "Windows PowerShell is required")
			def test_resource_sampler_parser_rejects_cpu_boundaries_and_empty_windows(self):
			sampler = Path(__file__).resolve().parent / "monitor_media_host_responsiveness.ps1"

			fields = (
			"timestamp_utc", "previous_valid_timestamp_utc", "interval_seconds",
			"counter_warmup", "cpu_valid", "host_cpu_percent",
			"scan_phase_active", "scan_ffmpeg_present", "scan_ffmpeg_exit_observed",
			"scan_ffmpeg_pid", "scan_ffmpeg_command_line", "decoder",
			"logical_processor_count", "task_cpu_percent", "task_working_set_bytes",
			"task_pids", "root_priority", "ffmpeg_priorities", "ffmpeg_command_lines",
			"gpu_query_status", "gpu_query_raw_rows", "gpu_utilization_percent",
			"gpu_decoder_utilization_percent", "gpu_memory_used_mib",
			"nvidia_smi_compute_query_status", "nvidia_smi_compute_raw_rows",
			"nvidia_smi_compute_task_rows",
			"gpu_process_memory_query_status", "gpu_process_memory_instances",
			"task_gpu_dedicated_memory_mib",
			)
			scan_command = (
			"ffmpeg -hwaccel cuda -hwaccel_output_format cuda "
			"-c:v h264_cuvid -threads 4 "
			"-filter_threads 2 -filter_complex_threads 2 -i x.mp4 -map 0:v:0 "
			"-vf hwdownload,format=nv12,fps=fps=1:start_time=0 "
			"-pix_fmt rgb24 -f rawvideo -"
			)

			def make_row(current, previous, interval, cpu, *, warmup, present, exited):
			return {
			"timestamp_utc": current.isoformat().replace("+00:00", "Z"),
			"previous_valid_timestamp_utc": (
			"" if previous is None else previous.isoformat().replace("+00:00", "Z")
			),
			"interval_seconds": f"{interval:.6f}",
			"counter_warmup": str(warmup).lower(),
			"cpu_valid": str(not warmup).lower(),
			"host_cpu_percent": "" if warmup else f"{cpu:.6f}",
			"scan_phase_active": "true",
			"scan_ffmpeg_present": str(present).lower(),
			"scan_ffmpeg_exit_observed": str(exited).lower(),
			"scan_ffmpeg_pid": "4242",
			"scan_ffmpeg_command_line": scan_command,
			"decoder": "h264_cuvid",
			"logical_processor_count": "44",
			"task_cpu_percent": "" if warmup else "12.5",
			"task_working_set_bytes": "123456789",
			"task_pids": "101;4242",
			"root_priority": "BelowNormal",
			"ffmpeg_priorities": (
			"[]" if not present else '[{"pid":4242,"priority":"BelowNormal"}]'
			),
			"ffmpeg_command_lines": (
			'[{"pid":4242,"command_line":' + json.dumps(scan_command) + "}]"
			),
			"gpu_query_status": "ok",
			"gpu_query_raw_rows": '["20, 31, 2048"]',
			"gpu_utilization_percent": "20",
			"gpu_decoder_utilization_percent": "31",
			"gpu_memory_used_mib": "2048",
			"nvidia_smi_compute_query_status": "ok",
			"nvidia_smi_compute_raw_rows": '["4242, N/A"]',
			"nvidia_smi_compute_task_rows": '["4242, N/A"]',
			"gpu_process_memory_query_status": "ok",
			"gpu_process_memory_instances": (
			'[{"pid":4242,"instance_name":"pid_4242_luid_0",'
			'"counter_path":"path","dedicated_usage_bytes":1048576}]'
			),
			"task_gpu_dedicated_memory_mib": "1",
			}

			def analyze_case(root, name, intervals, cpu_values, mutate=None, omit_field=None):
			case = root / name
			case.mkdir()
			samples = case / "samples.csv"
			current = datetime(2026, 8, 4, tzinfo=timezone.utc)
			rows = [make_row(current, None, 0, 0, warmup=True, present=True, exited=False)]
			previous = current
			for interval, cpu in zip(intervals, cpu_values):
			current += timedelta(seconds=interval)
			rows.append(make_row(
			current, previous, interval, cpu,
			warmup=False, present=True, exited=False,
			))
			previous = current
			current += timedelta(seconds=0.5)
			rows.append(make_row(
			current, previous, 0.5, 20.0,
			warmup=False, present=False, exited=True,
			))
			if mutate is not None:
			mutate(rows)
			output_fields = tuple(field for field in fields if field != omit_field)
			with samples.open("w", newline="", encoding="utf-8") as handle:
			writer = csv.DictWriter(handle, fieldnames=output_fields, extrasaction="ignore")
			writer.writeheader()
			writer.writerows(rows)
			evidence = case / "evidence"
			result = subprocess.run(
			[
			shutil.which("powershell"), "-NoProfile", "-File", str(sampler),
			"-AnalyzeSamplesPath", str(samples), "-EvidenceDirectory", str(evidence),
			],
			capture_output=True,
			text=True,
			timeout=60,
			)
			self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
			return json.loads((evidence / "resource_summary.json").read_text(encoding="utf-8-sig"))

			with tempfile.TemporaryDirectory() as raw:
			root = Path(raw)
			low = analyze_case(root, "low", [0.5] * 22, [84.9] * 22)
			self.assertTrue(low["sampling_valid"])
			self.assertGreater(low["window_count"], 0)
			self.assertTrue(low["cpu_gate_pass"])

			equal = analyze_case(root, "equal", [0.5] * 22, [85.0] * 22)
			self.assertTrue(equal["sampling_valid"])
			self.assertFalse(equal["all_windows_host_cpu_strictly_below_85"])
			self.assertFalse(equal["cpu_gate_pass"])

			exact_ten = analyze_case(
			root, "exact-ten", [0.5] * 21, [85.0] * 20 + [20.0]
			)
			self.assertEqual(exact_ten["maximum_high_cpu_segment_seconds"], 10.0)
			self.assertFalse(exact_ten["cpu_gate_pass"])

			below_ten = analyze_case(
			root, "below-ten", [0.49995] * 20 + [0.5], [85.0] * 20 + [20.0]
			)
			self.assertEqual(below_ten["maximum_high_cpu_segment_seconds"], 9.999)

			sparse = analyze_case(root, "sparse", [1.0] * 12, [20.0] * 12)
			self.assertFalse(sparse["sampling_valid"])
			self.assertTrue(any("window_density_invalid" in item for item in sparse["sampling_problems"]))

			zero_window = analyze_case(root, "zero-window", [0.5] * 8, [20.0] * 8)
			self.assertFalse(zero_window["sampling_valid"])
			self.assertEqual(zero_window["window_count"], 0)

			gap = analyze_case(
			root, "gap", [0.5] * 10 + [1.6] + [0.5] * 10, [20.0] * 21
			)
			self.assertFalse(gap["sampling_valid"])
			self.assertGreater(gap["maximum_observed_interval_seconds"], 1.5)

			reversed_time = analyze_case(
			root, "time-reversal", [0.5] * 12, [20.0] * 12,
			mutate=lambda rows: rows[5].__setitem__("timestamp_utc", rows[3]["timestamp_utc"]),
			)
			self.assertFalse(reversed_time["sampling_valid"])
			self.assertTrue(any(
			"timestamp_not_strictly_increasing" in item
			for item in reversed_time["sampling_problems"]
			))

			interval_mismatch = analyze_case(
			root, "interval-mismatch", [0.5] * 12, [20.0] * 12,
			mutate=lambda rows: rows[4].__setitem__("interval_seconds", "9.0"),
			)
			self.assertFalse(interval_mismatch["sampling_valid"])
			self.assertTrue(any(
			"interval_mismatch" in item for item in interval_mismatch["sampling_problems"]
			))

			previous_mismatch = analyze_case(
			root, "previous-mismatch", [0.5] * 12, [20.0] * 12,
			mutate=lambda rows: rows[4].__setitem__(
			"previous_valid_timestamp_utc", rows[1]["timestamp_utc"]
			),
			)
			self.assertFalse(previous_mismatch["sampling_valid"])
			self.assertTrue(any(
			"previous_timestamp_mismatch" in item
			for item in previous_mismatch["sampling_problems"]
			))

			for omitted in (
			"gpu_query_raw_rows", "task_pids", "ffmpeg_command_lines",
			"nvidia_smi_compute_task_rows",
			):
			with self.subTest(omitted=omitted):
			missing_field = analyze_case(
			root, f"missing-{omitted}", [0.5] * 12, [20.0] * 12,
			omit_field=omitted,
			)
			self.assertFalse(missing_field["sampling_valid"])
			self.assertIn(
			f"required_column_missing_{omitted}",
			missing_field["sampling_problems"],
			)

			first_not_present = analyze_case(
			root, "first-coverage", [0.5] * 12, [20.0] * 12,
			mutate=lambda rows: rows[0].__setitem__("scan_ffmpeg_present", "false"),
			)
			self.assertFalse(first_not_present["sampling_valid"])
			self.assertIn("scan_phase_first_row_not_present", first_not_present["sampling_problems"])

			last_not_exit = analyze_case(
			root, "last-coverage", [0.5] * 12, [20.0] * 12,
			mutate=lambda rows: rows[-1].__setitem__("scan_ffmpeg_exit_observed", "false"),
			)
			self.assertFalse(last_not_exit["sampling_valid"])
			self.assertIn("scan_phase_last_row_not_exit_observation", last_not_exit["sampling_problems"])

			argv_rewrites = {
			"missing-output-format": lambda value: value.replace(
			"-hwaccel_output_format cuda ", ""
			),
			"wrong-output-format": lambda value: value.replace(
			"-hwaccel_output_format cuda", "-hwaccel_output_format qsv"
			),
			"late-output-format": lambda value: value.replace(
			"-hwaccel_output_format cuda ", ""
			).replace("-i x.mp4", "-i x.mp4 -hwaccel_output_format cuda"),
			"missing-hwdownload": lambda value: value.replace(
			"hwdownload,format=nv12,", ""
			),
			}
			for name, rewrite in argv_rewrites.items():
			with self.subTest(argv_case=name):
			invalid_argv = analyze_case(
			root, f"invalid-{name}", [0.5] * 12, [20.0] * 12,
			mutate=lambda rows, transform=rewrite: [
			row.__setitem__(
			"scan_ffmpeg_command_line",
			transform(row["scan_ffmpeg_command_line"]),
			)
			for row in rows
			],
			)
			self.assertFalse(invalid_argv["sampling_valid"])
			self.assertIn("nvdec_scan_argv_invalid", invalid_argv["sampling_problems"])

			@unittest.skipUnless(
			os.name == "nt" and shutil.which("powershell") and shutil.which("nvidia-smi"),
			"Windows PowerShell and nvidia-smi are required",
			)
			def test_resource_sampler_run_rejects_forged_nvdec_and_excludes_page_ffmpeg(self):
			sampler = Path(__file__).resolve().parent / "monitor_media_host_responsiveness.ps1"
			sampler_text = sampler.read_text(encoding="utf-8")
			process_start = sampler_text.index("[void]$Process.Start()")
			self.assertLess(
			sampler_text.index("$HostCpuCounter = New-Object"), process_start
			)
			self.assertLess(sampler_text.index("$HostCpuCounter.NextValue()"), process_start)
			self.assertLess(
			sampler_text.index("$GpuMemoryCategory = New-Object"), process_start
			)
			self.assertLess(sampler_text.index("$GpuMemoryCategory.GetInstanceNames()"), process_start)
			self.assertLess(
			sampler_text.index('$StartInfo.EnvironmentVariables["MBX_MEDIA_SAMPLER_READY"]'),
			process_start,
			)
			with tempfile.TemporaryDirectory() as raw:
			root = Path(raw)
			fake_ffmpeg = root / "ffmpeg.exe"
			shutil.copy2(sys.executable, fake_ffmpeg)
			child = root / "fake_ffmpeg_child.py"
			child.write_text(
			"import sys, time\n"
			"time.sleep(float(sys.argv[sys.argv.index('--sleep') + 1]))\n",
			encoding="utf-8",
			)
			target = root / "stub_media_target.py"
			target.write_text(
			textwrap.dedent(
			f"""
			import ctypes
			import os
			from pathlib import Path
			import subprocess
			import sys
			import time

			sys.stdout.reconfigure(encoding="utf-8")
			sys.stderr.reconfigure(encoding="utf-8")
			kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
			kernel32.SetPriorityClass(kernel32.GetCurrentProcess(), 0x00004000)
			creationflags = 0x00004000
			if os.environ.get("MBX_MEDIA_SAMPLER_READY") != "1":
			raise RuntimeError("sampler-ready handshake missing")
			fake_ffmpeg = Path({str(fake_ffmpeg)!r})
			child = Path({str(child)!r})
			scan = subprocess.Popen([
			str(fake_ffmpeg), str(child), "--sleep", "7.0",
			"-hwaccel", "cuda", "-c:v", "h264_cuvid",
			"-threads", "4", "-filter_threads", "2",
			"-filter_complex_threads", "2", "-i", "fake.mp4",
			"-map", "0:v:0", "-vf",
			"hwdownload,format=nv12,setpts=PTS-STARTPTS,"
			"fps=fps=1:start_time=0,scale=320:180",
			"-pix_fmt", "rgb24", "-f", "rawvideo", "-",
			], creationflags=creationflags)
			scan.wait()
			print("警告：NVDEC 不可用；改用受限 CPU。", file=sys.stderr, flush=True)
			print(
			"扫描解码路径：NVDEC/CUDA (h264_cuvid)；解码线程=4；滤镜线程=2",
			flush=True,
			)
			time.sleep(1.5)
			for index in range(24):
			churn = subprocess.Popen([
			str(fake_ffmpeg), str(child), "--sleep", "0.04",
			"-threads", "4", "-i", "fake.mp4",
			"-frames:v", "1", "-n", f"race-{{index}}.png",
			], creationflags=creationflags)
			churn.wait()
			page = subprocess.Popen([
			str(fake_ffmpeg), str(child), "--sleep", "4.0",
			"-threads", "4", "-i", "fake.mp4",
			"-frames:v", "1", "-n", "fake.png",
			], creationflags=creationflags)
			page.wait()
			"""
			).lstrip(),
			encoding="utf-8",
			)
			source = root / "source.stub"
			source.write_bytes(b"read-only source fingerprint")
			evidence = root / "evidence"
			result = subprocess.run(
			[
			shutil.which("powershell"), "-NoProfile", "-File", str(sampler),
			"-PythonPath", sys.executable,
			"-ScriptPath", str(target),
			"-VideoPath", str(source),
			"-OutputPath", str(root / "unused-output"),
			"-EvidenceDirectory", str(evidence),
			],
			capture_output=True,
			text=True,
			encoding="utf-8",
			errors="replace",
			env={
			**os.environ,
			"PYTHONIOENCODING": "utf-8",
			"PYTHONUTF8": "1",
			"MBX_MEDIA_SAMPLER_TEST_CIM_DELAY_MS": "100",
			"MBX_MEDIA_SAMPLER_TEST_FAIL_SAMPLE_ONCE": "1",
			},
			timeout=90,
			)
			self.assertEqual(result.returncode, 3, result.stderr + result.stdout)
			run_summary = json.loads(
			(evidence / "run_summary.json").read_text(encoding="utf-8-sig")
			)
			self.assertEqual(run_summary["target_exit_code"], 0)
			self.assertFalse(run_summary["resource_acceptance_pass"])
			self.assertTrue(
			run_summary["nvdec_command_log_evidence"],
			json.dumps(
			{
			"run_summary": run_summary,
			"stdout": (evidence / "stdout.log").read_text("utf-8"),
			},
			ensure_ascii=False,
			),
			)
			self.assertFalse(run_summary["nvdec_scan_argv_evidence"])
			self.assertGreater(run_summary["command_line_disappearance_race_count"], 0)
			self.assertEqual(run_summary["injected_sample_failure_count"], 1)
			self.assertEqual(
			len(run_summary["sampling_errors"]), 1, run_summary["sampling_errors"]
			)
			self.assertIn("injected post-scan sample failure", run_summary["sampling_errors"][0])
			self.assertTrue(run_summary["source_unchanged"])
			self.assertIn("NVDEC/CUDA (h264_cuvid)", (evidence / "stdout.log").read_text("utf-8"))
			self.assertIn("改用受限 CPU", (evidence / "stderr.log").read_text("utf-8"))

			with (evidence / "resource_samples.csv").open(
			newline="", encoding="utf-8-sig"
			) as handle:
			rows = list(csv.DictReader(handle))
			resource_summary = json.loads(
			(evidence / "resource_summary.json").read_text(encoding="utf-8-sig")
			)
			self.assertFalse(resource_summary["sampling_valid"], resource_summary)
			self.assertEqual(
			resource_summary["sampling_problems"], ["nvdec_scan_argv_invalid"]
			)
			self.assertTrue(resource_summary["all_windows_host_cpu_strictly_below_85"])
			self.assertTrue(resource_summary["window_count"] > 0)
			self.assertFalse(any(
			"previous_timestamp" in problem or "interval_mismatch" in problem
			for problem in resource_summary["sampling_problems"]
			))
			for index, row in enumerate(rows):
			if index == 0:
			self.assertEqual(row["previous_valid_timestamp_utc"], "")
			self.assertAlmostEqual(float(row["interval_seconds"]), 0.0, places=6)
			continue
			current = datetime.fromisoformat(row["timestamp_utc"].replace("Z", "+00:00"))
			previous = datetime.fromisoformat(
			rows[index - 1]["timestamp_utc"].replace("Z", "+00:00")
			)
			reported_previous = datetime.fromisoformat(
			row["previous_valid_timestamp_utc"].replace("Z", "+00:00")
			)
			self.assertLess(abs((reported_previous - previous).total_seconds()), 0.001)
			self.assertLess(
			abs(float(row["interval_seconds"]) - (current - previous).total_seconds()),
			0.001,
			)
			active_indexes = [
			index for index, row in enumerate(rows)
			if row["scan_phase_active"].lower() == "true"
			]
			self.assertGreaterEqual(
			len(active_indexes), 2,
			json.dumps({"rows": rows, "run_summary": run_summary}, ensure_ascii=False),
			)
			self.assertEqual(len({rows[index]["scan_ffmpeg_pid"] for index in active_indexes}), 1)
			last_active = active_indexes[-1]
			self.assertEqual(rows[last_active]["scan_ffmpeg_present"].lower(), "false")
			self.assertEqual(rows[last_active]["scan_ffmpeg_exit_observed"].lower(), "true")
			later_rows = rows[last_active + 1 :]
			self.assertTrue(later_rows)
			self.assertTrue(any("-frames:v" in row["ffmpeg_command_lines"] for row in later_rows))
			self.assertTrue(all(row["scan_phase_active"].lower() == "false" for row in later_rows))

			@unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
			def test_h264_h265_nvdec_and_cpu_product_chains_are_equivalent(self):
			ffmpeg = shutil.which("ffmpeg")
			capabilities = subprocess.check_output(
			[ffmpeg, "-hide_banner", "-encoders"], text=True, errors="replace"
			) + subprocess.check_output(
			[ffmpeg, "-hide_banner", "-decoders"], text=True, errors="replace"
			)
			required = ("libx264", "libx265", "h264_cuvid", "hevc_cuvid")
			missing = [name for name in required if name not in capabilities]
			if missing:
			self.skipTest(f"missing FFmpeg capabilities: {missing}")

			with tempfile.TemporaryDirectory() as raw:
			root = Path(raw)
			frames_dir = root / "frames"
			frames_dir.mkdir()
			size = (320, 180)

			def page(kind, animated=False):
			colors = {"a": (245, 245, 240), "b": (20, 45, 85), "c": (232, 210, 70)}
			image = Image.new("RGB", size, colors[kind])
			draw = ImageDraw.Draw(image)
			draw.rectangle((0, 0, 319, 30), fill=(25, 90, 160))
			draw.rectangle((30, 55, 285, 70), fill=(30, 30, 30))
			if kind != "a" or animated:
			draw.rectangle((30, 95, 250, 110), fill=(35, 35, 35))
			if animated:
			draw.rectangle((260, 140, 280, 160), fill=(210, 30, 30))
			return image

			transition_colors = ((100, 100, 100), (120, 25, 130), (20, 150, 90), (180, 65, 20))
			frames = [page("a")] * 3 + [Image.new("RGB", size, transition_colors[0])]
			frames += [page("a", True)] * 3 + [Image.new("RGB", size, transition_colors[1])]
			frames += [page("b")] * 3 + [Image.new("RGB", size, transition_colors[2])]
			frames += [page("a", True)] * 3 + [Image.new("RGB", size, transition_colors[3])]
			frames += [page("c")] * 3
			for index, image in enumerate(frames):
			image.save(frames_dir / f"frame_{index:03d}.png")

			codec_settings = {
			"h264": ("libx264", "h264_cuvid"),
			"hevc": ("libx265", "hevc_cuvid"),
			}
			evidence = []
			for codec, (encoder, decoder) in codec_settings.items():
			source = root / f"sequence-{codec}.mp4"
			command = [
			ffmpeg, "-hide_banner", "-loglevel", "error", "-framerate", "1",
			"-i", str(frames_dir / "frame_%03d.png"), "-threads", "4",
			"-c:v", encoder, "-preset", "ultrafast",
			]
			if codec == "h264":
			command.extend(("-qp", "0"))
			else:
			command.extend(("-x265-params", "lossless=1:pools=4"))
			command.extend(("-pix_fmt", "yuv420p", "-n", str(source)))
			subprocess.run(command, check=True, timeout=120)
			probe_staging = root / f"probe-{codec}"
			probe_staging.mkdir()
			info = slides.probe_video(source, probe_staging)
			self.assertEqual(info.codec_name, codec)
			self.assertEqual(info.pixel_format, "yuv420p")

			hardware_staging = root / f"scan-{codec}-hardware"
			cpu_staging = root / f"scan-{codec}-cpu"
			hardware_staging.mkdir()
			cpu_staging.mkdir()
			hardware = slides._scan_video_once(
			source, hardware_staging, info, decoder=decoder
			)
			cpu = slides._scan_video_once(source, cpu_staging, info, decoder=None)
			self.assertEqual(hardware.sample_count, cpu.sample_count)
			self.assertEqual(hardware.stable_candidate_count, cpu.stable_candidate_count)
			self.assertEqual(hardware.classification_counts, cpu.classification_counts)
			self.assertEqual(len(hardware.pages), len(cpu.pages))
			hardware_selected = slides.select_main_sequence(hardware.pages)
			cpu_selected = slides.select_main_sequence(cpu.pages)
			self.assertEqual(
			[page.first_seen_order for page in hardware_selected],
			[page.first_seen_order for page in cpu_selected],
			)
			self.assertEqual(len(hardware_selected), len(cpu_selected))
			self.assertGreaterEqual(len(hardware_selected), 3)
			for hardware_page, cpu_page in zip(hardware_selected, cpu_selected):
			self.assertAlmostEqual(
			hardware_page.current_timestamp,
			cpu_page.current_timestamp,
			delta=0.001,
			)

			product_summaries = []
			for label, selected in (("hardware", hardware_selected), ("cpu", cpu_selected)):
			output = root / f"product-{codec}-{label}"
			output.mkdir()
			pngs = slides.extract_original_pages(source, output, info, selected)
			pdf = output / "slides.pdf"
			slides.write_image_pdf(pngs, pdf)
			png_summary = []
			for path in pngs:
			with Image.open(path) as image:
			png_summary.append((path.name, image.mode, image.size))
			product_summaries.append((png_summary, pdf_rgb_pages(pdf)))
			self.assertEqual(product_summaries[0][0], product_summaries[1][0])
			self.assertEqual(
			[page[0] for page in product_summaries[0][1]],
			[page[0] for page in product_summaries[1][1]],
			)
			self.assertEqual(
			len(product_summaries[0][1]), len(product_summaries[1][1])
			)
			hardware_command = slides._scan_command(
			ffmpeg, source, *slides._scan_size(info), decoder
			)
			cpu_command = slides._scan_command(
			ffmpeg, source, *slides._scan_size(info), None
			)
			evidence.extend((
			{
			"codec": codec,
			"path": "nvdec",
			"decoder": decoder,
			"argv": hardware_command,
			"samples": hardware.sample_count,
			"stable_candidates": hardware.stable_candidate_count,
			"classifications": dict(hardware.classification_counts),
			"selected_pages": len(hardware_selected),
			"timestamps": [page.current_timestamp for page in hardware_selected],
			},
			{
			"codec": codec,
			"path": "cpu",
			"decoder": None,
			"argv": cpu_command,
			"samples": cpu.sample_count,
			"stable_candidates": cpu.stable_candidate_count,
			"classifications": dict(cpu.classification_counts),
			"selected_pages": len(cpu_selected),
			"timestamps": [page.current_timestamp for page in cpu_selected],
			},
			))
			print("MEDIA_HOST_SYNTHETIC_EVIDENCE=" + json.dumps(evidence, ensure_ascii=False))

			@unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
			def test_vfr_nonzero_pts_scan_and_exact_seek_share_relative_time(self):
			@@ -281,10 +1222,9 @@
			source = root / "source.mp4"
			source.write_bytes(b"stub")
			output = root / "result"
			page_feature = np.zeros((8, 8), np.uint8)
			logical_page = slides.LogicalPage(page_feature, page_feature, 0, 1.0, 0.2, 0)
			page = logical_page(0)
			with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
			slides, "scan_video", return_value=[logical_page]
			slides, "scan_video", return_value=[page]
			), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
			slides.subprocess, "Popen", side_effect=make_process
			):
			@@ -575,6 +1515,46 @@
			self.assertFalse(output.exists())
			self.assertFalse(any(root.glob(".result.staging-*")))

			def test_main_sequence_filter_runs_once_before_original_extraction(self):
			with tempfile.TemporaryDirectory() as raw:
			root = Path(raw)
			source = root / "source.mp4"
			source.write_bytes(b"stub")
			output = root / "result"
			candidates = [logical_page(index, palette(1)) for index in range(10)]
			selected = candidates[1:9]
			with mock.patch.object(
			slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)
			), mock.patch.object(
			slides, "scan_video", return_value=candidates
			), mock.patch.object(
			slides, "select_main_sequence", return_value=selected
			) as sequence_filter, mock.patch.object(
			slides, "extract_original_pages", return_value=[]
			) as original_extraction, mock.patch.object(slides, "write_image_pdf"):
			self.assertEqual(slides.extract_slides(str(source), str(output)), output)
			sequence_filter.assert_called_once_with(candidates)
			self.assertIs(original_extraction.call_args.args[3], selected)

			def test_selected_pages_are_renumbered_continuously(self):
			with tempfile.TemporaryDirectory() as raw:
			staging = Path(raw)
			selected = [logical_page(index, palette(1)) for index in range(3, 6)]

			def write_png(command, stderr_path, stage):
			destination = Path(command[-1])
			Image.new("RGB", (64, 36), (20, 40, 80)).save(destination)

			with mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
			slides, "_run_file_command", side_effect=write_png
			):
			outputs = slides.extract_original_pages(
			Path("source.mp4"), staging, slides.VideoInfo(64, 36, 10), selected
			)
			self.assertEqual([path.name for path in outputs], [
			"slide_001.png", "slide_002.png", "slide_003.png"
			])

			def test_commit_conflict_is_wrapped_and_staging_is_removed(self):
			with tempfile.TemporaryDirectory() as raw:
			root = Path(raw)
			@@ -582,7 +1562,7 @@
			source.write_bytes(b"stub")
			output = root / "result"
			with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
			slides, "scan_video", return_value=[]
			slides, "scan_video", return_value=[logical_page(0)]
			), mock.patch.object(slides, "extract_original_pages", return_value=[]), mock.patch.object(
			slides, "write_image_pdf"
			), mock.patch.object(slides.os, "rename", side_effect=FileExistsError("race")):