Gate the frame-rate limit on sustained cadence, not the largest gap between two frames

scripts/screenshots.py

@@ -152,10 +152,10 @@ def _transform_facts(metadata: object) -> CutFacts:
     if type(frame_count) is not int or frame_count < 1:
         frame_count = max(
             1,
-            int(getattr(metadata, "duration") * getattr(metadata, "peak_rate")),
+            int(getattr(metadata, "duration") * getattr(metadata, "sustained_rate")),
         )
     if not isinstance(frame_rate, Fraction) or frame_rate <= 0:
-        frame_rate = getattr(metadata, "peak_rate")
+        frame_rate = getattr(metadata, "sustained_rate")
     fps = max(1, round(float(frame_rate)))
     return CutFacts(
         cache_key="screenshot",

src/matteloop/core/timeline.py

@@ -212,7 +212,7 @@ def timeline_from_metadata(
 
 
 def _metadata_rate(metadata: object) -> Fraction:
-    for name in ("average_rate", "base_rate", "guessed_rate", "peak_rate"):
+    for name in ("average_rate", "base_rate", "guessed_rate", "sustained_rate"):
         value = getattr(metadata, name, None)
         if isinstance(value, Fraction) and value > 0:
             return value

src/matteloop/jobs/source.py

@@ -78,7 +78,7 @@ class SourceValidationProof:
     average_rate: Fraction | None
     base_rate: Fraction | None
     guessed_rate: Fraction | None
-    peak_rate: Fraction
+    sustained_rate: Fraction
     frame_count: int | None
     rotation: int
     pixel_aspect: Fraction
@@ -115,7 +115,7 @@ def to_payload(self) -> dict[str, object]:
             "average_rate": _optional_fraction_payload(self.average_rate),
             "base_rate": _optional_fraction_payload(self.base_rate),
             "guessed_rate": _optional_fraction_payload(self.guessed_rate),
-            "peak_rate": _fraction_payload(self.peak_rate),
+            "sustained_rate": _fraction_payload(self.sustained_rate),
             "frame_count": self.frame_count,
             "rotation": self.rotation,
             "pixel_aspect": _fraction_payload(self.pixel_aspect),
@@ -166,7 +166,7 @@ def from_payload(cls, payload: Mapping[str, object]) -> SourceValidationProof:
                 average_rate=_optional_fraction_from_payload(payload["average_rate"]),
                 base_rate=_optional_fraction_from_payload(payload["base_rate"]),
                 guessed_rate=_optional_fraction_from_payload(payload["guessed_rate"]),
-                peak_rate=_fraction_from_payload(payload["peak_rate"]),
+                sustained_rate=_fraction_from_payload(payload["sustained_rate"]),
                 frame_count=(
                     None
                     if frame_count_value is None
@@ -204,7 +204,7 @@ class SourceInfo:
     average_rate: Fraction | None
     base_rate: Fraction | None
     guessed_rate: Fraction | None
-    peak_rate: Fraction
+    sustained_rate: Fraction
     frame_count: int | None
     rotation: int
     pixel_aspect: Fraction
@@ -522,8 +522,8 @@ def _validate_proof_identity(
     if (
         proof.duration <= 0
         or proof.duration > MAX_SOURCE_DURATION
-        or proof.peak_rate <= 0
-        or proof.peak_rate > MAX_SOURCE_FPS
+        or proof.sustained_rate <= 0
+        or proof.sustained_rate > MAX_SOURCE_FPS
         or proof.time_base <= 0
         or proof.coded_width <= 0
         or proof.coded_height <= 0
@@ -756,7 +756,7 @@ def _frame_at_timestamp(
 @dataclass(frozen=True, slots=True)
 class _DerivedTimeline:
     duration: Fraction
-    peak_rate: Fraction
+    sustained_rate: Fraction
     frame_count: int
 
 
@@ -767,7 +767,7 @@ def _derive_timeline(
     nominal_rate: Fraction | None,
     is_cancelled: CancelCheck | None = None,
 ) -> _DerivedTimeline:
-    """Boundedly prove duration and maximum cadence from exact decoded PTS."""
+    """Boundedly prove duration and cadence from timestamps on the stream grid."""
     _raise_if_cancelled(is_cancelled)
     time_base = Fraction(stream.time_base)
     offset = (
@@ -839,19 +839,17 @@ def _derive_timeline(
                 "convert-source",
             )
         return _DerivedTimeline(Fraction(1, nominal_rate), nominal_rate, 1)
-    deltas = tuple(
-        later - earlier for earlier, later in zip(timestamps, timestamps[1:])
-    )
-    peak_rate = max(Fraction(1, delta) for delta in deltas)
-    if peak_rate > MAX_SOURCE_FPS:
+    deltas = tuple(b - a for a, b in zip(timestamps, timestamps[1:]))
+    duration = timestamps[-1] + deltas[-1]
+    sustained_rate = Fraction(len(timestamps), duration + 2 * time_base)
+    if sustained_rate > MAX_SOURCE_FPS:
         raise _source_error(
             ErrorCode.SOURCE_FPS_UNSUPPORTED,
             "source.probe.high-frame-rate",
             "decoded source cadence exceeds the 60 fps V1 limit",
             "convert-source-to-60fps",
         )
-    duration = timestamps[-1] + deltas[-1]
-    return _DerivedTimeline(duration, peak_rate, len(timestamps))
+    return _DerivedTimeline(duration, sustained_rate, len(timestamps))
 
 
 def _source_info(
@@ -907,7 +905,7 @@ def _source_info(
         )
         if duration <= 0:
             duration = derived.duration
-        validation_rate = max(validation_rate or Fraction(0), derived.peak_rate)
+        validation_rate = max(validation_rate or Fraction(0), derived.sustained_rate)
     else:
         validation_rate = max(validation_rate or Fraction(0), cfr_rate)
     if duration <= 0:
@@ -964,8 +962,8 @@ def _source_info(
     frames = declared_frame_count if declared_frame_count > 0 else None
     if frames is None and derived is not None:
         frames = derived.frame_count
-    peak_rate = derived.peak_rate if derived is not None else validation_rate
-    if peak_rate is None or peak_rate <= 0:
+    sustained_rate = derived.sustained_rate if derived is not None else validation_rate
+    if sustained_rate is None or sustained_rate <= 0:
         raise _source_error(
             ErrorCode.SOURCE_FPS_UNSUPPORTED,
             "source.probe.unknown-cadence",
@@ -988,7 +986,7 @@ def _source_info(
         average_rate=average_rate,
         base_rate=base_rate,
         guessed_rate=guessed_rate,
-        peak_rate=peak_rate,
+        sustained_rate=sustained_rate,
         frame_count=frames,
         rotation=rotation,
         pixel_aspect=pixel_aspect,
@@ -1014,7 +1012,7 @@ def _source_info(
         average_rate=average_rate,
         base_rate=base_rate,
         guessed_rate=guessed_rate,
-        peak_rate=peak_rate,
+        sustained_rate=sustained_rate,
         frame_count=frames,
         rotation=rotation,
         pixel_aspect=pixel_aspect,

src/matteloop/ui/source_presentation.py

@@ -281,7 +281,7 @@ def present_source_metadata(
     path = getattr(metadata, "path", None)
     full_path = str(path) if path is not None else None
     average_rate = getattr(metadata, "average_rate", None)
-    rate = average_rate or getattr(metadata, "peak_rate", None)
+    rate = average_rate or getattr(metadata, "sustained_rate", None)
     return {
         "source_filename": format_source_filename(path),
         "source_dimensions": format_source_dimensions(

tests/jobs/test_source.py

@@ -31,6 +31,7 @@
     _rotation_from_display_matrix,
     _rotation_from_metadata,
     _source_info,
+    _validate_proof_identity,
     decode_frame,
     probe_source,
 )
@@ -140,7 +141,7 @@ def test_probe_reports_exact_vfr_stream_metadata_and_ignores_fixture_sidecar(
     assert info.duration == Fraction(1, 5)
     assert info.time_base == Fraction(1, 10240)
     assert info.average_rate == Fraction(15)
-    assert info.peak_rate == Fraction(20)
+    assert info.sustained_rate == Fraction(30720, 2562)
     assert info.frame_count == 3
     assert info.pixel_aspect == Fraction(1)
 
@@ -345,25 +346,105 @@ def decode(self, selected_stream):
     derived = _derive_timeline(Container(), stream, Fraction(0), None)
 
     assert derived.duration == Fraction(1, 2)
-    assert derived.peak_rate == Fraction(10)
+    assert derived.sustained_rate == Fraction(30, 7)
     assert derived.frame_count == 3
 
 
-def test_vfr_peak_rate_is_scanned_even_with_average_and_guessed_rates(tmp_path):
+def test_millisecond_quantized_60fps_source_opens_and_proof_verifies(tmp_path):
+    gaps = (17, 17, 16) * 599 + (17, 16)
+    timestamps = [0]
+    for gap in gaps:
+        timestamps.append(timestamps[-1] + gap)
+    frames = [_yuv_frame(81, 90, 240, matrix=1, color_range=1) for _ in timestamps]
+    for frame, pts in zip(frames, timestamps):
+        frame.pts = pts
+        frame.time_base = Fraction(1, 1000)
+    stream = _color_stream(matrix=1)
+    stream.index = 0
+    stream.width = 1920
+    stream.height = 1080
+    stream.time_base = Fraction(1, 1000)
+    stream.start_time = 0
+    stream.duration = 30_000
+    stream.average_rate = Fraction(2_700_000, 45_001)
+    stream.base_rate = Fraction(60)
+    stream.guessed_rate = Fraction(60)
+    stream.frames = len(frames)
+
+    class Container:
+        duration = None
+
+        def seek(self, *args, **kwargs):
+            return None
+
+        def decode(self, selected_stream):
+            assert selected_stream is stream
+            yield from frames
+
+    info = _source_info(
+        tmp_path / "quantized-60fps.mp4",
+        SourceRevision(1, 2, 3, 4, 5),
+        Container(),
+        stream,
+        frames[0],
+    )
+
+    assert info.sustained_rate == Fraction(1_800_000, 30_001)
+    _validate_proof_identity(info.validation_proof, info.revision)
+
+
+def test_short_millisecond_quantized_60fps_source_opens(tmp_path):
+    timestamps = (1, 17, 34, 50)
+    frames = [_yuv_frame(81, 90, 240, matrix=1, color_range=1) for _ in timestamps]
+    for frame, pts in zip(frames, timestamps):
+        frame.pts = pts
+        frame.time_base = Fraction(1, 1000)
+    stream = _color_stream(matrix=1)
+    stream.index = 0
+    stream.width = 1920
+    stream.height = 1080
+    stream.time_base = Fraction(1, 1000)
+    stream.start_time = 0
+    stream.duration = 66
+    stream.average_rate = Fraction(60)
+    stream.base_rate = Fraction(60)
+    stream.guessed_rate = Fraction(60)
+    stream.frames = len(frames)
+
+    class Container:
+        duration = None
+
+        def seek(self, *args, **kwargs):
+            return None
+
+        def decode(self, selected_stream):
+            assert selected_stream is stream
+            yield from frames
+
+    _source_info(
+        tmp_path / "short-quantized-60fps.mp4",
+        SourceRevision(1, 2, 3, 4, 5),
+        Container(),
+        stream,
+        frames[0],
+    )
+
+
+def test_sustained_120fps_source_is_rejected(tmp_path):
     frames = [_yuv_frame(81, 90, 240, matrix=1, color_range=1) for _ in range(3)]
-    for frame, pts in zip(frames, (0, 1, 10)):
+    for frame, pts in zip(frames, range(3)):
         frame.pts = pts
-        frame.time_base = Fraction(1, 100)
+        frame.time_base = Fraction(1, 120)
     stream = _color_stream(matrix=1)
     stream.width = 16
     stream.height = 8
-    stream.time_base = Fraction(1, 100)
+    stream.time_base = Fraction(1, 120)
     stream.start_time = 0
-    stream.duration = 20
+    stream.duration = 3
     stream.average_rate = Fraction(30)
+    stream.base_rate = Fraction(120)
     stream.guessed_rate = Fraction(30)
-    stream.base_rate = Fraction(100)
-    stream.frames = 3
+    stream.frames = len(frames)
 
     class Container:
         duration = None
@@ -377,7 +458,7 @@ def decode(self, selected_stream):
 
     with pytest.raises(AppError) as error:
         _source_info(
-            tmp_path / "vfr.mp4",
+            tmp_path / "120fps.mp4",
             SourceRevision(1, 2, 3, 4, 5),
             Container(),
             stream,
@@ -387,6 +468,50 @@ def decode(self, selected_stream):
     assert error.value.code is ErrorCode.SOURCE_FPS_UNSUPPORTED
 
 
+def test_vfr_timestamp_scan_uses_sustained_cadence_even_with_declared_rates(
+    tmp_path,
+):
+    frames = [_yuv_frame(81, 90, 240, matrix=1, color_range=1) for _ in range(3)]
+    for frame, pts in zip(frames, (0, 1, 10)):
+        frame.pts = pts
+        frame.time_base = Fraction(1, 100)
+    stream = _color_stream(matrix=1)
+    stream.width = 16
+    stream.height = 8
+    stream.time_base = Fraction(1, 100)
+    stream.start_time = 0
+    stream.duration = 20
+    stream.average_rate = Fraction(30)
+    stream.guessed_rate = Fraction(30)
+    stream.base_rate = Fraction(100)
+    stream.frames = 3
+
+    decoded = False
+
+    class Container:
+        duration = None
+
+        def seek(self, *args, **kwargs):
+            return None
+
+        def decode(self, selected_stream):
+            nonlocal decoded
+            assert selected_stream is stream
+            decoded = True
+            yield from frames
+
+    info = _source_info(
+        tmp_path / "vfr.mp4",
+        SourceRevision(1, 2, 3, 4, 5),
+        Container(),
+        stream,
+        frames[0],
+    )
+
+    assert decoded
+    assert info.sustained_rate == Fraction(100, 7)
+
+
 def test_exact_consistent_cfr_metadata_is_the_only_scan_free_path(tmp_path):
     frame = _yuv_frame(81, 90, 240, matrix=1, color_range=1)
     frame.pts = 0
@@ -417,7 +542,7 @@ def decode(self, selected_stream):
     )
 
     assert info.duration == Fraction(3, 2)
-    assert info.peak_rate == Fraction(2)
+    assert info.sustained_rate == Fraction(2)
 
 
 def test_probe_rejects_a_source_narrower_or_shorter_than_the_minimum(tmp_path):

tests/ui/test_source_presentation.py

@@ -80,7 +80,7 @@ def test_present_leaves_missing_source_metadata_values_blank() -> None:
                 height=None,
                 duration=None,
                 average_rate=None,
-                peak_rate=None,
+                sustained_rate=None,
                 file_size=None,
             )
         )