feat: add capability-aware agent preflight

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)

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(

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,

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
 

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",

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"
 

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(

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

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(

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)

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(

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

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(