From 5665217eedecb3b1a6075fc35f6e693d073cedfb Mon Sep 17 00:00:00 2001 From: Sagar Kharal Date: Sun, 13 Sep 2026 07:26:03 +0530 Subject: [PATCH 1/3] fix(egocentric): derive episode identity and operator from the source --- examples/egocentric/prepare.py | 64 +++++++++++-- tests/test_egocentric_prepare.py | 152 ++++++++++++++++++++++++++++++- 2 files changed, 207 insertions(+), 9 deletions(-) diff --git a/examples/egocentric/prepare.py b/examples/egocentric/prepare.py index 88e65855..125677dc 100644 --- a/examples/egocentric/prepare.py +++ b/examples/egocentric/prepare.py @@ -49,6 +49,14 @@ class DatasetSource: license: str +@dataclass(frozen=True) +class SourceIdentity: + """The factory and worker that produced one source video, from its sidecar.""" + + factory_id: str + worker_id: str + + @dataclass(frozen=True) class SourceArchive: path: str @@ -143,6 +151,7 @@ def _parse_fault_segment(value: object, context: str) -> tuple[float, float]: def _expand_episode_plan( sources: tuple[SourceVideo, ...], episode_plan: EpisodePlan, + archive_sha256: str, ) -> tuple[PlannedEpisode, ...]: planned_faults_by_episode_number = { planned_fault.episode_number: planned_fault for planned_fault in episode_plan.faults @@ -163,9 +172,18 @@ def _expand_episode_plan( ) planned_fault = planned_faults_by_episode_number.get(episode_number) + # Mirror App's _source_artifact_directory_name: source basenames are not + # identities, so the id carries a digest of the pinned archive, the member + # name, and this episode's window start (one member yields many excerpt + # windows). Two shards or two windows can never share a landing filename, + # and the episode number stays in the id for readability. + source_stem = Path(source_video.member).stem + source_identity_digest = hashlib.sha256( + f"{archive_sha256}:{source_video.member}:{source_start_s}".encode() + ).hexdigest()[:12] episodes.append( PlannedEpisode( - episode_id=f"factory_051_episode_{episode_number:04d}", + episode_id=f"{source_stem}-{episode_number:04d}-{source_identity_digest}", source_member=source_video.member, source_start_s=source_start_s, duration_s=episode_plan.duration_s, @@ -282,7 +300,7 @@ def _load_manifest(manifest_path: Path) -> CorpusManifest: f"fault segment for episode {planned_fault.episode_number} ends after the episode" ) - episodes = _expand_episode_plan(sources, episode_plan) + episodes = _expand_episode_plan(sources, episode_plan, archive.sha256) return CorpusManifest( schema_version=2, @@ -348,13 +366,35 @@ def _extract_source_videos( manifest: CorpusManifest, archive_path: Path, data_root: Path, -) -> dict[str, Path]: +) -> tuple[dict[str, Path], dict[str, SourceIdentity]]: source_root = data_root / "source" source_root.mkdir(parents=True, exist_ok=True) source_paths: dict[str, Path] = {} + identities: dict[str, SourceIdentity] = {} with tarfile.open(archive_path, mode="r") as source_archive: for source_video in manifest.sources: destination_path = source_root / Path(source_video.member).name + sidecar_member = str(Path(source_video.member).with_suffix(".json").as_posix()) + try: + sidecar_stream = source_archive.extractfile(sidecar_member) + except KeyError as error: + raise RuntimeError( + f"missing sidecar {sidecar_member!r} for source video " + f"{source_video.member!r} in the source archive" + ) from error + if sidecar_stream is None: + raise RuntimeError( + f"missing sidecar {sidecar_member!r} for source video " + f"{source_video.member!r} in the source archive" + ) + with sidecar_stream: + sidecar = json.loads(sidecar_stream.read()) + for field in ("factory_id", "worker_id"): + if not isinstance(sidecar.get(field), str) or not sidecar[field]: + raise RuntimeError(f"sidecar {sidecar_member!r} is missing a usable {field!r}") + identities[source_video.member] = SourceIdentity( + factory_id=sidecar["factory_id"], worker_id=sidecar["worker_id"] + ) if destination_path.is_file(): _verify_sha256(destination_path, source_video.sha256) source_paths[source_video.member] = destination_path @@ -373,7 +413,7 @@ def _extract_source_videos( temporary_path.replace(destination_path) _verify_sha256(destination_path, source_video.sha256) source_paths[source_video.member] = destination_path - return source_paths + return source_paths, identities def _fault_frame_range(episode: PlannedEpisode) -> tuple[int, int] | None: @@ -488,10 +528,13 @@ def _transcode_episode_to_h264( return access_units -def _episode_metadata(manifest: CorpusManifest, episode: PlannedEpisode) -> dict[str, str]: +def _episode_metadata( + manifest: CorpusManifest, episode: PlannedEpisode, source_identity: SourceIdentity +) -> dict[str, str]: return { "task": episode.task, - "operator": "factory_051_worker_001", + "factory": source_identity.factory_id, + "operator": f"{source_identity.factory_id}_{source_identity.worker_id}", EPISODE_KEY_ROBOT_SOFTWARE_VERSION: "build-ai-gen-1", "source_dataset": manifest.dataset.repo_id, "source_revision": manifest.dataset.revision, @@ -509,6 +552,7 @@ def _write_video_episode( manifest: CorpusManifest, episode: PlannedEpisode, episode_index: int, + source_identity: SourceIdentity, ) -> None: access_units = _transcode_episode_to_h264(source_video_path, episode) episode_start_time_ns = EPISODE_START_TIME_NS + episode_index * 60_000_000_000 @@ -526,7 +570,10 @@ def _write_video_episode( message_encoding="protobuf", schema_id=schema_id, ) - writer.add_metadata(name=METADATA_RECORD_EPISODE, data=_episode_metadata(manifest, episode)) + writer.add_metadata( + name=METADATA_RECORD_EPISODE, + data=_episode_metadata(manifest, episode, source_identity), + ) writer.add_metadata( name="source-provenance/v1", data={ @@ -599,7 +646,7 @@ def _write_prepared_manifest( def prepare_corpus(manifest_path: Path, source_root: Path, output_root: Path) -> list[Path]: manifest = _load_manifest(manifest_path) archive_path = _ensure_source_archive(manifest, source_root) - source_paths = _extract_source_videos(manifest, archive_path, source_root) + source_paths, identities = _extract_source_videos(manifest, archive_path, source_root) landing_root = output_root / "landing" landing_root.mkdir(parents=True, exist_ok=True) @@ -612,6 +659,7 @@ def prepare_corpus(manifest_path: Path, source_root: Path, output_root: Path) -> manifest, episode, episode_index, + identities[episode.source_member], ) prepared_episode_paths.append(output_path) prepared_count = episode_index + 1 diff --git a/tests/test_egocentric_prepare.py b/tests/test_egocentric_prepare.py index d93d2898..72679153 100644 --- a/tests/test_egocentric_prepare.py +++ b/tests/test_egocentric_prepare.py @@ -1,8 +1,12 @@ """The egocentric converter lands H.264 directly and preserves its planted faults.""" +import hashlib import importlib.util +import io +import json import subprocess import sys +import tarfile from pathlib import Path from types import ModuleType @@ -124,7 +128,10 @@ def test_egocentric_h264_lands_once_and_faults_survive_transform( landing_path = tmp_path / f"{fault}.mcap" canonical_path = tmp_path / f"{fault}.canonical.mcap" - PREPARE._write_video_episode(moving_hevc_video, landing_path, _manifest(episode), episode, 0) + identity = PREPARE.SourceIdentity(factory_id="factory_002", worker_id="worker_001") + PREPARE._write_video_episode( + moving_hevc_video, landing_path, _manifest(episode), episode, 0, identity + ) write_canonical_episode(landing_path, canonical_path, TransformConfig()) landing_schemas, landing_payloads = _video_payloads(landing_path) @@ -145,3 +152,146 @@ def test_egocentric_h264_lands_once_and_faults_survive_transform( assert isinstance(freeze_total_seconds, float) assert freeze_total_seconds >= 2.0 assert decoded_frame_count == 200 + + +def _write_shard_tar( + tar_path: Path, + member_stem: str, + video_source: Path, + factory_id: str, + worker_id: str, +) -> str: + """One pinned shard tar: a single video plus its sidecar. + + Returns the archive sha256 so the manifest can pin it. + """ + video_member = f"{member_stem}.mp4" + sidecar_member = f"{member_stem}.json" + video_bytes = video_source.read_bytes() + sidecar = json.dumps( + { + "factory_id": factory_id, + "worker_id": worker_id, + "video_index": 0, + "duration_sec": 24.0, + "width": 160, + "height": 90, + "fps": 10.0, + "size_bytes": len(video_bytes), + "codec": "h265", + } + ) + with tarfile.open(tar_path, "w") as tar: + video_info = tarfile.TarInfo(video_member) + video_info.size = len(video_bytes) + tar.addfile(video_info, io.BytesIO(video_bytes)) + sidecar_info = tarfile.TarInfo(sidecar_member) + sidecar_info.size = len(sidecar.encode()) + tar.addfile(sidecar_info, io.BytesIO(sidecar.encode())) + return ( + video_member, + hashlib.sha256(video_bytes).hexdigest(), + hashlib.sha256(tar_path.read_bytes()).hexdigest(), + ) + + +def _manifest_json( + archive_path: str, + archive_sha256: str, + member: str, + member_sha256: str, + task: str, +) -> str: + manifest = { + "schema_version": 2, + "dataset": {"repo_id": "e/c", "revision": "abc123", "license": "apache-2.0"}, + "archive": {"path": archive_path, "sha256": archive_sha256}, + "sources": [ + { + "member": member, + "sha256": member_sha256, + "duration_s": 24.0, + "task": task, + } + ], + "episode_plan": { + "total_episodes": 1, + "duration_s": 20.0, + "first_source_start_s": 1.0, + "source_stride_s": 0.0, + "faults": [], + }, + } + return json.dumps(manifest, indent=2) + + +def _episode_provenance(landing_path: Path) -> dict[str, str]: + with Episode(landing_path) as episode: + return episode.metadata_records["episode/v1"] + + +def test_two_shards_coexist_in_one_output_root(tmp_path: Path, moving_hevc_video: Path) -> None: + """#519: two factories' shards into one output root must coexist. With the + old hardcoded ids the second prepare silently overwrote the first.""" + source_root = tmp_path / "source" + output_root = tmp_path / "corpus" + shard_specs = [ + ("factory002_worker001_00000", "factory_002", "worker_001"), + ("factory012_worker003_00000", "factory_012", "worker_003"), + ] + for index, (stem, factory_id, worker_id) in enumerate(shard_specs): + tar_path = source_root / "huggingface" / f"shard{index}.tar" + tar_path.parent.mkdir(parents=True, exist_ok=True) + member, member_sha, archive_sha = _write_shard_tar( + tar_path, stem, moving_hevc_video, factory_id, worker_id + ) + manifest_path = tmp_path / f"manifest-{index}.json" + manifest_path.write_text( + _manifest_json( + f"shard{index}.tar", archive_sha, member, member_sha, f"{factory_id} task" + ), + encoding="utf-8", + ) + PREPARE.prepare_corpus(manifest_path, source_root, output_root) + + landing_paths = sorted((output_root / "landing").glob("*.mcap")) + assert len(landing_paths) == len({p.name for p in landing_paths}) == 2, landing_paths + for landing_path in (output_root / "landing").glob("*.mcap"): + with Episode(landing_path) as episode: + metadata = episode.metadata_records["episode/v1"] + expected_operator = { + "factory002_worker001_00000": "factory_002_worker_001", + "factory012_worker003_00000": "factory_012_worker_003", + }[metadata["source_member"].rsplit(".", 1)[0]] + assert metadata["operator"] == expected_operator, metadata["operator"] + assert ( + metadata["factory"] + == expected_operator.split("_")[0] + "_" + expected_operator.split("_")[1] + ) + + +def test_single_shard_provenance_names_the_real_source( + tmp_path: Path, moving_hevc_video: Path +) -> None: + """One shard, one episode: operator and factory come from the sidecar.""" + source_root = tmp_path / "source" + output_root = tmp_path / "corpus" + tar_path = source_root / "huggingface" / "shard.tar" + tar_path.parent.mkdir(parents=True, exist_ok=True) + member, member_sha, archive_sha = _write_shard_tar( + tar_path, "factory002_worker001_00000", moving_hevc_video, "factory_002", "worker_001" + ) + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text( + _manifest_json("shard.tar", archive_sha, member, member_sha, "factory_002 task"), + encoding="utf-8", + ) + + report = PREPARE.prepare_corpus(manifest_path, source_root, output_root) + + assert len(report) == 1 + with Episode(report[0]) as episode: + metadata = episode.metadata_records["episode/v1"] + assert metadata["operator"] == "factory_002_worker_001" + assert metadata["factory"] == "factory_002" + assert metadata["source_member"] == member From b9fe5c7a0cb8f50b8e1a9c758f764f540836fb8f Mon Sep 17 00:00:00 2001 From: Sagar Kharal Date: Sun, 13 Sep 2026 08:16:07 +0530 Subject: [PATCH 2/3] tests(egocentric): fix shard helper return annotation --- tests/test_egocentric_prepare.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_egocentric_prepare.py b/tests/test_egocentric_prepare.py index 72679153..68984cd8 100644 --- a/tests/test_egocentric_prepare.py +++ b/tests/test_egocentric_prepare.py @@ -160,7 +160,7 @@ def _write_shard_tar( video_source: Path, factory_id: str, worker_id: str, -) -> str: +) -> tuple[str, str, str]: """One pinned shard tar: a single video plus its sidecar. Returns the archive sha256 so the manifest can pin it. From 18656278702084233ebce08aaadbc84d2abf82b3 Mon Sep 17 00:00:00 2001 From: Sagar Kharal Date: Mon, 14 Sep 2026 11:31:37 +0530 Subject: [PATCH 3/3] tests(egocentric): pin the digest collision and sidecar refusal mutations --- tests/test_egocentric_prepare.py | 113 +++++++++++++++++++++++++++++-- 1 file changed, 107 insertions(+), 6 deletions(-) diff --git a/tests/test_egocentric_prepare.py b/tests/test_egocentric_prepare.py index 68984cd8..c74b1972 100644 --- a/tests/test_egocentric_prepare.py +++ b/tests/test_egocentric_prepare.py @@ -4,6 +4,7 @@ import importlib.util import io import json +import re import subprocess import sys import tarfile @@ -154,12 +155,20 @@ def test_egocentric_h264_lands_once_and_faults_survive_transform( assert decoded_frame_count == 200 +def _exactly(message: str) -> str: + """A ``match=`` pattern pinning the whole message, metacharacters and all.""" + return rf"^{re.escape(message)}$" + + def _write_shard_tar( tar_path: Path, member_stem: str, video_source: Path, factory_id: str, worker_id: str, + *, + sidecar_fields: dict[str, object] | None = None, + include_sidecar: bool = True, ) -> tuple[str, str, str]: """One pinned shard tar: a single video plus its sidecar. @@ -168,8 +177,8 @@ def _write_shard_tar( video_member = f"{member_stem}.mp4" sidecar_member = f"{member_stem}.json" video_bytes = video_source.read_bytes() - sidecar = json.dumps( - { + if sidecar_fields is None: + sidecar_fields = { "factory_id": factory_id, "worker_id": worker_id, "video_index": 0, @@ -180,14 +189,15 @@ def _write_shard_tar( "size_bytes": len(video_bytes), "codec": "h265", } - ) + sidecar = json.dumps(sidecar_fields) with tarfile.open(tar_path, "w") as tar: video_info = tarfile.TarInfo(video_member) video_info.size = len(video_bytes) tar.addfile(video_info, io.BytesIO(video_bytes)) - sidecar_info = tarfile.TarInfo(sidecar_member) - sidecar_info.size = len(sidecar.encode()) - tar.addfile(sidecar_info, io.BytesIO(sidecar.encode())) + if include_sidecar: + sidecar_info = tarfile.TarInfo(sidecar_member) + sidecar_info.size = len(sidecar.encode()) + tar.addfile(sidecar_info, io.BytesIO(sidecar.encode())) return ( video_member, hashlib.sha256(video_bytes).hexdigest(), @@ -295,3 +305,94 @@ def test_single_shard_provenance_names_the_real_source( assert metadata["operator"] == "factory_002_worker_001" assert metadata["factory"] == "factory_002" assert metadata["source_member"] == member + + +def test_same_member_stem_from_two_shards_never_collides( + tmp_path: Path, moving_hevc_video: Path +) -> None: + """The digest in the episode id is load-bearing: source basenames are not + identities, so two different shards whose members share one stem must land + as two episodes instead of the second overwriting the first.""" + source_root = tmp_path / "source" + output_root = tmp_path / "corpus" + shared_stem = "factory002_worker001_00000" + shard_factories = ["factory_002", "factory_012"] + for index, factory_id in enumerate(shard_factories): + tar_path = source_root / "huggingface" / f"shard{index}.tar" + tar_path.parent.mkdir(parents=True, exist_ok=True) + member, member_sha, archive_sha = _write_shard_tar( + tar_path, shared_stem, moving_hevc_video, factory_id, "worker_001" + ) + manifest_path = tmp_path / f"manifest-{index}.json" + manifest_path.write_text( + _manifest_json( + f"shard{index}.tar", archive_sha, member, member_sha, f"{factory_id} task" + ), + encoding="utf-8", + ) + PREPARE.prepare_corpus(manifest_path, source_root, output_root) + + landing_paths = sorted((output_root / "landing").glob("*.mcap")) + assert len(landing_paths) == 2, [path.name for path in landing_paths] + assert len({path.name for path in landing_paths}) == len(landing_paths) + observed = { + landing_path.name: _episode_provenance(landing_path)["factory"] + for landing_path in landing_paths + } + assert set(observed.values()) == set(shard_factories), observed + + +@pytest.mark.parametrize( + ("sidecar_case", "expected_message"), + [ + ( + "absent", + "missing sidecar 'factory002_worker001_00000.json' for source video " + "'factory002_worker001_00000.mp4' in the source archive", + ), + ( + "empty_worker_id", + "sidecar 'factory002_worker001_00000.json' is missing a usable 'worker_id'", + ), + ], +) +def test_unusable_sidecar_refuses_the_source( + tmp_path: Path, + moving_hevc_video: Path, + sidecar_case: str, + expected_message: str, +) -> None: + """A missing or malformed sidecar must fail the prepare loudly, naming the + member and the unusable field, instead of stamping false provenance.""" + source_root = tmp_path / "source" + output_root = tmp_path / "corpus" + tar_path = source_root / "huggingface" / "shard.tar" + tar_path.parent.mkdir(parents=True, exist_ok=True) + stem = "factory002_worker001_00000" + if sidecar_case == "absent": + member, member_sha, archive_sha = _write_shard_tar( + tar_path, + stem, + moving_hevc_video, + "factory_002", + "worker_001", + include_sidecar=False, + ) + else: + member, member_sha, archive_sha = _write_shard_tar( + tar_path, + stem, + moving_hevc_video, + "factory_002", + "worker_001", + sidecar_fields={"factory_id": "factory_002", "worker_id": ""}, + ) + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text( + _manifest_json("shard.tar", archive_sha, member, member_sha, "factory_002 task"), + encoding="utf-8", + ) + + with pytest.raises(RuntimeError, match=_exactly(expected_message)): + PREPARE.prepare_corpus(manifest_path, source_root, output_root) + assert list((output_root / "landing").glob("*.mcap")) == []