fix: classify transform content refusals as source-unsupported, not infrastructure

docs/CATALOG.md

@@ -44,7 +44,7 @@ append at all:
 | `measurements` | measurement key | `key`, typed value columns (`value_double`, `value_text`, `value_bool`), the producing `check_name`/`check_version` |
 | `tags` | tag | `check_name`, `tag` |
 | `intervals` | labeled time span | `label`, `start_ns`, `end_ns` |
-| `ingest_failures` | attempt that produced NO episode row | `source_uri`, `stage`, `failure_kind` (`source-missing` / `source-unreadable` / `infrastructure`), `error_type`, `message`, `pipeline_version`, `orchestrator_run_id`, `recorded_at` |
+| `ingest_failures` | attempt that produced NO episode row | `source_uri`, `stage`, `failure_kind` (`source-missing` / `source-unreadable` / `source-unsupported` / `infrastructure`), `error_type`, `message`, `pipeline_version`, `orchestrator_run_id`, `recorded_at` |
 
 `ingest_failures` is the complement of `episodes`, and exists because
 `episode_id` is a hash of canonical bytes: a recording that never
@@ -55,6 +55,18 @@ next to the verbatim `error_type` and `message` rather than in place of them;
 anything unrecognized is `infrastructure`, never an accusation against the
 recording.
 
+`source-unsupported` was added after `ingest_failures` already existed in
+deployed catalogs. Rows written before an ingest binary carrying this change
+keep whatever kind that binary assigned at write time -- a content refusal
+recorded as `infrastructure` by an older binary is not retroactively
+reclassified, because the ledger is append-only and nothing rewrites past
+rows. An operator draining an old ledger by kind should expect
+`infrastructure` rows from before the cutover to include content refusals
+alongside genuine infrastructure trouble; `error_type` and `message` remain
+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:
 
 - **Content-addressed**: `episode_id` is a sha256 of the canonical file's
@@ -291,7 +303,9 @@ decision:
   an unfilled hole excludes the whole corpus. `skipped` is different and stays
   excluded: a step skipped because a critical check quarantined the episode has
   real work to do the moment that check is retuned. So does `error`: a crash is
-  infrastructure, so it is a retry.
+  infrastructure, so it is a retry -- unlike an ingest-time content refusal
+  (`ingest_failures.failure_kind = source-unsupported`), which never reaches
+  a check run at all and is not retried.
 
 An episode inherits a term of its own from that last case. A crash is a retry
 for the *check*, but the episode it was supposed to check is left in a third

docs/how-to/write-a-converter.md

@@ -17,7 +17,7 @@ Raw `sensor_msgs/msg/Image` and `foxglove.RawImage` channels are not supported.
 
 Those are the HFlow-specific acceptance rules. Other malformed records or payloads can still be rejected by the MCAP decoder, the selected message decoder, or ffmpeg. HFlow does not require source provenance metadata, canonical chunk layout, globally sorted messages, or canonical identifiers before ingest.
 
-When ingest cannot create an episode, inspect both `error_type` and `message` in the [`ingest_failures` table](../CATALOG.md), not only `failure_kind`. MCAP parse exceptions are classified as `source-unreadable`, but transform-level `ValueError` and `NotImplementedError` exceptions currently fall under the fallback `infrastructure` classification even when their messages identify an unsupported converter output. The classification logic is in [`classify_ingest_failure`](../../src/hflow/ingest_ledger.py).
+When ingest cannot create an episode, `failure_kind` in the [`ingest_failures` table](../CATALOG.md) now tells you which side of the boundary it was: `source-unsupported` means the transform read the file fine but refused on its contents (the cases named above -- unsupported image format, a passthrough video violation, a raw image channel, a derived-topic collision), `source-unreadable` means the file was not a readable MCAP at all, and `infrastructure` is everything else the engine did not recognize. Still inspect `error_type` and `message` alongside `failure_kind` for the specific reason -- the kind tells you where to look, not what went wrong. The classification logic is in [`classify_ingest_failure`](../../src/hflow/ingest_ledger.py).
 
 ## What HFlow adds
 

src/hflow/ingest_ledger.py

@@ -47,10 +47,13 @@
 class IngestFailureKind(StrEnum):
     """Whose problem this was, as far as the engine can honestly tell.
 
-    Three members, because three is what the evidence supports. The point of
+    Four members, because four is what the evidence supports. The point of
     the distinction is triage: a source that is bad stays bad and wants a
     human, while infrastructure trouble wants a retry -- and reporting one as
-    the other sends people to the wrong place.
+    the other sends people to the wrong place. The three source-side members
+    separate different flavors of "bad": missing entirely, unreadable as a
+    file at all, and readable but refusing on its contents (a transform
+    declining to bridge or transcode what it found).
 
     ``INFRASTRUCTURE`` is the default for anything unrecognized, never the
     other way round. Guessing "your data is bad" from a crash the engine did
@@ -60,6 +63,7 @@ class IngestFailureKind(StrEnum):
 
     SOURCE_MISSING = "source-missing"
     SOURCE_UNREADABLE = "source-unreadable"
+    SOURCE_UNSUPPORTED = "source-unsupported"
     INFRASTRUCTURE = "infrastructure"
 
 
@@ -85,9 +89,15 @@ def classify_ingest_failure(error: BaseException) -> IngestFailureKind:
     next to the evidence that would correct it rather than replacing it.
     """
     from hflow.app import SourceNotFound
+    from hflow.transform import SourceNotConforming
 
     if isinstance(error, SourceNotFound):
         return IngestFailureKind.SOURCE_MISSING
+    if isinstance(error, SourceNotConforming):
+        # A refusal about the recording's contents (an unsupported format, a
+        # canonical-convention violation) -- distinct from SOURCE_UNREADABLE,
+        # which is a file that isn't even a readable MCAP.
+        return IngestFailureKind.SOURCE_UNSUPPORTED
     try:
         from mcap.exceptions import McapError
     except ImportError:  # pragma: no cover - mcap is a hard dependency

src/hflow/transform.py

@@ -111,6 +111,25 @@
 _RAW_IMAGE_SCHEMAS = frozenset({"sensor_msgs/msg/Image", "foxglove.RawImage"})
 
 
+class SourceNotConforming(ValueError):
+    """The recording refuses transcoding on its contents, not the machine.
+
+    Raised where the transform has looked at what a channel actually carries
+    and found it outside what v1 supports or bridges -- an unsupported image
+    format, a passthrough video channel violating the canonical convention, an
+    unsupported raw image schema, a derived-topic collision. A subclass of
+    ``ValueError`` so existing callers matching on that type keep working.
+
+    This is the ONLY path into
+    :attr:`~hflow.ingest_ledger.IngestFailureKind.SOURCE_UNSUPPORTED`:
+    ``classify_ingest_failure`` tests ``isinstance(error, SourceNotConforming)``
+    and nothing else routes there, so a bug that raises a bare ``ValueError``
+    for a new refusal falls back to ``infrastructure`` (the safe default)
+    rather than silently landing here. Any new content refusal must raise
+    this type explicitly to be classified correctly.
+    """
+
+
 @dataclass(frozen=True)
 class TransformConfig:
     """Configuration for :func:`write_canonical_episode`. This is the
@@ -245,7 +264,7 @@ def _input_codec_for_image_format(image_format: str, topic: str) -> Literal["mjp
         return "png"
     if "jpeg" in lowered or "jpg" in lowered:
         return "mjpeg"
-    raise ValueError(
+    raise SourceNotConforming(
         f"camera topic {topic!r} carries unsupported compressed-image format "
         f"{image_format!r}; v1 transcodes jpeg and png"
     )
@@ -435,7 +454,7 @@ class _PassthroughVideoMessage:
 
 def _parse_passthrough_video_message(topic: str, decoded_message: Any) -> _PassthroughVideoMessage:
     if decoded_message.format != "h264":
-        raise ValueError(
+        raise SourceNotConforming(
             f"pass-through video topic {topic!r} carries format "
             f"{decoded_message.format!r}; the canonical convention requires 'h264' "
             "(re-encode upstream -- v1 does not transcode CompressedVideo)"
@@ -450,7 +469,7 @@ def _insert_missing_passthrough_video_aud(
     try:
         canonical_data = video_module.ensure_access_unit_delimiter(message.data)
     except ValueError as error:
-        raise ValueError(
+        raise SourceNotConforming(
             f"pass-through video topic {topic!r} violates the canonical convention: {error}"
         ) from error
     return _PassthroughVideoMessage(format=message.format, data=canonical_data)
@@ -468,24 +487,24 @@ def _validate_passthrough_video_payload(
     try:
         access_units = video_module.split_annex_b_stream(message.data)
     except ValueError as error:
-        raise ValueError(
+        raise SourceNotConforming(
             f"pass-through video topic {topic!r} violates the canonical convention: "
             f"{error} (re-encode upstream -- v1 does not transcode CompressedVideo)"
         ) from error
     if len(access_units) != 1:
-        raise ValueError(
+        raise SourceNotConforming(
             f"pass-through video topic {topic!r} has a message containing "
             f"{len(access_units)} access units; the canonical convention requires "
             "exactly one decodable frame per message"
         )
     access_unit = access_units[0]
     if access_unit.is_keyframe and not access_unit.has_parameter_sets:
-        raise ValueError(
+        raise SourceNotConforming(
             f"pass-through video topic {topic!r} has a keyframe without SPS/PPS; "
             "the canonical convention requires parameter sets on every keyframe"
         )
     if is_first_message and not access_unit.is_keyframe:
-        raise ValueError(
+        raise SourceNotConforming(
             f"pass-through video topic {topic!r} starts mid-GOP: its first message "
             "is not a keyframe, so the stream is not decodable from the start"
         )
@@ -552,7 +571,7 @@ def write_canonical_episode(
         infos = reader.channels()
         for info in infos.values():
             if info.schema_name in _RAW_IMAGE_SCHEMAS:
-                raise NotImplementedError(
+                raise SourceNotConforming(
                     f"raw image channel {info.topic!r} ({info.schema_name}) is not supported "
                     "in v1; record compressed images upstream"
                 )
@@ -561,7 +580,7 @@ def write_canonical_episode(
             if derived_topic in source_topic_names:
                 # A second channel on the topic would be MCAP-legal but make
                 # topic-keyed reads ambiguous forever; refuse loudly instead.
-                raise ValueError(
+                raise SourceNotConforming(
                     f"derived channel topic {derived_topic!r} collides with a source "
                     "channel topic; pick a topic not present in the source"
                 )

tests/test_ingest_ledger.py

@@ -0,0 +1,115 @@
+"""``classify_ingest_failure`` pins each refusal to the right ``IngestFailureKind``."""
+
+from pathlib import Path
+
+import numpy as np
+import pytest
+from mcap.exceptions import InvalidMagic
+from mcap.writer import Writer as StockWriter
+
+from hflow import transform
+from hflow.app import SourceNotFound
+from hflow.format import METADATA_RECORD_EPISODE
+from hflow.ingest_ledger import IngestFailureKind, classify_ingest_failure
+from hflow.resample import DerivedSeries
+from hflow.testing import SyntheticEpisodeSpec, synthesize_episode
+from hflow.transform import SourceNotConforming, write_canonical_episode
+
+
+def test_classify_source_not_found_as_source_missing() -> None:
+    error = SourceNotFound("episode 'missing.mcap' not found")
+    assert classify_ingest_failure(error) == IngestFailureKind.SOURCE_MISSING
+
+
+def test_classify_mcap_error_as_source_unreadable() -> None:
+    error = InvalidMagic(b"not-an-mcap-file")
+    assert classify_ingest_failure(error) == IngestFailureKind.SOURCE_UNREADABLE
+
+
+def test_classify_source_not_conforming_as_source_unsupported() -> None:
+    error = SourceNotConforming("x")
+    assert classify_ingest_failure(error) == IngestFailureKind.SOURCE_UNSUPPORTED
+
+
+def test_classify_unrecognized_error_as_infrastructure() -> None:
+    error = RuntimeError("unknown")
+    assert classify_ingest_failure(error) == IngestFailureKind.INFRASTRUCTURE
+
+
+def test_unsupported_compressed_image_format_classifies_as_source_unsupported() -> None:
+    with pytest.raises(SourceNotConforming) as raised:
+        transform._input_codec_for_image_format("bogus", "/cam")
+    assert classify_ingest_failure(raised.value) == IngestFailureKind.SOURCE_UNSUPPORTED
+
+
+def test_nonconforming_passthrough_video_classifies_as_source_unsupported(tmp_path: Path) -> None:
+    from foxglove_schemas_protobuf.CompressedVideo_pb2 import CompressedVideo
+    from mcap_protobuf.schema import build_file_descriptor_set
+
+    source = tmp_path / "h265.mcap"
+    with source.open("wb") as stream:
+        writer = StockWriter(stream)
+        writer.start(profile="", library="test")
+        schema_id = writer.register_schema(
+            name="foxglove.CompressedVideo",
+            encoding="protobuf",
+            data=build_file_descriptor_set(CompressedVideo).SerializeToString(),
+        )
+        channel_id = writer.register_channel(
+            topic="/cam", message_encoding="protobuf", schema_id=schema_id
+        )
+        message = CompressedVideo()
+        message.timestamp.FromNanoseconds(10**9)
+        message.frame_id = "cam"
+        message.data = b"\x00\x00\x00\x01\x40junk"
+        message.format = "h265"
+        writer.add_message(
+            channel_id, log_time=10**9, data=message.SerializeToString(), publish_time=10**9
+        )
+        writer.finish()
+
+    with pytest.raises(SourceNotConforming) as raised:
+        write_canonical_episode(source, tmp_path / "out.mcap")
+    assert classify_ingest_failure(raised.value) == IngestFailureKind.SOURCE_UNSUPPORTED
+
+
+def test_raw_image_schema_classifies_as_source_unsupported(tmp_path: Path) -> None:
+    source = tmp_path / "raw_image.mcap"
+    with source.open("wb") as stream:
+        writer = StockWriter(stream)
+        writer.start(profile="", library="test")
+        schema_id = writer.register_schema(
+            name="sensor_msgs/msg/Image", encoding="ros2msg", data=b"std_msgs/Header header"
+        )
+        channel_id = writer.register_channel(
+            topic="/raw_cam", message_encoding="cdr", schema_id=schema_id
+        )
+        writer.add_message(channel_id, log_time=1, data=b"", publish_time=1)
+        writer.add_metadata(METADATA_RECORD_EPISODE, {})
+        writer.finish()
+
+    with pytest.raises(SourceNotConforming) as raised:
+        write_canonical_episode(source, tmp_path / "out.mcap")
+    assert classify_ingest_failure(raised.value) == IngestFailureKind.SOURCE_UNSUPPORTED
+
+
+def test_derived_topic_collision_classifies_as_source_unsupported(tmp_path: Path) -> None:
+    cameraless_spec = SyntheticEpisodeSpec(
+        duration_s=2.0,
+        cameras=(),
+        joint_hz=50.0,
+        black_segment=None,
+        joint_jump_at_s=None,
+        timestamp_offset_segment=None,
+    )
+    source = synthesize_episode(tmp_path / "episode.mcap", cameraless_spec)
+    series = DerivedSeries(
+        timestamps_ns=np.asarray([1_755_000_000_500_000_000], dtype=np.int64),
+        values={"value": np.asarray([1.0])},
+    )
+
+    with pytest.raises(SourceNotConforming) as raised:
+        write_canonical_episode(
+            source, tmp_path / "out.mcap", derived=[("/joint_states", series, "v1")]
+        )
+    assert classify_ingest_failure(raised.value) == IngestFailureKind.SOURCE_UNSUPPORTED

tests/test_processing_regressions.py

@@ -24,7 +24,7 @@
 from hflow.format import METADATA_RECORD_EPISODE
 from hflow.reader import TopicInfo
 from hflow.testing import SyntheticEpisodeSpec, synthesize_episode
-from hflow.transform import write_canonical_episode
+from hflow.transform import SourceNotConforming, write_canonical_episode
 from hflow.video import estimate_fps_from_log_times
 
 ANNEX_B_START_CODE = b"\x00\x00\x00\x01"
@@ -124,7 +124,7 @@ def test_transform_rejects_nonconforming_passthrough_video(tmp_path: Path) -> No
             channel_id, log_time=10**9, data=message.SerializeToString(), publish_time=10**9
         )
         writer.finish()
-    with pytest.raises(ValueError, match="requires 'h264'"):
+    with pytest.raises(SourceNotConforming, match="requires 'h264'"):
         write_canonical_episode(source, tmp_path / "out.mcap")
 
 
@@ -174,7 +174,7 @@ def test_transform_rejects_each_passthrough_video_channel_starting_mid_gop(
         ],
     )
 
-    with pytest.raises(ValueError, match=r"/cam/right.*starts mid-GOP"):
+    with pytest.raises(SourceNotConforming, match=r"/cam/right.*starts mid-GOP"):
         write_canonical_episode(source, tmp_path / "out.mcap")
 
 
@@ -238,7 +238,7 @@ def test_transform_still_rejects_undelimited_video_starting_mid_gop(tmp_path: Pa
     source = tmp_path / "undelimited-mid-gop.mcap"
     _write_passthrough_video_source(source, [("/cam", 1, NON_KEYFRAME_WITHOUT_AUD)])
 
-    with pytest.raises(ValueError, match="starts mid-GOP"):
+    with pytest.raises(SourceNotConforming, match="starts mid-GOP"):
         write_canonical_episode(source, tmp_path / "out.mcap")
 
 
@@ -248,7 +248,7 @@ def test_transform_and_doctor_reject_multiple_pictures_without_auds(tmp_path: Pa
         source, [("/cam", 1, KEYFRAME_WITHOUT_AUD + NON_KEYFRAME_WITHOUT_AUD)]
     )
 
-    with pytest.raises(ValueError, match="message contains 2 pictures"):
+    with pytest.raises(SourceNotConforming, match="message contains 2 pictures"):
         write_canonical_episode(source, tmp_path / "out.mcap")
 
     report = diagnose(source)