feat(index): integrate bulk planning and indexing

INSTALLATION_GUIDE.md

@@ -245,6 +245,31 @@ vidxp search speech "the bread just came out of the oven"
 Add `--media-id <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 --all --modality scene
+```
+
+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 <job-id>` 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
 
 | Interface | Command |

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
@@ -235,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(

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
@@ -642,6 +645,86 @@ 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(IndexOptions):
+    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):
+    options: IndexOptions
+    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
 

src/vidxp/bulk_indexing.py

@@ -0,0 +1,182 @@
+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,
+    PlanBulkIndexCommand,
+    BulkIndexPlan,
+    BulkIndexTargetState,
+    MediaState,
+)
+
+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 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,
+    plan: BulkIndexPlan | None = None,
+    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:
+    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 {},
+            )
+        )
+
+    results: list[BulkIndexItemResult] = []
+
+    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:
+                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, **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,
+                    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",
+                job_id=job_id,
+                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",
+                job_id=job_id,
+                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),
+    )