diff --git a/README.md b/README.md index 401136f..f314c36 100644 --- a/README.md +++ b/README.md @@ -171,6 +171,12 @@ scenes, ask questions about a library, and create clips or actor overlays. Local agents can connect over stdio; remote agents can connect to a self-hosted VidXP server. +Agents can call `get_workspace` before acting to inspect registered media, +active-index coverage, model readiness, and the searchable, queryable, +inspectable, or renderable roles available for each video. Invalid capability +or media selections are rejected before a durable job is queued and include an +actionable next step. + - [Python, HTTP, and MCP installation](INSTALLATION_GUIDE.md) - [Optional capability packages](INSTALLATION_GUIDE.md#optional-dependency-extras) - [Coolify server setup](docs/deployment/coolify.md) diff --git a/src/vidxp/api_routes/jobs.py b/src/vidxp/api_routes/jobs.py index 72e85c6..22cc186 100644 --- a/src/vidxp/api_routes/jobs.py +++ b/src/vidxp/api_routes/jobs.py @@ -49,7 +49,6 @@ def submit_index( actor: Annotated[Principal, Depends(write_principal)], idempotency_key: HttpIdempotencyKey, ) -> Job: - service.application.require_models(command.modalities) return accepted( response, service.jobs.submit_index( diff --git a/src/vidxp/api_routes/platform.py b/src/vidxp/api_routes/platform.py index 59fc8fa..1e3be99 100644 --- a/src/vidxp/api_routes/platform.py +++ b/src/vidxp/api_routes/platform.py @@ -1,12 +1,14 @@ from typing import Annotated -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, Query from vidxp.api_routes.dependencies import context, read_principal from vidxp.application_models import ( CapabilityInfo, CapabilityList, + ListMediaCommand, RuntimeReadiness, + WorkspaceOverview, ) from vidxp.composition import HttpApplicationContext @@ -17,6 +19,25 @@ ) +@router.get( + "/workspace", + response_model=WorkspaceOverview, + operation_id="getWorkspace", + summary="Inspect media and usable capability roles", +) +def workspace( + service: Annotated[HttpApplicationContext, Depends(context)], + page_size: Annotated[int, Query(gt=0, le=100)] = 50, + cursor: Annotated[ + str | None, + Query(min_length=1, max_length=512), + ] = None, +) -> WorkspaceOverview: + return service.application.workspace( + ListMediaCommand(page_size=page_size, cursor=cursor) + ) + + @router.get( "/runtime/readiness", response_model=RuntimeReadiness, diff --git a/src/vidxp/application.py b/src/vidxp/application.py index dd25eaf..56bbb1e 100644 --- a/src/vidxp/application.py +++ b/src/vidxp/application.py @@ -56,6 +56,7 @@ from vidxp.core.contracts import ( IndexConfig, ) +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 @@ -89,6 +90,7 @@ def __init__( media: MediaService, artifacts: ArtifactService, index_status: Callable[[], dict[str, Any] | None], + active_snapshot: Callable[[], IndexSnapshot | None] | None = None, completed_upload_importer: Callable[[str], MediaAsset] | None = None, query_model: QueryModelPort | None = None, ) -> None: @@ -104,6 +106,7 @@ def __init__( artifacts=artifacts, index_status=index_status, model_cache=settings.model_cache, + active_snapshot=active_snapshot, ) self.settings = settings diff --git a/src/vidxp/application_boundary.py b/src/vidxp/application_boundary.py index 33caa4a..51c0ea1 100644 --- a/src/vidxp/application_boundary.py +++ b/src/vidxp/application_boundary.py @@ -43,6 +43,8 @@ MediaIdempotencyConflictError, MediaImportNotAllowedError, ) + + def _validation_details( exc: ValidationError, ) -> list[dict[str, JsonValue]]: @@ -113,7 +115,13 @@ def wrapped(*args: Any, **kwargs: Any) -> Any: errors=_validation_details(exc), ) from exc except CapabilityRequestError as exc: - raise InvalidRequestError() from exc + error = { + "type": "capability_request", + "location": [exc.field], + "message": str(exc), + **exc.details(), + } + raise InvalidRequestError(errors=[error]) from exc except InvalidMediaError as exc: raise ApplicationError( "media_invalid", diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index 807a0a0..be6ab3e 100644 --- a/src/vidxp/application_models.py +++ b/src/vidxp/application_models.py @@ -291,12 +291,29 @@ class CapabilityOperationInfo(ApplicationModel): output_schema: dict[str, JsonValue] +class CapabilityRole(StrEnum): + searchable = "searchable" + queryable = "queryable" + inspectable = "inspectable" + renderable = "renderable" + + +class CapabilityIdentityMode(StrEnum): + not_applicable = "not_applicable" + anonymous_clusters = "anonymous_clusters" + registered_entities = "registered_entities" + + class CapabilitySummary(ApplicationModel): name: str = Field(min_length=1) description: str = Field(min_length=1) install_extra: str = Field(min_length=1) supports_indexing: bool prepares_models: bool + roles: tuple[CapabilityRole, ...] = () + identity_mode: CapabilityIdentityMode = ( + CapabilityIdentityMode.not_applicable + ) provenance: CapabilityProvenance | None = None @@ -588,6 +605,47 @@ def _validate_media_id_window(self) -> "IndexStatusSummary": return self +class WorkspaceCapability(CapabilitySummary): + models_ready: bool | None = Field( + default=None, + description=( + "Whether required model artifacts are prepared. Null means the " + "capability does not prepare model artifacts." + ), + ) + + +class WorkspaceMediaCapability(ApplicationModel): + name: Identifier + indexed: bool + record_count: NonNegativeInt | None = None + roles: tuple[CapabilityRole, ...] = Field( + default=(), + description="Capability roles currently usable for this media item.", + ) + identity_mode: CapabilityIdentityMode = ( + CapabilityIdentityMode.not_applicable + ) + + +class WorkspaceMedia(ApplicationModel): + media_id: MediaId + original_filename: str = Field(min_length=1) + duration_seconds: float = Field(gt=0) + state: MediaState + in_active_snapshot: bool + capabilities: tuple[WorkspaceMediaCapability, ...] = () + + +class WorkspaceOverview(ApplicationModel): + capabilities: tuple[WorkspaceCapability, ...] = () + media: tuple[WorkspaceMedia, ...] = () + media_total: NonNegativeInt + next_cursor: str | None = None + index: IndexStatus + next_actions: tuple[str, ...] = () + + class FusionProfile(StrEnum): reciprocal_rank = "rrf_v1" diff --git a/src/vidxp/capabilities/actor/definition.py b/src/vidxp/capabilities/actor/definition.py index f8e1e1c..c9ba378 100644 --- a/src/vidxp/capabilities/actor/definition.py +++ b/src/vidxp/capabilities/actor/definition.py @@ -2,6 +2,7 @@ from typing import Any, Mapping +from vidxp.application_models import CapabilityIdentityMode, CapabilityRole from vidxp.capabilities.actor.config import ActorConfig, actor_config from vidxp.capabilities.actor.indexing import VISUAL_PROCESSOR from vidxp.capabilities.actor.models import ( @@ -77,6 +78,12 @@ def model_manifest( index_stage="visual_indexing", execution_group="visual", prepares_models=True, + roles=( + CapabilityRole.queryable, + CapabilityRole.inspectable, + CapabilityRole.renderable, + ), + identity_mode=CapabilityIdentityMode.anonymous_clusters, model_specs=(YUNET_MODEL, SFACE_MODEL), operations={ "cluster": OperationDefinition( diff --git a/src/vidxp/capabilities/contracts.py b/src/vidxp/capabilities/contracts.py index 9448077..ddb9ef4 100644 --- a/src/vidxp/capabilities/contracts.py +++ b/src/vidxp/capabilities/contracts.py @@ -18,7 +18,11 @@ from vidxp.core.indexing_common import ProgressCallback from vidxp.model_contracts import ArtifactSpec, ModelSpec from vidxp.ports import IndexReader, ModelRuntimePort -from vidxp.application_models import CapabilityProvenance +from vidxp.application_models import ( + CapabilityIdentityMode, + CapabilityProvenance, + CapabilityRole, +) CAPABILITY_CONTRACT_VERSION = 1 @@ -224,6 +228,10 @@ class CapabilityDefinition(_ContractModel): index_stage: str | None = None execution_group: str | None = None operations: Mapping[str, OperationDefinition] = Field(default_factory=dict) + roles: tuple[CapabilityRole, ...] = () + identity_mode: CapabilityIdentityMode = ( + CapabilityIdentityMode.not_applicable + ) model_specs: tuple[ModelSpec | ArtifactSpec, ...] = () prepares_models: bool = False @@ -250,6 +258,16 @@ def _freeze_operations( ) -> Mapping[str, OperationDefinition]: return MappingProxyType(dict(value)) + @field_validator("roles") + @classmethod + def _unique_roles( + cls, + value: tuple[CapabilityRole, ...], + ) -> tuple[CapabilityRole, ...]: + if len(value) != len(set(value)): + raise ValueError("Capability roles must be unique.") + return value + @model_validator(mode="after") def _require_complete_metadata(self) -> CapabilityDefinition: indexing_fields = ( @@ -341,6 +359,37 @@ def capability_install_hint(name: str) -> str: class CapabilityRequestError(ValueError): """Expected invalid capability selection or options.""" + def __init__( + self, + message: str, + *, + field: str = "capabilities", + reason: str = "capability_request_invalid", + requested: tuple[str, ...] = (), + available: tuple[str, ...] = (), + indexed: tuple[str, ...] = (), + next_action: str | None = None, + ) -> None: + self.field = field + self.reason = reason + self.requested = requested + self.available = available + self.indexed = indexed + self.next_action = next_action + super().__init__(message) + + def details(self) -> dict[str, Any]: + result: dict[str, Any] = {"reason": self.reason} + if self.requested: + result["requested"] = list(self.requested) + if self.available: + result["available"] = list(self.available) + if self.indexed: + result["indexed"] = list(self.indexed) + if self.next_action is not None: + result["next_action"] = self.next_action + return result + class CapabilityDependencyError(RuntimeError): def __init__( diff --git a/src/vidxp/capabilities/dialogue/definition.py b/src/vidxp/capabilities/dialogue/definition.py index 2d6b1c7..950bd8c 100644 --- a/src/vidxp/capabilities/dialogue/definition.py +++ b/src/vidxp/capabilities/dialogue/definition.py @@ -5,6 +5,7 @@ from packaging.requirements import Requirement from packaging.utils import canonicalize_name +from vidxp.application_models import CapabilityRole from vidxp.capabilities.contracts import ( CapabilityDefinition, CapabilityExecutor, @@ -96,6 +97,7 @@ def model_manifest( index_stage="dialogue_indexing", execution_group="dialogue", prepares_models=True, + roles=(CapabilityRole.searchable, CapabilityRole.queryable), model_specs=(QWEN3_EMBEDDING_MODEL, FASTER_WHISPER_MODEL), operations={ "search": OperationDefinition( diff --git a/src/vidxp/capabilities/registry.py b/src/vidxp/capabilities/registry.py index b18d9e9..5e82f82 100644 --- a/src/vidxp/capabilities/registry.py +++ b/src/vidxp/capabilities/registry.py @@ -187,7 +187,12 @@ def get(self, name: str) -> CapabilityDefinition: available = ", ".join(self.names()) raise CapabilityRequestError( f"Unknown capability {name!r}. " - f"Available capabilities: {available}." + f"Available capabilities: {available}.", + field="modalities", + reason="capability_unknown", + requested=(name,), + available=self.names(), + next_action="Choose a capability returned by get_workspace.", ) from exc def executor(self, name: str) -> CapabilityExecutor: @@ -222,7 +227,13 @@ def executor(self, name: str) -> CapabilityExecutor: def validate_names(self, names: Iterable[str]) -> tuple[str, ...]: selected = tuple(dict.fromkeys(str(name).strip() for name in names)) if not selected: - raise CapabilityRequestError("At least one capability is required.") + raise CapabilityRequestError( + "At least one capability is required.", + field="modalities", + reason="capability_required", + available=self.names(), + next_action="Choose a capability returned by get_workspace.", + ) for name in selected: self.get(name) return selected @@ -249,7 +260,15 @@ def validate_options( if unknown: raise CapabilityRequestError( "Options were supplied for disabled capabilities: " - + ", ".join(unknown) + + ", ".join(unknown), + field="capability_options", + reason="capability_options_disabled", + requested=tuple(unknown), + available=selected, + next_action=( + "Remove options for disabled capabilities or enable those " + "capabilities in modalities." + ), ) return { name: self.get(name) diff --git a/src/vidxp/capabilities/scene/definition.py b/src/vidxp/capabilities/scene/definition.py index aee444c..366ad4e 100644 --- a/src/vidxp/capabilities/scene/definition.py +++ b/src/vidxp/capabilities/scene/definition.py @@ -2,6 +2,7 @@ from typing import Any, Mapping +from vidxp.application_models import CapabilityRole from vidxp.capabilities.contracts import ( CapabilityDefinition, CapabilityExecutor, @@ -59,6 +60,7 @@ def model_manifest( index_stage="visual_indexing", execution_group="visual", prepares_models=True, + roles=(CapabilityRole.searchable, CapabilityRole.queryable), model_specs=(SIGLIP2_MODEL,), operations={ "search": OperationDefinition( diff --git a/src/vidxp/capability_service.py b/src/vidxp/capability_service.py index 9b3844b..7b0c64c 100644 --- a/src/vidxp/capability_service.py +++ b/src/vidxp/capability_service.py @@ -28,6 +28,8 @@ def _summary(self, name: str) -> CapabilitySummary: install_extra=definition.extra, supports_indexing=definition.collection_name is not None, prepares_models=definition.prepares_models, + roles=definition.roles, + identity_mode=definition.identity_mode, provenance=self.registry.provenance(name), ) diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index 8d0036a..14eb192 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -40,7 +40,6 @@ def create_index( not state.quiet and state.output_format == OutputFormat.rich ) selected = tuple(modalities) - state.service.require_models(selected) with IndexProgress(show_progress) as progress: job = state.jobs.submit_index( CreateIndexCommand( diff --git a/src/vidxp/composition.py b/src/vidxp/composition.py index f0c514b..856c2af 100644 --- a/src/vidxp/composition.py +++ b/src/vidxp/composition.py @@ -3,11 +3,16 @@ from dataclasses import dataclass from functools import cached_property from pathlib import Path +from typing import Callable from pydantic import ValidationError from vidxp.application import VidXPApplication -from vidxp.application_models import ApplicationError, ErrorCategory +from vidxp.application_models import ( + ApplicationError, + CreateIndexCommand, + ErrorCategory, +) from vidxp.artifact_service import ArtifactQueryService, ArtifactService from vidxp.authentication import Authenticator, create_authenticator from vidxp.authorization import AuthorizationPolicy @@ -94,7 +99,10 @@ def settings(self) -> VidXPSettings: @cached_property def jobs(self) -> JobService: assert self._settings is not None - return create_job_service(self._settings) + return create_job_service( + self._settings, + index_preflight=self.application.preflight_index, + ) def close(self) -> None: jobs = self.__dict__.get("jobs") @@ -159,12 +167,9 @@ def _server_chroma_url(settings: VidXPSettings) -> str | None: return BUNDLED_CHROMA_SERVER_URL -def _create_control_plane_components( - settings: VidXPSettings, -) -> _ControlPlaneComponents: - settings.layout.ensure_local_directories() +def _capability_registry(settings: VidXPSettings) -> CapabilityRegistry: server_mode = settings.mode == ApplicationMode.server - registry = create_capability_registry( + return create_capability_registry( external=settings.external_capabilities, allowlist=settings.capability_allowlist, platform_runtime_checks=( @@ -183,6 +188,13 @@ def _create_control_plane_components( else None ), ) + + +def _create_control_plane_components( + settings: VidXPSettings, +) -> _ControlPlaneComponents: + settings.layout.ensure_local_directories() + registry = _capability_registry(settings) catalog = ( SQLCatalog( workflow_database_url(settings), @@ -305,6 +317,7 @@ def create_application( media=components.media, artifacts=artifacts, index_status=backend.repository.status, + active_snapshot=components.snapshots.read_active, completed_upload_importer=( upload_service.import_completed if upload_service is not None @@ -319,6 +332,8 @@ def create_job_service( *, catalog: SQLCatalog | None = None, snapshots: LocalSnapshotRepository | None = None, + registry: CapabilityRegistry | None = None, + index_preflight: Callable[[CreateIndexCommand], None] | None = None, include_read_planner: bool = True, ) -> JobService: settings.layout.ensure_local_directories() @@ -332,6 +347,7 @@ def create_job_service( stop_executor = supervisor.stop return JobService( settings=settings, + index_preflight=index_preflight, backend=DBOSJobBackend( system_database_url=( None @@ -353,6 +369,7 @@ def create_job_service( read_planner=( LocalReadJobPlanner( layout=settings.layout, + registry=registry or _capability_registry(settings), index=LocalIndexReader( settings.layout, chroma_server_url=_server_chroma_url(settings), @@ -380,11 +397,14 @@ def create_control_plane_application( ), index_status=components.snapshots.status, model_cache=active_settings.model_cache, + active_snapshot=components.snapshots.read_active, ) jobs = create_job_service( active_settings, catalog=components.catalog, snapshots=components.snapshots, + registry=components.registry, + index_preflight=application.preflight_index, ) uploads = ( RemoteUploadService( diff --git a/src/vidxp/control_plane.py b/src/vidxp/control_plane.py index 991f48c..1897080 100644 --- a/src/vidxp/control_plane.py +++ b/src/vidxp/control_plane.py @@ -7,8 +7,10 @@ from vidxp.application_models import ( Artifact, CapabilityInfo, + CapabilityRole, CapabilitySummary, ComponentReadiness, + CreateIndexCommand, DependencyCheckResult, IndexStatus, InvalidRequestError, @@ -18,11 +20,16 @@ ModelUnavailableError, ResourceNotFoundError, RuntimeReadiness, + WorkspaceCapability, + WorkspaceMedia, + WorkspaceMediaCapability, + WorkspaceOverview, ) from vidxp.artifact_service import ArtifactQueryService 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.index_state import INDEX_STATUS_SCHEMA from vidxp.media_service import MediaService from vidxp.ports import LocalFileResource @@ -41,12 +48,14 @@ def __init__( artifacts: ArtifactQueryService, index_status: Callable[[], dict | None], model_cache: Path, + active_snapshot: Callable[[], IndexSnapshot | None] | None = None, ) -> None: self.layout = layout self.capabilities = capabilities self.media = media self.artifacts = artifacts self._read_index_status = index_status + self._read_active_snapshot = active_snapshot or (lambda: None) self.model_cache = model_cache @application_boundary @@ -105,6 +114,139 @@ def list_media(self, command: ListMediaCommand) -> MediaPage: except ValueError as exc: raise InvalidRequestError() from exc + @application_boundary + def workspace(self, command: ListMediaCommand) -> WorkspaceOverview: + page = self.list_media(command) + index = self.index_status() + snapshot = self._read_active_snapshot() + capabilities = self.list_capabilities() + readiness = self.model_readiness() + readiness_by_capability = { + capability.name: all( + check.ok + for check in readiness.checks + if check.capability == capability.name + ) + for capability in capabilities + if capability.prepares_models + } + projected_capabilities = tuple( + WorkspaceCapability( + **capability.model_dump(), + models_ready=readiness_by_capability.get(capability.name), + ) + for capability in capabilities + ) + media = tuple( + self._workspace_media( + asset, + capabilities=capabilities, + snapshot=snapshot, + ) + for asset in page.items + ) + indexed_media = ( + frozenset(snapshot.generations) if snapshot is not None else frozenset() + ) + indexed_capabilities = ( + { + name + for generation in snapshot.generations.values() + for name in generation.modalities + } + if snapshot is not None + else set() + ) + active_roles = { + role + for capability in capabilities + if capability.name in indexed_capabilities + for role in capability.roles + } + next_actions = [] + if page.total == 0: + next_actions.append("register_media") + if page.total > len(indexed_media) or any( + item.media_id not in indexed_media for item in page.items + ): + next_actions.append("index_media") + if CapabilityRole.searchable in active_roles: + next_actions.append("find_moments") + if CapabilityRole.queryable in active_roles: + next_actions.append("answer_video") + return WorkspaceOverview( + capabilities=projected_capabilities, + media=media, + media_total=page.total, + next_cursor=page.next_cursor, + index=index, + next_actions=tuple(next_actions), + ) + + @staticmethod + def _workspace_media( + asset: MediaAsset, + *, + capabilities: tuple[CapabilitySummary, ...], + snapshot: IndexSnapshot | None, + ) -> WorkspaceMedia: + generation = ( + snapshot.generations.get(asset.media_id) if snapshot is not None else None + ) + coverage = tuple( + WorkspaceMediaCapability( + name=capability.name, + indexed=( + generation is not None and capability.name in generation.modalities + ), + record_count=( + generation.record_counts.get(capability.name) + if generation is not None + and capability.name in generation.modalities + else None + ), + roles=( + capability.roles + if generation is not None + and capability.name in generation.modalities + else () + ), + identity_mode=capability.identity_mode, + ) + for capability in capabilities + ) + return WorkspaceMedia( + media_id=asset.media_id, + original_filename=asset.original_filename, + duration_seconds=asset.duration_seconds, + state=asset.state, + in_active_snapshot=generation is not None, + capabilities=coverage, + ) + + @application_boundary + def preflight_index(self, command: CreateIndexCommand) -> None: + selected = self.capabilities.registry.validate_names(command.modalities) + indexable = self.capabilities.registry.index_names() + unsupported = tuple(name for name in selected if name not in indexable) + if unsupported: + raise CapabilityRequestError( + "Indexing does not support these capabilities: " + + ", ".join(unsupported) + + ".", + field="modalities", + reason="capability_not_indexable", + requested=unsupported, + available=indexable, + next_action="Choose capabilities returned by get_workspace.", + ) + self.capabilities.registry.validate_options( + selected, + command.capability_options, + ) + self.get_media(command.media_id) + self.require_models(selected) + @application_boundary def open_media_content(self, media_id: str) -> LocalFileResource: return self.media.content(media_id) @@ -186,10 +328,7 @@ def runtime_readiness(self) -> RuntimeReadiness: components = self.control_plane_readiness() models = self.model_readiness() return RuntimeReadiness( - ready=( - all(component.ready for component in components) - and models.ok - ), + ready=(all(component.ready for component in components) and models.ok), runtime=None, components=components, dependencies=models, diff --git a/src/vidxp/frontend.py b/src/vidxp/frontend.py index ce15d84..dffa81c 100644 --- a/src/vidxp/frontend.py +++ b/src/vidxp/frontend.py @@ -43,7 +43,11 @@ def _configured_service( @lru_cache(maxsize=1) def _configured_jobs(settings: VidXPSettings | None = None) -> JobService: - return create_job_service(settings or _settings_from_arguments()) + active_settings = settings or _settings_from_arguments() + return create_job_service( + active_settings, + index_preflight=_configured_service(active_settings).preflight_index, + ) def _settings_from_arguments( @@ -340,7 +344,6 @@ def _run_indexing( media_id = media_ids[0] if len(media_ids) == 1 else None if media_id is None: raise ValueError("Select or import media before indexing.") - service.require_models(modalities) job = _configured_jobs().submit_index( CreateIndexCommand( media_id=media_id, diff --git a/src/vidxp/infrastructure/local_index.py b/src/vidxp/infrastructure/local_index.py index e333d3e..982e419 100644 --- a/src/vidxp/infrastructure/local_index.py +++ b/src/vidxp/infrastructure/local_index.py @@ -20,6 +20,7 @@ from vidxp.core.indexing_common import ProgressCallback from vidxp.core.manifest import MANIFEST_FILE, ManifestStore from vidxp.core.runner import index_video +from vidxp.core.snapshots import IndexSnapshot from vidxp.core.storage import ( ChromaClientFactory, IndexStorage, @@ -89,6 +90,15 @@ def active_config( config, _snapshot = self.repository.active_config(device=device) return config + def active_snapshot( + self, + index_directory: Path, + *, + device: str, + ) -> tuple[IndexConfig, IndexSnapshot]: + self._require_index_directory(index_directory) + return self.repository.active_config(device=device) + def config_for_snapshot( self, index_directory: Path, diff --git a/src/vidxp/job_service.py b/src/vidxp/job_service.py index dcad2df..0fac51f 100644 --- a/src/vidxp/job_service.py +++ b/src/vidxp/job_service.py @@ -89,10 +89,12 @@ def __init__( settings: VidXPSettings, backend: JobBackend, read_planner: ReadJobPlanner | None = None, + index_preflight: Callable[[CreateIndexCommand], None] | None = None, ) -> None: self.settings = settings self.backend = backend self.read_planner = read_planner + self.index_preflight = index_preflight def _read_job_planner(self) -> ReadJobPlanner: if self.read_planner is None: @@ -114,6 +116,8 @@ def submit_index( *, job_id: str | None = None, ) -> Job: + if self.index_preflight is not None: + self.index_preflight(command) return self.backend.submit( IndexJobRequest(command=command), queue=self._model_queue(), diff --git a/src/vidxp/mcp.py b/src/vidxp/mcp.py index e9a297f..89f9ae4 100644 --- a/src/vidxp/mcp.py +++ b/src/vidxp/mcp.py @@ -44,6 +44,7 @@ QueryVideoCommand, RuntimeReadiness, SearchCommand, + WorkspaceOverview, ) from vidxp.authentication import ( OIDCBearerAuthenticator, @@ -345,9 +346,11 @@ def create_mcp_server( ) ], instructions=( - "Call get_runtime_readiness before indexing. If selected model " + "Call get_workspace before planning index, search, query, or actor " + "work; it reports valid capability roles for each media item. Call " + "get_runtime_readiness before indexing. If selected model " "artifacts are missing, submit prepare_models and poll get_job " - "until it completes. Discover registered media with list_media. " + "until it completes. " "Register and upload new video through the HTTP/tus API, then use " "its media_id with start_indexing. get_index_status identifies the " "media included in the active index snapshot. For search_moments " @@ -418,6 +421,32 @@ async def read_matroska_artifact(artifact_id: ArtifactId) -> bytes: expected_mime_type="video/x-matroska", ) + @server.tool( + description=( + "Inspect registered media, the active index, installed capabilities, " + "and the searchable, queryable, inspectable, or renderable roles " + "currently usable for each media item. Call this before planning " + "index, search, query, or actor work." + ), + annotations=_READ_ONLY, + structured_output=True, + ) + async def get_workspace( + page_size: Annotated[int, Field(gt=0, le=100)] = 50, + cursor: Annotated[ + str | None, + Field(min_length=1, max_length=512), + ] = None, + ) -> WorkspaceOverview: + return await _invoke_async( + context, + default_principal=default_principal, + permission=RepositoryPermission.read, + operation=lambda _actor: context.application.workspace( + ListMediaCommand(page_size=page_size, cursor=cursor) + ), + ) + @server.tool( description="List installed VidXP capabilities.", annotations=_READ_ONLY, @@ -532,7 +561,6 @@ async def start_indexing( idempotency_key: IdempotencyKey, ) -> Job: def submit(actor: Principal) -> Job: - context.application.require_models(command.modalities) return context.jobs.submit_index( command, job_id=scoped_job_id( diff --git a/src/vidxp/read_job_planner.py b/src/vidxp/read_job_planner.py index 19079b8..6e70e95 100644 --- a/src/vidxp/read_job_planner.py +++ b/src/vidxp/read_job_planner.py @@ -9,8 +9,11 @@ QueryVideoCommand, SearchCommand, SearchJobRequest, + CapabilityRole, ) from vidxp.capabilities.contracts import CapabilityRequestError +from vidxp.capabilities.registry import CapabilityRegistry +from vidxp.core.snapshots import IndexSnapshot from vidxp.infrastructure.local_index import LocalIndexReader from vidxp.repository_layout import RepositoryLayout @@ -22,13 +25,15 @@ def __init__( self, *, layout: RepositoryLayout, + registry: CapabilityRegistry, index: LocalIndexReader | None = None, ) -> None: self.layout = layout + self.registry = registry self.index = index or LocalIndexReader(layout) def _active(self): - config = self.index.active_config( + config, snapshot = self.index.active_snapshot( self.layout.indexes, device="cpu", ) @@ -36,39 +41,138 @@ def _active(self): raise RuntimeError( "The active index did not provide an immutable reference." ) - return config, IndexSnapshotReference( - snapshot_id=config.snapshot_id, - snapshot_sha256=config.snapshot_sha256, + return ( + config, + IndexSnapshotReference( + snapshot_id=config.snapshot_id, + snapshot_sha256=config.snapshot_sha256, + ), + snapshot, ) - @staticmethod - def _require_capability(capability: str, config) -> None: - if capability not in config.enabled_modalities: + def _select_capabilities( + self, + requested: tuple[str, ...], + *, + indexed: tuple[str, ...], + role: CapabilityRole, + operation: str, + ) -> tuple[str, ...]: + explicit = bool(requested) + selected = self.registry.validate_names(requested) if explicit else indexed + absent = tuple(name for name in selected if name not in indexed) + if absent: + raise CapabilityRequestError( + f"{operation} capabilities are not present in the active index: " + + ", ".join(absent) + + ".", + field="modalities", + reason="capability_not_indexed", + requested=absent, + available=tuple( + name for name in indexed if role in self.registry.get(name).roles + ), + indexed=indexed, + next_action=( + "Choose an indexed capability returned by get_workspace or " + "index the media with the requested capability." + ), + ) + supported = tuple( + name for name in selected if role in self.registry.get(name).roles + ) + unsupported = tuple(name for name in selected if name not in supported) + if explicit and unsupported: + available = tuple( + name for name in indexed if role in self.registry.get(name).roles + ) raise CapabilityRequestError( - f"The {capability} capability is not present in this index." + f"{operation} does not support these indexed capabilities: " + + ", ".join(unsupported) + + ".", + field="modalities", + reason="capability_role_unsupported", + requested=unsupported, + available=available, + indexed=indexed, + next_action=( + f"Choose a {role.value} capability returned by get_workspace." + ), ) + if not supported: + raise CapabilityRequestError( + f"The active index has no {role.value} capabilities.", + field="modalities", + reason="capability_role_unavailable", + indexed=indexed, + next_action=( + f"Index media with a capability whose role is {role.value}." + ), + ) + return supported + + @staticmethod + def _require_media( + media_id: str | None, + snapshot: IndexSnapshot, + ) -> None: + if media_id is None or media_id in snapshot.generations: + return + raise CapabilityRequestError( + "The selected media item is not present in the active index snapshot.", + field="media_id", + reason="media_not_indexed", + requested=(media_id,), + available=tuple(sorted(snapshot.generations)[:100]), + next_action=( + "Choose indexed media returned by get_workspace or index this " + "media item first." + ), + ) @application_boundary def plan_search(self, command: SearchCommand) -> SearchJobRequest: - config, reference = self._active() - for capability in command.modalities: - self._require_capability(capability, config) - return SearchJobRequest(command=command, snapshot=reference) + config, reference, snapshot = self._active() + selected = self._select_capabilities( + command.modalities, + indexed=config.enabled_modalities, + role=CapabilityRole.searchable, + operation="Search", + ) + self._require_media(command.media_id, snapshot) + return SearchJobRequest( + command=command.model_copy(update={"modalities": selected}), + snapshot=reference, + ) @application_boundary def plan_query(self, command: QueryVideoCommand) -> QueryJobRequest: - config, reference = self._active() - for capability in command.modalities: - self._require_capability(capability, config) - return QueryJobRequest(command=command, snapshot=reference) + config, reference, snapshot = self._active() + selected = self._select_capabilities( + command.modalities, + indexed=config.enabled_modalities, + role=CapabilityRole.queryable, + operation="Query", + ) + self._require_media(command.media_id, snapshot) + return QueryJobRequest( + command=command.model_copy(update={"modalities": selected}), + snapshot=reference, + ) @application_boundary def plan_actor_overlay( self, command: CreateActorOverlayCommand, ) -> ActorOverlayJobRequest: - config, reference = self._active() - self._require_capability("actor", config) + config, reference, _snapshot = self._active() + selected = self._select_capabilities( + ("actor",), + indexed=config.enabled_modalities, + role=CapabilityRole.renderable, + operation="Actor overlay", + ) + assert selected == ("actor",) return ActorOverlayJobRequest( command=command, snapshot=reference, diff --git a/tests/test_api.py b/tests/test_api.py index 097c9b4..dab53f5 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -25,11 +25,13 @@ JobKind, JobQueue, JobState, + IndexStatus, MediaAsset, Principal, SearchCommand, QueryVideoCommand, UploadIntent, + WorkspaceOverview, ) from vidxp.composition import ( HttpApplicationContext, @@ -278,6 +280,28 @@ def test_static_bearer_is_enforced_before_dispatch(self): self.assertEqual(accepted.json(), {"items": []}) context.application.list_capabilities.assert_called_once_with() + def test_workspace_endpoint_returns_actionable_repository_state(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + context.application.workspace.return_value = WorkspaceOverview( + media_total=0, + index=IndexStatus( + schema_version=2, + state="missing", + stage="status", + message="No index.", + ), + next_actions=("register_media",), + ) + with TestClient(create_app(context=context)) as client: + response = client.get("/api/v1/workspace?page_size=25") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["media_total"], 0) + self.assertEqual(response.json()["next_actions"], ["register_media"]) + command = context.application.workspace.call_args.args[0] + self.assertEqual(command.page_size, 25) + def test_repository_scopes_are_enforced_per_operation(self): with TemporaryDirectory() as directory: context = self.context( @@ -378,7 +402,6 @@ def test_job_submission_is_thin_idempotent_delegation(self): self.assertEqual(command.media_id, MEDIA_ID) self.assertEqual(command.modalities, ("scene",)) self.assertEqual(command.scene_sample_fps, 0.5) - context.application.require_models.assert_called_once_with(("scene",)) expected_job_id = scoped_job_id( context, context.authenticator.authenticate(None), @@ -441,7 +464,7 @@ def test_failed_model_preparation_job_is_structured_over_http(self): def test_missing_models_fail_before_job_submission(self): with TemporaryDirectory() as directory: context = self.context(Path(directory)) - context.application.require_models.side_effect = ApplicationError( + context.jobs.submit_index.side_effect = ApplicationError( "model_unavailable", ErrorCategory.unavailable, "Run vidxp prepare --modalities scene.", @@ -466,7 +489,7 @@ def test_missing_models_fail_before_job_submission(self): response.json()["error"]["details"]["remediation"], "vidxp prepare --modalities scene", ) - context.jobs.submit_index.assert_not_called() + context.jobs.submit_index.assert_called_once() def test_job_submission_requires_an_idempotency_key(self): with TemporaryDirectory() as directory: diff --git a/tests/test_control_plane.py b/tests/test_control_plane.py new file mode 100644 index 0000000..a6ae69b --- /dev/null +++ b/tests/test_control_plane.py @@ -0,0 +1,152 @@ +import unittest +from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import Mock + +from vidxp.application_models import ( + ApplicationError, + CapabilityIdentityMode, + CapabilityRole, + CreateIndexCommand, + ListMediaCommand, + MediaAsset, + MediaPage, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.capability_service import CapabilityService +from vidxp.control_plane import ControlPlaneApplication +from vidxp.core.media import MediaState, MediaStream +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.repository_layout import RepositoryLayout + + +MEDIA_ID = "123456781234423481234567890abcde" +OTHER_MEDIA_ID = "223456781234423481234567890abcde" +GENERATION_ID = "323456781234423481234567890abcde" +SNAPSHOT_ID = "423456781234423481234567890abcde" + + +def media_asset(media_id: str, filename: str) -> MediaAsset: + return MediaAsset( + media_id=media_id, + video_id=media_id, + original_filename=filename, + sha256="a" * 64, + byte_size=10, + detected_mime_type="video/mp4", + container="mp4", + duration_seconds=2, + streams=( + MediaStream( + index=0, + kind="video", + codec="h264", + width=1, + height=1, + ), + ), + state=MediaState.ready, + created_at=datetime.now(timezone.utc), + ) + + +class ControlPlaneWorkspaceTests(unittest.TestCase): + def test_index_preflight_rejects_unknown_capability_with_next_action(self): + with TemporaryDirectory() as directory: + root = Path(directory) + media = Mock() + application = ControlPlaneApplication( + layout=RepositoryLayout(root=root), + capabilities=CapabilityService(create_capability_registry()), + media=media, + artifacts=Mock(), + index_status=lambda: None, + model_cache=root / "models", + ) + + with self.assertRaises(ApplicationError) as raised: + application.preflight_index( + CreateIndexCommand( + media_id=MEDIA_ID, + modalities=("unknown",), + ) + ) + + error = raised.exception.to_dict()["details"]["errors"][0] + self.assertEqual(error["reason"], "capability_unknown") + self.assertEqual(error["requested"], ["unknown"]) + self.assertIn("get_workspace", error["next_action"]) + media.get.assert_not_called() + + def test_workspace_projects_index_coverage_roles_and_next_actions(self): + indexed = media_asset(MEDIA_ID, "indexed.mp4") + unindexed = media_asset(OTHER_MEDIA_ID, "new.mp4") + snapshot = IndexSnapshot( + snapshot_id=SNAPSHOT_ID, + created_at=datetime.now(timezone.utc), + config_fingerprint="b" * 64, + configuration={}, + generations={ + MEDIA_ID: GenerationReference( + generation_id=GENERATION_ID, + media_id=MEDIA_ID, + manifest_sha256="c" * 64, + input_sha256="d" * 64, + config_fingerprint="e" * 64, + modalities=("scene", "actor"), + record_counts={"scene": 12, "actor": 4}, + store_size_bytes_at_commit=100, + ) + }, + ) + media = Mock() + media.list.return_value = MediaPage( + items=(indexed, unindexed), + total=2, + ) + with TemporaryDirectory() as directory: + root = Path(directory) + application = ControlPlaneApplication( + layout=RepositoryLayout(root=root), + capabilities=CapabilityService(create_capability_registry()), + media=media, + artifacts=Mock(), + index_status=lambda: { + "schema_version": 2, + "state": "ready", + "stage": "status", + "message": "Index ready.", + }, + active_snapshot=lambda: snapshot, + model_cache=root / "models", + ) + + workspace = application.workspace(ListMediaCommand()) + + self.assertEqual(workspace.media_total, 2) + self.assertEqual( + workspace.next_actions, + ("index_media", "find_moments", "answer_video"), + ) + indexed_projection = workspace.media[0] + self.assertTrue(indexed_projection.in_active_snapshot) + by_name = { + capability.name: capability + for capability in indexed_projection.capabilities + } + self.assertEqual(by_name["scene"].record_count, 12) + self.assertEqual( + by_name["scene"].roles, + (CapabilityRole.searchable, CapabilityRole.queryable), + ) + self.assertEqual(by_name["actor"].record_count, 4) + self.assertEqual( + by_name["actor"].identity_mode, + CapabilityIdentityMode.anonymous_clusters, + ) + self.assertEqual(workspace.media[1].capabilities[0].roles, ()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_frontend.py b/tests/test_frontend.py index ee426c4..9991b0a 100644 --- a/tests/test_frontend.py +++ b/tests/test_frontend.py @@ -337,7 +337,7 @@ def test_indexing_submits_selected_scene_sample_rate(self): command = jobs.submit_index.call_args.args[0] self.assertEqual(command.scene_sample_fps, 2.0) - service.require_models.assert_called_once_with(("scene",)) + service.require_models.assert_not_called() def test_indexing_omits_scene_sample_rate_without_scene(self): jobs = Mock() diff --git a/tests/test_job_contracts.py b/tests/test_job_contracts.py index 34f67aa..018e757 100644 --- a/tests/test_job_contracts.py +++ b/tests/test_job_contracts.py @@ -79,6 +79,7 @@ def test_public_job_contract_has_no_path_or_storage_fields(self): def test_job_service_routes_model_work_without_reimplementing_it(self): backend = Mock() + preflight = Mock() backend.submit.return_value = Job( job_id=JOB_ID, kind=JobKind.index, @@ -91,6 +92,7 @@ def test_job_service_routes_model_work_without_reimplementing_it(self): runtime_backend="cpu", ), backend=backend, + index_preflight=preflight, ) job = service.submit_index( @@ -102,6 +104,7 @@ def test_job_service_routes_model_work_without_reimplementing_it(self): self.assertEqual(job.job_id, JOB_ID) request = backend.submit.call_args.args[0] + preflight.assert_called_once_with(request.command) self.assertEqual(request.kind, JobKind.index) self.assertEqual(request.command.media_id, MEDIA_ID) self.assertEqual( @@ -109,6 +112,34 @@ def test_job_service_routes_model_work_without_reimplementing_it(self): JobQueue.cpu, ) + def test_index_preflight_failure_never_reaches_the_job_backend(self): + backend = Mock() + command = CreateIndexCommand( + media_id=MEDIA_ID, + modalities=("scene",), + ) + preflight = Mock( + side_effect=ApplicationError( + "invalid_request", + ErrorCategory.validation, + "The capability is not usable.", + ) + ) + service = JobService( + settings=VidXPSettings( + repository_root=Path("repository"), + runtime_backend="cpu", + ), + backend=backend, + index_preflight=preflight, + ) + + with self.assertRaises(ApplicationError): + service.submit_index(command) + + preflight.assert_called_once_with(command) + backend.submit.assert_not_called() + def test_job_list_cursor_is_bounded(self): with self.assertRaises(ValidationError): ListJobsCommand(cursor="x" * 513) diff --git a/tests/test_mcp.py b/tests/test_mcp.py index 39de1e7..2639628 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -34,6 +34,7 @@ MediaPage, Principal, QueryVideoCommand, + WorkspaceOverview, ) from vidxp.authentication import ( AuthenticatedBearer, @@ -103,6 +104,11 @@ def context( stage="status", message="No index.", ) + application.workspace.return_value = WorkspaceOverview( + media_total=0, + index=application.index_status.return_value, + next_actions=("register_media",), + ) jobs = Mock(spec=JobService) readiness = Mock() readiness.ready.return_value = True @@ -132,6 +138,7 @@ async def test_curated_tools_have_structured_output_schemas(self): self.assertEqual( [tool.name for tool in discovered.tools], [ + "get_workspace", "list_capabilities", "get_capability", "get_runtime_readiness", @@ -166,6 +173,26 @@ async def test_curated_tools_have_structured_output_schemas(self): self.assertEqual(result.structured_content, {"items": []}) self.assertFalse(result.is_error) + async def test_workspace_tool_projects_actionable_repository_state(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + server = create_mcp_server( + context, + default_principal=Principal( + subject="local", + scopes=frozenset({"*"}), + ), + ) + async with Client(server) as client: + result = await client.call_tool("get_workspace", {}) + + self.assertEqual(result.structured_content["media_total"], 0) + self.assertEqual( + result.structured_content["next_actions"], + ["register_media"], + ) + context.application.workspace.assert_called_once() + def test_stdio_help_and_config_are_ready_to_copy(self): config = stdio_client_config( command=r"C:\VidXP\vidxp-mcp.exe", @@ -217,7 +244,7 @@ def test_stdio_check_performs_handshake_and_tool_probe(self): rendered = output.getvalue() self.assertIn("OK VidXP MCP", rendered) self.assertIn("Index state: missing", rendered) - self.assertIn("Tools: 16", rendered) + self.assertIn("Tools: 17", rendered) self.assertIn("get_index_status", rendered) async def test_server_info_exposes_vidxp_branding(self): @@ -280,15 +307,12 @@ async def test_index_submission_uses_shared_stable_idempotency(self): calls[0].kwargs["job_id"], calls[1].kwargs["job_id"], ) - self.assertEqual( - context.application.require_models.call_args.args[0], - ("scene",), - ) + self.assertEqual(calls[0].args[0].modalities, ("scene",)) async def test_missing_models_fail_before_index_submission(self): with TemporaryDirectory() as directory: context = self.context(Path(directory)) - context.application.require_models.side_effect = ApplicationError( + context.jobs.submit_index.side_effect = ApplicationError( "model_unavailable", ErrorCategory.unavailable, "Run vidxp prepare --modalities scene.", @@ -321,7 +345,7 @@ async def test_missing_models_fail_before_index_submission(self): '"remediation":"vidxp prepare --modalities scene"', result.content[0].text, ) - context.jobs.submit_index.assert_not_called() + context.jobs.submit_index.assert_called_once() async def test_query_video_submits_the_shared_durable_command(self): with TemporaryDirectory() as directory: @@ -628,6 +652,7 @@ async def test_stdio_entrypoint_serves_the_same_curated_surface(self): self.assertEqual( [tool.name for tool in discovered.tools], [ + "get_workspace", "list_capabilities", "get_capability", "get_runtime_readiness", @@ -697,7 +722,7 @@ async def test_streamable_http_works_with_the_official_remote_client(self): server.should_exit = True await serving - self.assertEqual(len(discovered.tools), 16) + self.assertEqual(len(discovered.tools), 17) self.assertEqual(result.structured_content, {"items": []}) async def test_oidc_verifier_projects_the_shared_validated_token(self): diff --git a/tests/test_read_job_planner.py b/tests/test_read_job_planner.py index bb12274..0aba9be 100644 --- a/tests/test_read_job_planner.py +++ b/tests/test_read_job_planner.py @@ -1,12 +1,16 @@ import unittest +from datetime import datetime, timezone from unittest.mock import Mock from vidxp.application_models import ( + ApplicationError, CreateActorOverlayCommand, QueryVideoCommand, SearchCommand, ) +from vidxp.capabilities.registry import create_capability_registry from vidxp.core.contracts import IndexConfig +from vidxp.core.snapshots import GenerationReference, IndexSnapshot from vidxp.read_job_planner import LocalReadJobPlanner from vidxp.repository_layout import RepositoryLayout @@ -27,10 +31,32 @@ def setUp(self): storage_directory=self.layout.index_store, collection_names={"scene": "scene", "actor": "actor"}, ) + self.snapshot = IndexSnapshot( + snapshot_id=SNAPSHOT_ID, + created_at=datetime.now(timezone.utc), + config_fingerprint="b" * 64, + configuration={}, + generations={ + MEDIA_ID: GenerationReference( + generation_id=GENERATION_ID, + media_id=MEDIA_ID, + manifest_sha256="c" * 64, + input_sha256="d" * 64, + config_fingerprint="e" * 64, + modalities=("scene", "actor"), + record_counts={"scene": 2, "actor": 1}, + store_size_bytes_at_commit=100, + ) + }, + ) self.index = Mock() - self.index.active_config.return_value = self.config + self.index.active_snapshot.return_value = ( + self.config, + self.snapshot, + ) self.planner = LocalReadJobPlanner( layout=self.layout, + registry=create_capability_registry(), index=self.index, ) @@ -68,6 +94,45 @@ def test_query_job_carries_the_same_logical_snapshot_reference(self): self.assertEqual(request.snapshot.snapshot_sha256, "a" * 64) self.index.open_store.assert_not_called() + def test_omitted_search_capabilities_exclude_non_searchable_actor(self): + request = self.planner.plan_search(SearchCommand(query="a taxi")) + + self.assertEqual(request.command.modalities, ("scene",)) + + def test_explicit_actor_search_fails_before_a_job_can_be_submitted(self): + with self.assertRaises(ApplicationError) as raised: + self.planner.plan_search( + SearchCommand(modalities=("actor",), query="Harry") + ) + + error = raised.exception.to_dict()["details"] + self.assertEqual(error["errors"][0]["reason"], "capability_role_unsupported") + self.assertEqual(error["errors"][0]["requested"], ["actor"]) + self.assertEqual(error["errors"][0]["available"], ["scene"]) + + def test_actor_remains_available_to_grounded_query(self): + request = self.planner.plan_query( + QueryVideoCommand( + question="When does this actor appear?", + modalities=("actor",), + ) + ) + + self.assertEqual(request.command.modalities, ("actor",)) + + def test_media_outside_active_snapshot_fails_during_planning(self): + with self.assertRaises(ApplicationError) as raised: + self.planner.plan_search( + SearchCommand( + media_id="423456781234423481234567890abcde", + query="a taxi", + ) + ) + + error = raised.exception.to_dict()["details"]["errors"][0] + self.assertEqual(error["reason"], "media_not_indexed") + self.assertEqual(error["available"], [MEDIA_ID]) + if __name__ == "__main__": unittest.main()