feat(media): wire pending and failed ingest states

docs/architecture/platform.md

@@ -297,15 +297,17 @@ Remote API/MCP responses never expose these internal paths.
 - original filename
 - byte size
 - declared and detected MIME/container
-- duration, streams and codecs from ffprobe
+- duration, streams and codecs from ffprobe when the asset is `ready`
 - managed storage key or approved external source reference
 - repository and owner/principal where applicable
-- ingest state
+- ingest state (`pending`, `ready`, or `failed`)
 - associated `video_id`
 - creation and retention timestamps
 
 Checksum is calculated once during ingest and reused for deduplication and indexing.
-Untrusted content is not published into the catalog until ffprobe validation succeeds.
+The catalog may record `pending` or `failed` ingest metadata. Untrusted bytes are
+not published into managed storage until ffprobe validation succeeds, and only
+`ready` media can be indexed or materialized.
 
 ### 9.2 Local CLI and desktop
 
@@ -405,8 +407,8 @@ retain explicit cancellation and manual cleanup for abandoned tus resources.
 The hook endpoint is private to the Compose network. Client authorization is read
 from the hook request body and redacted; client tokens are never stored in tus
 metadata. Only the tus upload route is public. Hooks remain enqueue-only; recovery
-runs in the API's existing ingestion coordinator. A completed upload is not a
-`MediaAsset` until durable probe/import succeeds.
+runs in the API's existing ingestion coordinator. A completed upload is not
+`ready` media until durable probe/import succeeds.
 
 The supported server topology uses tusd filestore on a named quarantine volume
 shared read-only with the hook service and worker. The API intentionally does not

src/vidxp/application_models.py

@@ -364,13 +364,26 @@ class MediaAsset(ApplicationModel):
     sha256: Sha256
     byte_size: int = Field(gt=0)
     declared_mime_type: MimeType | None = None
-    detected_mime_type: MimeType
-    container: str = Field(min_length=1)
-    duration_seconds: float = Field(gt=0)
-    streams: tuple[MediaStream, ...] = Field(min_length=1)
+    detected_mime_type: MimeType | None = None
+    container: str | None = Field(default=None, min_length=1)
+    duration_seconds: float | None = Field(default=None, gt=0)
+    streams: tuple[MediaStream, ...] = ()
     state: MediaState
     created_at: AwareDatetime
 
+    @model_validator(mode="after")
+    def _require_ready_probe(self) -> "MediaAsset":
+        if self.state != MediaState.ready:
+            return self
+        if (
+            self.detected_mime_type is None
+            or self.container is None
+            or self.duration_seconds is None
+            or not any(stream.kind == "video" for stream in self.streams)
+        ):
+            raise ValueError("ready media must contain a video stream")
+        return self
+
 
 class ListMediaCommand(ApplicationModel):
     page_size: int = Field(
@@ -767,7 +780,7 @@ class WorkspaceMediaCapability(ApplicationModel):
 class WorkspaceMedia(ApplicationModel):
     media_id: MediaId
     original_filename: str = Field(min_length=1)
-    duration_seconds: float = Field(gt=0)
+    duration_seconds: float | None = Field(default=None, gt=0)
     state: MediaState
     in_active_snapshot: bool
     capabilities: tuple[WorkspaceMediaCapability, ...] = ()

src/vidxp/cli_commands/media.py

@@ -94,7 +94,11 @@ def list_media(
         table.add_row(
             asset.media_id,
             asset.original_filename,
-            f"{asset.duration_seconds:.3f}s",
+            (
+                "-"
+                if asset.duration_seconds is None
+                else f"{asset.duration_seconds:.3f}s"
+            ),
             f"{asset.byte_size:,}",
             asset.state.value
         )

src/vidxp/control_plane.py

@@ -244,7 +244,7 @@ def preflight_index(self, command: CreateIndexCommand) -> None:
             selected,
             command.capability_options,
         )
-        self.get_media(command.media_id)
+        self.media.require_record(command.media_id)
         self.require_models(selected)
 
     @application_boundary

src/vidxp/core/media.py

@@ -42,7 +42,9 @@ class MediaUnavailableError(FileNotFoundError):
 
 
 class MediaState(StrEnum):
+    pending = "pending"
     ready = "ready"
+    failed = "failed"
 
 
 class _MediaModel(BaseModel):
@@ -132,10 +134,10 @@ class MediaRecord(_MediaModel):
     original_filename: str = Field(min_length=1, max_length=255)
     byte_size: int = Field(gt=0)
     declared_mime_type: MimeType | None = None
-    detected_mime_type: MimeType
-    container: str = Field(min_length=1)
-    duration_seconds: float = Field(gt=0)
-    streams: tuple[MediaStream, ...] = Field(min_length=1)
+    detected_mime_type: MimeType | None = None
+    container: str | None = Field(default=None, min_length=1)
+    duration_seconds: float | None = Field(default=None, gt=0)
+    streams: tuple[MediaStream, ...] = ()
     storage_key: str = Field(min_length=1)
     state: MediaState = MediaState.ready
     created_at: AwareDatetime
@@ -151,8 +153,15 @@ def _validate_storage_key(cls, value: str) -> str:
         return validate_storage_key(value)
 
     @model_validator(mode="after")
-    def _require_video_stream(self) -> "MediaRecord":
-        if not any(stream.kind == "video" for stream in self.streams):
+    def _require_ready_probe(self) -> "MediaRecord":
+        if self.state != MediaState.ready:
+            return self
+        if (
+            self.detected_mime_type is None
+            or self.container is None
+            or self.duration_seconds is None
+            or not any(stream.kind == "video" for stream in self.streams)
+        ):
             raise ValueError("ready media must contain a video stream")
         return self
 

src/vidxp/frontend.py

@@ -28,6 +28,7 @@
 )
 from vidxp.branding import PROJECT_URL, icon_path
 from vidxp.composition import create_application, create_job_service
+from vidxp.core.media import MediaState
 from vidxp.index_state import IndexNotReadyError
 from vidxp.job_service import JobService
 from vidxp.settings import LocalExecutionSettings, VidXPSettings
@@ -750,7 +751,11 @@ def _import_local_video(service, raw_path):
 def _select_video(busy, media_id, media_page):
     service = _configured_service()
     st.subheader("Video")
-    assets = tuple(media_page.items) if media_page is not None else ()
+    assets = tuple(
+        asset
+        for asset in (media_page.items if media_page is not None else ())
+        if asset.state == MediaState.ready
+    )
     media_id = _default_media_id(media_id, assets)
     if media_id is not None:
         st.session_state[MEDIA_ID_KEY] = media_id

src/vidxp/infrastructure/sql_catalog.py

@@ -22,7 +22,7 @@
 from sqlalchemy.pool import NullPool
 
 from vidxp.core.artifacts import ArtifactRecord, ArtifactState
-from vidxp.core.media import MediaRecord, utc_now
+from vidxp.core.media import MediaRecord, MediaState, utc_now
 from vidxp.core.uploads import (
     UploadIntentRecord,
     UploadSessionFileRecord,
@@ -49,6 +49,7 @@
     UploadState.processing.value,
     UploadState.failed.value,
 }
+_REPLACEABLE_MEDIA_STATES = {MediaState.pending, MediaState.failed}
 _UPLOAD_QUOTA_ID = "1"
 _EXPECTED_VALUE_UNSET = object()
 
@@ -258,6 +259,30 @@ def put_media(self, record: MediaRecord) -> MediaRecord:
                 raise
         return record
 
+    def replace_media(self, record: MediaRecord) -> MediaRecord:
+        with self._write_transaction() as connection:
+            existing = self._media_by_id(connection, record.media_id)
+            if existing is None:
+                raise FileNotFoundError(
+                    f"Media {record.media_id} is not cataloged."
+                )
+            if existing.sha256 != record.sha256:
+                raise FileExistsError(
+                    f"Media {record.media_id} already has another record."
+                )
+            if existing == record:
+                return existing
+            if existing.state not in _REPLACEABLE_MEDIA_STATES:
+                raise FileExistsError(
+                    f"Media {record.media_id} already has another record."
+                )
+            connection.execute(
+                update(media)
+                .where(media.c.media_id == record.media_id)
+                .values(payload=record.model_dump(mode="json"))
+            )
+        return record
+
     @staticmethod
     def _media_by_id(
         connection: Connection,

src/vidxp/media_service.py

@@ -183,43 +183,72 @@ def _publish_import(
         declared_mime_type: str | None,
         staged: StagedMedia,
     ) -> MediaAsset:
-        if existing := self.catalog.get_media_by_checksum(staged.sha256):
+        existing = self.catalog.get_media_by_checksum(staged.sha256)
+        if existing is not None and existing.state == MediaState.ready:
             self.store.publish(
                 staged.model_copy(
                     update={"storage_key": existing.storage_key}
                 )
             )
             return media_asset(existing)
-        probe = self.probe.probe(staged.path)
-        stored = self.store.publish(staged)
-        media_id = uuid4().hex
-        record = MediaRecord(
+        media_id = existing.media_id if existing is not None else uuid4().hex
+        pending = MediaRecord(
             media_id=media_id,
             video_id=media_id,
-            sha256=stored.sha256,
+            sha256=staged.sha256,
             original_filename=original_filename,
-            byte_size=stored.byte_size,
+            byte_size=staged.byte_size,
             declared_mime_type=declared_mime_type,
-            detected_mime_type=probe.detected_mime_type,
-            container=probe.container,
-            duration_seconds=probe.duration_seconds,
-            streams=probe.streams,
-            storage_key=stored.storage_key,
-            state=MediaState.ready,
-            created_at=utc_now(),
+            storage_key=staged.storage_key,
+            state=MediaState.pending,
+            created_at=(
+                existing.created_at if existing is not None else utc_now()
+            ),
+        )
+        if existing is None:
+            pending = self.catalog.put_media(pending)
+            if pending.state == MediaState.ready:
+                self.store.publish(
+                    staged.model_copy(
+                        update={"storage_key": pending.storage_key}
+                    )
+                )
+                return media_asset(pending)
+        elif existing != pending:
+            pending = self.catalog.replace_media(pending)
+        try:
+            probe = self.probe.probe(staged.path)
+        except BaseException:
+            self._mark_failed(pending)
+            raise
+        try:
+            stored = self.store.publish(staged)
+        except BaseException:
+            self._mark_failed(pending)
+            raise
+        ready = pending.model_copy(
+            update={
+                "detected_mime_type": probe.detected_mime_type,
+                "container": probe.container,
+                "duration_seconds": probe.duration_seconds,
+                "streams": probe.streams,
+                "storage_key": stored.storage_key,
+                "state": MediaState.ready,
+            }
         )
         try:
-            authoritative = self.catalog.put_media(record)
+            authoritative = self.catalog.replace_media(ready)
         except BaseException:
             try:
                 retained = self.catalog.get_media_by_checksum(stored.sha256)
             except Exception:
                 retained = None
-            if retained is None:
+            if retained is None or retained.state != MediaState.ready:
                 try:
                     self.store.delete(stored.storage_key)
                 except OSError:
                     pass
+            self._mark_failed(pending)
             raise
         if authoritative.storage_key != stored.storage_key:
             try:
@@ -228,8 +257,21 @@ def _publish_import(
                 pass
         return media_asset(authoritative)
 
+    def _mark_failed(self, pending: MediaRecord) -> None:
+        if pending.state == MediaState.failed:
+            return
+        try:
+            self.catalog.replace_media(
+                pending.model_copy(update={"state": MediaState.failed})
+            )
+        except Exception:
+            pass
+
     def get(self, media_id: str) -> MediaAsset:
-        return media_asset(self.require_record(media_id))
+        record = self.catalog.get_media(media_id)
+        if record is None:
+            raise MediaUnavailableError("The media asset is unavailable.")
+        return media_asset(record)
 
     def list(self, command: ListMediaCommand) -> MediaPage:
         scope = hashlib.sha256(

src/vidxp/ports.py

@@ -97,6 +97,8 @@ def get_media_by_checksum(self, sha256: str) -> MediaRecord | None: ...
 
     def put_media(self, record: MediaRecord) -> MediaRecord: ...
 
+    def replace_media(self, record: MediaRecord) -> MediaRecord: ...
+
     def list_media(
         self,
         *,

tests/test_cli.py

@@ -429,6 +429,42 @@ def test_media_list_shows_media_state(self):
         self.assertIn("State", result.output)
         self.assertIn("ready", result.output)
 
+    def test_media_list_shows_pending_and_failed_states(self):
+        failed_id = "223456781234423481234567890abcde"
+        self.service.list_media.return_value = MediaPage(
+            items=(
+                MediaAsset(
+                    schema_version=1,
+                    media_id=MEDIA_ID,
+                    video_id=MEDIA_ID,
+                    original_filename="pending.mp4",
+                    sha256="1" * 64,
+                    byte_size=5,
+                    state=MediaState.pending,
+                    created_at=datetime.now(timezone.utc),
+                ),
+                MediaAsset(
+                    schema_version=1,
+                    media_id=failed_id,
+                    video_id=failed_id,
+                    original_filename="failed.mp4",
+                    sha256="2" * 64,
+                    byte_size=7,
+                    state=MediaState.failed,
+                    created_at=datetime.now(timezone.utc),
+                ),
+            ),
+            next_cursor=None,
+            total=2,
+        )
+
+        result = self.invoke(["media", "list"])
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        self.assertIn("pending", result.output)
+        self.assertIn("failed", result.output)
+        self.assertIn("-", result.output)
+
     def test_ui_share_uses_streamlit_wildcard_bind_and_warns(self):
         with (
             patch(