fix(egocentric): derive episode identity and operator from the source

[Sagar-024]

fix(egocentric): derive episode identity and operator from the source

Closes #519. Found in the multi-factory field run (discussion #503): the example hardcodes every episode id to factory_051 and the operator to factory_051_worker_001, so a second factory's shard silently overwrites the first corpus and the survivors carry provenance from a worker who never recorded them.

What changed

• Episode ids derive from the source member stem plus a digest of the pinned
  archive sha256, the member name, and the episode's window start. This mirrors the collision-resistant pattern App.process uses for run directories (_source_artifact_directory_name, app.py:311), extended with the window start because one source member yields many excerpt windows. Two shards or two windows can never share a landing filename.
• operator is composed from the per-clip sidecar's factory_id and
  worker_id. The sidecar rides in the same pinned tar as the video (identical stem, .json suffix) and is read in the same _extract_source_videos pass: one extract call, no new IO. A missing or unusable sidecar fails loudly, naming the member and the field.
• A factory metadata field records factory_id explicitly on every
  episode.

Three assumptions, redirect me if any are wrong

1. The identity digest seeds from the archive sha256, the member name, and the window start. If the digest should seed from something else (for example a content hash of the video bytes), say so and I will re-cut it. 2. operator keeps the factory_X_worker_Y composition so existing consumers see the same shape, now truthful per source. 3. A new factory key is added to episode/v1. The record already carries example-specific keys (injected_fault, task_completion), and the canonical transform plus checks accept the field (the suite covers the full canonical path with it present).

Invariance proof

The MCAP message payloads are byte-identical before and after the change. The full default-shard corpus (96 episodes) was rebuilt with the new code and compared against the existing landing episode-for-episode, matched by source member and window start: 96 of 96 payload byte-identical. Only ids, operator, the new factory field, and filenames changed.

Honest note: the first implementation of this fix collapsed those 96 windows into 11 files (one per member, the exact silent-overwrite shape this issue describes, reintroduced by my own id scheme). The default-shard rebuild caught it, and the window start moved into the digest seed.

Tests

• Two-shard coexistence: two factories' shards into one output root produce
  two coexisting episodes with distinct derived ids and truthful provenance (the old code produced one, silently destroying the other).
• Single-shard provenance: operator and factory name the real sidecar values.
• The suite also pins the derived-id shape: each landing filename carries the
  source member stem.

Validation: ruff check, ruff format, ty on the changed files; 4 passed in tests/test_egocentric_prepare.py; zero non-ASCII and zero em dashes in the changed files.

[kstonekuan]

The invariance proof and the near-miss note are the right way to report this.

Two mutations pass that should not:

digest dropped from the id entirely      4 passed
sidecar field validation removed         4 passed

Your two shards use different member stems (factory002_worker001_00000, factory012_worker003_00000), so {source_stem}-{episode_number} already separates them and the digest never does the work. Its stated job is that source basenames are not identities. Either cover the case where that bites, or drop it.

The sidecar refusals have no case at all.

Not a gap, for the record: dropping the window start from the seed also passes, and that is correct now, since episode_number is global and disambiguates on its own. Your near-miss is not reachable by that route any more.

[Sagar-024]

Both mutation gaps from your review are now pinned.

Mutation A, digest dropped: the existing shards used different member stems, so {source_stem}-{episode_number} already separated them and the digest never fired. Added test_same_member_stem_from_two_shards_never_collides: two different archives whose members share one stem, prepared into a shared output root. With the digest dropped, both episodes land on the same filename and the second silently overwrites the first (red, one file left). Restored, two distinct episodes coexist with their own provenance (green). The old two-shards test still passes under the mutant, reproducing your finding.

Mutation B, sidecar validation removed: added test_unusable_sidecar_refuses_the_source, parametrized over both refusal branches (sidecar member absent; sidecar present with an empty worker_id). With the validation removed, the empty-field case no longer raises and stamps operator "factory_002_" (red). Restored, both branches refuse loudly naming the member and field (green).

Verified red then green for each. Full suite 1925 passed, 6 skipped, 0 failed. ruff check, ruff format, ty clean on the changed file; 0 non-ASCII, 0 em dashes. Only tests/test_egocentric_prepare.py changed; prepare.py untouched.

[kstonekuan]

LGTM, merging.

Both gaps pinned, and I checked a third: making the sidecar member lookup never raise also fails test_unusable_sidecar_refuses_the_source, so both refusal branches hold, not just the field check.

test_same_member_stem_from_two_shards_never_collides is the one that gives the digest a reason to exist. Two archives sharing a member stem is exactly what "source basenames are not identities" means, and it was the only case that could show it.