Allow explicitly audited Ego4D bbox tails

docs/ego4d_local_baselines.md

@@ -190,11 +190,11 @@ worker 及其进程组，不杀无关训练。已完成的 `.pt` 保留，重跑
 `clip_end_sec=299.9333333333333`，但 bbox 仍到 `8999`，越界索引为 `[8998, 8999]`。
 因此只兼容一个末尾索引还不够；这第二例同样是标注结束位置与媒体长度一致、bbox 尾部多出记录。
 
-提取器只在以下条件同时满足时忽略末尾记录：所有越界索引都属于 `N、N+1`，
+提取器默认只在以下条件同时满足时忽略末尾记录：所有越界索引都属于 `N、N+1`，
 `N=视频帧数=clip_end_frame`，标注帧起点为 0、时间起点在 1 微秒内为 0，
 且标注时长与 `N / source_fps` 相差不超过 0.1 毫秒。
 容差限制的是**末尾索引位置**，不是框数：即使仅一条记录，出现在 `N+2` 或更远处也不接受。
-负索引、缺少确认边界、较大越界或时长不匹配仍会报错，并打印具体范围；不自动跳过 clip。
+负索引、缺少确认边界、超过显式容差的越界或时长不匹配仍会报错，并打印具体范围；不自动跳过 clip。
 这是基于已核实样例的受限兼容规则，不代表所有 Ego4D 越界记录都可忽略。
 
 日志显示忽略的末尾索引和框数；新 `.pt` 的 `box_frame_audit` 与预览目录的
@@ -203,6 +203,58 @@ worker 及其进程组，不杀无关训练。已完成的 `.pt` 保留，重跑
 `0.2..299.8s`，2 FPS 的最后一个输入采样时刻仍为 `299.5s`。
 该处理不改编码器/池化配置，之前完成且通过校验的特征可以继续复用；同步修复后重跑原 8 卡命令即可。
 
+#### 更长但边界一致的尾部：显式授权，不修改默认值
+
+用户随后报告 `d2cb4045-8ef1-4288-8398-ab6d4d6a0709`：视频为 8987 帧、30 FPS，
+`clip_end_frame=8987`、`clip_end_sec=299.5666666666666`，但 bbox 延伸到 8999。
+两种长度信息一致，越界的 8987–8999 共 13 个索引都没有对应的视频帧。
+这解释了默认两帧规则为什么失败；不是 GPU OOM，也不意味着应平移全部框。
+本地没有该远程视频，本次用用户提供的边界建立 CPU 回归 fixture，未重新解码实际文件。
+
+新增参数 `--ego4d-max-terminal-overrun-frames K`，只授权 `[N, N+K)` 内的尾部索引：
+
+- 默认 `K=2` 不变；`K=0` 禁用所有尾部兼容；负值被拒绝。
+- 仍必须满足零起点、`N=source_frames=clip_end_frame` 和同样的时长一致性校验。
+- 上限按最远索引计算，不按框数或越界索引数量计算；本例 `K=13` 不接受索引 9000 或更远处。
+- 原始视频、标注 JSON 和 target 不改动；仅从本次内存 bbox 映射排除无对应视频帧的尾部。
+- 媒体仍需覆盖原始 target/全局特征要求的时长。参数不能修复截断的视频、补帧、缩短 target 或绕过全局/局部时间线校验。
+- 参数及实际尾部长度保存在每个新文件的 `box_frame_audit`，不改变编码器配置/旧特征复用条件。
+
+同步代码后，使用上面已经配置好的 `MULTIGPU_ARGS` 续跑：
+
+```bash
+python extract_ego4d_local_multigpu.py "${MULTIGPU_ARGS[@]}" \
+  --ego4d-max-terminal-overrun-frames 13
+```
+
+若在新终端中运行，先重新定义上一节的完整 `MULTIGPU_ARGS` 数组。无需软链接，不使用 `--overwrite`。
+当前失败会停止本次启动的其他 workers；已原子保存且复用检查通过的 `.pt` 不会重提，
+中途被停止、尚未保存的 clip 会重新计算。不要对更大的越界直接设无限容差，应先核对新日志。
+
+新生成的该 clip 可以检查审计和网格长度：
+
+```bash
+python - <<'PY'
+from pathlib import Path
+import numpy as np
+import torch
+
+root = Path('/data1/yfzx/wangx/Datasets/ego4d')
+clip_id = 'd2cb4045-8ef1-4288-8398-ab6d4d6a0709'
+p = torch.load(root / 'omni_features/local_qwen_1s' / f'{clip_id}.pt',
+               map_location='cpu', weights_only=False)
+target = np.load(root / 'ego4d_target_frame' / f'{clip_id}.npy', mmap_mode='r')
+assert len(p['visual']) == len(p['audio']) == len(p['times']) == len(target)
+assert not p['partial']
+audit = p['box_frame_audit']
+assert audit['source_frames'] == audit['declared_bounds']['clip_end_frame'] == 8987
+assert audit['ignored_frame_indices'] == list(range(8987, 9000))
+print('visual:', tuple(p['visual'].shape), 'audio:', tuple(p['audio'].shape))
+print('target:', target.shape, 'last cutoff:', float(p['times'][-1]))
+print('bbox audit:', audit)
+PY
+```
+
 提取完成后，原训练入口也直接用同一个原始全局特征根，无需 flat 索引：
 
 ```bash

extract_turn_local_features.py

@@ -27,8 +27,9 @@ from local_turn_features import (
 
 
 TESTED_TRANSFORMERS_VERSIONS = ("4.52.3", "4.57.6")
-# Verified Ego4D exports can retain one or two bbox frame indices after the
+# Conservative default for previously verified exports. Longer, audited tails
-# media endpoint. This bounds frame positions, not the number of people/boxes.
+# require an explicit CLI limit; endpoint/duration checks remain mandatory.
+# This bounds frame positions, not the number of people/boxes.
 EGO4D_MAX_TERMINAL_FRAME_OVERRUN = 2
 
 
@@ -193,14 +194,23 @@ def load_ego4d_boxes(paths: list[Path], names: list[str]) -> dict[str, Any]:
     return result
 
 
-def validate_ego4d_box_frames(name: str, annotation: dict[str, Any], source):
+def validate_ego4d_box_frames(
+    name: str, annotation: dict[str, Any], source, *,
+    max_terminal_frame_overrun: int = EGO4D_MAX_TERMINAL_FRAME_OVERRUN,
+):
     """Ignore a bounded, boundary-confirmed tail; never shift/clamp boxes."""
+    if (isinstance(max_terminal_frame_overrun, bool)
+            or not isinstance(max_terminal_frame_overrun, int)
+            or max_terminal_frame_overrun < 0):
+        raise ValueError("max_terminal_frame_overrun must be a nonnegative integer")
     frames, bounds = annotation["frames"], annotation["bounds"]
     count = source.num_frames
     outside = sorted(frame for frame in frames if not 0 <= frame < count)
+    terminal_overrun = max(0, max(frames, default=-1) - count + 1)
     audit = {
-        "policy": "clip-local-zero-based; ignore-confirmed-tail-up-to-2-frames",
+        "policy": f"clip-local-zero-based; ignore-confirmed-tail-up-to-{max_terminal_frame_overrun}-frames",
-        "max_terminal_frame_overrun": EGO4D_MAX_TERMINAL_FRAME_OVERRUN,
+        "max_terminal_frame_overrun": max_terminal_frame_overrun,
+        "annotation_terminal_overrun_frames": terminal_overrun,
         "source_frames": count, "source_fps": source.source_fps,
         "declared_bounds": dict(bounds),
         "annotation_frame_range": [min(frames), max(frames)] if frames else None,
@@ -215,9 +225,10 @@ def validate_ego4d_box_frames(name: str, annotation: dict[str, Any], source):
         and source.source_fps > 0 and abs(start_sec) <= 1e-6
         and abs((end_sec - start_sec) - count / source.source_fps) <= 1e-4
     )
-    # Allow only N and N+1. Counting outliers alone would accidentally permit
+    # The default allows N and N+1. An explicit larger limit authorizes only
-    # sparse but arbitrarily distant boxes; the maximum index must be bounded.
+    # that bounded tail, never arbitrary outliers. Counting outliers alone
-    small_terminal_tail = count <= outside[0] <= outside[-1] < count + EGO4D_MAX_TERMINAL_FRAME_OVERRUN
+    # would accidentally permit sparse but arbitrarily distant boxes.
+    small_terminal_tail = count <= outside[0] <= outside[-1] < count + max_terminal_frame_overrun
     if (not small_terminal_tail or bounds.get("clip_start_frame") != 0
             or bounds.get("clip_end_frame") != count or not duration_matches):
         raise ValueError(
@@ -225,10 +236,13 @@ def validate_ego4d_box_frames(name: str, annotation: dict[str, Any], source):
             f"source_frames={count}, source_fps={source.source_fps}, "
             f"annotation_range={audit['annotation_frame_range']}, "
             f"out_of_range_count={len(outside)}, first_10={outside[:10]}, "
-            f"declared_bounds={bounds}. Only terminal indices in "
+            f"declared_bounds={bounds}; terminal_overrun_frames={terminal_overrun}. "
-            f"[{count}, {count + EGO4D_MAX_TERMINAL_FRAME_OVERRUN - 1}] "
+            f"Current tail limit is {max_terminal_frame_overrun} frame(s): require "
-            "with source_frames == clip_end_frame, zero origin and matching duration are allowed; "
+            f"0 <= frame - {count} < {max_terminal_frame_overrun}, "
-            "check annotation/video alignment instead of clamping or skipping the clip."
+            "source_frames == clip_end_frame, zero origin and matching duration. "
+            "A larger audited tail requires explicit --ego4d-max-terminal-overrun-frames; "
+            "this never bypasses endpoint/duration checks. Check annotation/video alignment "
+            "instead of clamping or skipping the clip."
         )
     ignored_boxes = sum(len(frames[frame]) for frame in outside)
     audit.update(ignored_frame_indices=outside, ignored_box_count=ignored_boxes)
@@ -264,7 +278,10 @@ def build_media_and_boxes(args: argparse.Namespace, name: str, end: float, ego_b
                 raise ValueError(f"Box/video frame inventory mismatch: {path}")
             annotations.append(mapping)
     else:
-        mapping, audit = validate_ego4d_box_frames(name, ego_boxes[name], media.clips[0])
+        mapping, audit = validate_ego4d_box_frames(
+            name, ego_boxes[name], media.clips[0],
+            max_terminal_frame_overrun=args.ego4d_max_terminal_overrun_frames,
+        )
         annotations = [mapping]
         ego_boxes[name]["frame_audit"] = audit
     return media, annotations
@@ -517,6 +534,12 @@ def parse_args(argv=None):
     p.add_argument("--num-shards", type=int, default=1, help="Disjoint round-robin shards after all sample selection")
     p.add_argument("--shard-index", type=int, default=0, help="Zero-based shard index")
     p.add_argument("--ego4d-annotations", type=Path, nargs="+")
+    p.add_argument(
+        "--ego4d-max-terminal-overrun-frames", type=int,
+        default=EGO4D_MAX_TERMINAL_FRAME_OVERRUN,
+        help=("Maximum audited bbox tail beyond the decoded Ego4D endpoint (default: 2; 0: reject all). "
+              "Requires matching declared frame/time bounds; never shifts boxes or removes target rows."),
+    )
     p.add_argument("--model-id", default="Qwen/Qwen2.5-Omni-3B")
     p.add_argument("--encoder-id", help="Stable checkpoint identity for relocated copies; default: model-id")
     p.add_argument("--device", default="cuda:0")
@@ -551,6 +574,11 @@ def parse_args(argv=None):
     p.add_argument("--dry-run", action="store_true", help="Validate references/targets and list work; no model, media decode or writes")
     p.add_argument("--no-progress", action="store_true")
     args = p.parse_args(argv)
+    if args.ego4d_max_terminal_overrun_frames < 0:
+        p.error("--ego4d-max-terminal-overrun-frames must be nonnegative")
+    if (args.dataset != "ego4d"
+            and args.ego4d_max_terminal_overrun_frames != EGO4D_MAX_TERMINAL_FRAME_OVERRUN):
+        p.error("--ego4d-max-terminal-overrun-frames is only supported with --dataset ego4d")
     if args.num_shards <= 0 or not 0 <= args.shard_index < args.num_shards:
         p.error("Require --num-shards > 0 and 0 <= --shard-index < --num-shards")
     if args.manifest is not None and args.dataset != "ego4d":

test_ego4d_box_frames.py

@@ -1,4 +1,4 @@
-"""CPU regressions for verified one/two-frame Ego4D terminal bbox overruns."""
+"""CPU regressions for bounded, explicitly authorized Ego4D bbox tails."""
 
 from __future__ import annotations
 
@@ -24,15 +24,19 @@ from test_local_turn_features import fixture
 
 UID = "31488ff6-05ef-4362-9786-35981b06f307"
 TWO_FRAME_UID = "ac7d3d16-d74d-4fe3-a1e0-cab0b1adf492"
+THIRTEEN_FRAME_UID = "d2cb4045-8ef1-4288-8398-ab6d4d6a0709"
 
 
 def terminal_clip(*, uid=UID, count=8999, tail_frames=1):
+    num_steps = count // 6  # 30 FPS source, 5 Hz targets.
+    num_video_frames = (num_steps * 2 + 4) // 5  # ceil(duration * 2 FPS).
+    last_grid_frame = (num_video_frames - 1) * 15
     return {
         "clip_uid": uid, "clip_start_frame": 0, "clip_end_frame": count,
         "clip_start_sec": 0, "clip_end_sec": count / 30.,
         "persons": [{"person_id": "p1", "tracking_paths": [{"track": [
             {"frame": frame, "clip_frame": None, "x": 0, "y": 0, "width": 2, "height": 2}
-            for frame in (0, 8985, count - 1, *range(count, count + tail_frames))
+            for frame in (0, last_grid_frame, count - 1, *range(count, count + tail_frames))
         ]}]}],
     }
 
@@ -123,13 +127,70 @@ class Ego4DBoxFrameTests(unittest.TestCase):
                 extract.validate_ego4d_box_frames(
                     UID, terminal_annotation(), SimpleNamespace(num_frames=8999, source_fps=fps))
 
+    def test_thirteen_frame_tail_requires_explicit_limit_and_keeps_valid_boxes(self):
+        annotation = terminal_annotation(uid=THIRTEEN_FRAME_UID, count=8987, tail_frames=13)
+        original = copy.deepcopy(annotation)
+        source = SimpleNamespace(num_frames=8987, source_fps=30.)
+        with self.assertRaisesRegex(ValueError, "terminal_overrun_frames=13"):
+            extract.validate_ego4d_box_frames(THIRTEEN_FRAME_UID, annotation, source)
+        with contextlib.redirect_stdout(io.StringIO()):
+            frames, audit = extract.validate_ego4d_box_frames(
+                THIRTEEN_FRAME_UID, annotation, source, max_terminal_frame_overrun=13)
+        self.assertEqual(annotation, original)
+        self.assertEqual(frames, {frame: boxes for frame, boxes in original["frames"].items() if frame < 8987})
+        self.assertIn(8986, frames)  # Never discard/replace the last valid frame.
+        self.assertEqual(audit["ignored_frame_indices"], list(range(8987, 9000)))
+        self.assertEqual(audit["ignored_box_count"], 13)
+        self.assertEqual(audit["max_terminal_frame_overrun"], 13)
+        self.assertEqual(audit["annotation_terminal_overrun_frames"], 13)
+
+    def test_explicit_limit_does_not_bypass_endpoint_duration_or_far_outliers(self):
+        source = SimpleNamespace(num_frames=8987, source_fps=30.)
+        for key, value in (("clip_start_frame", 1), ("clip_end_frame", 9000),
+                           ("clip_start_sec", .1), ("clip_end_sec", 300.),
+                           ("clip_end_sec", float("nan"))):
+            annotation = terminal_annotation(count=8987, tail_frames=13)
+            annotation["bounds"][key] = value
+            with self.subTest(key=key, value=value), self.assertRaises(ValueError):
+                extract.validate_ego4d_box_frames(
+                    THIRTEEN_FRAME_UID, annotation, source, max_terminal_frame_overrun=13)
+        for frame in (-1, 9000, 12000):
+            annotation = terminal_annotation(count=8987, tail_frames=0)
+            annotation["frames"][frame] = []
+            with self.subTest(frame=frame), self.assertRaises(ValueError):
+                extract.validate_ego4d_box_frames(
+                    THIRTEEN_FRAME_UID, annotation, source, max_terminal_frame_overrun=13)
+        with self.assertRaises(ValueError):
+            extract.validate_ego4d_box_frames(
+                THIRTEEN_FRAME_UID, terminal_annotation(count=8987, tail_frames=13),
+                SimpleNamespace(num_frames=8986, source_fps=30.), max_terminal_frame_overrun=13)
+
+    def test_zero_limit_disables_tail_compatibility_and_invalid_limits_fail(self):
+        with self.assertRaisesRegex(ValueError, "Current tail limit is 0"):
+            extract.validate_ego4d_box_frames(
+                UID, terminal_annotation(), self.source, max_terminal_frame_overrun=0)
+        annotation = terminal_annotation(tail_frames=0)
+        frames, audit = extract.validate_ego4d_box_frames(
+            UID, annotation, self.source, max_terminal_frame_overrun=0)
+        self.assertIs(frames, annotation["frames"])
+        self.assertEqual(audit["ignored_frame_indices"], [])
+        for limit in (-1, 1.5, True):
+            with self.subTest(limit=limit), self.assertRaises(ValueError):
+                extract.validate_ego4d_box_frames(
+                    UID, annotation, self.source, max_terminal_frame_overrun=limit)
+
     def test_full_extraction_keeps_1499_rows_saves_audit_and_reuses(self):
         self._assert_full_extraction_and_reuse(UID, 8999, 1)
 
     def test_two_frame_tail_extraction_matches_manual_removal_of_unsampled_boxes(self):
         self._assert_full_extraction_and_reuse(TWO_FRAME_UID, 8998, 2)
 
-    def _assert_full_extraction_and_reuse(self, uid, count, tail_frames):
+    def test_thirteen_frame_tail_extraction_keeps_target_grid_and_reuses(self):
+        self._assert_full_extraction_and_reuse(THIRTEEN_FRAME_UID, 8987, 13, explicit_limit=13)
+
+    def _assert_full_extraction_and_reuse(self, uid, count, tail_frames, explicit_limit=None):
+        num_steps = count // 6
+        last_video_time = ((num_steps * 2 + 4) // 5 - 1) / 2
         image = Image.new("RGB", (2, 2))
         source = SimpleNamespace(num_frames=count, source_fps=30., crop=None, dense_start=0.,
                                  path=Path(f"{uid}.mp4"), width=2, height=2)
@@ -150,7 +211,7 @@ class Ego4DBoxFrameTests(unittest.TestCase):
             root = Path(temporary)
             for folder in ("global", "targets", "v1/clips", "v1/annotations"):
                 (root / folder).mkdir(parents=True)
-            reference, _, targets = fixture(n=1499, name=uid, dataset="ego4d")
+            reference, _, targets = fixture(n=num_steps, name=uid, dataset="ego4d")
             torch.save(reference, root / "global" / f"{uid}.pt")
             np.save(root / "targets" / f"{uid}.npy", targets)
             (root / "v1/clips" / f"{uid}.mp4").write_bytes(b"fake-video")
@@ -162,6 +223,10 @@ class Ego4DBoxFrameTests(unittest.TestCase):
             argv = ["--dataset", "ego4d", "--data-root", str(root), "--feature-root", str(root / "global"),
                     "--target-root", str(root / "targets"), "--output-root", str(root / "local"),
                     "--samples", uid, "--no-progress"]
+            if explicit_limit is not None:
+                before = extract.requested_local_config(extract.parse_args(argv))
+                argv += ["--ego4d-max-terminal-overrun-frames", str(explicit_limit)]
+                self.assertEqual(extract.requested_local_config(extract.parse_args(argv)), before)
             with patch.dict(sys.modules, {"transformers": SimpleNamespace(__version__="4.57.6"),
                                           "streaming_turn_model_directly": fake_module}), \
                  patch.object(extract, "FrozenQwenLocalEncoder", return_value=encoder) as model, \
@@ -170,16 +235,17 @@ class Ego4DBoxFrameTests(unittest.TestCase):
                 model.assert_called_once()
                 destination = root / "local" / f"{uid}.pt"
                 result = torch.load(destination, weights_only=False)
-                self.assertEqual(result["visual"].shape, (1499, 6))
+                self.assertEqual(result["visual"].shape, (num_steps, 6))
-                self.assertEqual(result["audio"].shape, (1499, 5))
+                self.assertEqual(result["audio"].shape, (num_steps, 5))
                 self.assertTrue(torch.equal(result["times"], reference["times"]))
                 self.assertTrue(torch.equal(result["label_indices"], reference["label_indices"]))
-                self.assertAlmostEqual(result["times"][-1].item(), 299.8)
+                self.assertAlmostEqual(result["times"][-1].item(), num_steps / 5)
+                self.assertLessEqual(num_steps / 5, count / 30)
                 self.assertFalse(result["partial"])
                 ignored = list(range(count, count + tail_frames))
                 self.assertEqual(result["box_frame_audit"]["ignored_frame_indices"], ignored)
-                self.assertEqual(result["persons"]["frame_times"][-1].item(), 299.5)
+                self.assertEqual(result["persons"]["frame_times"][-1].item(), last_video_time)
-                self.assertEqual(result["persons"]["records"][-1]["source_frame_index"], 8985)
+                self.assertEqual(result["persons"]["records"][-1]["source_frame_index"], int(last_video_time * 30))
                 load_local_features(destination, reference, sample_id=uid, dataset="ego4d", modalities="av")
                 # Previously completed valid sidecars have no audit. Their
                 # vectors/config remain reusable: only unsampled boxes changed.

test_ego4d_local_multigpu.py

@@ -64,6 +64,16 @@ class MultiGPUExtractionTests(unittest.TestCase):
         command = plans[0]["command"]
         self.assertEqual(command[command.index("--num-shards") + 1], "1")
 
+    def test_explicit_bbox_tail_limit_is_forwarded_to_all_workers(self):
+        option = "--ego4d-max-terminal-overrun-frames"
+        args, extraction, forwarded = launch.parse_args(self.argv + [option, "13"])
+        self.assertEqual(extraction.ego4d_max_terminal_overrun_frames, 13)
+        for plan in launch.plan_workers(args, extraction, forwarded):
+            command = plan["command"]
+            self.assertEqual(command[command.index(option) + 1], "13")
+        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
+            launch.parse_args(self.argv + [option, "-1"])
+
     def test_rejects_unsafe_or_ambiguous_launcher_options(self):
         for extra in (["--gpus", "0", "0"], ["--gpus", "-1"], ["--cpu-threads", "0"],
                       ["--preview-only"], ["--max-steps", "20"], ["--overwrite"],