feat(index): add bulk indexing for many media items

INSTALLATION_GUIDE.md

@@ -245,6 +245,21 @@ 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 --modality scene
+```
+
+Videos the active index already covers are skipped, so the command is safe to
+repeat after importing more. Add `--media-id <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 |

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())

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