From 3e389605b11b2b67cb3d0e05fb0d356292ce722a Mon Sep 17 00:00:00 2001 From: Faizan Naeem Date: Tue, 11 Aug 2026 23:18:36 +0500 Subject: [PATCH 1/2] feat(media): wire pending and failed ingest states --- docs/architecture/platform.md | 12 ++- src/vidxp/application_models.py | 23 ++++- src/vidxp/cli_commands/index.py | 6 +- src/vidxp/cli_commands/media.py | 6 +- src/vidxp/core/media.py | 21 ++-- src/vidxp/frontend.py | 8 +- src/vidxp/infrastructure/sql_catalog.py | 27 +++++- src/vidxp/media_service.py | 71 ++++++++++---- src/vidxp/ports.py | 2 + tests/test_cli.py | 36 +++++++ tests/test_media_catalog.py | 52 ++++++++++ tests/test_media_services.py | 124 +++++++++++++++++++++--- 12 files changed, 337 insertions(+), 51 deletions(-) diff --git a/docs/architecture/platform.md b/docs/architecture/platform.md index 275b7ec..d4251ff 100644 --- a/docs/architecture/platform.md +++ b/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 diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index d6c64e6..585205d 100644 --- a/src/vidxp/application_models.py +++ b/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, ...] = () diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index 14eb192..009e1a8 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -248,7 +248,11 @@ def index_list( 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:,}", ) Console().print(table) diff --git a/src/vidxp/cli_commands/media.py b/src/vidxp/cli_commands/media.py index bc86552..dcd3234 100644 --- a/src/vidxp/cli_commands/media.py +++ b/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 ) diff --git a/src/vidxp/core/media.py b/src/vidxp/core/media.py index 95b6baf..6fbf7e7 100644 --- a/src/vidxp/core/media.py +++ b/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 diff --git a/src/vidxp/frontend.py b/src/vidxp/frontend.py index 0260d7e..2b4469b 100644 --- a/src/vidxp/frontend.py +++ b/src/vidxp/frontend.py @@ -708,8 +708,12 @@ def _select_video(busy, media_id, media_page): else 0 ), format_func=lambda value: ( - f"{asset_by_id[value].original_filename} " - f"({asset_by_id[value].duration_seconds:.1f}s)" + asset_by_id[value].original_filename + if asset_by_id[value].duration_seconds is None + else ( + f"{asset_by_id[value].original_filename} " + f"({asset_by_id[value].duration_seconds:.1f}s)" + ) ), disabled=busy, ) diff --git a/src/vidxp/infrastructure/sql_catalog.py b/src/vidxp/infrastructure/sql_catalog.py index 076c722..c433665 100644 --- a/src/vidxp/infrastructure/sql_catalog.py +++ b/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, diff --git a/src/vidxp/media_service.py b/src/vidxp/media_service.py index af33145..0464932 100644 --- a/src/vidxp/media_service.py +++ b/src/vidxp/media_service.py @@ -183,39 +183,63 @@ 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: - authoritative = self.catalog.put_media(record) + probe = self.probe.probe(staged.path) + except BaseException: + self._mark_failed(pending) + raise + stored = self.store.publish(staged) + 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.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: @@ -228,8 +252,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( diff --git a/src/vidxp/ports.py b/src/vidxp/ports.py index 66475d4..18c3ad2 100644 --- a/src/vidxp/ports.py +++ b/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, *, diff --git a/tests/test_cli.py b/tests/test_cli.py index 12d826e..f88d467 100644 --- a/tests/test_cli.py +++ b/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( diff --git a/tests/test_media_catalog.py b/tests/test_media_catalog.py index 57bbef5..a9df9fd 100644 --- a/tests/test_media_catalog.py +++ b/tests/test_media_catalog.py @@ -7,6 +7,8 @@ from tempfile import TemporaryDirectory from unittest.mock import patch +from pydantic import ValidationError + from vidxp.core.artifacts import ArtifactKind, ArtifactRecord from vidxp.core.media import ( MediaImportLimitError, @@ -59,6 +61,24 @@ def media_record( ) +def incomplete_media_record( + *, + state: MediaState, + media_id: str = MEDIA_ID, + checksum: str = "1" * 64, +) -> MediaRecord: + return MediaRecord( + media_id=media_id, + video_id=media_id, + sha256=checksum, + original_filename="video.mp4", + byte_size=5, + storage_key=f"objects/{checksum[:2]}/{checksum}.mp4", + state=state, + created_at=datetime.now(timezone.utc), + ) + + class LocalCatalogTests(unittest.TestCase): def test_catalog_enforces_sqlite_integrity_and_schema_version(self): with TemporaryDirectory() as directory: @@ -124,6 +144,38 @@ def test_catalog_persists_and_deduplicates_media_by_checksum(self): self.assertEqual(reopened.put_media(duplicate), record) self.assertEqual(reopened.list_media(limit=10), (record,)) + def test_catalog_replaces_pending_and_failed_media(self): + with TemporaryDirectory() as directory: + catalog = LocalCatalog(Path(directory) / "catalog.sqlite3") + pending = incomplete_media_record(state=MediaState.pending) + self.assertEqual(catalog.put_media(pending), pending) + + failed = pending.model_copy(update={"state": MediaState.failed}) + self.assertEqual(catalog.replace_media(failed), failed) + self.assertEqual(catalog.get_media(MEDIA_ID), failed) + + ready = media_record() + self.assertEqual(catalog.replace_media(ready), ready) + self.assertEqual(catalog.get_media(MEDIA_ID), ready) + with self.assertRaises(FileExistsError): + catalog.replace_media( + ready.model_copy( + update={"original_filename": "other.mp4"} + ) + ) + + def test_ready_media_requires_a_video_stream(self): + with self.assertRaises(ValidationError): + incomplete_media_record(state=MediaState.ready) + self.assertEqual( + incomplete_media_record(state=MediaState.pending).state, + MediaState.pending, + ) + self.assertEqual( + incomplete_media_record(state=MediaState.failed).state, + MediaState.failed, + ) + def test_artifact_requires_cataloged_media_and_survives_reopen(self): with TemporaryDirectory() as directory: database = Path(directory) / "catalog.sqlite3" diff --git a/tests/test_media_services.py b/tests/test_media_services.py index 5121994..7468b70 100644 --- a/tests/test_media_services.py +++ b/tests/test_media_services.py @@ -21,6 +21,7 @@ from vidxp.core.contracts import CancellationToken, IndexCancelledError from vidxp.core.media import ( MediaProbe, + MediaUnavailableError, QuarantinedMedia, MediaRecord, MediaState, @@ -73,6 +74,19 @@ def record() -> MediaRecord: ) +def pending_record(*, state: MediaState = MediaState.pending) -> MediaRecord: + return MediaRecord( + media_id=MEDIA_ID, + video_id=MEDIA_ID, + sha256="1" * 64, + original_filename="video.mp4", + byte_size=5, + storage_key="objects/11/video.mp4", + state=state, + created_at=datetime.now(timezone.utc), + ) + + class MediaServiceTests(unittest.TestCase): def service(self, root: Path): catalog = Mock() @@ -88,6 +102,8 @@ def service(self, root: Path): store=store, probe=probe, ) + catalog.put_media.side_effect = lambda item: item + catalog.replace_media.side_effect = lambda item: item return service, catalog, store, probe def test_import_probes_staging_before_publishing_and_cataloging(self): @@ -125,7 +141,6 @@ def test_import_probes_staging_before_publishing_and_cataloging(self): ), ) catalog.get_media_by_checksum.return_value = None - catalog.put_media.side_effect = lambda item: item with patch("vidxp.media_service.uuid4") as identifier: identifier.return_value.hex = MEDIA_ID result = service.import_local(ImportMediaCommand(path=source)) @@ -135,9 +150,18 @@ def test_import_probes_staging_before_publishing_and_cataloging(self): probe.probe.assert_called_once_with(staged.path) store.publish.assert_called_once_with(staged) catalog.put_media.assert_called_once() + self.assertEqual( + catalog.put_media.call_args.args[0].state, + MediaState.pending, + ) + catalog.replace_media.assert_called_once() + self.assertEqual( + catalog.replace_media.call_args.args[0].state, + MediaState.ready, + ) store.discard.assert_called_once_with(staged) - def test_invalid_probe_never_publishes_or_catalogs_media(self): + def test_invalid_probe_catalogs_failed_media_without_publishing(self): with TemporaryDirectory() as directory: root = Path(directory) source = root / "video.mp4" @@ -157,7 +181,16 @@ def test_invalid_probe_never_publishes_or_catalogs_media(self): service.import_local(ImportMediaCommand(path=source)) store.publish.assert_not_called() - catalog.put_media.assert_not_called() + catalog.put_media.assert_called_once() + self.assertEqual( + catalog.put_media.call_args.args[0].state, + MediaState.pending, + ) + catalog.replace_media.assert_called_once() + self.assertEqual( + catalog.replace_media.call_args.args[0].state, + MediaState.failed, + ) store.discard.assert_called_once_with(staged) def test_quarantined_import_reuses_the_same_ingestion_pipeline(self): @@ -195,7 +228,6 @@ def test_quarantined_import_reuses_the_same_ingestion_pipeline(self): ), ) catalog.get_media_by_checksum.return_value = None - catalog.put_media.side_effect = lambda item: item with patch("vidxp.media_service.uuid4") as identifier: identifier.return_value.hex = MEDIA_ID result = service.import_quarantined( @@ -249,16 +281,17 @@ def test_quarantined_import_completes_durable_idempotency_record(self): catalog.reserve_media_import.return_value = None catalog.get_media_by_checksum.return_value = None imported = record() - catalog.put_media.return_value = imported catalog.get_media.return_value = imported - result = service.import_quarantined( - QuarantinedMedia( - path=source, - original_filename="upload.mp4", - ), - request_key="request-key", - ) + with patch("vidxp.media_service.uuid4") as identifier: + identifier.return_value.hex = MEDIA_ID + result = service.import_quarantined( + QuarantinedMedia( + path=source, + original_filename="upload.mp4", + ), + request_key="request-key", + ) self.assertEqual(result.media_id, MEDIA_ID) fingerprint = catalog.reserve_media_import.call_args.args[1] @@ -381,7 +414,7 @@ def test_catalog_failure_rolls_back_new_managed_content(self): ), ) catalog.get_media_by_checksum.return_value = None - catalog.put_media.side_effect = RuntimeError("catalog failed") + catalog.replace_media.side_effect = RuntimeError("catalog failed") with self.assertRaisesRegex(RuntimeError, "catalog failed"): service.import_local(ImportMediaCommand(path=source)) @@ -424,6 +457,71 @@ def test_media_pages_are_bounded_and_cursor_scoped(self): 2, ) + def test_failed_checksum_is_retried_to_ready(self): + with TemporaryDirectory() as directory: + root = Path(directory) + source = root / "video.mp4" + source.write_bytes(b"video") + service, catalog, store, probe = self.service(root) + staged = StagedMedia( + sha256="1" * 64, + byte_size=5, + storage_key="objects/11/video.mp4", + path=root / "staged.tmp", + ) + stored = StoredMedia( + sha256=staged.sha256, + byte_size=5, + storage_key=staged.storage_key, + local_path=root / "managed.mp4", + ) + store.stage_local.return_value = staged + store.publish.return_value = stored + probe.probe.return_value = MediaProbe( + detected_mime_type="video/mp4", + container="mp4", + duration_seconds=2, + streams=( + MediaStream( + index=0, + kind="video", + codec="h264", + width=1, + height=1, + ), + ), + ) + catalog.get_media_by_checksum.return_value = pending_record( + state=MediaState.failed + ) + + result = service.import_local(ImportMediaCommand(path=source)) + + self.assertEqual(result.media_id, MEDIA_ID) + self.assertEqual(result.state, MediaState.ready) + catalog.put_media.assert_not_called() + self.assertEqual( + [call.args[0].state for call in catalog.replace_media.call_args_list], + [MediaState.pending, MediaState.ready], + ) + + def test_get_returns_non_ready_media(self): + with TemporaryDirectory() as directory: + service, catalog, _store, _probe = self.service(Path(directory)) + catalog.get_media.return_value = pending_record(state=MediaState.failed) + + result = service.get(MEDIA_ID) + + self.assertEqual(result.state, MediaState.failed) + + def test_require_record_rejects_non_ready_media(self): + with TemporaryDirectory() as directory: + service, catalog, _store, _probe = self.service(Path(directory)) + catalog.get_media.return_value = pending_record() + + with self.assertRaises(MediaUnavailableError): + service.require_record(MEDIA_ID) + class ArtifactServiceTests(unittest.TestCase): def test_actor_overlay_uses_catalog_media_and_store_owned_destination(self): From 0ab8f0c68c28ab7587a5d4715ab1ce5c51ad06d8 Mon Sep 17 00:00:00 2001 From: Faizan Naeem Date: Thu, 13 Aug 2026 15:53:03 +0500 Subject: [PATCH 2/2] fix(media): address pending/failed ingest review feedback --- src/vidxp/cli_commands/index.py | 6 +--- src/vidxp/control_plane.py | 2 +- src/vidxp/frontend.py | 15 +++++---- src/vidxp/media_service.py | 7 +++- tests/test_control_plane.py | 37 ++++++++++++++++++-- tests/test_frontend.py | 60 +++++++++++++++++++++++++++++++++ tests/test_media_services.py | 54 ++++++++++++++++++++++++++++- 7 files changed, 164 insertions(+), 17 deletions(-) diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index 009e1a8..14eb192 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -248,11 +248,7 @@ def index_list( table.add_row( asset.media_id, asset.original_filename, - ( - "-" - if asset.duration_seconds is None - else f"{asset.duration_seconds:.3f}s" - ), + f"{asset.duration_seconds:.3f}s", f"{asset.byte_size:,}", ) Console().print(table) diff --git a/src/vidxp/control_plane.py b/src/vidxp/control_plane.py index 1897080..08278d0 100644 --- a/src/vidxp/control_plane.py +++ b/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 diff --git a/src/vidxp/frontend.py b/src/vidxp/frontend.py index 2b4469b..1bfff31 100644 --- a/src/vidxp/frontend.py +++ b/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 @@ -690,7 +691,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 @@ -708,12 +713,8 @@ def _select_video(busy, media_id, media_page): else 0 ), format_func=lambda value: ( - asset_by_id[value].original_filename - if asset_by_id[value].duration_seconds is None - else ( - f"{asset_by_id[value].original_filename} " - f"({asset_by_id[value].duration_seconds:.1f}s)" - ) + f"{asset_by_id[value].original_filename} " + f"({asset_by_id[value].duration_seconds:.1f}s)" ), disabled=busy, ) diff --git a/src/vidxp/media_service.py b/src/vidxp/media_service.py index 0464932..e208efa 100644 --- a/src/vidxp/media_service.py +++ b/src/vidxp/media_service.py @@ -221,7 +221,11 @@ def _publish_import( except BaseException: self._mark_failed(pending) raise - stored = self.store.publish(staged) + 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, @@ -244,6 +248,7 @@ def _publish_import( self.store.delete(stored.storage_key) except OSError: pass + self._mark_failed(pending) raise if authoritative.storage_key != stored.storage_key: try: diff --git a/tests/test_control_plane.py b/tests/test_control_plane.py index a6ae69b..05a7d8d 100644 --- a/tests/test_control_plane.py +++ b/tests/test_control_plane.py @@ -16,7 +16,7 @@ from vidxp.capabilities.registry import create_capability_registry from vidxp.capability_service import CapabilityService from vidxp.control_plane import ControlPlaneApplication -from vidxp.core.media import MediaState, MediaStream +from vidxp.core.media import MediaState, MediaStream, MediaUnavailableError from vidxp.core.snapshots import GenerationReference, IndexSnapshot from vidxp.repository_layout import RepositoryLayout @@ -77,7 +77,40 @@ def test_index_preflight_rejects_unknown_capability_with_next_action(self): self.assertEqual(error["reason"], "capability_unknown") self.assertEqual(error["requested"], ["unknown"]) self.assertIn("get_workspace", error["next_action"]) - media.get.assert_not_called() + media.require_record.assert_not_called() + + def test_index_preflight_rejects_non_ready_media(self): + with TemporaryDirectory() as directory: + root = Path(directory) + media = Mock() + media.require_record.side_effect = MediaUnavailableError( + "The media asset is unavailable." + ) + application = ControlPlaneApplication( + layout=RepositoryLayout(root=root), + capabilities=CapabilityService(create_capability_registry()), + media=media, + artifacts=Mock(), + index_status=lambda: None, + model_cache=root / "models", + ) + + for state in (MediaState.pending, MediaState.failed): + with self.subTest(state=state): + media.require_record.reset_mock() + with self.assertRaises(ApplicationError) as raised: + application.preflight_index( + CreateIndexCommand( + media_id=MEDIA_ID, + modalities=("scene",), + ) + ) + + self.assertEqual( + raised.exception.to_dict()["category"], + "not_found", + ) + media.require_record.assert_called_once_with(MEDIA_ID) def test_workspace_projects_index_coverage_roles_and_next_actions(self): indexed = media_asset(MEDIA_ID, "indexed.mp4") diff --git a/tests/test_frontend.py b/tests/test_frontend.py index 9991b0a..42a6f1a 100644 --- a/tests/test_frontend.py +++ b/tests/test_frontend.py @@ -21,6 +21,7 @@ from vidxp.capabilities.registry import create_capability_registry from vidxp.capability_service import CapabilityService from vidxp.control_plane import ControlPlaneApplication +from vidxp.core.media import MediaState from vidxp.repository_layout import RepositoryLayout from vidxp.settings import LocalExecutionSettings, VidXPSettings @@ -201,6 +202,7 @@ def test_busy_video_layout_keeps_controls_and_preview_stable(self): media_id=MEDIA_ID, original_filename="video.mp4", duration_seconds=27.2, + state=MediaState.ready, ), ), next_cursor=None, @@ -245,6 +247,64 @@ def test_busy_video_layout_keeps_controls_and_preview_stable(self): self.assertTrue(uploader.call_args.kwargs["disabled"]) video.assert_called_once_with("video.mp4", width=560) + def test_registered_video_selector_lists_only_ready_media(self): + service = Mock() + ready_id = MEDIA_ID + pending_id = "223456781234423481234567890abcde" + media_page = SimpleNamespace( + items=( + SimpleNamespace( + media_id=ready_id, + original_filename="ready.mp4", + duration_seconds=12.0, + state=MediaState.ready, + ), + SimpleNamespace( + media_id=pending_id, + original_filename="pending.mp4", + duration_seconds=None, + state=MediaState.pending, + ), + ), + next_cursor=None, + ) + with ( + patch.object( + frontend, + "_configured_service", + return_value=service, + ), + patch.object(frontend.st, "session_state", {}), + patch.object(frontend.st, "subheader"), + patch.object( + frontend.st, + "selectbox", + return_value=ready_id, + ) as selectbox, + patch.object( + frontend.st, + "expander", + return_value=nullcontext(), + ), + patch.object(frontend.st, "caption"), + patch.object(frontend.st, "text_input", return_value=""), + patch.object(frontend.st, "button", return_value=False), + patch.object( + frontend.st, + "file_uploader", + return_value=None, + ), + patch.object(frontend.st, "video"), + ): + _uploaded, media_id = frontend._select_video( + False, + pending_id, + media_page, + ) + + self.assertEqual(media_id, ready_id) + self.assertEqual(selectbox.call_args.args[1], (ready_id,)) + def test_local_path_import_uses_the_shared_application_command(self): service = Mock() service.import_media.return_value = SimpleNamespace(media_id=MEDIA_ID) diff --git a/tests/test_media_services.py b/tests/test_media_services.py index 7468b70..d2dcb28 100644 --- a/tests/test_media_services.py +++ b/tests/test_media_services.py @@ -414,12 +414,64 @@ def test_catalog_failure_rolls_back_new_managed_content(self): ), ) catalog.get_media_by_checksum.return_value = None - catalog.replace_media.side_effect = RuntimeError("catalog failed") + + def replace_side_effect(record): + if record.state == MediaState.ready: + raise RuntimeError("catalog failed") + return record + + catalog.replace_media.side_effect = replace_side_effect with self.assertRaisesRegex(RuntimeError, "catalog failed"): service.import_local(ImportMediaCommand(path=source)) store.delete.assert_called_once_with(stored.storage_key) + failed_calls = [ + call.args[0].state + for call in catalog.replace_media.call_args_list + if call.args[0].state == MediaState.failed + ] + self.assertEqual(failed_calls, [MediaState.failed]) + + def test_publish_failure_catalogs_failed_media_without_publishing(self): + with TemporaryDirectory() as directory: + root = Path(directory) + source = root / "video.mp4" + source.write_bytes(b"video") + service, catalog, store, probe = self.service(root) + staged = StagedMedia( + sha256="1" * 64, + byte_size=5, + storage_key="objects/11/video.mp4", + path=root / "staged.tmp", + ) + store.stage_local.return_value = staged + probe.probe.return_value = MediaProbe( + detected_mime_type="video/mp4", + container="mp4", + duration_seconds=2, + streams=( + MediaStream( + index=0, + kind="video", + codec="h264", + width=1, + height=1, + ), + ), + ) + catalog.get_media_by_checksum.return_value = None + store.publish.side_effect = RuntimeError("publish failed") + + with self.assertRaisesRegex(RuntimeError, "publish failed"): + service.import_local(ImportMediaCommand(path=source)) + + catalog.replace_media.assert_called_once() + self.assertEqual( + catalog.replace_media.call_args.args[0].state, + MediaState.failed, + ) + store.discard.assert_called_once_with(staged) def test_media_pages_are_bounded_and_cursor_scoped(self): with TemporaryDirectory() as directory: