From 0153adb9ac69e49749c040c3d95fcc4dad279e1f Mon Sep 17 00:00:00 2001 From: umair Date: Tue, 8 Sep 2026 13:53:48 +0500 Subject: [PATCH 1/4] feat(indexing): add bulk video indexing command and coordinator --- src/vidxp/bulk_indexing.py | 201 +++++++++ src/vidxp/cli_commands/index.py | 178 +++++++- tests/test_bulk_indexing.py | 697 ++++++++++++++++++++++++++++++++ 3 files changed, 1075 insertions(+), 1 deletion(-) create mode 100644 src/vidxp/bulk_indexing.py create mode 100644 tests/test_bulk_indexing.py diff --git a/src/vidxp/bulk_indexing.py b/src/vidxp/bulk_indexing.py new file mode 100644 index 00000000..6cbe5e2b --- /dev/null +++ b/src/vidxp/bulk_indexing.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Callable, Sequence + +from vidxp.application_models import ( + ApplicationError, + CreateIndexCommand, + ListMediaCommand, + MediaAsset, + MediaState, +) +from vidxp.core.snapshots import IndexSnapshot + +if TYPE_CHECKING: + from vidxp.application import VidXPApplication + from vidxp.control_plane import ControlPlaneApplication + from vidxp.job_service import JobService + + +@dataclass(frozen=True) +class BulkIndexItemResult: + media_id: str + filename: str + status: str + job_id: str | None = None + error_code: str | None = None + error_message: str | None = None + + +@dataclass(frozen=True) +class BulkIndexSummary: + total: int + indexed: int + skipped: int + failed: int + queued: int = 0 + results: tuple[BulkIndexItemResult, ...] = () + + +def _resolve_all_media( + application: VidXPApplication | ControlPlaneApplication, +) -> list[MediaAsset]: + media_list: list[MediaAsset] = [] + cursor: str | None = None + while True: + page = application.list_media( + ListMediaCommand( + page_size=100, + cursor=cursor, + state=MediaState.ready, + ) + ) + media_list.extend(page.items) + if not page.next_cursor or not page.items: + break + cursor = page.next_cursor + return media_list + + +def _is_already_indexed( + snapshot: IndexSnapshot | None, + media_id: str, + requested_modalities: Sequence[str] | None = None, +) -> bool: + if snapshot is None: + return False + generation = snapshot.generations.get(media_id) + if generation is None: + return False + if requested_modalities is not None: + return set(requested_modalities).issubset(set(generation.modalities)) + return True + + +def run_bulk_index( + application: VidXPApplication | ControlPlaneApplication, + jobs: JobService, + media_ids: Sequence[str] | None = None, + *, + all_eligible: bool = False, + skip_indexed: bool = True, + detach: bool = False, + modalities: Sequence[str] | None = None, + frame_stride: int = 1, + scene_sample_fps: float | None = None, + capability_options: dict[str, dict] | None = None, + on_item_start: Callable[[str, str], None] | None = None, + on_item_progress: Callable[[str, Any], None] | None = None, + on_item_complete: Callable[[BulkIndexItemResult], None] | None = None, +) -> BulkIndexSummary: + items: list[tuple[str, str]] = [] + if all_eligible: + all_media = _resolve_all_media(application) + items = [(asset.media_id, asset.original_filename) for asset in all_media] + elif media_ids: + for mid in media_ids: + asset = application.get_media(mid) + items.append((asset.media_id, asset.original_filename)) + + read_snapshot = getattr(application, "_read_active_snapshot", None) + snapshot: IndexSnapshot | None = ( + read_snapshot() if callable(read_snapshot) else None + ) + + if modalities is not None: + cmd_modalities = tuple(modalities) + elif hasattr(application, "select_index_modalities"): + cmd_modalities = application.select_index_modalities(None) + elif hasattr(application, "list_capabilities"): + cmd_modalities = tuple( + c.name + for c in application.list_capabilities() + if getattr(c, "supports_indexing", True) + ) + else: + cmd_modalities = () + + results: list[BulkIndexItemResult] = [] + + for media_id, filename in items: + if skip_indexed and _is_already_indexed(snapshot, media_id, modalities): + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="skipped", + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + continue + + if on_item_start is not None: + on_item_start(media_id, filename) + + command = CreateIndexCommand( + media_id=media_id, + modalities=cmd_modalities, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=capability_options or {}, + ) + + try: + job = jobs.submit_index(command) + if detach: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="queued", + job_id=job.job_id, + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + else: + def _progress(current: Any) -> None: + if on_item_progress is not None: + on_item_progress(media_id, current) + + job = jobs.wait(job.job_id, progress=_progress) + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="indexed", + job_id=job.job_id, + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + except ApplicationError as exc: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="failed", + error_code=exc.code, + error_message=str(exc), + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + except Exception as exc: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="failed", + error_code="unexpected_error", + error_message=str(exc), + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + + return BulkIndexSummary( + total=len(results), + indexed=sum(1 for r in results if r.status == "indexed"), + skipped=sum(1 for r in results if r.status == "skipped"), + failed=sum(1 for r in results if r.status == "failed"), + queued=sum(1 for r in results if r.status == "queued"), + results=tuple(results), + ) diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index 14eb192a..af23423d 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -1,15 +1,17 @@ from __future__ import annotations -from typing import Annotated, Iterable +from typing import Annotated, Any, Iterable import typer from rich.console import Console +from rich.markup import escape from rich.table import Table from vidxp.application_models import ( CreateIndexCommand, RemoveIndexCommand, ) +from vidxp.bulk_indexing import run_bulk_index from vidxp.cli_support import ( CLIState, IndexProgress, @@ -154,6 +156,180 @@ def index_create( ) +@app.command("bulk") +def index_bulk( + ctx: typer.Context, + media_ids: Annotated[ + list[str] | None, + typer.Argument( + help="Registered media identifiers to index.", + ), + ] = None, + all_eligible: Annotated[ + bool, + typer.Option( + "--all", + help="Index all eligible registered media in the catalog.", + ), + ] = False, + reindex: Annotated[ + bool, + typer.Option( + "--reindex", + help="Reindex media even if already present in the active index.", + ), + ] = False, + modalities: Annotated[ + list[str] | None, + typer.Option( + "--modality", + "-m", + help="Modality to index; repeat to select more than one.", + ), + ] = None, + frame_stride: Annotated[ + int, + typer.Option( + "--frame-stride", + min=1, + help=( + "Materialize every Nth frame for actor and legacy visual " + "indexing." + ), + ), + ] = 1, + scene_sample_fps: Annotated[ + float | None, + typer.Option( + "--scene-sample-fps", + min=0.01, + help=( + "Target scene samples per second; lower-FPS media uses every " + "available frame." + ), + ), + ] = None, + capability_options: Annotated[ + list[str] | None, + typer.Option( + "--option", + help=( + "Capability setting as CAPABILITY.KEY=VALUE; " + "repeat for multiple settings." + ), + ), + ] = None, + detach: Annotated[ + bool, + typer.Option( + "--detach", + help="Return after the durable job is queued.", + ), + ] = False, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Index multiple media items or all eligible media in the catalog.""" + + if not media_ids and not all_eligible: + raise typer.BadParameter("Provide either media IDs or pass --all.") + if media_ids and all_eligible: + raise typer.BadParameter("Pass media IDs or --all, not both.") + + state = state_from_context(ctx) + indexable = tuple( + capability.name + for capability in state.service.list_capabilities() + if capability.supports_indexing + ) + selected = selected_modalities(modalities, indexable) + parsed_options = parse_capability_options(capability_options) + output_fmt = effective_output_format(state, json_output) + show_progress = not state.quiet and output_fmt == OutputFormat.rich + + with IndexProgress(show_progress) as progress: + def on_item_start(media_id: str, filename: str) -> None: + if show_progress: + progress.update({ + "stage": "indexing", + "message": f"Indexing {filename} ({media_id[:8]}...)", + }) + + def on_item_progress(media_id: str, current: Any) -> None: + if show_progress: + if hasattr(current, "progress") and current.progress is not None: + progress.update(current.progress.model_dump(mode="python")) + elif isinstance(current, dict): + progress.update(current) + + summary = run_bulk_index( + application=state.service, + jobs=state.jobs, + media_ids=media_ids, + all_eligible=all_eligible, + skip_indexed=not reindex, + detach=detach, + modalities=selected, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=parsed_options, + on_item_start=on_item_start, + on_item_progress=on_item_progress, + ) + + if output_fmt == OutputFormat.json: + payload = { + "total": summary.total, + "indexed": summary.indexed, + "skipped": summary.skipped, + "failed": summary.failed, + "queued": summary.queued, + "results": [ + { + "media_id": r.media_id, + "filename": r.filename, + "status": r.status, + "job_id": r.job_id, + "error_code": r.error_code, + "error_message": r.error_message, + } + for r in summary.results + ], + } + emit_json(payload) + else: + table = Table(title="Bulk indexing summary") + table.add_column("Media ID") + table.add_column("Filename") + table.add_column("Status") + table.add_column("Job ID") + table.add_column("Error") + for r in summary.results: + error_str = ( + f"[{r.error_code}] {r.error_message}" + if r.error_code + else (r.error_message or "—") + ) + table.add_row( + escape(r.media_id), + escape(r.filename), + escape(r.status), + escape(r.job_id or "—"), + escape(error_str), + ) + Console().print(table) + typer.echo( + f"Total: {summary.total}, Indexed: {summary.indexed}, " + f"Skipped: {summary.skipped}, Failed: {summary.failed}, " + f"Queued: {summary.queued}." + ) + + if summary.failed > 0: + raise typer.Exit(code=1) + + @app.command("remove") def index_remove( ctx: typer.Context, diff --git a/tests/test_bulk_indexing.py b/tests/test_bulk_indexing.py new file mode 100644 index 00000000..6a6ba2e6 --- /dev/null +++ b/tests/test_bulk_indexing.py @@ -0,0 +1,697 @@ +from __future__ import annotations + +import json +import unittest +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from unittest.mock import Mock, patch + +from typer.testing import CliRunner + +from vidxp import cli +from vidxp.application_models import ( + ApplicationError, + CreateIndexCommand, + DependencyCheckResult, + ErrorCategory, + IndexJobResult, + IndexResult, + Job, + JobKind, + JobProgress, + JobQueue, + JobState, + ListMediaCommand, + MediaAsset, + MediaPage, + MediaState, + MediaStream, +) +from vidxp.bulk_indexing import ( + BulkIndexItemResult, + BulkIndexSummary, + _is_already_indexed, + _resolve_all_media, + run_bulk_index, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.capability_service import CapabilityService +from vidxp.composition import LocalApplicationContext +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.repositories import RepositoryConfig, RepositoryRegistry + +MEDIA_ID_1 = "123456781234423481234567890abcde" +MEDIA_ID_2 = "223456781234423481234567890abcde" +MEDIA_ID_3 = "323456781234423481234567890abcde" +JOB_ID_1 = "423456781234423481234567890abcde" +JOB_ID_2 = "523456781234423481234567890abcde" +SNAPSHOT_ID = "623456781234423481234567890abcde" +GENERATION_ID = "723456781234423481234567890abcde" + + +def make_media(media_id: str, filename: str = "video.mp4") -> MediaAsset: + return MediaAsset( + schema_version=1, + media_id=media_id, + video_id=media_id, + original_filename=filename, + sha256="1" * 64, + byte_size=1024, + detected_mime_type="video/mp4", + container="mp4", + duration_seconds=10.0, + streams=( + MediaStream( + index=0, + kind="video", + codec="h264", + width=640, + height=480, + ), + ), + state=MediaState.ready, + created_at=datetime.now(timezone.utc), + ) + + +def make_job( + job_id: str, + state: JobState = JobState.succeeded, + media_id: str = MEDIA_ID_1, +) -> Job: + result = None + if state == JobState.succeeded: + result = IndexJobResult( + result=IndexResult( + media_id=media_id, + generation_id=GENERATION_ID, + snapshot_id=SNAPSHOT_ID, + active_media_count=1, + record_counts={"scene": 1}, + ) + ) + return Job( + job_id=job_id, + kind=JobKind.index, + state=state, + queue=JobQueue.cpu, + result=result, + progress=JobProgress( + stage="indexing", + current=1, + total=1, + message="Done", + updated_at=datetime.now(timezone.utc), + ), + ) + + +def make_snapshot(generations: dict[str, tuple[str, ...]]) -> IndexSnapshot: + gen_refs: dict[str, GenerationReference] = {} + for mid, modalities in generations.items(): + gen_refs[mid] = GenerationReference( + generation_id=GENERATION_ID, + media_id=mid, + manifest_sha256="1" * 64, + input_sha256="2" * 64, + config_fingerprint="3" * 64, + modalities=modalities, + record_counts={m: 1 for m in modalities}, + store_size_bytes_at_commit=1024, + ) + return IndexSnapshot( + schema_version=1, + snapshot_id=SNAPSHOT_ID, + created_at=datetime.now(timezone.utc), + config_fingerprint="0" * 64, + configuration={}, + generations=gen_refs, + ) + + +class BulkIndexingHelperTests(unittest.TestCase): + def test_resolve_all_media_paginates_cursor(self): + media_1 = make_media(MEDIA_ID_1, "first.mp4") + media_2 = make_media(MEDIA_ID_2, "second.mp4") + media_3 = make_media(MEDIA_ID_3, "third.mp4") + + app = Mock() + app.list_media.side_effect = [ + MediaPage(items=(media_1, media_2), total=3, next_cursor="c1"), + MediaPage(items=(media_3,), total=3, next_cursor=None), + ] + + result = _resolve_all_media(app) + + self.assertEqual(result, [media_1, media_2, media_3]) + self.assertEqual(app.list_media.call_count, 2) + app.list_media.assert_any_call( + ListMediaCommand( + page_size=100, + cursor=None, + state=MediaState.ready, + ) + ) + app.list_media.assert_any_call( + ListMediaCommand( + page_size=100, + cursor="c1", + state=MediaState.ready, + ) + ) + + def test_resolve_all_media_empty_catalog(self): + app = Mock() + app.list_media.return_value = MediaPage( + items=(), total=0, next_cursor=None + ) + + result = _resolve_all_media(app) + + self.assertEqual(result, []) + self.assertEqual(app.list_media.call_count, 1) + + def test_is_already_indexed_snapshot_none(self): + self.assertFalse(_is_already_indexed(None, MEDIA_ID_1)) + + def test_is_already_indexed_media_not_in_generations(self): + snapshot = make_snapshot({MEDIA_ID_2: ("scene",)}) + self.assertFalse(_is_already_indexed(snapshot, MEDIA_ID_1)) + + def test_is_already_indexed_no_modalities_requested(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.assertTrue(_is_already_indexed(snapshot, MEDIA_ID_1, None)) + + def test_is_already_indexed_matching_modalities_subset(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene", "speech", "actor")}) + self.assertTrue( + _is_already_indexed(snapshot, MEDIA_ID_1, ("scene", "speech")) + ) + + def test_is_already_indexed_missing_requested_modality(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.assertFalse( + _is_already_indexed(snapshot, MEDIA_ID_1, ("scene", "speech")) + ) + + +class RunBulkIndexTests(unittest.TestCase): + def setUp(self): + self.app = Mock() + self.jobs = Mock() + self.media_1 = make_media(MEDIA_ID_1, "one.mp4") + self.media_2 = make_media(MEDIA_ID_2, "two.mp4") + self.app.get_media.side_effect = lambda mid: ( + self.media_1 if mid == MEDIA_ID_1 else self.media_2 + ) + self.app._read_active_snapshot.return_value = None + + def test_bulk_index_multiple_media_ids_success(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + started: list[tuple[str, str]] = [] + completed: list[BulkIndexItemResult] = [] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + modalities=["scene"], + on_item_start=lambda mid, fn: started.append((mid, fn)), + on_item_complete=lambda r: completed.append(r), + ) + + self.assertEqual( + summary, + BulkIndexSummary( + total=2, + indexed=2, + skipped=0, + failed=0, + queued=0, + results=( + BulkIndexItemResult( + media_id=MEDIA_ID_1, + filename="one.mp4", + status="indexed", + job_id=JOB_ID_1, + ), + BulkIndexItemResult( + media_id=MEDIA_ID_2, + filename="two.mp4", + status="indexed", + job_id=JOB_ID_2, + ), + ), + ), + ) + self.assertEqual( + started, + [(MEDIA_ID_1, "one.mp4"), (MEDIA_ID_2, "two.mp4")], + ) + self.assertEqual(len(completed), 2) + self.assertEqual(self.jobs.submit_index.call_count, 2) + self.assertEqual(self.jobs.wait.call_count, 2) + + def test_bulk_index_all_eligible_paginates(self): + self.app.list_media.side_effect = [ + MediaPage(items=(self.media_1,), total=2, next_cursor="c1"), + MediaPage(items=(self.media_2,), total=2, next_cursor=None), + ] + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + all_eligible=True, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 2) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.failed, 0) + + def test_bulk_index_skips_already_indexed(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.app._read_active_snapshot.return_value = snapshot + + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.return_value = job_2 + self.jobs.wait.return_value = job_2 + + started: list[tuple[str, str]] = [] + completed: list[BulkIndexItemResult] = [] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + skip_indexed=True, + modalities=["scene"], + on_item_start=lambda mid, fn: started.append((mid, fn)), + on_item_complete=lambda r: completed.append(r), + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.skipped, 1) + self.assertEqual(summary.failed, 0) + self.assertEqual(summary.results[0].status, "skipped") + self.assertEqual(summary.results[0].media_id, MEDIA_ID_1) + self.assertEqual(summary.results[1].status, "indexed") + self.assertEqual(summary.results[1].media_id, MEDIA_ID_2) + # started should NOT include skipped item + self.assertEqual(started, [(MEDIA_ID_2, "two.mp4")]) + self.assertEqual(len(completed), 2) + self.jobs.submit_index.assert_called_once() + + def test_bulk_index_reindex_flag_overrides_skip(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.app._read_active_snapshot.return_value = snapshot + + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + skip_indexed=False, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.results[0].status, "indexed") + + def test_bulk_index_detach_queues_jobs(self): + job_1 = make_job(JOB_ID_1, state=JobState.queued, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + detach=True, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.queued, 1) + self.assertEqual(summary.indexed, 0) + self.assertEqual(summary.results[0].status, "queued") + self.assertEqual(summary.results[0].job_id, JOB_ID_1) + self.jobs.wait.assert_not_called() + + def test_bulk_index_error_resilience(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [ + ApplicationError( + "transcription_failed", + ErrorCategory.unavailable, + "Model crashed during transcription.", + ), + job_2, + ] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + modalities=["scene"], + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.failed, 1) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.results[0].status, "failed") + self.assertEqual( + summary.results[0].error_code, "transcription_failed" + ) + self.assertEqual( + summary.results[0].error_message, + "Model crashed during transcription.", + ) + self.assertEqual(summary.results[1].status, "indexed") + self.assertEqual(summary.results[1].job_id, JOB_ID_2) + + def test_bulk_index_generic_exception_resilience(self): + self.jobs.submit_index.side_effect = RuntimeError("Disk IO error") + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.failed, 1) + self.assertEqual(summary.results[0].status, "failed") + self.assertEqual(summary.results[0].error_code, "unexpected_error") + self.assertIn("Disk IO error", summary.results[0].error_message or "") + + + def test_bulk_index_forwards_options_and_command_fields(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + + progress_events: list[tuple[str, Any]] = [] + + def fake_wait(job_id, progress=None): + if progress: + progress(job_1) + return job_1 + + self.jobs.wait.side_effect = fake_wait + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=["scene"], + frame_stride=3, + scene_sample_fps=1.5, + capability_options={"scene": {"threshold": 0.7}}, + on_item_progress=lambda mid, curr: progress_events.append((mid, curr)), + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertIsInstance(submitted_command, CreateIndexCommand) + self.assertEqual(submitted_command.media_id, MEDIA_ID_1) + self.assertEqual(submitted_command.modalities, ("scene",)) + self.assertEqual(submitted_command.frame_stride, 3) + self.assertEqual(submitted_command.scene_sample_fps, 1.5) + self.assertEqual( + submitted_command.capability_options, + {"scene": {"threshold": 0.7}}, + ) + self.assertEqual(len(progress_events), 1) + self.assertEqual(progress_events[0][0], MEDIA_ID_1) + + def test_bulk_index_default_modalities_resolves_from_application(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + self.app.select_index_modalities.return_value = ("scene", "speech") + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=None, + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertEqual(submitted_command.modalities, ("scene", "speech")) + + def test_bulk_index_default_modalities_resolves_from_list_capabilities(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + del self.app.select_index_modalities + cap1 = Mock() + cap1.name = "scene" + cap1.supports_indexing = True + cap2 = Mock() + cap2.name = "summary" + cap2.supports_indexing = False + self.app.list_capabilities.return_value = [cap1, cap2] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=None, + ) + + self.assertEqual(summary.total, 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertEqual(submitted_command.modalities, ("scene",)) + + def test_bulk_index_empty_items(self): + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[], + modalities=["scene"], + ) + self.assertEqual(summary.total, 0) + self.assertEqual(summary.indexed, 0) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.failed, 0) + self.assertEqual(summary.queued, 0) + self.assertEqual(summary.results, ()) + + +class CliBulkIndexTests(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + self.service = Mock() + self.service.registry = create_capability_registry() + self.service.list_capabilities.return_value = CapabilityService( + self.service.registry + ).list() + self.service.index_directory = Path("repo/indexes") + self.service.layout.root = Path("repo") + self.service.model_cache = Path("model-cache") + self.service.runtime.backends.requested = "cpu" + self.service.model_readiness.return_value = DependencyCheckResult( + ok=True, + modalities=(), + checks=(), + ) + self.service._read_active_snapshot.return_value = None + + self.media_1 = make_media(MEDIA_ID_1, "one.mp4") + self.media_2 = make_media(MEDIA_ID_2, "two.mp4") + self.service.get_media.side_effect = lambda mid: ( + self.media_1 if mid == MEDIA_ID_1 else self.media_2 + ) + self.service.list_media.return_value = MediaPage( + items=(self.media_1, self.media_2), total=2, next_cursor=None + ) + + self.jobs = Mock() + self.registry = Mock(spec=RepositoryRegistry) + self.registry.path = Path("repositories.json") + self.repository = RepositoryConfig( + "default", + Path("repo"), + device="cpu", + configured=False, + ) + + def invoke(self, arguments, *, media_runtime_initialized=True): + with ( + patch.object( + cli, + "create_local_application", + return_value=LocalApplicationContext( + application=self.service, + jobs=self.jobs, + repositories=self.registry, + repository=self.repository, + ), + ) as create_local_application, + patch( + "vidxp.cli_support.media_runtime_is_initialized", + return_value=media_runtime_initialized, + ), + ): + result = self.runner.invoke(cli.app, arguments) + self.create_local_application = create_local_application + return result + + def test_cli_requires_media_ids_or_all(self): + result = self.invoke(["index", "bulk"]) + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("Provide either media IDs or pass --all.", result.output) + + def test_cli_rejects_both_media_ids_and_all(self): + result = self.invoke(["index", "bulk", MEDIA_ID_1, "--all"]) + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("Pass media IDs or --all, not both.", result.output) + + def test_cli_bulk_index_success_json(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke( + [ + "index", + "bulk", + MEDIA_ID_1, + MEDIA_ID_2, + "--modality", + "scene", + "--json", + ] + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 2) + self.assertEqual(payload["indexed"], 2) + self.assertEqual(payload["failed"], 0) + self.assertEqual(payload["skipped"], 0) + self.assertEqual(len(payload["results"]), 2) + self.assertEqual(payload["results"][0]["media_id"], MEDIA_ID_1) + self.assertEqual(payload["results"][0]["status"], "indexed") + self.assertEqual(payload["results"][0]["job_id"], JOB_ID_1) + + def test_cli_bulk_index_all_success_table(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke(["index", "bulk", "--all", "--modality", "scene"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("Bulk indexing summary", result.output) + self.assertIn("Total: 2, Indexed: 2, Skipped: 0, Failed: 0, Queued: 0.", result.output) + + def test_cli_bulk_index_forwards_options_and_flags(self): + job_1 = make_job(JOB_ID_1, state=JobState.queued, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + + result = self.invoke( + [ + "index", + "bulk", + MEDIA_ID_1, + "--modality", + "scene", + "--frame-stride", + "4", + "--scene-sample-fps", + "2.5", + "--option", + "scene.threshold=0.8", + "--detach", + "--reindex", + "--json", + ] + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 1) + self.assertEqual(payload["queued"], 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertEqual(submitted_command.frame_stride, 4) + self.assertEqual(submitted_command.scene_sample_fps, 2.5) + self.assertEqual( + submitted_command.capability_options, + {"scene": {"threshold": 0.8}}, + ) + + def test_cli_bulk_index_with_failure_exits_code_1(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.side_effect = ApplicationError( + "model_error", + ErrorCategory.unavailable, + "Failed to load model weights.", + ) + + result = self.invoke( + ["index", "bulk", MEDIA_ID_1, "--modality", "scene", "--json"] + ) + + self.assertEqual(result.exit_code, 1, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 1) + self.assertEqual(payload["failed"], 1) + self.assertEqual(payload["results"][0]["status"], "failed") + self.assertEqual(payload["results"][0]["error_code"], "model_error") + + def test_cli_bulk_index_all_json(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke(["index", "bulk", "--all", "--modality", "scene", "--json"]) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 2) + self.assertEqual(payload["indexed"], 2) + + def test_cli_bulk_index_with_failure_table_output(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.side_effect = ApplicationError( + "model_error", + ErrorCategory.unavailable, + "Failed to load model weights.", + ) + + result = self.invoke(["index", "bulk", MEDIA_ID_1, "--modality", "scene"]) + + self.assertEqual(result.exit_code, 1, result.output) + self.assertIn("Bulk indexing summary", result.output) + self.assertIn("model_error", result.output) + self.assertIn("Failed: 1", result.output) From 8282213688cf30a1a778479cafea1a202eacd152 Mon Sep 17 00:00:00 2001 From: Hasnain Ibrar Date: Tue, 8 Sep 2026 19:10:30 +0300 Subject: [PATCH 2/4] feat(index): add bulk indexing for many media items `vidxp index create` handled one media item per invocation, so indexing a repository meant driving it once per video and tracking the results by hand. Add a transport-neutral planning operation and a thin CLI adapter over it: - `Application.plan_bulk_index` resolves a selection to per-media targets and decides which ones the active snapshot already covers. The plan is read-only, so callers can show it before committing to any work. - `vidxp index bulk` indexes every registered media item, or a selection passed with repeated `--media-id`. `--plan-only` shows the decision without indexing, and `--reindex` plans covered media anyway. Media is skipped when the active snapshot holds a generation for it, that generation covers every requested modality, and its recorded input checksum still matches the registered media. Replacing a video's content or asking for a modality the generation lacks therefore plans it again. Media that is not in the ready state is reported as skipped rather than silently dropped. No new indexing behavior. Each pending target is submitted through the existing `submit_index` durable job, one job per media item, matching how `IngestionCoordinator` already sequences ingestion. That is what gives the batch its guarantees: a failure isolates to its own media, earlier successes stay committed, and rerunning the command retries only what is still missing because completed media is then skipped. The command exits non-zero when any media failed. Co-Authored-By: Claude Opus 5 (1M context) --- INSTALLATION_GUIDE.md | 15 ++ src/vidxp/application.py | 93 ++++++++++- src/vidxp/application_models.py | 79 +++++++++ src/vidxp/cli_commands/index.py | 169 +++++++++++++++++++ tests/test_bulk_index.py | 285 ++++++++++++++++++++++++++++++++ 5 files changed, 640 insertions(+), 1 deletion(-) create mode 100644 tests/test_bulk_index.py diff --git a/INSTALLATION_GUIDE.md b/INSTALLATION_GUIDE.md index a35ba255..490fd0ac 100644 --- a/INSTALLATION_GUIDE.md +++ b/INSTALLATION_GUIDE.md @@ -245,6 +245,21 @@ vidxp search speech "the bread just came out of the oven" Add `--media-id ` to a search command to restrict results to one video. Without it, VidXP searches all indexed videos in the active repository. +### Index more than one video + +Index every registered video that the active index does not already cover: + +```bash +vidxp index bulk --modality scene +``` + +Videos the active index already covers are skipped, so the command is safe to +repeat after importing more. Add `--media-id ` once per video to index +a specific selection, `--plan-only` to see what would be indexed and skipped +without indexing, and `--reindex` to index covered videos again. A video that +fails does not stop the rest; rerun the command to retry only what is still +missing. + ### Start an installed interface | Interface | Command | diff --git a/src/vidxp/application.py b/src/vidxp/application.py index afcf735b..a52651b1 100644 --- a/src/vidxp/application.py +++ b/src/vidxp/application.py @@ -32,6 +32,12 @@ PrepareModelsCommand, PrepareModelsResult, RemoveIndexCommand, + BulkIndexPlan, + BulkIndexSkipReason, + BulkIndexTarget, + BulkIndexTargetState, + ListMediaCommand, + PlanBulkIndexCommand, ResourceNotFoundError, RuntimeReadiness, QueryAnswer, @@ -56,10 +62,11 @@ from vidxp.capabilities.registry import CapabilityRegistry from vidxp.capability_service import CapabilityService from vidxp.capabilities.schemas import SearchResult +from vidxp.core.media import MediaState from vidxp.core.contracts import ( IndexConfig, ) -from vidxp.core.snapshots import IndexSnapshot +from vidxp.core.snapshots import GenerationReference, IndexSnapshot from vidxp.execution import ExecutionContext, execution_context from vidxp.ports import IndexBackend, ModelRuntimePort, QueryModelPort from vidxp.query_service import GroundedQueryService @@ -275,6 +282,90 @@ def create_index( ) return IndexResult.model_validate(result) + @application_boundary + def plan_bulk_index(self, command: PlanBulkIndexCommand) -> BulkIndexPlan: + """Decide which registered media still need indexing. + + The plan is a read-only decision. Callers submit one ordinary indexing + operation per pending target, so a failure isolates to its own media + and can be retried without disturbing the rest of the selection. + """ + + selected = self.registry.validate_names(command.modalities) + non_indexable = [ + name for name in selected if self.registry.get(name).collection_name is None + ] + if non_indexable: + raise CapabilityRequestError( + "One or more selected capabilities do not support indexing." + ) + snapshot = self._read_active_snapshot() + generations = {} if snapshot is None else snapshot.generations + requested = frozenset(selected) + targets = tuple( + self._bulk_index_target( + asset, + generation=generations.get(asset.media_id), + requested=requested, + reindex=command.reindex, + ) + for asset in self._bulk_index_selection(command.media_ids) + ) + return BulkIndexPlan(targets=targets, modalities=selected) + + def _bulk_index_selection( + self, + media_ids: tuple[str, ...], + ) -> tuple[MediaAsset, ...]: + if media_ids: + return tuple(self.get_media(media_id) for media_id in media_ids) + assets: list[MediaAsset] = [] + cursor: str | None = None + while True: + page = self.list_media( + ListMediaCommand(page_size=100, cursor=cursor) + ) + assets.extend(page.items) + cursor = page.next_cursor + if cursor is None: + break + return tuple(assets) + + @staticmethod + def _bulk_index_target( + asset: MediaAsset, + *, + generation: GenerationReference | None, + requested: frozenset[str], + reindex: bool, + ) -> BulkIndexTarget: + if asset.state != MediaState.ready: + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.skipped, + reason=BulkIndexSkipReason.media_not_ready, + ) + covered = ( + not reindex + and generation is not None + and requested <= frozenset(generation.modalities) + and generation.input_sha256 == asset.sha256 + ) + if covered and generation is not None: + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.skipped, + reason=BulkIndexSkipReason.already_indexed, + generation_id=generation.generation_id, + ) + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.pending, + ) + @application_boundary def indexing_in_progress(self) -> bool: return self.index_backend.indexing_in_progress(self._base_config()) diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index 395bde42..249dc755 100644 --- a/src/vidxp/application_models.py +++ b/src/vidxp/application_models.py @@ -642,6 +642,85 @@ class IndexResult(ApplicationModel): record_counts: dict[str, NonNegativeInt] = Field(default_factory=dict) +class BulkIndexTargetState(StrEnum): + pending = "pending" + skipped = "skipped" + + +class BulkIndexSkipReason(StrEnum): + already_indexed = "already_indexed" + media_not_ready = "media_not_ready" + + +class BulkIndexTarget(ApplicationModel): + media_id: MediaId + original_filename: str = Field(min_length=1) + state: BulkIndexTargetState + reason: BulkIndexSkipReason | None = Field( + default=None, + description="Why the media was skipped. Absent for pending targets.", + ) + generation_id: IndexGenerationId | None = Field( + default=None, + description=( + "Generation already covering this media in the active snapshot." + ), + ) + + @model_validator(mode="after") + def _reason_matches_state(self) -> "BulkIndexTarget": + if self.state == BulkIndexTargetState.skipped and self.reason is None: + raise ValueError("A skipped target requires a reason.") + if self.state == BulkIndexTargetState.pending and self.reason is not None: + raise ValueError("A pending target cannot carry a skip reason.") + return self + + +class PlanBulkIndexCommand(ApplicationModel): + media_ids: tuple[MediaId, ...] = Field( + default=(), + description=( + "Registered media to consider. Empty selects every registered " + "media item in the repository." + ), + ) + modalities: tuple[str, ...] + reindex: bool = Field( + default=False, + description=( + "Plan already-indexed media for indexing instead of skipping it." + ), + ) + + @field_validator("media_ids") + @classmethod + def _unique_media_ids(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if len(set(value)) != len(value): + raise ValueError("media_ids must not repeat a media identifier.") + return value + + +class BulkIndexPlan(ApplicationModel): + targets: tuple[BulkIndexTarget, ...] = () + modalities: tuple[str, ...] + + @property + def pending(self) -> tuple[BulkIndexTarget, ...]: + return tuple( + target + for target in self.targets + if target.state == BulkIndexTargetState.pending + ) + + @property + def skipped(self) -> tuple[BulkIndexTarget, ...]: + return tuple( + target + for target in self.targets + if target.state == BulkIndexTargetState.skipped + ) + + class RemoveIndexCommand(ApplicationModel): media_id: MediaId diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index 14eb192a..4b466070 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -7,7 +7,9 @@ from rich.table import Table from vidxp.application_models import ( + BulkIndexTargetState, CreateIndexCommand, + PlanBulkIndexCommand, RemoveIndexCommand, ) from vidxp.cli_support import ( @@ -154,6 +156,173 @@ def index_create( ) +@app.command("bulk") +def index_bulk( + ctx: typer.Context, + media_ids: Annotated[ + list[str] | None, + typer.Option( + "--media-id", + help=( + "Registered media identifier to index; repeat to select more " + "than one. Omit to select every registered media item." + ), + ), + ] = None, + modalities: Annotated[ + list[str] | None, + typer.Option( + "--modality", + "-m", + help="Modality to index; repeat to select more than one.", + ), + ] = None, + reindex: Annotated[ + bool, + typer.Option( + "--reindex", + help="Index already-indexed media instead of skipping it.", + ), + ] = False, + plan_only: Annotated[ + bool, + typer.Option( + "--plan-only", + help="Show what would be indexed and skipped without indexing.", + ), + ] = False, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Index many media items, skipping those the active snapshot covers.""" + + state = state_from_context(ctx) + indexable = tuple( + capability.name + for capability in state.service.list_capabilities() + if capability.supports_indexing + ) + selected = selected_modalities(modalities, indexable) + plan = state.service.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=tuple(media_ids or ()), + modalities=selected, + reindex=reindex, + ) + ) + output_format = effective_output_format(state, json_output) + pending = plan.pending + outcomes: list[dict] = [ + { + "media_id": target.media_id, + "original_filename": target.original_filename, + "state": target.state.value, + "reason": None if target.reason is None else target.reason.value, + "generation_id": target.generation_id, + "job_id": None, + } + for target in plan.skipped + ] + + if plan_only: + outcomes.extend( + { + "media_id": target.media_id, + "original_filename": target.original_filename, + "state": target.state.value, + "reason": None, + "generation_id": None, + "job_id": None, + } + for target in pending + ) + else: + for position, target in enumerate(pending, start=1): + if output_format == OutputFormat.rich and not state.quiet: + typer.echo( + f"[{position}/{len(pending)}] Indexing " + f"{target.original_filename} ({target.media_id})." + ) + entry = { + "media_id": target.media_id, + "original_filename": target.original_filename, + "state": "indexed", + "reason": None, + "generation_id": None, + "job_id": None, + } + try: + job = state.jobs.submit_index( + CreateIndexCommand( + media_id=target.media_id, + modalities=plan.modalities, + ) + ) + entry["job_id"] = job.job_id + completed = state.jobs.wait(job.job_id) + entry["job_id"] = completed.job_id + # One media item failing must not abandon the rest of the + # batch, so every error is recorded and the loop continues. + except Exception as exc: + entry["state"] = "failed" + entry["reason"] = str(exc) + outcomes.append(entry) + + failed = [entry for entry in outcomes if entry["state"] == "failed"] + payload = { + "planned": len(pending), + "skipped": len(plan.skipped), + "indexed": len( + [entry for entry in outcomes if entry["state"] == "indexed"] + ), + "failed": len(failed), + "plan_only": plan_only, + "modalities": list(plan.modalities), + "items": outcomes, + } + if output_format == OutputFormat.json: + emit_json(payload) + else: + table = Table( + title="Planned media" if plan_only else "Bulk indexing results" + ) + table.add_column("Filename") + table.add_column("Outcome") + table.add_column("Detail") + for target in plan.skipped: + table.add_row( + target.original_filename, + "skipped", + "" if target.reason is None else target.reason.value, + ) + if plan_only: + for target in pending: + table.add_row(target.original_filename, "would index", "") + else: + for entry in outcomes: + if entry["state"] == BulkIndexTargetState.skipped.value: + continue + table.add_row( + entry["original_filename"], + entry["state"], + entry["reason"] or entry["job_id"] or "", + ) + Console().print(table) + typer.echo( + f"Selected {len(plan.targets)} media item(s): " + f"{payload['skipped']} skipped, " + + ( + f"{payload['planned']} would be indexed." + if plan_only + else f"{payload['indexed']} indexed, {payload['failed']} failed." + ) + ) + if failed: + raise typer.Exit(code=1) + + @app.command("remove") def index_remove( ctx: typer.Context, diff --git a/tests/test_bulk_index.py b/tests/test_bulk_index.py new file mode 100644 index 00000000..f3c6633e --- /dev/null +++ b/tests/test_bulk_index.py @@ -0,0 +1,285 @@ +import unittest +from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import Mock + +from vidxp.application import VidXPApplication +from vidxp.application_models import ( + BulkIndexSkipReason, + BulkIndexTargetState, + InvalidRequestError, + MediaAsset, + MediaPage, + PlanBulkIndexCommand, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.core.media import MediaState, MediaStream +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.runtime import ModelRuntime +from vidxp.repository_layout import RepositoryLayout +from vidxp.settings import VidXPSettings + + +FIRST_MEDIA_ID = "123456781234423481234567890abcde" +SECOND_MEDIA_ID = "223456781234423481234567890abcde" +GENERATION_ID = "323456781234423481234567890abcde" +SNAPSHOT_ID = "423456781234423481234567890abcde" +FIRST_SHA256 = "a" * 64 +SECOND_SHA256 = "b" * 64 +CONFIG_FINGERPRINT = "c" * 64 +MANIFEST_SHA256 = "d" * 64 + + +def media_asset( + media_id: str, + *, + sha256: str = FIRST_SHA256, + filename: str = "clip.mp4", + state: MediaState = MediaState.ready, +) -> MediaAsset: + return MediaAsset( + media_id=media_id, + video_id=media_id, + original_filename=filename, + sha256=sha256, + byte_size=1024, + detected_mime_type="video/mp4", + container="mov,mp4,m4a,3gp,3g2,mj2", + duration_seconds=3.0, + streams=(MediaStream(index=0, kind="video", codec="h264"),), + state=state, + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + + +def generation( + media_id: str, + *, + input_sha256: str = FIRST_SHA256, + modalities: tuple[str, ...] = ("scene",), +) -> GenerationReference: + return GenerationReference( + generation_id=GENERATION_ID, + media_id=media_id, + manifest_sha256=MANIFEST_SHA256, + input_sha256=input_sha256, + config_fingerprint=CONFIG_FINGERPRINT, + modalities=modalities, + record_counts={name: 1 for name in modalities}, + store_size_bytes_at_commit=2048, + ) + + +def snapshot(*references: GenerationReference) -> IndexSnapshot: + return IndexSnapshot( + snapshot_id=SNAPSHOT_ID, + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + config_fingerprint=CONFIG_FINGERPRINT, + configuration={"enabled_modalities": ["scene"]}, + generations={ + reference.media_id: reference for reference in references + }, + ) + + +class BulkIndexPlanTests(unittest.TestCase): + def application( + self, + root: str | Path, + *, + assets: tuple[MediaAsset, ...], + active_snapshot: IndexSnapshot | None = None, + page_size: int | None = None, + ) -> VidXPApplication: + settings = VidXPSettings( + repository_root=Path(root), + runtime_backend="cpu", + ) + media_service = Mock() + by_id = {asset.media_id: asset for asset in assets} + media_service.get.side_effect = lambda media_id: by_id[media_id] + + def list_media(command): + if page_size is None: + return MediaPage(items=assets, total=len(assets)) + start = 0 if command.cursor is None else int(command.cursor) + window = assets[start : start + page_size] + following = start + page_size + return MediaPage( + items=window, + total=len(assets), + next_cursor=( + str(following) if following < len(assets) else None + ), + ) + + media_service.list.side_effect = list_media + return VidXPApplication( + settings=settings, + layout=RepositoryLayout(root=Path(root)), + registry=create_capability_registry(), + runtime=ModelRuntime(settings), + index_backend=Mock(), + media=media_service, + artifacts=Mock(), + index_status=lambda: None, + active_snapshot=lambda: active_snapshot, + ) + + def test_media_covered_by_the_active_snapshot_is_skipped(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot(generation(FIRST_MEDIA_ID)), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.skipped), 1) + self.assertEqual(plan.pending, ()) + skipped = plan.skipped[0] + self.assertEqual(skipped.reason, BulkIndexSkipReason.already_indexed) + self.assertEqual(skipped.generation_id, GENERATION_ID) + + def test_media_missing_a_requested_modality_is_planned(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot( + generation(FIRST_MEDIA_ID, modalities=("scene",)) + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene", "speech")) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_replaced_media_content_is_planned_again(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID, sha256=SECOND_SHA256),), + active_snapshot=snapshot( + generation(FIRST_MEDIA_ID, input_sha256=FIRST_SHA256) + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_reindex_plans_media_the_snapshot_already_covers(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot(generation(FIRST_MEDIA_ID)), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",), reindex=True) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_media_that_is_not_ready_is_skipped(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, state=MediaState.pending), + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.skipped), 1) + self.assertEqual( + plan.skipped[0].reason, + BulkIndexSkipReason.media_not_ready, + ) + + def test_an_empty_selection_covers_every_registered_media_item(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.targets), 2) + self.assertEqual( + [target.state for target in plan.targets], + [BulkIndexTargetState.pending] * 2, + ) + + def test_an_explicit_selection_ignores_other_media(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=(SECOND_MEDIA_ID,), + modalities=("scene",), + ) + ) + self.assertEqual(len(plan.targets), 1) + self.assertEqual(plan.targets[0].media_id, SECOND_MEDIA_ID) + + def test_every_page_of_registered_media_is_selected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + page_size=1, + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual( + [target.media_id for target in plan.targets], + [FIRST_MEDIA_ID, SECOND_MEDIA_ID], + ) + + def test_an_unknown_capability_is_rejected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + ) + with self.assertRaises(InvalidRequestError): + application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("nonexistent",)) + ) + + def test_planning_reads_no_media_when_the_selection_is_rejected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + ) + with self.assertRaises(InvalidRequestError): + application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("nonexistent",)) + ) + application.media.list.assert_not_called() + + +if __name__ == "__main__": + unittest.main() From bd162579c3a46ae5a7c97b682442e93d5f353505 Mon Sep 17 00:00:00 2001 From: Talha Amjad Date: Mon, 14 Sep 2026 16:43:11 +0500 Subject: [PATCH 3/4] fix(index): integrate bulk planning with reusable job execution --- INSTALLATION_GUIDE.md | 24 +++++-- src/vidxp/application.py | 124 ++------------------------------ src/vidxp/application_models.py | 22 +++--- src/vidxp/bulk_indexing.py | 77 ++++++++------------ src/vidxp/cli_commands/index.py | 76 +++++++++++++------- src/vidxp/control_plane.py | 124 ++++++++++++++++++++++++++++++-- tests/test_bulk_index.py | 41 ++++++++--- tests/test_bulk_indexing.py | 116 ++++++++++++++---------------- 8 files changed, 316 insertions(+), 288 deletions(-) diff --git a/INSTALLATION_GUIDE.md b/INSTALLATION_GUIDE.md index 490fd0ac..27b4f59b 100644 --- a/INSTALLATION_GUIDE.md +++ b/INSTALLATION_GUIDE.md @@ -250,15 +250,25 @@ video. Without it, VidXP searches all indexed videos in the active repository. Index every registered video that the active index does not already cover: ```bash -vidxp index bulk --modality scene +vidxp index bulk --all --modality scene ``` -Videos the active index already covers are skipped, so the command is safe to -repeat after importing more. Add `--media-id ` once per video to index -a specific selection, `--plan-only` to see what would be indexed and skipped -without indexing, and `--reindex` to index covered videos again. A video that -fails does not stop the rest; rerun the command to retry only what is still -missing. +Videos with matching content and indexing settings are skipped. To select specific +videos, pass their media IDs after `bulk` instead of `--all`. Add `--plan-only` +to preview the decisions without indexing, or `--reindex` to rebuild matching +videos. Bulk indexing accepts the same sampling and capability options as +`index create`. + +Each video uses its own indexing job. If one fails, the others continue and +successful results remain available. Rerun the same command to retry missing +results, or use `vidxp jobs retry ` with the failed job ID from the +summary. `--detach` returns after submission and reports job IDs; it does not +wait for indexing to succeed. + +The existing repository rule still applies: videos in one active index must +use the same indexing settings. Bulk indexing does not migrate an existing +multi-video index to a different profile or remove old results to make room +for one. Such jobs report the existing profile-compatibility error. ### Start an installed interface diff --git a/src/vidxp/application.py b/src/vidxp/application.py index bd68ff0e..71b40d2d 100644 --- a/src/vidxp/application.py +++ b/src/vidxp/application.py @@ -1,5 +1,7 @@ from __future__ import annotations +from dataclasses import replace + from contextlib import contextmanager from pathlib import Path from shutil import which @@ -32,12 +34,6 @@ PrepareModelsCommand, PrepareModelsResult, RemoveIndexCommand, - BulkIndexPlan, - BulkIndexSkipReason, - BulkIndexTarget, - BulkIndexTargetState, - ListMediaCommand, - PlanBulkIndexCommand, ResourceNotFoundError, RuntimeReadiness, QueryAnswer, @@ -62,11 +58,10 @@ from vidxp.capabilities.registry import CapabilityRegistry from vidxp.capability_service import CapabilityService from vidxp.capabilities.schemas import SearchResult -from vidxp.core.media import MediaState from vidxp.core.contracts import ( IndexConfig, ) -from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.core.snapshots import IndexSnapshot from vidxp.execution import ExecutionContext, execution_context from vidxp.ports import IndexBackend, ModelRuntimePort, QueryModelPort from vidxp.query_service import GroundedQueryService @@ -242,36 +237,11 @@ def create_index( execution: ExecutionContext | None = None, ) -> IndexResult: active_execution = execution_context(execution) - selected = self.registry.validate_names(command.modalities) - non_indexable = [ - name for name in selected if self.registry.get(name).collection_name is None - ] - if non_indexable: - raise CapabilityRequestError( - "One or more selected capabilities do not support indexing." - ) + config = replace(self._index_config(command, media_id=command.media_id), device=self.device) + selected = config.enabled_modalities media = self.media.require_record(command.media_id) content = self.media.content(command.media_id) self.layout.ensure_local_directories() - capability_options = { - name: dict(options) for name, options in command.capability_options.items() - } - if command.scene_sample_fps is not None: - capability_options.setdefault("scene", {})["sample_fps"] = ( - command.scene_sample_fps - ) - config = IndexConfig.local( - video_id=command.media_id, - enabled_modalities=selected, - frame_stride=command.frame_stride, - storage_directory=self.index_directory, - collection_names=self.registry.collection_names(selected), - capability_options=self.registry.validate_options( - selected, - capability_options, - ), - device=self.device, - ) with self.runtime.scheduler.indexing(): with self._capability_dependencies(selected): result = self.index_backend.create( @@ -285,90 +255,6 @@ def create_index( ) return IndexResult.model_validate(result) - @application_boundary - def plan_bulk_index(self, command: PlanBulkIndexCommand) -> BulkIndexPlan: - """Decide which registered media still need indexing. - - The plan is a read-only decision. Callers submit one ordinary indexing - operation per pending target, so a failure isolates to its own media - and can be retried without disturbing the rest of the selection. - """ - - selected = self.registry.validate_names(command.modalities) - non_indexable = [ - name for name in selected if self.registry.get(name).collection_name is None - ] - if non_indexable: - raise CapabilityRequestError( - "One or more selected capabilities do not support indexing." - ) - snapshot = self._read_active_snapshot() - generations = {} if snapshot is None else snapshot.generations - requested = frozenset(selected) - targets = tuple( - self._bulk_index_target( - asset, - generation=generations.get(asset.media_id), - requested=requested, - reindex=command.reindex, - ) - for asset in self._bulk_index_selection(command.media_ids) - ) - return BulkIndexPlan(targets=targets, modalities=selected) - - def _bulk_index_selection( - self, - media_ids: tuple[str, ...], - ) -> tuple[MediaAsset, ...]: - if media_ids: - return tuple(self.get_media(media_id) for media_id in media_ids) - assets: list[MediaAsset] = [] - cursor: str | None = None - while True: - page = self.list_media( - ListMediaCommand(page_size=100, cursor=cursor) - ) - assets.extend(page.items) - cursor = page.next_cursor - if cursor is None: - break - return tuple(assets) - - @staticmethod - def _bulk_index_target( - asset: MediaAsset, - *, - generation: GenerationReference | None, - requested: frozenset[str], - reindex: bool, - ) -> BulkIndexTarget: - if asset.state != MediaState.ready: - return BulkIndexTarget( - media_id=asset.media_id, - original_filename=asset.original_filename, - state=BulkIndexTargetState.skipped, - reason=BulkIndexSkipReason.media_not_ready, - ) - covered = ( - not reindex - and generation is not None - and requested <= frozenset(generation.modalities) - and generation.input_sha256 == asset.sha256 - ) - if covered and generation is not None: - return BulkIndexTarget( - media_id=asset.media_id, - original_filename=asset.original_filename, - state=BulkIndexTargetState.skipped, - reason=BulkIndexSkipReason.already_indexed, - generation_id=generation.generation_id, - ) - return BulkIndexTarget( - media_id=asset.media_id, - original_filename=asset.original_filename, - state=BulkIndexTargetState.pending, - ) - @application_boundary def indexing_in_progress(self) -> bool: return self.index_backend.indexing_in_progress(self._base_config()) diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index cbaa0a85..257edcf8 100644 --- a/src/vidxp/application_models.py +++ b/src/vidxp/application_models.py @@ -562,13 +562,7 @@ class MediaUploadSessionStatus(ApplicationModel): next_action: str = Field(min_length=1, max_length=1024) -class CreateIndexCommand(ApplicationModel): - media_id: MediaId = Field( - description=( - "Stable identifier returned by list_media, get_media, or a " - "completed upload." - ) - ) +class IndexOptions(ApplicationModel): modalities: tuple[str, ...] frame_stride: int = Field( default=1, @@ -628,12 +622,21 @@ def _canonicalize_scene_sampling(cls, value: Any) -> Any: return payload @model_validator(mode="after") - def _scene_sampling_requires_scene(self) -> "CreateIndexCommand": + def _scene_sampling_requires_scene(self) -> "IndexOptions": if self.scene_sample_fps is not None and "scene" not in self.modalities: raise ValueError("scene_sample_fps requires the scene modality.") return self +class CreateIndexCommand(IndexOptions): + media_id: MediaId = Field( + description=( + "Stable identifier returned by list_media, get_media, or a " + "completed upload." + ) + ) + + class IndexResult(ApplicationModel): media_id: MediaId generation_id: IndexGenerationId @@ -676,7 +679,7 @@ def _reason_matches_state(self) -> "BulkIndexTarget": return self -class PlanBulkIndexCommand(ApplicationModel): +class PlanBulkIndexCommand(IndexOptions): media_ids: tuple[MediaId, ...] = Field( default=(), description=( @@ -701,6 +704,7 @@ def _unique_media_ids(cls, value: tuple[str, ...]) -> tuple[str, ...]: class BulkIndexPlan(ApplicationModel): + options: IndexOptions targets: tuple[BulkIndexTarget, ...] = () modalities: tuple[str, ...] diff --git a/src/vidxp/bulk_indexing.py b/src/vidxp/bulk_indexing.py index 6cbe5e2b..db363fcb 100644 --- a/src/vidxp/bulk_indexing.py +++ b/src/vidxp/bulk_indexing.py @@ -8,9 +8,11 @@ CreateIndexCommand, ListMediaCommand, MediaAsset, + PlanBulkIndexCommand, + BulkIndexPlan, + BulkIndexTargetState, MediaState, ) -from vidxp.core.snapshots import IndexSnapshot if TYPE_CHECKING: from vidxp.application import VidXPApplication @@ -58,21 +60,6 @@ def _resolve_all_media( return media_list -def _is_already_indexed( - snapshot: IndexSnapshot | None, - media_id: str, - requested_modalities: Sequence[str] | None = None, -) -> bool: - if snapshot is None: - return False - generation = snapshot.generations.get(media_id) - if generation is None: - return False - if requested_modalities is not None: - return set(requested_modalities).issubset(set(generation.modalities)) - return True - - def run_bulk_index( application: VidXPApplication | ControlPlaneApplication, jobs: JobService, @@ -81,6 +68,7 @@ def run_bulk_index( all_eligible: bool = False, skip_indexed: bool = True, detach: bool = False, + plan: BulkIndexPlan | None = None, modalities: Sequence[str] | None = None, frame_stride: int = 1, scene_sample_fps: float | None = None, @@ -89,41 +77,33 @@ def run_bulk_index( on_item_progress: Callable[[str, Any], None] | None = None, on_item_complete: Callable[[BulkIndexItemResult], None] | None = None, ) -> BulkIndexSummary: - items: list[tuple[str, str]] = [] - if all_eligible: - all_media = _resolve_all_media(application) - items = [(asset.media_id, asset.original_filename) for asset in all_media] - elif media_ids: - for mid in media_ids: - asset = application.get_media(mid) - items.append((asset.media_id, asset.original_filename)) - - read_snapshot = getattr(application, "_read_active_snapshot", None) - snapshot: IndexSnapshot | None = ( - read_snapshot() if callable(read_snapshot) else None - ) - - if modalities is not None: - cmd_modalities = tuple(modalities) - elif hasattr(application, "select_index_modalities"): - cmd_modalities = application.select_index_modalities(None) - elif hasattr(application, "list_capabilities"): - cmd_modalities = tuple( - c.name - for c in application.list_capabilities() - if getattr(c, "supports_indexing", True) + if plan is None: + if bool(media_ids) == all_eligible: + raise ValueError("Provide media IDs or all_eligible, not both.") + selected = application.select_index_modalities( + tuple(modalities) if modalities is not None else None + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=tuple(media_ids or ()), + modalities=selected, + reindex=not skip_indexed, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=capability_options or {}, + ) ) - else: - cmd_modalities = () results: list[BulkIndexItemResult] = [] - for media_id, filename in items: - if skip_indexed and _is_already_indexed(snapshot, media_id, modalities): + for target in plan.targets: + media_id, filename = target.media_id, target.original_filename + if target.state == BulkIndexTargetState.skipped: item_result = BulkIndexItemResult( media_id=media_id, filename=filename, status="skipped", + error_message=target.reason.value if target.reason else None, ) results.append(item_result) if on_item_complete is not None: @@ -134,15 +114,13 @@ def run_bulk_index( on_item_start(media_id, filename) command = CreateIndexCommand( - media_id=media_id, - modalities=cmd_modalities, - frame_stride=frame_stride, - scene_sample_fps=scene_sample_fps, - capability_options=capability_options or {}, + media_id=media_id, **plan.options.model_dump(mode="python") ) + job_id = None try: job = jobs.submit_index(command) + job_id = job.job_id if detach: item_result = BulkIndexItemResult( media_id=media_id, @@ -154,6 +132,7 @@ def run_bulk_index( if on_item_complete is not None: on_item_complete(item_result) else: + def _progress(current: Any) -> None: if on_item_progress is not None: on_item_progress(media_id, current) @@ -173,6 +152,7 @@ def _progress(current: Any) -> None: media_id=media_id, filename=filename, status="failed", + job_id=job_id, error_code=exc.code, error_message=str(exc), ) @@ -184,6 +164,7 @@ def _progress(current: Any) -> None: media_id=media_id, filename=filename, status="failed", + job_id=job_id, error_code="unexpected_error", error_message=str(exc), ) diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index af23423d..d51a1593 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -9,6 +9,7 @@ from vidxp.application_models import ( CreateIndexCommand, + PlanBulkIndexCommand, RemoveIndexCommand, ) from vidxp.bulk_indexing import run_bulk_index @@ -38,9 +39,7 @@ def create_index( capability_options: dict[str, dict], detach: bool = False, ) -> dict: - show_progress = ( - not state.quiet and state.output_format == OutputFormat.rich - ) + show_progress = not state.quiet and state.output_format == OutputFormat.rich selected = tuple(modalities) with IndexProgress(show_progress) as progress: job = state.jobs.submit_index( @@ -56,9 +55,7 @@ def create_index( job = state.jobs.wait( job.job_id, progress=lambda current: ( - progress.update( - current.progress.model_dump(mode="python") - ) + progress.update(current.progress.model_dump(mode="python")) if current.progress is not None else None ), @@ -99,10 +96,7 @@ def index_create( typer.Option( "--frame-stride", min=1, - help=( - "Materialize every Nth frame for actor and legacy visual " - "indexing." - ), + help=("Materialize every Nth frame for actor and legacy visual indexing."), ), ] = 1, scene_sample_fps: Annotated[ @@ -172,6 +166,9 @@ def index_bulk( help="Index all eligible registered media in the catalog.", ), ] = False, + plan_only: Annotated[ + bool, typer.Option("--plan-only", help="Preview indexing and skip decisions.") + ] = False, reindex: Annotated[ bool, typer.Option( @@ -192,10 +189,7 @@ def index_bulk( typer.Option( "--frame-stride", min=1, - help=( - "Materialize every Nth frame for actor and legacy visual " - "indexing." - ), + help=("Materialize every Nth frame for actor and legacy visual indexing."), ), ] = 1, scene_sample_fps: Annotated[ @@ -249,13 +243,48 @@ def index_bulk( output_fmt = effective_output_format(state, json_output) show_progress = not state.quiet and output_fmt == OutputFormat.rich + if plan_only and detach: + raise typer.BadParameter("--plan-only cannot be combined with --detach.") + plan = state.service.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=tuple(media_ids or ()), + modalities=selected, + reindex=reindex, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=parsed_options, + ) + ) + if plan_only: + if output_fmt == OutputFormat.json: + emit_json(plan.model_dump(mode="json")) + else: + for target in plan.targets: + detail = target.reason.value if target.reason else "would index" + typer.echo(f"{target.original_filename}: {detail}") + typer.echo(f"{len(plan.pending)} to index, {len(plan.skipped)} skipped.") + return + + completed = 0 + + def on_item_complete(result) -> None: + nonlocal completed + completed += 1 + if show_progress: + typer.echo( + f"[{completed}/{len(plan.targets)}] {result.filename}: {result.status}" + ) + with IndexProgress(show_progress) as progress: + def on_item_start(media_id: str, filename: str) -> None: if show_progress: - progress.update({ - "stage": "indexing", - "message": f"Indexing {filename} ({media_id[:8]}...)", - }) + progress.update( + { + "stage": "indexing", + "message": f"Indexing {filename} ({media_id[:8]}...)", + } + ) def on_item_progress(media_id: str, current: Any) -> None: if show_progress: @@ -266,6 +295,8 @@ def on_item_progress(media_id: str, current: Any) -> None: summary = run_bulk_index( application=state.service, + plan=plan, + on_item_complete=on_item_complete, jobs=state.jobs, media_ids=media_ids, all_eligible=all_eligible, @@ -345,9 +376,7 @@ def index_remove( """Remove one media item from the active snapshot.""" state = state_from_context(ctx) - removed = state.service.remove_from_index( - RemoveIndexCommand(media_id=media_id) - ) + removed = state.service.remove_from_index(RemoveIndexCommand(media_id=media_id)) payload = {"removed": removed, "media_id": media_id} if effective_output_format(state, json_output) == OutputFormat.json: emit_json(payload) @@ -392,10 +421,7 @@ def index_list( assets = ( () if summary is None - else tuple( - state.service.get_media(media_id) - for media_id in summary.media_ids - ) + else tuple(state.service.get_media(media_id) for media_id in summary.media_ids) ) payload = { "state": status.state, diff --git a/src/vidxp/control_plane.py b/src/vidxp/control_plane.py index 7b84a08a..bc86329c 100644 --- a/src/vidxp/control_plane.py +++ b/src/vidxp/control_plane.py @@ -6,6 +6,12 @@ from vidxp.application_boundary import application_boundary from vidxp.application_models import ( Artifact, + IndexOptions, + PlanBulkIndexCommand, + BulkIndexPlan, + BulkIndexTarget, + BulkIndexTargetState, + BulkIndexSkipReason, CapabilityInfo, CapabilityRole, CapabilitySummary, @@ -30,7 +36,9 @@ from vidxp.capabilities.contracts import CapabilityRequestError from vidxp.capability_service import CapabilityService from vidxp.core.media import QuarantinedMedia -from vidxp.core.snapshots import IndexSnapshot +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.core.contracts import IndexConfig +from vidxp.core.media import MediaState from vidxp.index_state import INDEX_STATUS_SCHEMA from vidxp.media_service import MediaService from vidxp.ports import LocalFileResource @@ -59,6 +67,112 @@ def __init__( self._read_active_snapshot = active_snapshot or (lambda: None) self.model_cache = model_cache + def _index_config( + self, command: IndexOptions, *, media_id: str | None = None + ) -> IndexConfig: + selected = self.select_index_modalities(command.modalities) + registry = self.capabilities.registry + options = { + name: dict(values) for name, values in command.capability_options.items() + } + if command.scene_sample_fps is not None: + options.setdefault("scene", {})["sample_fps"] = command.scene_sample_fps + return IndexConfig.local( + video_id=media_id, + enabled_modalities=selected, + frame_stride=command.frame_stride, + storage_directory=self.layout.indexes, + collection_names=registry.collection_names(selected), + capability_options=registry.validate_options(selected, options), + ) + + @application_boundary + def plan_bulk_index(self, command: PlanBulkIndexCommand) -> BulkIndexPlan: + """Decide which registered media still need indexing. + + The plan is a read-only decision. Callers submit one ordinary indexing + operation per pending target, so a failure isolates to its own media + and can be retried without disturbing the rest of the selection. + """ + + config = self._index_config(command) + selected = config.enabled_modalities + snapshot = self._read_active_snapshot() + generations = {} if snapshot is None else snapshot.generations + requested = frozenset(selected) + targets = tuple( + self._bulk_index_target( + asset, + generation=generations.get(asset.media_id), + requested=requested, + reindex=command.reindex, + config_fingerprint=config.fingerprint(), + ) + for asset in self._bulk_index_selection(command.media_ids) + ) + return BulkIndexPlan( + targets=targets, + modalities=selected, + options=IndexOptions( + modalities=selected, + frame_stride=config.frame_stride, + capability_options=config.capability_options, + ), + ) + + def _bulk_index_selection( + self, + media_ids: tuple[str, ...], + ) -> tuple[MediaAsset, ...]: + if media_ids: + return tuple(self.get_media(media_id) for media_id in media_ids) + assets: list[MediaAsset] = [] + cursor: str | None = None + while True: + page = self.list_media(ListMediaCommand(page_size=100, cursor=cursor)) + assets.extend(page.items) + cursor = page.next_cursor + if cursor is None: + break + return tuple(assets) + + @staticmethod + def _bulk_index_target( + asset: MediaAsset, + *, + generation: GenerationReference | None, + requested: frozenset[str], + reindex: bool, + config_fingerprint: str, + ) -> BulkIndexTarget: + if asset.state != MediaState.ready: + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.skipped, + reason=BulkIndexSkipReason.media_not_ready, + ) + covered = ( + not reindex + and generation is not None + and requested <= frozenset(generation.modalities) + and generation.input_sha256 == asset.sha256 + and generation.config_fingerprint == config_fingerprint + ) + if covered and generation is not None: + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.skipped, + reason=BulkIndexSkipReason.already_indexed, + generation_id=generation.generation_id, + ) + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.pending, + ) + @application_boundary def import_uploaded_media( self, @@ -99,13 +213,9 @@ def select_index_modalities( registry = self.capabilities.registry indexable = registry.index_names() selected = ( - indexable - if requested is None - else registry.validate_names(requested) - ) - unsupported = tuple( - name for name in selected if name not in indexable + indexable if requested is None else registry.validate_names(requested) ) + unsupported = tuple(name for name in selected if name not in indexable) if unsupported: raise CapabilityRequestError( "Indexing does not support these capabilities: " diff --git a/tests/test_bulk_index.py b/tests/test_bulk_index.py index f3c6633e..ea6fce94 100644 --- a/tests/test_bulk_index.py +++ b/tests/test_bulk_index.py @@ -14,6 +14,7 @@ PlanBulkIndexCommand, ) from vidxp.capabilities.registry import create_capability_registry +from vidxp.core.contracts import IndexConfig from vidxp.core.media import MediaState, MediaStream from vidxp.core.snapshots import GenerationReference, IndexSnapshot from vidxp.runtime import ModelRuntime @@ -27,7 +28,12 @@ SNAPSHOT_ID = "423456781234423481234567890abcde" FIRST_SHA256 = "a" * 64 SECOND_SHA256 = "b" * 64 -CONFIG_FINGERPRINT = "c" * 64 +_registry = create_capability_registry() +CONFIG_FINGERPRINT = IndexConfig.local( + enabled_modalities=("scene",), + collection_names=_registry.collection_names(("scene",)), + capability_options=_registry.validate_options(("scene",), {}), +).fingerprint() MANIFEST_SHA256 = "d" * 64 @@ -77,9 +83,7 @@ def snapshot(*references: GenerationReference) -> IndexSnapshot: created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), config_fingerprint=CONFIG_FINGERPRINT, configuration={"enabled_modalities": ["scene"]}, - generations={ - reference.media_id: reference for reference in references - }, + generations={reference.media_id: reference for reference in references}, ) @@ -109,9 +113,7 @@ def list_media(command): return MediaPage( items=window, total=len(assets), - next_cursor=( - str(following) if following < len(assets) else None - ), + next_cursor=(str(following) if following < len(assets) else None), ) media_service.list.side_effect = list_media @@ -143,6 +145,27 @@ def test_media_covered_by_the_active_snapshot_is_skipped(self): self.assertEqual(skipped.reason, BulkIndexSkipReason.already_indexed) self.assertEqual(skipped.generation_id, GENERATION_ID) + def test_changed_configuration_is_planned_and_matches_execution_options(self): + application = self.application( + "unused", + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot(generation(FIRST_MEDIA_ID)), + ) + for options in ( + {"frame_stride": 3}, + {"scene_sample_fps": 2}, + {"capability_options": {"scene": {"batch_size": 8}}}, + ): + with self.subTest(options=options): + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",), **options) + ) + self.assertEqual(len(plan.pending), 1) + self.assertNotEqual( + application._index_config(plan.options).fingerprint(), + CONFIG_FINGERPRINT, + ) + def test_media_missing_a_requested_modality_is_planned(self): with TemporaryDirectory() as root: application = self.application( @@ -190,9 +213,7 @@ def test_media_that_is_not_ready_is_skipped(self): with TemporaryDirectory() as root: application = self.application( root, - assets=( - media_asset(FIRST_MEDIA_ID, state=MediaState.pending), - ), + assets=(media_asset(FIRST_MEDIA_ID, state=MediaState.pending),), ) plan = application.plan_bulk_index( PlanBulkIndexCommand(modalities=("scene",)) diff --git a/tests/test_bulk_indexing.py b/tests/test_bulk_indexing.py index 6a6ba2e6..84a896f8 100644 --- a/tests/test_bulk_indexing.py +++ b/tests/test_bulk_indexing.py @@ -5,6 +5,9 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any +from types import MethodType +from vidxp.control_plane import ControlPlaneApplication +from vidxp.core.contracts import IndexConfig from unittest.mock import Mock, patch from typer.testing import CliRunner @@ -31,7 +34,6 @@ from vidxp.bulk_indexing import ( BulkIndexItemResult, BulkIndexSummary, - _is_already_indexed, _resolve_all_media, run_bulk_index, ) @@ -114,8 +116,16 @@ def make_snapshot(generations: dict[str, tuple[str, ...]]) -> IndexSnapshot: generation_id=GENERATION_ID, media_id=mid, manifest_sha256="1" * 64, - input_sha256="2" * 64, - config_fingerprint="3" * 64, + input_sha256="1" * 64, + config_fingerprint=IndexConfig.local( + enabled_modalities=modalities, + collection_names=create_capability_registry().collection_names( + modalities + ), + capability_options=create_capability_registry().validate_options( + modalities, {} + ), + ).fingerprint(), modalities=modalities, record_counts={m: 1 for m in modalities}, store_size_bytes_at_commit=1024, @@ -163,42 +173,31 @@ def test_resolve_all_media_paginates_cursor(self): def test_resolve_all_media_empty_catalog(self): app = Mock() - app.list_media.return_value = MediaPage( - items=(), total=0, next_cursor=None - ) + app.list_media.return_value = MediaPage(items=(), total=0, next_cursor=None) result = _resolve_all_media(app) self.assertEqual(result, []) self.assertEqual(app.list_media.call_count, 1) - def test_is_already_indexed_snapshot_none(self): - self.assertFalse(_is_already_indexed(None, MEDIA_ID_1)) - - def test_is_already_indexed_media_not_in_generations(self): - snapshot = make_snapshot({MEDIA_ID_2: ("scene",)}) - self.assertFalse(_is_already_indexed(snapshot, MEDIA_ID_1)) - - def test_is_already_indexed_no_modalities_requested(self): - snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) - self.assertTrue(_is_already_indexed(snapshot, MEDIA_ID_1, None)) - - def test_is_already_indexed_matching_modalities_subset(self): - snapshot = make_snapshot({MEDIA_ID_1: ("scene", "speech", "actor")}) - self.assertTrue( - _is_already_indexed(snapshot, MEDIA_ID_1, ("scene", "speech")) - ) - def test_is_already_indexed_missing_requested_modality(self): - snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) - self.assertFalse( - _is_already_indexed(snapshot, MEDIA_ID_1, ("scene", "speech")) - ) +def bind_planner(app): + app.capabilities = CapabilityService(create_capability_registry()) + app.layout.indexes = Path("unused/indexes") + for name in ( + "select_index_modalities", + "_index_config", + "plan_bulk_index", + "_bulk_index_selection", + ): + setattr(app, name, MethodType(getattr(ControlPlaneApplication, name), app)) + app._bulk_index_target = ControlPlaneApplication._bulk_index_target class RunBulkIndexTests(unittest.TestCase): def setUp(self): self.app = Mock() + bind_planner(self.app) self.jobs = Mock() self.media_1 = make_media(MEDIA_ID_1, "one.mp4") self.media_2 = make_media(MEDIA_ID_2, "two.mp4") @@ -376,11 +375,10 @@ def test_bulk_index_error_resilience(self): self.assertEqual(summary.total, 2) self.assertEqual(summary.indexed, 1) self.assertEqual(summary.failed, 1) + self.assertEqual(summary.results[0].job_id, JOB_ID_1) self.assertEqual(summary.skipped, 0) self.assertEqual(summary.results[0].status, "failed") - self.assertEqual( - summary.results[0].error_code, "transcription_failed" - ) + self.assertEqual(summary.results[0].error_code, "transcription_failed") self.assertEqual( summary.results[0].error_message, "Model crashed during transcription.", @@ -404,7 +402,6 @@ def test_bulk_index_generic_exception_resilience(self): self.assertEqual(summary.results[0].error_code, "unexpected_error") self.assertIn("Disk IO error", summary.results[0].error_message or "") - def test_bulk_index_forwards_options_and_command_fields(self): job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) self.jobs.submit_index.return_value = job_1 @@ -426,7 +423,7 @@ def fake_wait(job_id, progress=None): modalities=["scene"], frame_stride=3, scene_sample_fps=1.5, - capability_options={"scene": {"threshold": 0.7}}, + capability_options={"scene": {"batch_size": 8}}, on_item_progress=lambda mid, curr: progress_events.append((mid, curr)), ) @@ -440,7 +437,7 @@ def fake_wait(job_id, progress=None): self.assertEqual(submitted_command.scene_sample_fps, 1.5) self.assertEqual( submitted_command.capability_options, - {"scene": {"threshold": 0.7}}, + {"scene": {"batch_size": 8}}, ) self.assertEqual(len(progress_events), 1) self.assertEqual(progress_events[0][0], MEDIA_ID_1) @@ -449,7 +446,9 @@ def test_bulk_index_default_modalities_resolves_from_application(self): job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) self.jobs.submit_index.return_value = job_1 self.jobs.wait.return_value = job_1 - self.app.select_index_modalities.return_value = ("scene", "speech") + self.app.select_index_modalities = lambda requested: ( + requested or ("scene", "speech") + ) summary = run_bulk_index( application=self.app, @@ -463,35 +462,12 @@ def test_bulk_index_default_modalities_resolves_from_application(self): submitted_command = self.jobs.submit_index.call_args.args[0] self.assertEqual(submitted_command.modalities, ("scene", "speech")) - def test_bulk_index_default_modalities_resolves_from_list_capabilities(self): - job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) - self.jobs.submit_index.return_value = job_1 - self.jobs.wait.return_value = job_1 - del self.app.select_index_modalities - cap1 = Mock() - cap1.name = "scene" - cap1.supports_indexing = True - cap2 = Mock() - cap2.name = "summary" - cap2.supports_indexing = False - self.app.list_capabilities.return_value = [cap1, cap2] - - summary = run_bulk_index( - application=self.app, - jobs=self.jobs, - media_ids=[MEDIA_ID_1], - modalities=None, - ) - - self.assertEqual(summary.total, 1) - submitted_command = self.jobs.submit_index.call_args.args[0] - self.assertEqual(submitted_command.modalities, ("scene",)) - def test_bulk_index_empty_items(self): + self.app.list_media.return_value = MediaPage(items=(), total=0) summary = run_bulk_index( application=self.app, jobs=self.jobs, - media_ids=[], + all_eligible=True, modalities=["scene"], ) self.assertEqual(summary.total, 0) @@ -506,6 +482,7 @@ class CliBulkIndexTests(unittest.TestCase): def setUp(self): self.runner = CliRunner() self.service = Mock() + bind_planner(self.service) self.service.registry = create_capability_registry() self.service.list_capabilities.return_value = CapabilityService( self.service.registry @@ -561,6 +538,15 @@ def invoke(self, arguments, *, media_runtime_initialized=True): self.create_local_application = create_local_application return result + def test_cli_preview_does_not_submit_jobs(self): + result = self.invoke( + ["index", "bulk", "--all", "--modality", "scene", "--plan-only", "--json"] + ) + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(len(payload["targets"]), 2) + self.jobs.submit_index.assert_not_called() + def test_cli_requires_media_ids_or_all(self): result = self.invoke(["index", "bulk"]) self.assertEqual(result.exit_code, 2, result.output) @@ -610,7 +596,9 @@ def test_cli_bulk_index_all_success_table(self): self.assertEqual(result.exit_code, 0, result.output) self.assertIn("Bulk indexing summary", result.output) - self.assertIn("Total: 2, Indexed: 2, Skipped: 0, Failed: 0, Queued: 0.", result.output) + self.assertIn( + "Total: 2, Indexed: 2, Skipped: 0, Failed: 0, Queued: 0.", result.output + ) def test_cli_bulk_index_forwards_options_and_flags(self): job_1 = make_job(JOB_ID_1, state=JobState.queued, media_id=MEDIA_ID_1) @@ -628,7 +616,7 @@ def test_cli_bulk_index_forwards_options_and_flags(self): "--scene-sample-fps", "2.5", "--option", - "scene.threshold=0.8", + "scene.batch_size=8", "--detach", "--reindex", "--json", @@ -644,7 +632,7 @@ def test_cli_bulk_index_forwards_options_and_flags(self): self.assertEqual(submitted_command.scene_sample_fps, 2.5) self.assertEqual( submitted_command.capability_options, - {"scene": {"threshold": 0.8}}, + {"scene": {"batch_size": 8}}, ) def test_cli_bulk_index_with_failure_exits_code_1(self): @@ -673,7 +661,9 @@ def test_cli_bulk_index_all_json(self): self.jobs.submit_index.side_effect = [job_1, job_2] self.jobs.wait.side_effect = [job_1, job_2] - result = self.invoke(["index", "bulk", "--all", "--modality", "scene", "--json"]) + result = self.invoke( + ["index", "bulk", "--all", "--modality", "scene", "--json"] + ) self.assertEqual(result.exit_code, 0, result.output) payload = json.loads(result.output) From ed259e0015130f27303982820655f103f4ba94da Mon Sep 17 00:00:00 2001 From: Talha Amjad Date: Mon, 14 Sep 2026 16:55:34 +0500 Subject: [PATCH 4/4] test(cli): ignore ANSI styling in bulk error assertions --- tests/test_bulk_indexing.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/test_bulk_indexing.py b/tests/test_bulk_indexing.py index 84a896f8..c7ec5f37 100644 --- a/tests/test_bulk_indexing.py +++ b/tests/test_bulk_indexing.py @@ -10,6 +10,7 @@ from vidxp.core.contracts import IndexConfig from unittest.mock import Mock, patch +from click import unstyle from typer.testing import CliRunner from vidxp import cli @@ -550,12 +551,12 @@ def test_cli_preview_does_not_submit_jobs(self): def test_cli_requires_media_ids_or_all(self): result = self.invoke(["index", "bulk"]) self.assertEqual(result.exit_code, 2, result.output) - self.assertIn("Provide either media IDs or pass --all.", result.output) + self.assertIn("Provide either media IDs or pass --all.", unstyle(result.output)) def test_cli_rejects_both_media_ids_and_all(self): result = self.invoke(["index", "bulk", MEDIA_ID_1, "--all"]) self.assertEqual(result.exit_code, 2, result.output) - self.assertIn("Pass media IDs or --all, not both.", result.output) + self.assertIn("Pass media IDs or --all, not both.", unstyle(result.output)) def test_cli_bulk_index_success_json(self): job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1)