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.