fix(catalog): preserve recurring outcomes as latest

docs/CATALOG.md

@@ -27,8 +27,8 @@ is the point. Either way, `report.catalog_entry` tells you what happened:
 ```python
 entry = report.catalog_entry
 entry.episode_id  # content address of the canonical file
-entry.run_fingerprint  # content hash of versions + the observable run outcome
-entry.written  # False when this exact run was already recorded
+entry.run_fingerprint  # outcome hash, with an occurrence suffix when needed
+entry.written  # False when this occurrence was already recorded
 ```
 
 For a targeted backfill, `step_names=` records only the selected registered
@@ -83,15 +83,17 @@ the ground truth for any row, which is exactly why they are kept verbatim
 next to the heuristic. Only rows written by an ingest binary that already
 raises `SourceNotConforming` (see below) can carry `source-unsupported`.
 
-Three durability rules govern writes:
+These durability rules govern writes:
 
 - **Content-addressed**: `episode_id` is a sha256 of the canonical file's
   bytes. Re-ingesting an unchanged episode dedupes; a reprocessed episode
   (new `pipeline_version`) is a distinct fact.
 - **Create-if-absent**: an append whose `(episode_id, run_fingerprint)` file
-  already exists is a no-op (`written=False`). The fingerprint includes the
-  observable outcome: exact retries deduplicate, while a successful retry
-  after an error appends the repaired result.
+  already exists is a no-op (`written=False`). Consecutive identical outcomes
+  deduplicate. If an outcome recurs after a different append for the same
+  episode or source, it gets a new occurrence fingerprint and timestamp:
+  error → success → identical error ends with `status = 'unverified'`, and
+  success → error → identical success ends with `status = 'ok'`.
 - **Append, never overwrite**: when a check's results are no longer comparable,
   bump its explicit version. Re-running then adds rows under the new
   `check_version` next to the old ones. The corpus is assumed permanently
@@ -100,7 +102,7 @@ Three durability rules govern writes:
   orchestrated run recorded the row (the generated Airflow DAGs pass their
   stage sub-DAG's own run id; a local run records NULL). It is deliberately
   outside `run_fingerprint`, or a rerun producing the same outcome would stop
-  deduplicating. So it names the run that FIRST recorded an outcome, and since
+  deduplicating. So it names the run that FIRST recorded an occurrence, and since
   filters read the latest row per episode, selecting on it answers "whose work
   is the current answer" rather than "which runs ever touched this".
 
@@ -112,6 +114,47 @@ crashed earlier attempt left with a stale timestamp -- so concurrent
 duplicate appends and retried tasks converge instead of stitching two runs'
 rows together.
 
+### Occurrences and delayed retries
+
+The outcome hash still depends only on content and versions. First occurrences
+keep their existing filenames. A recurring outcome's `run_fingerprint` is
+`<outcome-hash>.<scope-hash>`, where the scope hashes the preceding append's
+episode identity and fingerprint. It contains no random value and does not
+change canonical bytes, `episode_id`, `pipeline_version`, or check versions.
+The preceding append is selected using the same timestamp/fingerprint ordering
+as curation. Selective-stage runs participate in this history too, so a full
+run cannot replay an old success over a later metadata-only failure. Checks
+omitted from a selective run retain their own latest recorded results.
+
+Without an explicit execution identity, identical current content is treated
+as a retry; identical historical content is treated as a new occurrence. Content
+alone cannot tell a delayed retry from a fresh execution. Callers that need
+retries to remain idempotent across intervening executions must persist an
+`execution_id` **before the first attempt** and reuse it:
+
+```python
+report = app.process("episode_0001.mcap", execution_id="durable-task-attempt-42")
+```
+
+`Catalog.append_episode()` accepts the same argument. With an explicit id, the
+scope hashes that id instead of the predecessor. The same id and outcome always
+address the original append, even after another execution commits or a writer
+crashes before returning. A new execution must get a new id. If a retry actually
+produces a different outcome, that different content still gets its own append.
+Do not reuse an id across fresh executions that may return to an older outcome.
+The scheduler's provenance-only `orchestrator_run_id` is not used as this token.
+
+This is a change to append identity semantics, with no Parquet schema change:
+catalog format version 1 remains readable, existing first-occurrence hashes
+remain valid, and historical timestamps are never advanced. Upgrade all writers
+to obtain these recurrence semantics; old writers still deduplicate globally.
+Already suppressed executions cannot be recovered from the catalog; rerun the
+checks to record current state. Implicit appends now scan episode history
+(including syncing the bucket mirror); explicit ids avoid this scan. This is
+not a globally serialized execution log: concurrent different outcomes retain
+the existing `recorded_at`/fingerprint ordering and require synchronized clocks
+for meaningful cross-worker time ordering.
+
 ## Querying
 
 ### Explore in DuckDB UI

src/hflow/app.py

@@ -2249,6 +2249,7 @@ def process(
         step_names: Iterable[str] | None = None,
         quarantine_history: QuarantineHistory | None = None,
         orchestrator_run_id: str | None = None,
+        execution_id: str | None = None,
         _registered_step_selection: RegisteredStepSelection | None = None,
         _prepared_process_configuration: _PreparedProcessConfiguration | None = None,
     ) -> ProcessReport:
@@ -2302,6 +2303,12 @@ def process(
         equivalent. The dev loop passes nothing and records NULL. Provenance
         only, never part of any identity hash (see
         :meth:`hflow.catalog.Catalog.append_episode`).
+
+        ``execution_id`` optionally identifies a catalog append across delayed
+        retries, independently of intervening runs. Persist it before the first
+        attempt and reuse it only for retries of that execution. Without it,
+        consecutive identical outcomes deduplicate and a recurring outcome
+        after another append becomes current again.
         """
         if _prepared_process_configuration is not None:
             if (
@@ -2788,6 +2795,7 @@ def render_contact_sheets(media_episode: Episode) -> EnrichmentResult:
                 quarantine_tags=report.quarantine_tags,
                 source_uri=source_identifier,
                 orchestrator_run_id=orchestrator_run_id,
+                execution_id=execution_id,
                 time_bounds=episode_time_bounds,
             )
 

src/hflow/catalog.py

@@ -14,7 +14,8 @@
 - **Create-if-absent appends**: each append writes one Parquet file per table
   named ``<episode_id>-<run_fingerprint>.parquet``; if the file already
   exists the append is a no-op. The fingerprint includes the observable
-  outcome, so an exact replay deduplicates while a repaired retry appends.
+  outcome and, when it recurs, its predecessor. Consecutive identical
+  outcomes deduplicate; returning to an older outcome appends a new fact.
 - **Append, never overwrite**: re-running a changed check (new
   ``check_version``) adds new-version rows; curation picks or pins versions.
 
@@ -111,8 +112,8 @@
 # Appends this process already verified (or repaired) as recorded_at-aligned,
 # keyed by (location, file_stem). Once aligned a stem can never go stale
 # again -- the episodes file is immutable and every post-commit dependent
-# write carries its recorded_at -- so replays skip straight back to the
-# single existence check instead of re-reading every dependent file each time.
+# write carries its recorded_at -- so replays need not re-read every
+# dependent file after resolving the occurrence's identity.
 _reconciled_append_stems: set[tuple[str, str]] = set()
 
 _FORMAT_MARKER_NAME = "format_version"
@@ -603,12 +604,12 @@ def _run_fingerprint(
     check_rows: Sequence[CheckRunRow],
     quarantine_tags: Sequence[str],
 ) -> str:
-    """Identify one observable run outcome while keeping exact retries idempotent.
+    """Identify observable content, independently of when it occurred.
 
     Step versions identify intended behavior, not whether a particular
     attempt timed out or what it measured. Outcome data belongs in the append
-    identity so a successful retry after a transient error is preserved,
-    while replaying the exact same result remains a no-op.
+    identity so a successful retry after a transient error is preserved.
+    Catalog._append_fingerprint scopes this content to an occurrence.
     """
     check_outcomes: list[dict[str, object]] = []
     for row in check_rows:
@@ -818,9 +819,22 @@ def append_episode(
         source_uri: str | None = None,
         uri: str | None = None,
         orchestrator_run_id: str | None = None,
+        execution_id: str | None = None,
         time_bounds: EpisodeTimeBounds | None = None,
     ) -> AppendResult:
-        """Record one outcome; replaying that exact outcome is idempotent.
+        """Record an outcome; consecutive identical outcomes are idempotent.
+
+        Without ``execution_id``, the latest append for this episode or source
+        defines the retry boundary. An older outcome recurring after another
+        append gets a new, deterministic fingerprint scoped to that predecessor.
+        Selective-stage appends participate in the same history.
+
+        For retries that may arrive after intervening executions, persist and
+        reuse an explicit ``execution_id`` before the first append attempt.
+        Together with the outcome it identifies the append independently of
+        current state, including after a crash before this method returns.
+        Use a new id for new executions; reusing an id and outcome deliberately
+        replays the original append without making it current again.
 
         ``time_bounds`` is the episode's own time axis (``Episode.time_bounds``),
         recorded as the ``start_ns``/``end_ns`` columns. It is a fact about the
@@ -840,7 +854,7 @@ def append_episode(
 
         The consequence is worth stating rather than discovering. A replay
         returns ``written=False`` above without touching the stored row, so
-        this column names the run that FIRST recorded an outcome, not every
+        this column names the run that FIRST recorded an occurrence, not every
         run that has since produced it. That is the honest reading of an
         append that did nothing. ``None`` (the local dev loop, any caller
         outside the runtime) records NULL.
@@ -854,6 +868,8 @@ def append_episode(
         # it has to compare equal to what the orchestrator's own API reports.
         if orchestrator_run_id is not None and not orchestrator_run_id.strip():
             orchestrator_run_id = None
+        if execution_id is not None and not execution_id.strip():
+            raise ValueError("execution_id must be non-empty when supplied")
         episode_id = content_episode_id(canonical_path)
         # One normalized shape feeds every consumer below -- the run
         # fingerprint, the replay repair pass, and the dependent-table
@@ -869,12 +885,18 @@ def append_episode(
             for row in check_rows
         ]
         _raise_if_measurement_keys_shadow_episode_columns(check_rows)
-        run_fingerprint = _run_fingerprint(
+        outcome_fingerprint = _run_fingerprint(
             episode_id,
             stamps.pipeline_version,
             check_rows,
             quarantine_tags,
         )
+        run_fingerprint = self._append_fingerprint(
+            episode_id=episode_id,
+            outcome_fingerprint=outcome_fingerprint,
+            source_uri=source_uri,
+            execution_id=execution_id,
+        )
         file_stem = f"{episode_id}-{run_fingerprint}"
 
         # Create-if-absent: the episodes file is written last (manifest-last,
@@ -999,6 +1021,56 @@ def append_episode(
             episode_id=episode_id, run_fingerprint=run_fingerprint, written=episodes_created
         )
 
+    def _append_fingerprint(
+        self,
+        *,
+        episode_id: str,
+        outcome_fingerprint: str,
+        source_uri: str | None,
+        execution_id: str | None,
+    ) -> str:
+        """Keep outcome content stable while identifying each recurrence.
+
+        A suffix is an identity scope, never a clock or random salt. All tables
+        continue joining on the same (episode_id, run_fingerprint), and the
+        existing create-if-absent commit and repair protocol applies unchanged.
+        Racing identical appends observing the same predecessor choose the same
+        stem. Explicit execution ids also survive intervening commits.
+        """
+        if execution_id is not None:
+            scope = ["execution", execution_id]
+        else:
+            self.sync_for_read(("episodes",))
+            if not any(self.table_dir("episodes").glob("*.parquet")):
+                return outcome_fingerprint
+            connection = duckdb.connect()
+            try:
+                latest = connection.execute(
+                    "SELECT episode_id, run_fingerprint FROM read_parquet(?, union_by_name=true) "
+                    "WHERE episode_id = ? OR (source_uri IS NOT NULL AND source_uri = ?) "
+                    "ORDER BY recorded_at DESC, run_fingerprint DESC LIMIT 1",
+                    [str(self.table_dir("episodes") / "*.parquet"), episode_id, source_uri],
+                ).fetchone()
+            finally:
+                connection.close()
+            if latest is None:
+                # Let replay reconciliation diagnose an empty commit marker.
+                return outcome_fingerprint
+            latest_episode_id, latest_fingerprint = map(str, latest)
+            if (
+                latest_episode_id == episode_id
+                and latest_fingerprint.split(".", 1)[0] == outcome_fingerprint
+            ):
+                return latest_fingerprint
+            # Preserve existing v1 filenames for first occurrences, including
+            # crash debris whose episode commit marker has not landed yet.
+            original_key = f"episodes/{episode_id}-{outcome_fingerprint}.parquet"
+            if not self.location.exists(original_key):
+                return outcome_fingerprint
+            scope = ["after", latest_episode_id, latest_fingerprint]
+        suffix = hashlib.sha256(json.dumps(scope, separators=(",", ":")).encode()).hexdigest()[:16]
+        return f"{outcome_fingerprint}.{suffix}"
+
     def _reconcile_replayed_append(
         self,
         *,
@@ -1024,9 +1096,8 @@ def _reconcile_replayed_append(
         with the same committed ``recorded_at``, so concurrent replays
         converge instead of fighting.
 
-        Verified stems are memoized per process: the steady-state replay
-        (the idempotent-dedupe hot path) pays this pass's dependent file reads at
-        most once, then returns to the single existence check.
+        Verified stems are memoized per process: a replay pays this pass's
+        dependent file reads at most once per occurrence.
         """
         memo_key = (str(self.location), file_stem)
         if memo_key in _reconciled_append_stems:

tests/test_bucket_create_if_absent.py

@@ -51,7 +51,7 @@ def gated_store_file_if_absent(
         monkeypatch.setattr(BucketStorageRoot, "store_file_if_absent", gated_store_file_if_absent)
 
 
-def _append_outcome(catalog: Catalog, canonical_path: Path) -> AppendResult:
+def _append_outcome(catalog: Catalog, canonical_path: Path, value: float = 1.0) -> AppendResult:
     stamps = EpisodeStamps(
         schema_version="1",
         pipeline_version="abc123def456",
@@ -64,7 +64,7 @@ def _append_outcome(catalog: Catalog, canonical_path: Path) -> AppendResult:
         critical=False,
         status=hflow.CheckStatus.MEASURED,
         duration_s=0.01,
-        measurements={"score": 1.0},
+        measurements={"score": value},
         tags=["seen"],
         intervals=[hflow.Interval(start_ns=0, end_ns=10, label="span")],
     )
@@ -76,22 +76,32 @@ def _append_outcome(catalog: Catalog, canonical_path: Path) -> AppendResult:
     )
 
 
+@pytest.mark.parametrize("recurring", [False, True])
 def test_concurrent_bucket_catalog_appends_publish_one_complete_outcome(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
     bucket_over_tmp: tuple[BucketStorageRoot, Path],
+    recurring: bool,
 ) -> None:
     bucket_root, remote_dir = bucket_over_tmp
     catalog = Catalog(bucket_root.child("catalog"))
     canonical = tmp_path / "episode.canonical.mcap"
     canonical.write_bytes(b"canonical episode")
+    if recurring:
+        assert _append_outcome(catalog, canonical).written
+        assert _append_outcome(catalog, canonical, value=2.0).written
     race = _CreateIfAbsentRace(
         monkeypatch,
         lambda _root, relative: relative.startswith("episodes/"),
     )
 
-    def append(_label: str) -> AppendResult:
-        return _append_outcome(catalog, canonical)
+    def append(label: str) -> AppendResult:
+        # Separate workers must discover the same predecessor from the store,
+        # without relying on a shared, already-warm mirror.
+        worker_root = BucketStorageRoot(
+            bucket_root.child("catalog").url, mirror=tmp_path / f"worker-{label}"
+        )
+        return _append_outcome(Catalog(worker_root), canonical)
 
     with ThreadPoolExecutor(max_workers=2) as pool:
         results = list(pool.map(append, ("A", "B")))