diff --git a/README.md b/README.md
index be59222..00b7f95 100644
--- a/README.md
+++ b/README.md
@@ -610,6 +610,9 @@ response = coll.generate_text(
     model_name="pro",  # basic, pro, or ultra
     response_type="text"  # text or json
 )
+
+# Large prompts are uploaded automatically with a unique filename and
+# sent as prompt_url instead of inline JSON to avoid request payload limits.
 ```
 
 ### Video Dubbing and Translation
diff --git a/openapi.yaml b/openapi.yaml
index 417b87b..c378110 100644
--- a/openapi.yaml
+++ b/openapi.yaml
@@ -1915,6 +1915,14 @@ paths:
                 aspect_ratio:
                   type: string
                   example: "16:9"
+                model_name:
+                  type: string
+                  description: Use `flux` for FLUX self-inference image generation.
+                  example: "flux"
+                config:
+                  type: object
+                  description: Model configuration. For FLUX supports size, num_inference_steps, guidance_scale, negative_prompt, and seed.
+                  additionalProperties: true
                 callback_url:
                   type: string
                   example: "https://webhook.example.com/callback"
@@ -1983,16 +1991,32 @@ paths:
             schema:
               type: object
               required:
-                - prompt
                 - audio_type
               properties:
                 prompt:
                   type: string
+                  description: Prompt for music or sound effect generation.
                   example: "Generate upbeat background music"
+                text:
+                  type: string
+                  description: Text to convert to speech when `audio_type` is `voice`.
+                  example: "Hello, welcome to VideoDB."
                 audio_type:
                   type: string
-                  enum: [speech, sound_effect, music]
-                  example: "music"
+                  enum: [voice, sound_effect, music]
+                  example: "voice"
+                voice_name:
+                  type: string
+                  description: Voice name for hosted text-to-speech.
+                  example: "Default"
+                model_name:
+                  type: string
+                  description: Use `omnivoice` for OmniVoice self-inference text-to-speech.
+                  example: "omnivoice"
+                config:
+                  type: object
+                  description: Voice configuration. For OmniVoice supports instructions, ref_audio, ref_text, response_format, speed, language, and max_new_tokens.
+                  additionalProperties: true
                 callback_url:
                   type: string
                   example: "https://webhook.example.com/callback"
@@ -2004,6 +2028,39 @@ paths:
               schema:
                 $ref: '#/components/schemas/AsyncResponse'
 
+  /job/{job_id}:
+    get:
+      summary: Get self-inference generation job status
+      security:
+        - ApiKeyAuth: []
+      parameters:
+        - name: job_id
+          in: path
+          required: true
+          schema:
+            type: string
+            example: "550e8400-e29b-41d4-a716-446655440000"
+      responses:
+        '200':
+          description: Generation job status or final generated asset
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  success:
+                    type: boolean
+                  status:
+                    type: string
+                    enum: [processing, done, failed]
+                  data:
+                    type: object
+                    additionalProperties: true
+                  message:
+                    type: string
+        '404':
+          description: Job not found
+
   /collection/{collection_id}/generate/text/:
     post:
       summary: Generate text using AI
@@ -2022,27 +2079,45 @@ paths:
           application/json:
             schema:
               type: object
-              required:
-                - prompt
+              description: Provide either `prompt` for inline text or `prompt_url` for a previously uploaded prompt file.
               properties:
                 prompt:
                   type: string
+                  description: Inline prompt text. Use for smaller prompts.
                   example: "Summarize the content of this video"
+                prompt_url:
+                  type: string
+                  description: URL to a previously uploaded plain-text prompt file. Use for larger prompts.
+                  example: "https://storage.googleapis.com/.../generate_text_prompt_123abc.txt"
                 video_id:
                   type: string
                   example: "m-12345"
                 model_name:
                   type: string
-                  example: "gpt-4"
+                  description: Hosted tier (`mini`, `basic`, `pro`, `ultra`) or self-hosted sandbox model name.
+                  example: "Qwen/Qwen3.5-27B"
+                sandbox_id:
+                  type: string
+                  description: Sandbox ID to use when `model_name` is a self-hosted model.
+                  example: "bx-123abc"
                 max_tokens:
                   type: integer
                   example: 500
                 temperature:
                   type: number
                   example: 0.7
+                model_config:
+                  type: object
+                  description: Additional self-hosted model configuration.
+                  additionalProperties: true
                 callback_url:
                   type: string
                   example: "https://webhook.example.com/callback"
+              oneOf:
+                - required:
+                    - prompt
+                - required:
+                    - prompt_url
       responses:
         '200':
           description: Text generation started or completed
@@ -5602,4 +5677,4 @@ tags:
   - name: Transcode
     description: Media transcoding operations
   - name: Assets
-    description: Cross-collection asset listing
\ No newline at end of file
+    description: Cross-collection asset listing
diff --git a/tests/test_search_v2_migration_warnings.py b/tests/test_search_v2_migration_warnings.py
new file mode 100644
index 0000000..2af8290
--- /dev/null
+++ b/tests/test_search_v2_migration_warnings.py
@@ -0,0 +1,358 @@
+import pytest
+
+import videodb.search as search_module
+from videodb.collection import Collection
+from videodb.exceptions import SearchError
+from videodb.video import Video
+
+
+WARNING = {
+    "code": "search_v2_migration_dual_read",
+    "message": "search() now uses Search V2 and may also check older indexes during migration.",
+    "docs": [{"label": "Search V2 search", "url": "https://example.test/search"}],
+}
+ASK_WARNING = {
+    "code": "ask_requires_search_v2_index",
+    "message": "ask() uses Search V2 indexes only.",
+    "docs": [{"label": "Ask", "url": "https://example.test/ask"}],
+}
+SEMANTIC_WARNING = {
+    "code": "semantic_search_v2_migration_dual_read",
+    "message": "semantic_search() may also check older indexes during migration.",
+    "docs": [{"label": "Semantic search", "url": "https://example.test/semantic"}],
+}
+
+
+class FakeConnection:
+    def __init__(self):
+        self.calls = []
+
+    def post(self, path, data=None, **kwargs):
+        self.calls.append((path, data or {}))
+        if path == "compile":
+            return {
+                "stream_url": "https://stream.example.test/compiled.m3u8",
+                "player_url": "https://player.example.test/watch?v=compiled",
+            }
+        if path.endswith("/ask"):
+            return {"answer": "I do not have enough information to answer.", "sources": [], "warnings": [ASK_WARNING]}
+        if path.endswith("/semantic-search"):
+            return {"results": _results("semantic"), "warnings": [SEMANTIC_WARNING]}
+        if path.endswith("/query"):
+            return {"results": _results("query"), "warnings": [WARNING]}
+        if path.endswith("/aggregate"):
+            return {"results": [{"label": "object", "count": 2}], "warnings": [WARNING]}
+        if path.endswith("/search/v2"):
+            if (data or {}).get("mode") == "deepsearch":
+                return {"response_type": "deepsearch", "results": _results("deepsearch"), "warnings": [WARNING]}
+            return {"response_type": "shots", "results": _results("v2"), "warnings": [WARNING]}
+        if path.endswith("/search"):
+            return {"results": _results("legacy")}
+        raise AssertionError(f"unexpected path: {path}")
+
+
+class FakeCollectionConnection(FakeConnection):
+    pass
+
+
+def _results(text):
+    return [
+        {
+            "collection_id": "c1",
+            "video_id": "v1",
+            "length": 30,
+            "title": "video",
+            "docs": [{"start": 1, "end": 3, "text": text, "score": 0.8}],
+        }
+    ]
+
+
+@pytest.fixture(autouse=True)
+def reset_warning_state():
+    search_module._LEGACY_SEARCH_WARNING_EMITTED.clear()
+    yield
+    search_module._LEGACY_SEARCH_WARNING_EMITTED.clear()
+
+
+def test_search_v2_server_warnings_are_emitted_and_exposed():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2") as emitted:
+        response = video.search("find cars", filter={"legacy_key": "legacy_value"})
+
+    assert len(emitted) == 1
+    assert conn.calls[-1] == (
+        "video/v1/search/v2",
+        {"query": "find cars", "filter": {"legacy_key": "legacy_value"}},
+    )
+    assert response.warnings == [WARNING]
+    assert response.results.warnings == [WARNING]
+    assert response.shots[0].text == "v2"
+
+
+def test_deepsearch_server_warnings_are_emitted_and_exposed_on_search_response():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2"):
+        response = video.search("find cars", mode="deepsearch")
+
+    assert conn.calls[-1] == ("video/v1/search/v2", {"query": "find cars", "mode": "deepsearch"})
+    assert response.response_type == "deepsearch"
+    assert response.warnings == [WARNING]
+    assert response.results.warnings == [WARNING]
+
+
+def test_ask_server_warnings_are_emitted_and_exposed_on_ask_response():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"ask\(\) uses Search V2 indexes only"):
+        response = video.ask("what happened?")
+
+    assert conn.calls[-1] == (
+        "video/v1/ask",
+        {"question": "what happened?", "top_k": 15, "mode": "default", "include_sources": False},
+    )
+    assert response.answer == "I do not have enough information to answer."
+    assert response.warnings == [ASK_WARNING]
+
+
+def test_search_response_compile_delegates_for_shots():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2"):
+        response = video.search("find cars")
+
+    assert response.compile() == "https://stream.example.test/compiled.m3u8"
+    assert conn.calls[-1] == (
+        "compile",
+        [{"video_id": "v1", "collection_id": "c1", "shots": [(1, 3)]}],
+    )
+    assert response.results.stream_url == "https://stream.example.test/compiled.m3u8"
+
+
+def test_search_response_play_delegates_for_shots(monkeypatch):
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+    played = []
+
+    monkeypatch.setattr(search_module, "play_stream", lambda url: played.append(url) or f"played:{url}")
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2"):
+        response = video.search("find cars")
+
+    assert response.play() == "played:https://stream.example.test/compiled.m3u8"
+    assert played == ["https://stream.example.test/compiled.m3u8"]
+
+
+def test_search_response_embed_code_delegates_for_deepsearch():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2"):
+        response = video.search("find cars", mode="deepsearch")
+
+    html = response.get_embed_code(width="640", height=360, title="Result", allow_fullscreen=False)
+    assert response.response_type == "deepsearch"
+    assert 'src="https://player.example.test/embed?v=compiled"' in html
+    assert 'width="640"' in html
+    assert 'height="360"' in html
+    assert 'title="Result"' in html
+    assert "allowfullscreen" not in html
+
+
+def test_search_response_media_helpers_raise_for_non_shot_response():
+    response = search_module.SearchResponse(
+        FakeConnection(),
+        response_type="aggregate",
+        results=[{"label": "car", "count": 2}],
+    )
+
+    with pytest.raises(SearchError, match="does not contain shot results"):
+        response.compile()
+    with pytest.raises(SearchError, match="does not contain shot results"):
+        response.play()
+    with pytest.raises(SearchError, match="does not contain shot results"):
+        response.get_embed_code()
+
+
+def test_semantic_search_accepts_plural_selectors_emits_and_exposes_warnings():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"semantic_search\(\) may also check older indexes"):
+        response = video.semantic_search("find cars", index_names=["scene"], index_ids=["idx-v2"])
+
+    assert conn.calls[-1] == (
+        "video/v1/semantic-search",
+        {
+            "query": "find cars",
+            "index_names": ["scene"],
+            "index_ids": ["idx-v2"],
+            "top_k": 10,
+            "score_threshold": None,
+            "filter": None,
+            "return_fields": None,
+        },
+    )
+    assert response.warnings == [SEMANTIC_WARNING]
+    assert response.shots[0].text == "semantic"
+
+
+def test_semantic_search_rejects_singular_selectors_in_sdk_signature():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.raises(TypeError, match="unexpected keyword argument 'index_name'"):
+        video.semantic_search("find cars", index_name="scene")
+    with pytest.raises(TypeError, match="unexpected keyword argument 'index_id'"):
+        video.semantic_search("find cars", index_id="idx-old")
+    assert conn.calls == []
+
+
+def test_query_supports_singular_index_name_emits_and_exposes_warnings():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2"):
+        response = video.query(index_name="scene")
+
+    assert conn.calls[-1] == (
+        "video/v1/query",
+        {
+            "index_name": "scene",
+            "index_id": None,
+            "filter": None,
+            "limit": 100,
+            "return_fields": None,
+            "sort": None,
+        },
+    )
+    assert response.warnings == [WARNING]
+    assert response.shots[0].text == "query"
+
+
+def test_query_supports_singular_index_id():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2"):
+        response = video.query(index_id="idx-v2")
+
+    assert conn.calls[-1][0] == "video/v1/query"
+    assert conn.calls[-1][1]["index_name"] is None
+    assert conn.calls[-1][1]["index_id"] == "idx-v2"
+    assert response.warnings == [WARNING]
+
+
+def test_query_rejects_plural_selectors_in_sdk_signature():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.raises(TypeError, match="unexpected keyword argument 'index_ids'"):
+        video.query(index_ids=["idx-v2"])
+    assert conn.calls == []
+
+
+def test_aggregate_keeps_singular_index_name_emits_and_returns_raw_warning_payload():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2"):
+        response = video.aggregate(index_name="scene")
+
+    assert conn.calls[-1] == (
+        "video/v1/aggregate",
+        {
+            "index_name": "scene",
+            "index_id": None,
+            "filter": None,
+            "group_by": None,
+            "metric": "count",
+            "limit": 100,
+            "sort": None,
+        },
+    )
+    assert response["warnings"] == [WARNING]
+
+
+def test_aggregate_keeps_singular_index_id():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"search\(\) now uses Search V2"):
+        response = video.aggregate(index_id="idx-v2")
+
+    assert conn.calls[-1][0] == "video/v1/aggregate"
+    assert conn.calls[-1][1]["index_name"] is None
+    assert conn.calls[-1][1]["index_id"] == "idx-v2"
+    assert response["warnings"] == [WARNING]
+
+
+def test_search_index_id_routes_video_to_legacy_with_python_warning_and_scene_index_id_mapping():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match="This search used legacy search because legacy parameters were provided") as emitted:
+        response = video.search("find cars", index_id="old-scene-index")
+
+    assert len(emitted) == 1
+    assert conn.calls[-1][0] == "video/v1/search"
+    assert conn.calls[-1][1]["scene_index_id"] == "old-scene-index"
+    assert response.warnings == []
+    assert response.shots[0].text == "legacy"
+
+
+def test_search_index_id_routes_collection_to_legacy_with_python_warning_and_scene_index_id_mapping():
+    conn = FakeCollectionConnection()
+    collection = Collection(conn, id="c1")
+
+    with pytest.warns(UserWarning, match="This search used legacy search because legacy parameters were provided"):
+        response = collection.search("find cars", index_id="old-scene-index")
+
+    assert conn.calls[-1][0] == "collection/c1/search"
+    assert conn.calls[-1][1]["scene_index_id"] == "old-scene-index"
+    assert response.warnings == []
+
+
+def test_explicit_legacy_search_warns_once_and_uses_old_endpoint():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.warns(UserWarning, match=r"legacy_search\(\) searches older spoken-word and scene indexes only") as emitted:
+        video.legacy_search("find cars")
+        video.legacy_search("find people")
+
+    assert len(emitted) == 1
+    assert [call[0] for call in conn.calls] == ["video/v1/search", "video/v1/search"]
+
+
+def test_search_rejects_mixed_legacy_and_v2_params_before_http():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.raises(ValueError, match="Cannot mix legacy search parameters with Search V2 parameters"):
+        video.search("find cars", index_id="old-scene-index", top_k=5)
+    assert conn.calls == []
+
+
+def test_search_rejects_index_selectors_before_http():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.raises(ValueError, match=r"search\(\) chooses indexes automatically"):
+        video.search("find cars", index_names=["scene"])
+    with pytest.raises(ValueError, match=r"search\(\) chooses indexes automatically"):
+        video.search("find cars", index_ids=["idx-v2"])
+    assert conn.calls == []
+
+
+def test_search_rejects_internal_deepsearch_config_before_http():
+    conn = FakeConnection()
+    video = Video(conn, id="v1", collection_id="c1")
+
+    with pytest.raises(ValueError, match=r"deepsearch_config is not a public search\(\) option"):
+        video.search("find cars", deepsearch_config={"internal": True})
+    assert conn.calls == []
diff --git a/videodb/__about__.py b/videodb/__about__.py
index b4566c7..7f103dd 100644
--- a/videodb/__about__.py
+++ b/videodb/__about__.py
@@ -2,7 +2,7 @@
 
 
 
-__version__ = "0.4.5"
+__version__ = "0.5.0"
 __title__ = "videodb"
 __author__ = "videodb"
 __email__ = "contact@videodb.io"
diff --git a/videodb/__init__.py b/videodb/__init__.py
index 59d2a1b..2282292 100644
--- a/videodb/__init__.py
+++ b/videodb/__init__.py
@@ -8,6 +8,8 @@
 from videodb._constants import (
     VIDEO_DB_API,
     IndexType,
+    IndexCapability,
+    FieldGroup,
     SceneExtractionType,
     MediaType,
     SearchType,
@@ -23,8 +25,16 @@
     ReframeMode,
     SegmentationType,
     RTStreamChannelType,
+    SandboxTier,
+    SandboxStatus,
 )
 from videodb.client import Connection
+from videodb.search import AskResponse, SearchResponse, SearchResult
+from videodb.understanding import Understanding, UnderstandingAnalyzer
+from videodb.job import GenerationJob
+from videodb.sandbox import Sandbox
+from videodb.sandbox_models import SandboxModel
+from videodb.voice_clone import VoiceClone
 from videodb.capture_session import CaptureSession
 from videodb.websocket_client import WebSocketConnection
 from videodb.capture import CaptureClient, Channel, AudioChannel, VideoChannel, Channels, ChannelList
@@ -42,6 +52,10 @@
 __all__ = [
     "connect",
     "CaptureSession",
+    "GenerationJob",
+    "Sandbox",
+    "SandboxModel",
+    "VoiceClone",
     "WebSocketConnection",
     "CaptureClient",
     "Channel",
@@ -53,7 +67,14 @@
     "AuthenticationError",
     "InvalidRequestError",
     "IndexType",
+    "IndexCapability",
+    "FieldGroup",
     "SearchError",
+    "SearchResult",
+    "SearchResponse",
+    "AskResponse",
+    "Understanding",
+    "UnderstandingAnalyzer",
     "play_stream",
     "build_iframe_embed_code",
     "MediaType",
@@ -71,6 +92,8 @@
     "ReframeMode",
     "SegmentationType",
     "RTStreamChannelType",
+    "SandboxTier",
+    "SandboxStatus",
 ]
 
 
diff --git a/videodb/_constants.py b/videodb/_constants.py
index f7de578..b13a889 100644
--- a/videodb/_constants.py
+++ b/videodb/_constants.py
@@ -36,6 +36,23 @@ class IndexType:
     scene = "scene"
 
 
+class IndexCapability:
+    """Retrieval capabilities an index can be built for (``use_for``)."""
+
+    semantic = "semantic"
+    query = "query"
+    aggregate = "aggregate"
+
+
+class FieldGroup:
+    """Field groups that map artifact fields to retrieval capabilities."""
+
+    semantic = "semantic"
+    filter = "filter"
+    aggregate = "aggregate"
+    sort = "sort"
+
+
 class SceneExtractionType:
     shot_based = "shot"
     time_based = "time"
@@ -80,7 +97,14 @@ class ApiPath:
     upload_url = "upload_url"
     transcription = "transcription"
     index = "index"
+    indexes = "indexes"
+    records = "records"
+    understand = "understand"
     search = "search"
+    ask = "ask"
+    semantic_search = "semantic-search"
+    query = "query"
+    aggregate = "aggregate"
     compile = "compile"
     workflow = "workflow"
     timeline = "timeline"
@@ -116,11 +140,36 @@ class ApiPath:
     token = "token"
     websocket = "websocket"
     export = "export"
+    job = "job"
+    sandbox = "sandbox"
+    voice_clone = "voice_clone"
+    understand = "understand"
+    indexes = "indexes"
+    identities = "identities"
+    merge = "merge"
+    split = "split"
+    async_response = "async-response"
+
 
 
 class Status:
     processing = "processing"
     in_progress = "in progress"
+    complete = "complete"
+
+
+class SandboxTier:
+    small = "small"
+    medium = "medium"
+
+
+class SandboxStatus:
+    provisioning = "provisioning"
+    active = "active"
+    stopping = "stopping"
+    stopped = "stopped"
+    failed = "failed"
+    alert = "alert"
 
 
 class MeetingStatus:
@@ -135,6 +184,8 @@ class HttpClientDefaultValues:
     timeout = 30
     backoff_factor = 0.1
     status_forcelist = [502, 503, 504]
+    max_poll_time = 500
+    poll_interval = 5
 
 
 class MaxSupported:
diff --git a/videodb/_upload.py b/videodb/_upload.py
index 0291abe..0a25e5d 100644
--- a/videodb/_upload.py
+++ b/videodb/_upload.py
@@ -1,9 +1,9 @@
-import requests
-
-from typing import Optional
+import os
+from typing import Optional, Union
 from urllib.parse import urlparse
+
+import requests
 from requests import HTTPError
-import os
 
 
 from videodb._constants import (
@@ -20,6 +20,31 @@ def _is_url(path: str) -> bool:
     return all([parsed.scheme in ("http", "https"), parsed.netloc])
 
 
+def upload_bytes(
+    _connection,
+    content: Union[str, bytes],
+    name: str,
+    content_type: str = "application/octet-stream",
+    collection_id: Optional[str] = None,
+) -> str:
+    """Upload in-memory content using a presigned upload URL and return the object URL."""
+    collection_id = collection_id or _connection.collection_id
+    upload_url_data = _connection.get(
+        path=f"{ApiPath.collection}/{collection_id}/{ApiPath.upload_url}",
+        params={"name": name},
+    )
+    upload_url = upload_url_data.get("upload_url")
+
+    try:
+        files = {"file": (name, content, content_type)}
+        response = requests.post(upload_url, files=files)
+        response.raise_for_status()
+    except requests.exceptions.RequestException as e:
+        raise VideodbError("Error while uploading content", cause=e)
+
+    return upload_url
+
+
 def upload(
     _connection,
     source: Optional[str] = None,
@@ -71,16 +96,13 @@ def upload(
     if file_path:
         try:
             name = os.path.splitext(os.path.basename(file_path))[0] if not name else name
-            upload_url_data = _connection.get(
-                path=f"{ApiPath.collection}/{collection_id}/{ApiPath.upload_url}",
-                params={"name": name},
-            )
-            upload_url = upload_url_data.get("upload_url")
             with open(file_path, "rb") as file:
-                files = {"file": (name, file)}
-                response = requests.post(upload_url, files=files)
-                response.raise_for_status()
-                url = upload_url
+                url = upload_bytes(
+                    _connection=_connection,
+                    content=file,
+                    name=name,
+                    collection_id=collection_id,
+                )
 
         except FileNotFoundError as e:
             raise VideodbError("File not found", cause=e)
diff --git a/videodb/_utils/_http_client.py b/videodb/_utils/_http_client.py
index ab571f6..6507e38 100644
--- a/videodb/_utils/_http_client.py
+++ b/videodb/_utils/_http_client.py
@@ -1,8 +1,8 @@
 """Http client Module."""
 
 import logging
+import time
 import requests
-import backoff
 
 from tqdm import tqdm
 from typing import (
@@ -65,6 +65,8 @@ def __init__(
         self.base_url = base_url
         self.show_progress = False
         self.progress_bar = None
+        self.max_poll_time = HttpClientDefaultValues.max_poll_time
+        self.poll_interval = HttpClientDefaultValues.poll_interval
         logger.debug(f"Initialized http client with base url: {self.base_url}")
 
     def _make_request(
@@ -73,6 +75,7 @@ def _make_request(
         path: str,
         base_url: Optional[str] = None,
         headers: Optional[dict] = None,
+        wait: bool = True,
         **kwargs,
     ):
         """Make a request to the api
@@ -81,6 +84,7 @@ def _make_request(
         :param str path: The path to make the request to
         :param str base_url: (optional) The base url to use for the request
         :param dict headers: (optional) The headers to use for the request
+        :param bool wait: If False, return raw data without polling (default True)
         :param kwargs: The keyword arguments to pass to the request method
         :return: json response from the request
         """
@@ -90,6 +94,11 @@ def _make_request(
             request_headers = {**self.session.headers, **(headers or {})}
             response = method(url, headers=request_headers, timeout=timeout, **kwargs)
             response.raise_for_status()
+            if not wait:
+                data = response.json().get("data")
+                if data is not None:
+                    return data
+                return response.json()
             return self._parse_response(response)
 
         except requests.exceptions.RequestException as e:
@@ -133,23 +142,31 @@ def _handle_request_error(self, e: requests.exceptions.RequestException) -> None
                 f"Invalid request: {str(e)}", e.response
             ) from None
 
-    @backoff.on_exception(
-        backoff.constant, Exception, max_time=500, interval=5, logger=None, jitter=None
-    )
     def _get_output(self, url: str):
-        """Get the output from an async request"""
-        response_json = self.session.get(url).json()
-        if (
-            response_json.get("status") == Status.in_progress
-            or response_json.get("status") == Status.processing
-        ):
+        """Poll an output URL until the job completes or times out."""
+        start = time.monotonic()
+        while True:
+            response_json = self.session.get(url).json()
+            status = response_json.get("status")
+            if status not in (Status.in_progress, Status.processing):
+                break
+
             percentage = response_json.get("data", {}).get("percentage")
             if percentage and self.show_progress and self.progress_bar:
                 self.progress_bar.n = int(percentage)
                 self.progress_bar.update(0)
 
-            logger.debug("Waiting for processing to complete")
-            raise Exception("Stuck on processing status") from None
+            elapsed = time.monotonic() - start
+            if elapsed >= self.max_poll_time:
+                raise RequestTimeoutError(
+                    f"Polling timed out after {int(elapsed)}s "
+                    f"(max_poll_time={self.max_poll_time})",
+                    None,
+                )
+
+            logger.debug("Waiting for processing to complete (%.0fs elapsed)", elapsed)
+            time.sleep(self.poll_interval)
+
         if self.show_progress and self.progress_bar:
             self.progress_bar.n = 100
             self.progress_bar.update(0)
@@ -209,19 +226,31 @@ def _format_headers(self, headers: dict):
             formatted_headers[f"x-{key}"] = value
         return formatted_headers
 
+    def _apply_poll_overrides(self, kwargs):
+        """Extract and apply per-call poll overrides from kwargs."""
+        self.max_poll_time = kwargs.pop(
+            "max_poll_time", HttpClientDefaultValues.max_poll_time
+        )
+        self.poll_interval = kwargs.pop(
+            "poll_interval", HttpClientDefaultValues.poll_interval
+        )
+
     def get(
-        self, path: str, show_progress: Optional[bool] = False, **kwargs
+        self, path: str, show_progress: Optional[bool] = False, wait: bool = True, **kwargs
     ) -> requests.Response:
         """Make a get request"""
         self.show_progress = show_progress
-        return self._make_request(method=self.session.get, path=path, **kwargs)
+        self._apply_poll_overrides(kwargs)
+        return self._make_request(method=self.session.get, path=path, wait=wait, **kwargs)
 
     def post(
-        self, path: str, data=None, show_progress: Optional[bool] = False, **kwargs
+        self, path: str, data=None, show_progress: Optional[bool] = False,
+        wait: bool = True, **kwargs,
     ) -> requests.Response:
         """Make a post request"""
         self.show_progress = show_progress
-        return self._make_request(self.session.post, path, json=data, **kwargs)
+        self._apply_poll_overrides(kwargs)
+        return self._make_request(self.session.post, path, json=data, wait=wait, **kwargs)
 
     def put(self, path: str, data=None, **kwargs) -> requests.Response:
         """Make a put request"""
diff --git a/videodb/client.py b/videodb/client.py
index 7eac7bb..28f1d4d 100644
--- a/videodb/client.py
+++ b/videodb/client.py
@@ -1,4 +1,5 @@
 import logging
+import requests
 
 from typing import (
     Optional,
@@ -11,6 +12,7 @@
     TranscodeMode,
     VideoConfig,
     AudioConfig,
+    HttpClientDefaultValues,
 )
 
 from videodb.collection import Collection
@@ -19,6 +21,8 @@
 from videodb.audio import Audio
 from videodb.image import Image
 from videodb.meeting import Meeting
+from videodb.sandbox import Sandbox
+from videodb.voice_clone import VoiceClone
 from videodb.capture_session import CaptureSession
 from videodb.websocket_client import WebSocketConnection
 
@@ -264,6 +268,165 @@ def get_transcode_details(self, job_id: str) -> dict:
         """
         return self.get(path=f"{ApiPath.transcode}/{job_id}")
 
+    def get_job_status(self, job_id: str) -> dict:
+        """Get status/details for a self-inference generation job.
+
+        This preserves and normalizes the job response wrapper instead of using
+        the standard SDK response parser, because callers need the top-level
+        job status while polling.
+
+        :param str job_id: ID of the generation job
+        :return: Normalized job response with success, status, data, message
+        :rtype: dict
+        """
+        try:
+            url = f"{self.base_url}/{ApiPath.job}/{job_id}"
+            response = self.session.get(url, timeout=HttpClientDefaultValues.timeout)
+            response.raise_for_status()
+            response_json = response.json()
+        except requests.exceptions.RequestException as e:
+            self._handle_request_error(e)
+        except ValueError:
+            from videodb.exceptions import InvalidRequestError
+
+            raise InvalidRequestError("Invalid request: Unable to parse job response") from None
+
+        data = response_json.get("data") or {}
+        status = response_json.get("status") or data.get("status")
+        success = response_json.get("success", False)
+
+        if not status:
+            status = "done" if success else "failed"
+
+        return {
+            "success": success,
+            "status": status,
+            "data": data,
+            "message": response_json.get("message"),
+        }
+
+    def wait_for_job(
+        self,
+        job_id: str,
+        timeout: int = 600,
+        interval: int = 5,
+        result_type: str = None,
+    ):
+        """Poll a self-inference generation job until completion.
+
+        :param str job_id: ID of the generation job
+        :param int timeout: Maximum seconds to wait
+        :param int interval: Seconds between polls
+        :param str result_type: Optional expected result type, "audio" or "image"
+        :return: Generated SDK asset or final job data
+        """
+        from videodb.job import GenerationJob
+
+        return GenerationJob(
+            self, job_id=job_id, result_type=result_type
+        ).wait(timeout=timeout, interval=interval)
+
+    def create_sandbox(
+        self,
+        tier: Optional[str] = None,
+        name: Optional[str] = None,
+        callback_url: Optional[str] = None,
+    ) -> "Sandbox":
+        """Create a new sandbox (GPU compute pool).
+
+        :param str tier: Sandbox tier — "small" or "medium" (default: server decides)
+        :param str name: Human-readable name (auto-generated if not provided)
+        :param str callback_url: URL to receive sandbox lifecycle webhooks
+        :return: :class:`Sandbox <Sandbox>` object in provisioning state
+        :rtype: :class:`videodb.sandbox.Sandbox`
+        """
+        data = self.post(
+            path=ApiPath.sandbox,
+            data={"tier": tier, "name": name, "callback_url": callback_url},
+        )
+        return Sandbox(self, **(data or {}))
+
+    def get_sandbox(self, sandbox_id: str) -> "Sandbox":
+        """Get a sandbox by ID.
+
+        :param str sandbox_id: The sandbox ID
+        :return: :class:`Sandbox <Sandbox>` object
+        :rtype: :class:`videodb.sandbox.Sandbox`
+        """
+        data = self.get(path=f"{ApiPath.sandbox}/{sandbox_id}")
+        return Sandbox(self, **(data or {}))
+
+    def list_sandboxes(self, status: Optional[str] = None) -> List["Sandbox"]:
+        """List all sandboxes, optionally filtered by status.
+
+        :param str status: Filter by sandbox status (optional)
+        :return: List of :class:`Sandbox <Sandbox>` objects
+        :rtype: list[:class:`videodb.sandbox.Sandbox`]
+        """
+        params = {}
+        if status:
+            params["status"] = status
+        data = self.get(path=ApiPath.sandbox, params=params)
+        sandboxes_data = (data or {}).get("sandboxes", [])
+        return [Sandbox(self, **s) for s in sandboxes_data]
+
+    def create_voice_clone(
+        self,
+        ref_audio_id: str,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        ref_text: Optional[str] = None,
+        language: Optional[str] = None,
+        collection_id: Optional[str] = None,
+    ) -> "VoiceClone":
+        """Create a reusable voice clone from an existing audio asset.
+
+        :param str ref_audio_id: Source audio ID to use as the voice reference.
+        :param str name: Human-readable name (optional).
+        :param str description: Description (optional).
+        :param str ref_text: Text spoken in the reference audio (optional).
+        :param str language: Language code, e.g. ``"en"`` (optional).
+        :param str collection_id: Collection associated with the source audio (optional).
+        :return: :class:`VoiceClone <VoiceClone>` object.
+        :rtype: :class:`videodb.voice_clone.VoiceClone`
+        """
+        data = self.post(
+            path=ApiPath.voice_clone,
+            data={
+                "ref_audio_id": ref_audio_id,
+                "name": name,
+                "description": description,
+                "ref_text": ref_text,
+                "language": language,
+                "collection_id": collection_id,
+            },
+        )
+        return VoiceClone(self, **(data or {}))
+
+    def get_voice_clone(self, voice_clone_id: str) -> "VoiceClone":
+        """Get a voice clone by ID."""
+        data = self.get(path=f"{ApiPath.voice_clone}/{voice_clone_id}")
+        return VoiceClone(self, **(data or {}))
+
+    def list_voice_clones(
+        self,
+        page: int = 1,
+        page_size: int = 20,
+        language: Optional[str] = None,
+    ) -> List["VoiceClone"]:
+        """List voice clones for the current user."""
+        params = {"page": page, "page_size": page_size}
+        if language:
+            params["language"] = language
+        data = self.get(path=ApiPath.voice_clone, params=params)
+        voice_clones = (data or {}).get("voice_clones", [])
+        return [VoiceClone(self, **v) for v in voice_clones]
+
+    def delete_voice_clone(self, voice_clone_id: str) -> None:
+        """Delete a voice clone by ID."""
+        self.delete(path=f"{ApiPath.voice_clone}/{voice_clone_id}")
+        return None
+
     def upload(
         self,
         source: Optional[str] = None,
@@ -494,3 +657,12 @@ def generate_client_token(self, expires_in: int = 86400) -> str:
             data={"expires_in": expires_in},
         )
         return response.get("token")
+
+    def get_async_response(self, id: str) -> dict:
+        """Get the details of an async response.
+
+        :param str id: ID of the async response
+        :return: Details of the async response
+        :rtype: dict
+        """
+        return self.get(path=f"{ApiPath.async_response}/{id}", wait=False)
diff --git a/videodb/collection.py b/videodb/collection.py
index 7f57fc3..c8e1ea7 100644
--- a/videodb/collection.py
+++ b/videodb/collection.py
@@ -1,8 +1,11 @@
+import json
 import logging
+import uuid
 
-from typing import Optional, Union, List, Dict, Any, Literal
+from typing import Optional, Union, List, Dict, Any, Literal, Tuple
 from videodb._upload import (
     upload,
+    upload_bytes,
 )
 from videodb._constants import (
     ApiPath,
@@ -14,13 +17,25 @@
 from videodb.video import Video
 from videodb.audio import Audio
 from videodb.image import Image
+from videodb.job import GenerationJob
+from videodb.voice_clone import VoiceClone
 from videodb.meeting import Meeting
 from videodb.capture_session import CaptureSession
 from videodb.rtstream import RTStream, RTStreamSearchResult, RTStreamShot
-from videodb.search import SearchFactory, SearchResult
+from videodb.search import (
+    AskResponse,
+    SearchFactory,
+    SearchResponse,
+    SearchResult,
+    warn_explicit_legacy_search_once,
+    warn_legacy_search_once,
+    warn_response_warnings_once,
+)
 
 logger = logging.getLogger(__name__)
 
+MAX_GENERATE_TEXT_PAYLOAD_SIZE = 250 * 1024
+
 
 class Collection:
     """Collection class to interact with the Collection.
@@ -133,6 +148,54 @@ def delete_audio(self, audio_id: str) -> None:
             path=f"{ApiPath.audio}/{audio_id}", params={"collection_id": self.id}
         )
 
+    def create_voice_clone(
+        self,
+        ref_audio_id: str,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        ref_text: Optional[str] = None,
+        language: Optional[str] = None,
+    ) -> VoiceClone:
+        """Create a reusable voice clone from an audio in this collection.
+
+        :param str ref_audio_id: Source audio ID to use as the voice reference.
+        :param str name: Human-readable name (optional).
+        :param str description: Description (optional).
+        :param str ref_text: Text spoken in the reference audio (optional).
+        :param str language: Language code, e.g. ``"en"`` (optional).
+        :return: :class:`VoiceClone <VoiceClone>` object.
+        :rtype: :class:`videodb.voice_clone.VoiceClone`
+        """
+        return self._connection.create_voice_clone(
+            ref_audio_id=ref_audio_id,
+            name=name,
+            description=description,
+            ref_text=ref_text,
+            language=language,
+            collection_id=self.id,
+        )
+
+    def get_voice_clone(self, voice_clone_id: str) -> VoiceClone:
+        """Get a voice clone by ID."""
+        return self._connection.get_voice_clone(voice_clone_id)
+
+    def list_voice_clones(
+        self,
+        page: int = 1,
+        page_size: int = 20,
+        language: Optional[str] = None,
+    ) -> List[VoiceClone]:
+        """List user voice clones, optionally filtered by language."""
+        return self._connection.list_voice_clones(
+            page=page,
+            page_size=page_size,
+            language=language,
+        )
+
+    def delete_voice_clone(self, voice_clone_id: str) -> None:
+        """Delete a voice clone by ID."""
+        return self._connection.delete_voice_clone(voice_clone_id)
+
     def get_images(self) -> List[Image]:
         """Get all the images in the collection.
 
@@ -274,25 +337,51 @@ def generate_image(
         prompt: str,
         aspect_ratio: Optional[Literal["1:1", "9:16", "16:9", "4:3", "3:4"]] = "1:1",
         callback_url: Optional[str] = None,
-    ) -> Image:
+        model_name: Optional[str] = None,
+        config: Optional[dict] = None,
+        sandbox_id: Optional[str] = None,
+        wait: bool = False,
+        poll_interval: int = 5,
+        timeout: int = 600,
+    ) -> Union[Image, GenerationJob]:
         """Generate an image from a prompt.
 
         :param str prompt: Prompt for the image generation
-        :param str aspect_ratio: Aspect ratio of the image (optional)
+        :param str aspect_ratio: Aspect ratio of the image (optional, hosted models)
         :param str callback_url: URL to receive the callback (optional)
-        :return: :class:`Image <Image>` object
-        :rtype: :class:`videodb.image.Image`
+        :param str model_name: Model name. Use ``"black-forest-labs/FLUX.1-dev"`` for FLUX self-inference.
+        :param dict config: Model configuration. Used by FLUX.
+        :param str sandbox_id: ID of the sandbox to route the self-inference job to (optional).
+        :param bool wait: If True, wait for self-inference jobs and return Image.
+        :param int poll_interval: Seconds between job polls when wait=True.
+        :param int timeout: Maximum seconds to wait when wait=True.
+        :return: :class:`Image <Image>` or :class:`GenerationJob <GenerationJob>`
+        :rtype: Union[:class:`videodb.image.Image`, :class:`videodb.job.GenerationJob`]
         """
+        payload = {
+            "prompt": prompt,
+            "aspect_ratio": aspect_ratio,
+            "callback_url": callback_url,
+        }
+        if model_name:
+            payload["model_name"] = model_name
+        if config is not None:
+            payload["config"] = config
+        if sandbox_id:
+            payload["sandbox_id"] = sandbox_id
+
         image_data = self._connection.post(
             path=f"{ApiPath.collection}/{self.id}/{ApiPath.generate}/{ApiPath.image}",
-            data={
-                "prompt": prompt,
-                "aspect_ratio": aspect_ratio,
-                "callback_url": callback_url,
-            },
+            data=payload,
         )
-        if image_data:
-            return Image(self._connection, **image_data)
+        if not image_data:
+            return None
+        if image_data.get("job_id"):
+            job = GenerationJob.from_data(
+                self._connection, image_data, result_type="image"
+            )
+            return job.wait(timeout=timeout, interval=poll_interval) if wait else job
+        return Image(self._connection, **image_data)
 
     def generate_music(
         self, prompt: str, duration: int = 5, callback_url: Optional[str] = None
@@ -352,28 +441,55 @@ def generate_voice(
         voice_name: str = "Default",
         config: dict = {},
         callback_url: Optional[str] = None,
-    ) -> Audio:
+        model_name: str = "elevenlabs",
+        sandbox_id: Optional[str] = None,
+        voice_clone_id: Optional[str] = None,
+        clone_voice_id: Optional[str] = None,
+        wait: bool = False,
+        poll_interval: int = 5,
+        timeout: int = 600,
+    ) -> Union[Audio, GenerationJob]:
         """Generate voice from text.
 
         :param str text: Text to convert to voice
         :param str voice_name: Name of the voice to use
         :param dict config: Configuration for the voice generation
         :param str callback_url: URL to receive the callback (optional)
-        :return: :class:`Audio <Audio>` object
-        :rtype: :class:`videodb.audio.Audio`
+        :param str model_name: Model name. Use ``"k2-fsa/OmniVoice"`` for OmniVoice.
+        :param str sandbox_id: ID of the sandbox to route the self-inference job to (optional).
+        :param str voice_clone_id: ID of a reusable voice clone to use for OmniVoice (optional).
+        :param str clone_voice_id: Alias for ``voice_clone_id`` (optional).
+        :param bool wait: If True, wait for self-inference jobs and return Audio.
+        :param int poll_interval: Seconds between job polls when wait=True.
+        :param int timeout: Maximum seconds to wait when wait=True.
+        :return: :class:`Audio <Audio>` or :class:`GenerationJob <GenerationJob>`
+        :rtype: Union[:class:`videodb.audio.Audio`, :class:`videodb.job.GenerationJob`]
         """
+        if voice_clone_id and clone_voice_id and voice_clone_id != clone_voice_id:
+            raise ValueError("voice_clone_id and clone_voice_id cannot both be different")
+        resolved_voice_clone_id = voice_clone_id or clone_voice_id
+
         audio_data = self._connection.post(
             path=f"{ApiPath.collection}/{self.id}/{ApiPath.generate}/{ApiPath.audio}",
             data={
                 "text": text,
                 "audio_type": "voice",
                 "voice_name": voice_name,
+                "model_name": model_name,
                 "config": config,
                 "callback_url": callback_url,
+                "sandbox_id": sandbox_id,
+                "voice_clone_id": resolved_voice_clone_id,
             },
         )
-        if audio_data:
-            return Audio(self._connection, **audio_data)
+        if not audio_data:
+            return None
+        if audio_data.get("job_id"):
+            job = GenerationJob.from_data(
+                self._connection, audio_data, result_type="audio"
+            )
+            return job.wait(timeout=timeout, interval=poll_interval) if wait else job
+        return Audio(self._connection, **audio_data)
 
     def generate_video(
         self,
@@ -417,25 +533,68 @@ def generate_video(
     def generate_text(
         self,
         prompt: str,
-        model_name: Literal["basic", "pro", "ultra"] = "basic",
+        model_name: str = "basic",
         response_type: Literal["text", "json"] = "text",
+        wait: bool = True,
+        callback_url: Optional[str] = None,
+        sandbox_id: Optional[str] = None,
+        max_tokens: Optional[int] = None,
+        temperature: Optional[float] = None,
+        model_config: Optional[Dict[str, Any]] = None,
     ) -> Union[str, dict]:
         """Generate text from a prompt using genai offering.
 
+        Small prompts are sent inline as JSON. When the serialized request body
+        approaches the observed gateway limit (~256 KB), the prompt is uploaded
+        via the collection presigned upload URL and referenced as ``prompt_url``
+        to avoid API Gateway/Lambda payload size limits.
+
         :param str prompt: Prompt for the text generation
-        :param str model_name: Model name to use ("basic", "pro" or "ultra")
+        :param str model_name: Model name to use ("basic", "pro", "ultra", or a self-hosted model)
         :param str response_type: Desired response type ("text" or "json")
-        :return: Generated text response
+        :param bool wait: Wait for the text generation to complete (default: True)
+        :param str callback_url: URL to receive the callback (optional)
+        :param str sandbox_id: Sandbox ID to use for self-hosted models (optional)
+        :param int max_tokens: Maximum tokens to generate (optional)
+        :param float temperature: Sampling temperature (optional)
+        :param dict model_config: Additional self-hosted model configuration (optional)
+        :return: Generated text response if wait is False, otherwise job id of the text generation
         :rtype: Union[str, dict]
         """
+        payload = {
+            "prompt": prompt,
+            "model_name": model_name,
+            "response_type": response_type,
+            "callback_url": callback_url,
+            "sandbox_id": sandbox_id,
+            "max_tokens": max_tokens,
+            "temperature": temperature,
+            "model_config": model_config,
+        }
 
-        return self._connection.post(
-            path=f"{ApiPath.collection}/{self.id}/{ApiPath.generate}/{ApiPath.text}",
-            data={
-                "prompt": prompt,
+        payload_size = len(json.dumps(payload).encode("utf-8"))
+        if payload_size > MAX_GENERATE_TEXT_PAYLOAD_SIZE:
+            payload = {
+                "prompt_url": upload_bytes(
+                    _connection=self._connection,
+                    content=prompt,
+                    name=f"generate_text_prompt_{uuid.uuid4().hex}.txt",
+                    content_type="text/plain; charset=utf-8",
+                    collection_id=self.id,
+                ),
                 "model_name": model_name,
                 "response_type": response_type,
-            },
+                "callback_url": callback_url,
+                "sandbox_id": sandbox_id,
+                "max_tokens": max_tokens,
+                "temperature": temperature,
+                "model_config": model_config,
+            }
+
+        return self._connection.post(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.generate}/{ApiPath.text}",
+            data=payload,
+            wait=wait,
         )
 
     def dub_video(
@@ -463,6 +622,180 @@ def dub_video(
             return Video(self._connection, **dub_data)
 
     def search(
+        self,
+        query: str,
+        *args,
+        config: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> Union[SearchResponse, SearchResult, RTStreamSearchResult]:
+        """Search the collection.
+
+        New search is used by default. Calls that use legacy-shaped parameters are
+        routed to :meth:`legacy_search` with a warning.
+        """
+        old_params = {
+            "search_type",
+            "index_type",
+            "result_threshold",
+            "dynamic_score_percentage",
+            "scene_index_id",
+            "index_id",
+            "algorithm",
+            "sort_docs_on",
+            "namespace",
+        }
+        new_params = {
+            "top_k",
+            "mode",
+            "return_fields",
+            "include_clip",
+            "session_id",
+            "config",
+        }
+        unsupported_params = {"index_name", "index_names", "index_ids"}
+
+        if config is not None:
+            kwargs["config"] = config
+
+        if args:
+            legacy_arg_names = [
+                "search_type",
+                "index_type",
+                "result_threshold",
+                "score_threshold",
+                "dynamic_score_percentage",
+                "filter",
+                "sort_docs_on",
+                "namespace",
+                "scene_index_id",
+            ]
+            for name, value in zip(legacy_arg_names, args):
+                kwargs.setdefault(name, value)
+
+        has_old = bool(args) or any(k in kwargs and kwargs[k] is not None for k in old_params)
+        has_new = any(k in kwargs and kwargs[k] is not None for k in new_params)
+        has_unsupported = any(k in kwargs and kwargs[k] is not None for k in unsupported_params)
+
+        if kwargs.get("deepsearch_config") is not None:
+            raise ValueError(
+                "deepsearch_config is not a public search() option. Use mode='deepsearch', top_k, session_id, and return_fields for DeepSearch requests."
+            )
+        if has_old and (has_new or has_unsupported):
+            raise ValueError(
+                "Cannot mix legacy search parameters with Search V2 parameters. "
+                "Use legacy_search(...) for older indexes, or remove legacy parameters and use Search V2 search(...)."
+            )
+        if has_unsupported:
+            raise ValueError(
+                "search() chooses indexes automatically and does not accept index selectors. "
+                "Use semantic_search() for semantic index selection, query() for structured filtering, or aggregate() for counts and facets."
+            )
+
+        if has_old:
+            warn_legacy_search_once()
+            return self.legacy_search(query=query, _skip_warning=True, **kwargs)
+
+        return self._new_search(query=query, **kwargs)
+
+    def _new_search(self, query: str, **kwargs) -> SearchResponse:
+        payload = {"query": query, **{k: v for k, v in kwargs.items() if v is not None}}
+        search_data = self._connection.post(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.search}/v2",
+            data=payload,
+        )
+        return SearchResponse(self._connection, **search_data)
+
+    def ask(
+        self,
+        question: str,
+        top_k: int = 15,
+        mode: str = "default",
+        include_sources: bool = False,
+    ) -> AskResponse:
+        ask_data = self._connection.post(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.ask}",
+            data={
+                "question": question,
+                "top_k": top_k,
+                "mode": mode,
+                "include_sources": include_sources,
+            },
+        )
+        return AskResponse(self._connection, **ask_data)
+
+    def semantic_search(
+        self,
+        query: str,
+        index_names: Optional[Union[List[str], str]] = None,
+        top_k: int = 10,
+        score_threshold: Optional[float] = None,
+        filter: Optional[Union[List, Dict]] = None,
+        return_fields: Optional[Union[List, Dict, str]] = None,
+        index_ids: Optional[Union[List[str], str]] = None,
+    ) -> SearchResult:
+        search_data = self._connection.post(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.semantic_search}",
+            data={
+                "query": query,
+                "index_names": index_names,
+                "index_ids": index_ids,
+                "top_k": top_k,
+                "score_threshold": score_threshold,
+                "filter": filter,
+                "return_fields": return_fields,
+            },
+        )
+        return SearchResult(self._connection, **search_data)
+
+    def query(
+        self,
+        index_name: Optional[str] = None,
+        filter: Optional[Union[List, Dict]] = None,
+        limit: int = 100,
+        return_fields: Optional[Union[List, Dict, str]] = None,
+        sort: Optional[Union[str, List[Tuple[str, str]]]] = None,
+        index_id: Optional[str] = None,
+    ) -> SearchResult:
+        query_data = self._connection.post(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.query}",
+            data={
+                "index_name": index_name,
+                "index_id": index_id,
+                "filter": filter,
+                "limit": limit,
+                "return_fields": return_fields,
+                "sort": sort,
+            },
+        )
+        return SearchResult(self._connection, **query_data)
+
+    def aggregate(
+        self,
+        index_name: Optional[str] = None,
+        filter: Optional[Union[List, Dict]] = None,
+        group_by: Optional[str] = None,
+        metric: str = "count",
+        limit: int = 100,
+        sort: Optional[Union[str, List[Tuple[str, str]]]] = None,
+        index_id: Optional[str] = None,
+    ) -> Union[Dict, List[Dict]]:
+        aggregate_data = self._connection.post(
+            path=f"{ApiPath.collection}/{self.id}/{ApiPath.aggregate}",
+            data={
+                "index_name": index_name,
+                "index_id": index_id,
+                "filter": filter,
+                "group_by": group_by,
+                "metric": metric,
+                "limit": limit,
+                "sort": sort,
+            },
+        )
+        if isinstance(aggregate_data, dict):
+            warn_response_warnings_once(aggregate_data.get("warnings") or [])
+        return aggregate_data
+
+    def legacy_search(
         self,
         query: str,
         search_type: Optional[str] = SearchType.semantic,
@@ -474,6 +807,9 @@ def search(
         sort_docs_on: Optional[str] = None,
         namespace: Optional[str] = None,
         scene_index_id: Optional[str] = None,
+        index_id: Optional[str] = None,
+        algorithm: Optional[str] = None,
+        _skip_warning: bool = False,
     ) -> Union[SearchResult, RTStreamSearchResult]:
         """Search for a query in the collection.
 
@@ -493,6 +829,11 @@ def search(
         :rtype: Union[:class:`videodb.search.SearchResult`,
             :class:`videodb.rtstream.RTStreamSearchResult`]
         """
+        if not _skip_warning:
+            warn_explicit_legacy_search_once()
+        if scene_index_id is None and index_id is not None:
+            scene_index_id = index_id
+
         if namespace == "rtstream":
             data = {"query": query}
             if scene_index_id is not None:
@@ -539,6 +880,8 @@ def search(
             dynamic_score_percentage=dynamic_score_percentage,
             sort_docs_on=sort_docs_on,
             filter=filter,
+            scene_index_id=scene_index_id,
+            algorithm=algorithm,
         )
 
     def search_title(self, query) -> List[Video]:
diff --git a/videodb/editor.py b/videodb/editor.py
index 1dec1e8..85f0981 100644
--- a/videodb/editor.py
+++ b/videodb/editor.py
@@ -1,12 +1,12 @@
 import json
 import logging
-import requests
 import warnings
 
 from typing import List, Optional, Union
 from enum import Enum
 
 from videodb._constants import ApiPath
+from videodb._upload import upload_bytes
 from videodb._utils._video import build_iframe_embed_code
 from videodb.exceptions import InvalidRequestError
 
@@ -1148,26 +1148,20 @@ def _upload_timeline_data(self, json_str: str) -> str:
         :rtype: str
         :raises InvalidRequestError: If upload fails
         """
-        # Get a presigned upload URL
-        upload_url_data = self.connection.get(
-            path=f"{ApiPath.collection}/{self.connection.collection_id}/{ApiPath.upload_url}",
-            params={"name": "timeline_data.json"},
-        )
-        upload_url = upload_url_data.get("upload_url")
-
-        # Upload the JSON data as a file
         try:
-            files = {"file": ("timeline_data.json", json_str, "application/json")}
-            response = requests.post(upload_url, files=files)
-            response.raise_for_status()
-        except requests.exceptions.RequestException as e:
+            return upload_bytes(
+                _connection=self.connection,
+                content=json_str,
+                name="timeline_data.json",
+                content_type="application/json",
+                collection_id=self.connection.collection_id,
+            )
+        except Exception as e:
             raise InvalidRequestError(
                 f"Failed to upload timeline data: {str(e)}",
                 getattr(e, "response", None),
             ) from None
 
-        return upload_url
-
     def download_stream(self, stream_url: str) -> dict:
         """Download a stream from the timeline.
 
diff --git a/videodb/index.py b/videodb/index.py
new file mode 100644
index 0000000..731bb04
--- /dev/null
+++ b/videodb/index.py
@@ -0,0 +1,276 @@
+import time
+
+from typing import List, Optional
+
+from videodb._constants import ApiPath
+
+
+INDEX_TERMINAL_STATUSES = {"ready", "failed"}
+
+
+class FieldSchema:
+    """Schema details for a single indexed field.
+
+    :ivar str type: Data type of the field (e.g. ``"string"``, ``"string_array"``, ``"number"``, ``"text"``, ``"boolean"``)
+    :ivar list groups: Field groups this field belongs to (e.g. ``["semantic", "filter"]``)
+    :ivar list operators: Filter operators supported by the field (for filterable fields)
+    """
+
+    def __init__(
+        self,
+        type: Optional[str] = None,
+        groups: Optional[List[str]] = None,
+        operators: Optional[List[str]] = None,
+    ) -> None:
+        self.type = type
+        self.groups = groups or []
+        self.operators = operators or []
+
+    def __repr__(self) -> str:
+        return (
+            f"FieldSchema("
+            f"type={self.type}, "
+            f"groups={self.groups}, "
+            f"operators={self.operators})"
+        )
+
+    def __getitem__(self, key):
+        return self.__dict__[key]
+
+
+class IndexRecord:
+    """A single indexed record (one temporal scene of an index).
+
+    :ivar str video_id: ID of the video the record belongs to
+    :ivar str understanding_id: ID of the understanding run that produced the record
+    :ivar str scene_id: ID of the scene within the artifact
+    :ivar float start: Start time of the scene in seconds
+    :ivar float end: End time of the scene in seconds
+    :ivar dict data: Indexed field values for the scene
+    :ivar str segment_id: Deprecated alias of ``scene_id``
+    :ivar float start_sec: Deprecated alias of ``start``
+    :ivar float end_sec: Deprecated alias of ``end``
+    """
+
+    def __init__(
+        self,
+        video_id: Optional[str] = None,
+        understanding_id: Optional[str] = None,
+        scene_id: Optional[str] = None,
+        start: Optional[float] = None,
+        end: Optional[float] = None,
+        data: Optional[dict] = None,
+        segment_id: Optional[str] = None,
+        start_sec: Optional[float] = None,
+        end_sec: Optional[float] = None,
+    ) -> None:
+        self.video_id = video_id
+        self.understanding_id = understanding_id
+        self.scene_id = scene_id if scene_id is not None else segment_id
+        self.start = start if start is not None else start_sec
+        self.end = end if end is not None else end_sec
+        self.data = data or {}
+        self.segment_id = self.scene_id
+        self.start_sec = self.start
+        self.end_sec = self.end
+
+    def __repr__(self) -> str:
+        return (
+            f"IndexRecord("
+            f"video_id={self.video_id}, "
+            f"scene_id={self.scene_id}, "
+            f"start={self.start}, "
+            f"end={self.end}, "
+            f"data={self.data})"
+        )
+
+    def __getitem__(self, key):
+        return self.__dict__[key]
+
+
+class RecordPage:
+    """A page of indexed records returned by :meth:`Index.records`.
+
+    :ivar list[IndexRecord] records: Records in this page
+    :ivar str next_cursor: Cursor for the next page, ``None`` if there are no more records
+    """
+
+    def __init__(
+        self,
+        records: Optional[List[IndexRecord]] = None,
+        next_cursor: Optional[str] = None,
+    ) -> None:
+        self.records = records or []
+        self.next_cursor = next_cursor
+
+    def __repr__(self) -> str:
+        return (
+            f"RecordPage(records={len(self.records)}, next_cursor={self.next_cursor})"
+        )
+
+    def __iter__(self):
+        return iter(self.records)
+
+    def __getitem__(self, key):
+        return self.records[key]
+
+
+class Index:
+    """Index manifest for a retrieval-ready index built from an understanding artifact.
+
+    Note: Users should not initialize this class directly.
+    Instead use :meth:`Video.index() <videodb.video.Video.index>`,
+    :meth:`Video.get_index() <videodb.video.Video.get_index>`, or
+    :meth:`Video.list_indexes() <videodb.video.Video.list_indexes>`.
+
+    :ivar str index_id: Unique identifier for the index
+    :ivar str video_id: ID of the video this index belongs to
+    :ivar str collection_id: ID of the collection this index belongs to
+    :ivar str name: User-facing name of the index
+    :ivar str status: Build status of the index (e.g. ``"building"``, ``"ready"``, ``"failed"``)
+    :ivar list use_for: Retrieval capabilities the index supports
+        (subset of ``"semantic"``, ``"query"``, ``"aggregate"``)
+    :ivar source: Source artifact reference or records the index was built from
+    :ivar int record_count: Number of records in the index
+    :ivar dict fields: Field groups mapping (``semantic``, ``filter``,
+        ``aggregate``, ``sort``) to lists of field names
+    :ivar dict field_schema: Mapping of field name to :class:`FieldSchema <FieldSchema>`
+    """
+
+    def __init__(
+        self, _connection, video_id: str, collection_id: str = None, **kwargs
+    ) -> None:
+        self._connection = _connection
+        self.video_id = video_id
+        self.collection_id = collection_id
+        self.update_from_response(kwargs)
+
+    def update_from_response(self, data: dict) -> "Index":
+        data = data or {}
+        self.index_id = data.get("index_id") or getattr(self, "index_id", None)
+        self.name = data.get("name")
+        self.status = data.get("status")
+        self.error = data.get("error")
+        self.use_for = data.get("use_for", [])
+        self.source = data.get("source")
+        self.record_count = data.get("record_count")
+        self.fields = data.get("fields", {})
+        self.field_schema = {
+            field: FieldSchema(
+                type=schema.get("type"),
+                groups=schema.get("groups"),
+                operators=schema.get("operators"),
+            )
+            for field, schema in (data.get("field_schema") or {}).items()
+        }
+        return self
+
+    def __repr__(self) -> str:
+        return (
+            f"Index("
+            f"index_id={self.index_id}, "
+            f"video_id={self.video_id}, "
+            f"name={self.name}, "
+            f"status={self.status}, "
+            + (f"error={self.error}, " if self.error else "")
+            + f"use_for={self.use_for}, "
+            f"record_count={self.record_count})"
+        )
+
+    def __getitem__(self, key):
+        return self.__dict__[key]
+
+    @property
+    def is_complete(self) -> bool:
+        """Return True when the index build is in a terminal status."""
+        return self.status in INDEX_TERMINAL_STATUSES
+
+    @property
+    def is_successful(self) -> bool:
+        """Return True when the index build completed successfully."""
+        return self.status == "ready"
+
+    def refresh(self) -> "Index":
+        """Refresh the index manifest and build status from the API."""
+        data = self._connection.get(
+            path=f"{ApiPath.video}/{self.video_id}/{ApiPath.indexes}/{self.index_id}",
+            params={"collection_id": self.collection_id}
+            if self.collection_id
+            else None,
+        )
+        return self.update_from_response(data)
+
+    def wait_until_complete(
+        self,
+        timeout: int = 1800,
+        poll_interval: int = 10,
+    ) -> "Index":
+        """Poll this index until it reaches a terminal status.
+
+        Terminal statuses are ``ready`` and ``failed``.
+
+        :param int timeout: Maximum time to wait, in seconds
+        :param int poll_interval: Seconds between status checks
+        :raises TimeoutError: If the build does not complete before timeout
+        :return: This index with refreshed status
+        """
+        deadline = time.time() + timeout
+        while True:
+            self.refresh()
+            if self.is_complete:
+                return self
+            if time.time() >= deadline:
+                raise TimeoutError(f"Index {self.index_id} did not complete within {timeout}s")
+            time.sleep(poll_interval)
+
+    def records(self, limit: int = 20, cursor: Optional[str] = None) -> RecordPage:
+        """Preview the records stored in the index.
+
+        Intended for inspection and debugging. Records are paginated via a cursor.
+
+        :param int limit: (optional) Maximum number of records to return (default: 20)
+        :param str cursor: (optional) Cursor returned by a previous page to fetch the next page
+        :return: A page of indexed records, :class:`RecordPage <RecordPage>` object
+        :rtype: :class:`videodb.index.RecordPage`
+        """
+        params = {"limit": limit, "collection_id": self.collection_id}
+        if cursor is not None:
+            params["cursor"] = cursor
+        records_data = self._connection.get(
+            path=f"{ApiPath.video}/{self.video_id}/{ApiPath.indexes}/{self.index_id}/{ApiPath.records}",
+            params={key: value for key, value in params.items() if value is not None},
+        )
+        if not records_data:
+            return RecordPage()
+        records = [
+            IndexRecord(
+                video_id=record.get("video_id"),
+                understanding_id=record.get("understanding_id"),
+                scene_id=record.get("scene_id"),
+                start=record.get("start"),
+                end=record.get("end"),
+                data=record.get("data"),
+                segment_id=record.get("segment_id"),
+                start_sec=record.get("start_sec"),
+                end_sec=record.get("end_sec"),
+            )
+            for record in records_data.get("records", [])
+        ]
+        return RecordPage(records=records, next_cursor=records_data.get("next_cursor"))
+
+    def delete(self) -> None:
+        """Delete the index.
+
+        Removes the index's retrieval structures. It does not delete the original
+        video or stored understanding artifacts.
+
+        :raises InvalidRequestError: If the delete fails
+        :return: None if the delete is successful
+        :rtype: None
+        """
+        self._connection.delete(
+            path=f"{ApiPath.video}/{self.video_id}/{ApiPath.indexes}/{self.index_id}",
+            params={"collection_id": self.collection_id}
+            if self.collection_id
+            else None,
+        )
diff --git a/videodb/job.py b/videodb/job.py
new file mode 100644
index 0000000..f0ba7c4
--- /dev/null
+++ b/videodb/job.py
@@ -0,0 +1,111 @@
+import time
+from typing import Optional
+
+from videodb.audio import Audio
+from videodb.image import Image
+from videodb.exceptions import InvalidRequestError, RequestTimeoutError
+
+
+class GenerationJob:
+    """A self-inference generation job.
+
+    OmniVoice text-to-speech and FLUX image generation return a VideoDB job
+    after the initial server-side async task has dispatched work to
+    inference-core. Use :meth:`wait` to poll until the final asset is ready.
+    """
+
+    def __init__(
+        self,
+        connection,
+        job_id: str,
+        output_url: Optional[str] = None,
+        status: str = "processing",
+        result_type: Optional[str] = None,
+        data: Optional[dict] = None,
+    ) -> None:
+        self._connection = connection
+        self.id = job_id
+        self.job_id = job_id
+        self.output_url = output_url
+        self.status = status or "processing"
+        self.result_type = result_type
+        self.data = data or {}
+
+    def __repr__(self) -> str:
+        return (
+            f"GenerationJob("
+            f"job_id={self.job_id}, "
+            f"status={self.status}, "
+            f"result_type={self.result_type})"
+        )
+
+    @classmethod
+    def from_data(cls, connection, data: dict, result_type: Optional[str] = None):
+        """Create a job from an API job payload."""
+        data = data or {}
+        job_id = data.get("job_id") or data.get("id")
+        if not job_id:
+            raise InvalidRequestError("Invalid generation job response: missing job_id")
+        job_type = data.get("job_type")
+        inferred_result_type = result_type
+        if not inferred_result_type:
+            if job_type == "tts":
+                inferred_result_type = "audio"
+            elif job_type == "image":
+                inferred_result_type = "image"
+        return cls(
+            connection=connection,
+            job_id=job_id,
+            output_url=data.get("output_url"),
+            status=data.get("status", "processing"),
+            result_type=inferred_result_type,
+            data=data,
+        )
+
+    def refresh(self):
+        """Refresh this job's status from VideoDB."""
+        job = self._connection.get_job_status(self.job_id)
+        self.status = job.get("status", self.status)
+        self.data = job.get("data") or {}
+        self.output_url = self.data.get("output_url", self.output_url)
+        if not self.result_type:
+            job_type = self.data.get("job_type")
+            if job_type == "tts":
+                self.result_type = "audio"
+            elif job_type == "image":
+                self.result_type = "image"
+        return self
+
+    def wait(self, timeout: int = 600, interval: int = 5):
+        """Poll this job until it completes and return the generated asset.
+
+        :param int timeout: Maximum seconds to wait
+        :param int interval: Seconds between polls
+        :return: :class:`Audio <videodb.audio.Audio>`,
+            :class:`Image <videodb.image.Image>`, or final job data if the
+            result type cannot be inferred
+        """
+        deadline = time.time() + timeout
+        while True:
+            self.refresh()
+            if self.status != "processing":
+                break
+            if time.time() >= deadline:
+                raise RequestTimeoutError(
+                    f"Generation job {self.job_id} did not complete within {timeout} seconds"
+                )
+            time.sleep(interval)
+
+        if self.status == "failed":
+            raise InvalidRequestError(f"Generation job {self.job_id} failed")
+
+        return self._to_asset()
+
+    def _to_asset(self):
+        data = self.data or {}
+        asset_id = data.get("id")
+        if self.result_type == "audio" or (asset_id and asset_id.startswith("a-")):
+            return Audio(self._connection, **data)
+        if self.result_type == "image" or (asset_id and asset_id.startswith("img-")):
+            return Image(self._connection, **data)
+        return data
diff --git a/videodb/rtstream.py b/videodb/rtstream.py
index 0cf35c6..93f1928 100644
--- a/videodb/rtstream.py
+++ b/videodb/rtstream.py
@@ -235,6 +235,7 @@ class RTStreamSceneIndex:
     :ivar str prompt: Prompt for scene extraction
     :ivar str name: Name of the scene index
     :ivar str status: Status of the scene index
+    :ivar str sandbox_id: Sandbox ID used for self-hosted inference
     """
 
     def __init__(
@@ -248,6 +249,7 @@ def __init__(
         self.prompt = kwargs.get("prompt", None)
         self.name = kwargs.get("name", None)
         self.status = kwargs.get("status", None)
+        self.sandbox_id = kwargs.get("sandbox_id", None)
 
     def __repr__(self) -> str:
         return (
@@ -258,7 +260,8 @@ def __repr__(self) -> str:
             f"extraction_config={self.extraction_config}, "
             f"prompt={self.prompt}, "
             f"name={self.name}, "
-            f"status={self.status})"
+            f"status={self.status}, "
+            f"sandbox_id={self.sandbox_id})"
         )
 
     def get_scenes(self, start: int = None, end: int = None, page=1, page_size=100):
@@ -368,6 +371,246 @@ def disable_alert(self, alert_id):
         )
 
 
+class RTStreamUnderstanding:
+    """RTStreamUnderstanding class to interact with a continuous understanding job.
+
+    Produced by :meth:`RTStream.understand`. It runs VLM analysis over stream
+    windows and, when ``store=True``, persists the output so it can be indexed
+    later. Understanding is independent of scene indexing.
+
+    :ivar str id: Understanding id (``und-...``)
+    :ivar str rtstream_id: ID of the parent RTStream
+    :ivar str status: Job status (``running`` | ``stopped`` | ``failed``)
+    :ivar bool store: Whether output is persisted for later indexing
+    :ivar dict segmentation: Time segmentation, e.g. ``{"type": "time", "window": "10s"}``
+    :ivar list analyzers: Analyzer specs for this understanding
+    :ivar dict outputs: Named output source descriptors, e.g. ``outputs["scene"]``
+    """
+
+    def __init__(
+        self,
+        _connection,
+        understanding_id: str = None,
+        rtstream_id: str = None,
+        id: str = None,
+        **kwargs,
+    ) -> None:
+        self._connection = _connection
+        self.id = understanding_id or id
+        self.rtstream_id = rtstream_id
+        self.status = kwargs.get("status", None)
+        self.store = kwargs.get("store", True)
+        self.segmentation = kwargs.get("segmentation", {})
+        self.analyzers = kwargs.get("analyzers", [])
+        self.outputs = kwargs.get("outputs", {})
+
+    def __repr__(self) -> str:
+        return (
+            f"RTStreamUnderstanding("
+            f"id={self.id}, "
+            f"rtstream_id={self.rtstream_id}, "
+            f"status={self.status}, "
+            f"store={self.store}, "
+            f"analyzers={len(self.analyzers)})"
+        )
+
+    def refresh(self) -> "RTStreamUnderstanding":
+        """Reload this understanding from the server.
+
+        :return: This understanding, updated
+        :rtype: :class:`RTStreamUnderstanding <RTStreamUnderstanding>`
+        """
+        data = self._connection.get(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.understand}/{self.id}"
+        )
+        if data:
+            data.setdefault("understanding_id", self.id)
+            data.setdefault("rtstream_id", self.rtstream_id)
+            self.__init__(self._connection, **data)
+        return self
+
+    def start(self):
+        """Resume processing new stream windows.
+
+        :return: None
+        :rtype: None
+        """
+        self._connection.patch(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.understand}/{self.id}/{ApiPath.status}",
+            data={"action": "start"},
+        )
+        self.status = "running"
+
+    def stop(self):
+        """Pause processing new stream windows. Existing records remain available.
+
+        :return: None
+        :rtype: None
+        """
+        self._connection.patch(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.understand}/{self.id}/{ApiPath.status}",
+            data={"action": "stop"},
+        )
+        self.status = "stopped"
+
+    def get_records(
+        self,
+        start: float,
+        end: float,
+        output: str = "scene",
+        page: int = 1,
+        page_size: int = 100,
+    ):
+        """Get understanding output records for a time range.
+
+        :param float start: Start Unix timestamp
+        :param float end: End Unix timestamp
+        :param str output: Analyzer output name (default: ``"scene"``)
+        :param int page: Page number (default: 1)
+        :param int page_size: Records per page (default: 100)
+        :return: Records payload with ``records`` and ``next_page``
+        :rtype: dict
+        """
+        params = {
+            "start": start,
+            "end": end,
+            "output": output,
+            "page": page,
+            "page_size": page_size,
+        }
+        return self._connection.get(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.understand}/{self.id}/{ApiPath.records}",
+            params={k: v for k, v in params.items() if v is not None},
+        )
+
+
+class RTStreamIndex:
+    """RTStreamIndex — a continuous index over an understanding output.
+
+    Produced by :meth:`RTStream.index`. Materializes an understanding's stored
+    output into a searchable index; has its own lifecycle, separate from the
+    understanding.
+
+    :ivar str id: Index id (``idx-...``)
+    :ivar str rtstream_id: ID of the parent RTStream
+    :ivar str name: Index name
+    :ivar str status: ``running`` | ``stopped`` | ``failed``
+    :ivar list use_for: Index capabilities, e.g. ``["semantic"]``
+    :ivar str source_understanding_id: Understanding this index consumes
+    :ivar str output: Analyzer output name being indexed (e.g. ``"scene"``)
+    """
+
+    def __init__(self, _connection, index_id=None, rtstream_id=None, id=None, **kwargs):
+        self._connection = _connection
+        self.id = index_id or id
+        self.rtstream_id = rtstream_id
+        self.name = kwargs.get("name")
+        self.status = kwargs.get("status")
+        self.use_for = kwargs.get("use_for") or ["semantic"]
+        self.source_understanding_id = kwargs.get("source_understanding_id")
+        self.output = kwargs.get("output", "scene")
+
+    def __repr__(self) -> str:
+        return (
+            f"RTStreamIndex("
+            f"id={self.id}, "
+            f"rtstream_id={self.rtstream_id}, "
+            f"status={self.status}, "
+            f"use_for={self.use_for}, "
+            f"source_understanding_id={self.source_understanding_id})"
+        )
+
+    def refresh(self) -> "RTStreamIndex":
+        """Reload this index from the server."""
+        data = self._connection.get(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}"
+        )
+        if data:
+            data.setdefault("index_id", self.id)
+            data.setdefault("rtstream_id", self.rtstream_id)
+            self.__init__(self._connection, **data)
+        return self
+
+    def start(self):
+        """Resume materializing new understanding output into the index."""
+        self._connection.patch(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.status}",
+            data={"action": "start"},
+        )
+        self.status = "running"
+
+    def stop(self):
+        """Pause materializing. Existing indexed records remain searchable."""
+        self._connection.patch(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.status}",
+            data={"action": "stop"},
+        )
+        self.status = "stopped"
+
+    def get_records(self, start=None, end=None, page: int = 1, page_size: int = 100):
+        """Get materialized index records.
+
+        :param int start: Start Unix timestamp (optional)
+        :param int end: End Unix timestamp (optional)
+        :param int page: Page number
+        :param int page_size: Records per page
+        :return: Records payload
+        :rtype: dict
+        """
+        params = {"page": page, "page_size": page_size}
+        if start is not None:
+            params["start"] = start
+        if end is not None:
+            params["end"] = end
+        return self._connection.get(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.records}",
+            params=params,
+        )
+
+    def create_alert(self, event_id, callback_url, ws_connection_id=None) -> str:
+        """Attach an event alert to this index.
+
+        :param str event_id: ID of the event
+        :param str callback_url: URL to receive the alert callback
+        :param str ws_connection_id: WebSocket connection ID for real-time alerts (optional)
+        :return: Alert ID
+        :rtype: str
+        """
+        data = {"event_id": event_id, "callback_url": callback_url}
+        if ws_connection_id:
+            data["ws_connection_id"] = ws_connection_id
+        alert_data = self._connection.post(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.alert}",
+            data=data,
+        )
+        return (alert_data or {}).get("alert_id")
+
+    def list_alerts(self):
+        """List all alerts on this index.
+
+        :return: List of alerts
+        :rtype: List[dict]
+        """
+        alert_data = self._connection.get(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.alert}"
+        )
+        return (alert_data or {}).get("alerts", [])
+
+    def enable_alert(self, alert_id):
+        """Enable an alert on this index."""
+        self._connection.patch(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.alert}/{alert_id}/{ApiPath.status}",
+            data={"action": "enable"},
+        )
+
+    def disable_alert(self, alert_id):
+        """Disable an alert on this index."""
+        self._connection.patch(
+            f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.alert}/{alert_id}/{ApiPath.status}",
+            data={"action": "disable"},
+        )
+
+
 class RTStream:
     """RTStream class to interact with the RTStream
 
@@ -571,6 +814,7 @@ def index_scenes(
         model_config={},
         name=None,
         ws_connection_id: Optional[str] = None,
+        sandbox_id: Optional[str] = None,
     ):
         """Index scenes from the rtstream.
 
@@ -581,6 +825,7 @@ def index_scenes(
         :param dict model_config: Configuration for the model
         :param str name: Name of the scene index
         :param str ws_connection_id: WebSocket connection ID for real-time updates (optional)
+        :param str sandbox_id: ID of the sandbox to route self-hosted inference to (optional)
         :return: Scene index, :class:`RTStreamSceneIndex <RTStreamSceneIndex>` object
         :rtype: :class:`videodb.rtstream.RTStreamSceneIndex`
         """
@@ -594,6 +839,8 @@ def index_scenes(
         }
         if ws_connection_id:
             data["ws_connection_id"] = ws_connection_id
+        if sandbox_id:
+            data["sandbox_id"] = sandbox_id
 
         index_data = self._connection.post(
             f"{ApiPath.rtstream}/{self.id}/{ApiPath.index}/{ApiPath.scene}",
@@ -610,6 +857,7 @@ def index_scenes(
             prompt=index_data.get("prompt"),
             name=index_data.get("name"),
             status=index_data.get("status"),
+            sandbox_id=index_data.get("sandbox_id"),
         )
 
     def index_spoken_words(
@@ -621,6 +869,7 @@ def index_spoken_words(
         model_config: dict = {},
         name: str = None,
         ws_connection_id: Optional[str] = None,
+        sandbox_id: Optional[str] = None,
     ):
         """Index spoken words from the rtstream transcript.
 
@@ -632,6 +881,7 @@ def index_spoken_words(
         :param dict model_config: Configuration for the model
         :param str name: Name of the spoken words index
         :param str ws_connection_id: WebSocket connection ID for real-time updates (optional)
+        :param str sandbox_id: ID of the sandbox to route self-hosted inference to (optional)
         :return: Scene index, :class:`RTStreamSceneIndex <RTStreamSceneIndex>` object
         :rtype: :class:`videodb.rtstream.RTStreamSceneIndex`
         """
@@ -650,6 +900,8 @@ def index_spoken_words(
         }
         if ws_connection_id:
             data["ws_connection_id"] = ws_connection_id
+        if sandbox_id:
+            data["sandbox_id"] = sandbox_id
 
         index_data = self._connection.post(
             f"{ApiPath.rtstream}/{self.id}/{ApiPath.index}/{ApiPath.scene}",
@@ -666,6 +918,7 @@ def index_spoken_words(
             prompt=index_data.get("prompt"),
             name=index_data.get("name"),
             status=index_data.get("status"),
+            sandbox_id=index_data.get("sandbox_id"),
         )
 
     def index_audio(
@@ -676,6 +929,7 @@ def index_audio(
         model_config: dict = {},
         name: str = None,
         ws_connection_id: Optional[str] = None,
+        sandbox_id: Optional[str] = None,
     ):
         """Index audio from the rtstream transcript.
 
@@ -687,6 +941,7 @@ def index_audio(
         :param dict model_config: Configuration for the model
         :param str name: Name of the audio index
         :param str ws_connection_id: WebSocket connection ID for real-time updates (optional)
+        :param str sandbox_id: ID of the sandbox to route self-hosted inference to (optional)
         :return: Scene index, :class:`RTStreamSceneIndex <RTStreamSceneIndex>` object
         :rtype: :class:`videodb.rtstream.RTStreamSceneIndex`
         """
@@ -708,6 +963,8 @@ def index_audio(
         }
         if ws_connection_id:
             data["ws_connection_id"] = ws_connection_id
+        if sandbox_id:
+            data["sandbox_id"] = sandbox_id
 
         index_data = self._connection.post(
             f"{ApiPath.rtstream}/{self.id}/{ApiPath.index}/{ApiPath.scene}",
@@ -724,6 +981,7 @@ def index_audio(
             prompt=index_data.get("prompt"),
             name=index_data.get("name"),
             status=index_data.get("status"),
+            sandbox_id=index_data.get("sandbox_id"),
         )
 
     def index_visuals(
@@ -734,6 +992,7 @@ def index_visuals(
         model_config: dict = {},
         name: str = None,
         ws_connection_id: Optional[str] = None,
+        sandbox_id: Optional[str] = None,
     ):
         """Index visuals (scenes) from the rtstream.
 
@@ -746,6 +1005,7 @@ def index_visuals(
         :param dict model_config: Configuration for the model
         :param str name: Name of the visual index
         :param str ws_connection_id: WebSocket connection ID for real-time updates (optional)
+        :param str sandbox_id: ID of the sandbox to route self-hosted inference to (optional)
         :return: Scene index, :class:`RTStreamSceneIndex <RTStreamSceneIndex>` object
         :rtype: :class:`videodb.rtstream.RTStreamSceneIndex`
         """
@@ -767,6 +1027,8 @@ def index_visuals(
         }
         if ws_connection_id:
             data["ws_connection_id"] = ws_connection_id
+        if sandbox_id:
+            data["sandbox_id"] = sandbox_id
 
         index_data = self._connection.post(
             f"{ApiPath.rtstream}/{self.id}/{ApiPath.index}/{ApiPath.scene}",
@@ -783,6 +1045,7 @@ def index_visuals(
             prompt=index_data.get("prompt"),
             name=index_data.get("name"),
             status=index_data.get("status"),
+            sandbox_id=index_data.get("sandbox_id"),
         )
 
     def list_scene_indexes(self):
@@ -804,6 +1067,7 @@ def list_scene_indexes(self):
                 prompt=index.get("prompt"),
                 name=index.get("name"),
                 status=index.get("status"),
+                sandbox_id=index.get("sandbox_id"),
             )
             for index in index_data.get("scene_indexes", [])
         ]
@@ -827,8 +1091,136 @@ def get_scene_index(self, index_id: str) -> RTStreamSceneIndex:
             prompt=index_data.get("prompt"),
             name=index_data.get("name"),
             status=index_data.get("status"),
+            sandbox_id=index_data.get("sandbox_id"),
         )
 
+    def understand(
+        self,
+        segmentation: Dict = None,
+        analyzers: List[Dict] = None,
+        store: bool = True,
+        ws_connection_id: str = None,
+    ) -> "RTStreamUnderstanding":
+        """Start a continuous VLM understanding job on the stream.
+
+        Understanding is independent of indexing: it produces VLM output per
+        stream window and (when ``store=True``) persists it so it can be indexed
+        later. Initial support is one ``vlm`` analyzer with time segmentation.
+
+        :param dict segmentation: Time segmentation, e.g. ``{"type": "time", "window": "10s"}``
+        :param list analyzers: Exactly one VLM analyzer spec, e.g.
+            ``[{"type": "vlm", "name": "scene", "sampling": {"frame_count": 5}, "config": {"prompt": "...", "model": "basic"}}]``
+        :param bool store: Persist output for later indexing (default: True)
+        :param str ws_connection_id: WebSocket connection ID for real-time updates (optional)
+        :return: The understanding job, :class:`RTStreamUnderstanding <RTStreamUnderstanding>` object
+        :rtype: :class:`videodb.rtstream.RTStreamUnderstanding`
+        """
+        data = {
+            "segmentation": segmentation or {},
+            "analyzers": analyzers or [],
+            "store": store,
+        }
+        if ws_connection_id:
+            data["ws_connection_id"] = ws_connection_id
+        understanding_data = self._connection.post(
+            f"{ApiPath.rtstream}/{self.id}/{ApiPath.understand}",
+            data=data,
+        )
+        if not understanding_data:
+            return None
+        understanding_data.setdefault("rtstream_id", self.id)
+        return RTStreamUnderstanding(
+            _connection=self._connection, **understanding_data
+        )
+
+    def get_understanding(self, understanding_id: str) -> "RTStreamUnderstanding":
+        """Get an understanding job by id.
+
+        :param str understanding_id: ID of the understanding job
+        :return: The understanding job, :class:`RTStreamUnderstanding <RTStreamUnderstanding>` object
+        :rtype: :class:`videodb.rtstream.RTStreamUnderstanding`
+        """
+        if not understanding_id:
+            raise ValueError("understanding_id is required")
+        understanding_data = self._connection.get(
+            f"{ApiPath.rtstream}/{self.id}/{ApiPath.understand}/{understanding_id}"
+        )
+        if not understanding_data:
+            return None
+        understanding_data.setdefault("rtstream_id", self.id)
+        return RTStreamUnderstanding(
+            _connection=self._connection, **understanding_data
+        )
+
+    def list_understanding(self) -> List["RTStreamUnderstanding"]:
+        """List all understanding jobs on the stream.
+
+        :return: List of understanding jobs
+        :rtype: List[:class:`RTStreamUnderstanding <RTStreamUnderstanding>`]
+        """
+        data = self._connection.get(
+            f"{ApiPath.rtstream}/{self.id}/{ApiPath.understand}"
+        )
+        results = (data or {}).get("understandings") or []
+        for item in results:
+            item.setdefault("rtstream_id", self.id)
+        return [
+            RTStreamUnderstanding(_connection=self._connection, **item)
+            for item in results
+        ]
+
+    def index(self, source, name=None, use_for=None) -> "RTStreamIndex":
+        """Materialize an understanding output into a searchable index.
+
+        :param dict source: understanding output descriptor, e.g.
+            ``understanding.outputs["scene"]``
+        :param str name: index name (optional)
+        :param list use_for: capabilities; defaults to ``["semantic"]``
+        :return: The index, :class:`RTStreamIndex <RTStreamIndex>` object
+        :rtype: :class:`videodb.rtstream.RTStreamIndex`
+        """
+        data = {"source": source}
+        if name is not None:
+            data["name"] = name
+        if use_for is not None:
+            data["use_for"] = use_for
+        index_data = self._connection.post(
+            f"{ApiPath.rtstream}/{self.id}/{ApiPath.indexes}", data=data
+        )
+        if not index_data:
+            return None
+        index_data.setdefault("rtstream_id", self.id)
+        return RTStreamIndex(_connection=self._connection, **index_data)
+
+    def get_index(self, index_id: str) -> "RTStreamIndex":
+        """Get an index by id.
+
+        :param str index_id: ID of the index
+        :return: :class:`RTStreamIndex <RTStreamIndex>` object
+        :rtype: :class:`videodb.rtstream.RTStreamIndex`
+        """
+        if not index_id:
+            raise ValueError("index_id is required")
+        index_data = self._connection.get(
+            f"{ApiPath.rtstream}/{self.id}/{ApiPath.indexes}/{index_id}"
+        )
+        if not index_data:
+            return None
+        index_data.setdefault("rtstream_id", self.id)
+        return RTStreamIndex(_connection=self._connection, **index_data)
+
+    def list_indexes(self) -> List["RTStreamIndex"]:
+        """List all indexes on the stream.
+
+        :return: List of indexes
+        :rtype: List[:class:`RTStreamIndex <RTStreamIndex>`]
+        """
+        data = self._connection.get(f"{ApiPath.rtstream}/{self.id}/{ApiPath.indexes}")
+        results = (data or {}).get("indexes") or []
+        for item in results:
+            item.setdefault("rtstream_id", self.id)
+        return [RTStreamIndex(_connection=self._connection, **item) for item in results]
+
     def get_transcript(
         self,
         page=1,
diff --git a/videodb/sandbox.py b/videodb/sandbox.py
new file mode 100644
index 0000000..8451e8d
--- /dev/null
+++ b/videodb/sandbox.py
@@ -0,0 +1,105 @@
+import time
+
+from videodb._constants import ApiPath, SandboxStatus
+from videodb.exceptions import InvalidRequestError, RequestTimeoutError
+
+TERMINAL_STATUSES = (SandboxStatus.stopped, SandboxStatus.failed)
+
+
+class Sandbox:
+    """A persistent GPU compute pool for running inference jobs."""
+
+    def __init__(self, _connection, sandbox_id=None, id=None, tier=None, status=None,
+                 name=None, created_at=None, started_at=None, stopped_at=None, **kwargs):
+        self._connection = _connection
+        self.id = sandbox_id or id
+        self.tier = tier
+        self.status = status
+        self.name = name
+        self.created_at = created_at
+        self.started_at = started_at
+        self.stopped_at = stopped_at
+
+    def __repr__(self):
+        return f"Sandbox(id={self.id}, tier={self.tier}, status={self.status}, name={self.name})"
+
+    def _update(self, data):
+        if not data:
+            return
+        self.id = data.get("sandbox_id", data.get("id", self.id))
+        self.tier = data.get("tier", self.tier)
+        self.status = data.get("status", self.status)
+        self.name = data.get("name", self.name)
+        self.created_at = data.get("created_at", self.created_at)
+        self.started_at = data.get("started_at", self.started_at)
+        self.stopped_at = data.get("stopped_at", self.stopped_at)
+
+    def refresh(self):
+        """Fetch latest sandbox state from the server."""
+        data = self._connection.get(path=f"{ApiPath.sandbox}/{self.id}")
+        self._update(data or {})
+        return self
+
+    def wait_for_ready(self, timeout=300, interval=5):
+        """Poll until the sandbox is active.
+
+        :param int timeout: Maximum seconds to wait (default 300)
+        :param int interval: Seconds between polls (default 5)
+        :return: self
+        :raises RequestTimeoutError: If timeout is exceeded
+        :raises InvalidRequestError: If sandbox enters a terminal state
+        """
+        deadline = time.time() + timeout
+        while True:
+            self.refresh()
+            if self.status == SandboxStatus.active:
+                return self
+            if self.status in TERMINAL_STATUSES:
+                raise InvalidRequestError(
+                    f"Sandbox {self.id} entered terminal state: {self.status}"
+                )
+            if time.time() >= deadline:
+                raise RequestTimeoutError(
+                    f"Sandbox {self.id} not ready within {timeout}s"
+                )
+            time.sleep(interval)
+
+    def stop(self, grace=True):
+        """Stop this sandbox.
+
+        :param bool grace: Wait for running jobs to finish before teardown (default True)
+        :return: self
+        """
+        data = self._connection.post(
+            path=f"{ApiPath.sandbox}/{self.id}/stop",
+            data={"grace": grace},
+        )
+        self._update(data.get("data", {}))
+        return self
+
+    def wait_for_stop(self, timeout=120, interval=5):
+        """Poll until the sandbox is stopped.
+
+        :param int timeout: Maximum seconds to wait (default 120)
+        :param int interval: Seconds between polls (default 5)
+        :return: self
+        :raises RequestTimeoutError: If timeout is exceeded
+        """
+        deadline = time.time() + timeout
+        while True:
+            self.refresh()
+            if self.status in TERMINAL_STATUSES:
+                return self
+            if time.time() >= deadline:
+                raise RequestTimeoutError(
+                    f"Sandbox {self.id} did not stop within {timeout}s"
+                )
+            time.sleep(interval)
+
+    @property
+    def is_active(self):
+        return self.status == SandboxStatus.active
+
+    @property
+    def is_ready(self):
+        return self.status in (SandboxStatus.provisioning, SandboxStatus.active)
diff --git a/videodb/sandbox_models.py b/videodb/sandbox_models.py
new file mode 100644
index 0000000..261175c
--- /dev/null
+++ b/videodb/sandbox_models.py
@@ -0,0 +1,23 @@
+from enum import Enum
+
+
+class SandboxModel(str, Enum):
+    """Models that can run on VideoDB Sandbox compute."""
+
+    # GenAI
+    FLUX = "black-forest-labs/FLUX.1-dev"
+    OMNIVOICE = "k2-fsa/OmniVoice"
+
+    # Small tier
+    GEMMA_4_E2B = "google/gemma-4-E2B-it"
+    QWEN_9B = "Qwen/Qwen3.5-9B"
+    WHISPER_LARGE_V3_TURBO = "openai/whisper-large-v3-turbo"
+    OMNIVOICE_CANONICAL = "k2-fsa/OmniVoice"
+    STABLE_AUDIO_OPEN = "stabilityai/stable-audio-open-1.0"
+
+    # Medium tier
+    GEMMA_4_26B = "google/gemma-4-26B-A4B-it"
+    QWEN_27B = "Qwen/Qwen3.5-27B"
+    FLUX_CANONICAL = "black-forest-labs/FLUX.1-dev"
+    GEMMA_4_31B = "google/gemma-4-31B-it"
+
diff --git a/videodb/scene.py b/videodb/scene.py
index 7bd651a..7affac4 100644
--- a/videodb/scene.py
+++ b/videodb/scene.py
@@ -64,12 +64,14 @@ def describe(
         prompt: Optional[str] = None,
         model_name: Optional[str] = None,
         model_config: Optional[Dict] = None,
+        sandbox_id: Optional[str] = None,
     ) -> None:
         """Describe the scene.
 
         :param str prompt: (optional) The prompt to use for the description
         :param str model_name: (optional) The model to use for the description
         :param dict model_config: (optional) The model configuration for the description
+        :param str sandbox_id: (optional) ID of the sandbox to route the job to
         :return: The description of the scene
         :rtype: str
         """
@@ -81,6 +83,7 @@ def describe(
                 "prompt": prompt,
                 "model_name": model_name,
                 "model_config": model_config,
+                "sandbox_id": sandbox_id,
             },
         )
         self.description = description_data.get("description", None)
diff --git a/videodb/search.py b/videodb/search.py
index 7cb87f9..32cb102 100644
--- a/videodb/search.py
+++ b/videodb/search.py
@@ -1,4 +1,6 @@
 from abc import ABC, abstractmethod
+import warnings
+
 from videodb._utils._video import play_stream, build_iframe_embed_code
 from videodb._constants import (
     IndexType,
@@ -14,6 +16,60 @@
 from videodb.shot import Shot
 
 
+_LEGACY_SEARCH_WARNING = (
+    "This search used legacy search because legacy parameters were provided. "
+    "Use legacy_search(...) to keep searching older indexes, or remove legacy parameters and create Search V2 indexes. "
+    "Docs: https://videodb-docs-indexing-search-v2.mintlify.app/api-reference/search-v2/legacy-search"
+)
+_EXPLICIT_LEGACY_SEARCH_WARNING = (
+    "legacy_search() searches older spoken-word and scene indexes only. "
+    "Create Search V2 indexes to use search(), semantic_search(), query(), aggregate(), and ask(). "
+    "Docs: https://videodb-docs-indexing-search-v2.mintlify.app/pages/understand/indexing-pipelines/create-an-index"
+)
+_LEGACY_SEARCH_WARNING_EMITTED = set()
+
+
+def _warn_once(key: str, message: str, stacklevel: int = 4):
+    if key in _LEGACY_SEARCH_WARNING_EMITTED:
+        return
+    _LEGACY_SEARCH_WARNING_EMITTED.add(key)
+    warnings.warn(message, UserWarning, stacklevel=stacklevel)
+
+
+def warn_legacy_search_once():
+    _warn_once("legacy_params", _LEGACY_SEARCH_WARNING)
+
+
+def warn_explicit_legacy_search_once():
+    _warn_once("legacy_search", _EXPLICIT_LEGACY_SEARCH_WARNING)
+
+
+def _warning_docs_text(warning: dict) -> str:
+    docs = warning.get("docs") or []
+    parts = []
+    for item in docs:
+        if not isinstance(item, dict):
+            continue
+        label = item.get("label")
+        url = item.get("url")
+        if label and url:
+            parts.append(f"{label}: {url}")
+        elif url:
+            parts.append(str(url))
+    return f" Docs: {'; '.join(parts)}" if parts else ""
+
+
+def warn_response_warnings_once(response_warnings, stacklevel: int = 5):
+    if isinstance(response_warnings, dict):
+        response_warnings = [response_warnings]
+    for warning in response_warnings or []:
+        if not isinstance(warning, dict):
+            continue
+        code = str(warning.get("code") or warning.get("message") or "response_warning")
+        message = str(warning.get("message") or f"VideoDB warning: {code}").strip()
+        _warn_once(f"response:{code}", f"{message}{_warning_docs_text(warning)}", stacklevel=stacklevel)
+
+
 class SearchResult:
     """SearchResult class to interact with search results
 
@@ -29,13 +85,16 @@ def __init__(self, _connection, **kwargs):
         self.stream_url = None
         self.player_url = None
         self.collection_id = "default"
+        self.warnings = kwargs.get("warnings") or []
+        if kwargs.get("_emit_warnings", True):
+            warn_response_warnings_once(self.warnings, stacklevel=5)
         self._results = kwargs.get("results", [])
         self._format_results()
 
     def _format_results(self):
         for result in self._results:
             self.collection_id = result.get("collection_id")
-            for doc in result.get("docs"):
+            for doc in result.get("docs") or []:
                 self.shots.append(
                     Shot(
                         self._connection,
@@ -49,7 +108,7 @@ def _format_results(self):
                         scene_index_id=doc.get("scene_index_id"),
                         scene_index_name=doc.get("scene_index_name"),
                         metadata=doc.get("metadata"),
-                        stream_url=doc.get("stream_link"),
+                        stream_url=doc.get("stream_link") or doc.get("stream_url"),
                         player_url=doc.get("player_url"),
                     )
                 )
@@ -63,6 +122,15 @@ def __repr__(self) -> str:
             f"shots={self.shots})"
         )
 
+    def __iter__(self):
+        return iter(self.shots)
+
+    def __len__(self):
+        return len(self.shots)
+
+    def __getitem__(self, index):
+        return self.shots[index]
+
     def get_shots(self) -> List[Shot]:
         return self.shots
 
@@ -139,6 +207,141 @@ def get_embed_code(
         )
 
 
+class AskResponse:
+    """Response returned by ``ask()``.
+
+    :ivar str answer: Text answer generated from retrieved video context.
+    :ivar list[Shot] sources: Source shots selected by the LLM when requested.
+    """
+
+    def __init__(self, _connection, **kwargs):
+        self._connection = _connection
+        self.answer = kwargs.get("answer") or ""
+        self.warnings = kwargs.get("warnings") or []
+        warn_response_warnings_once(self.warnings, stacklevel=5)
+        self.sources = SearchResult(
+            _connection,
+            results=kwargs.get("sources") or [],
+            warnings=self.warnings,
+            _emit_warnings=False,
+        ).shots
+
+    def __repr__(self) -> str:
+        return f"AskResponse(answer={self.answer!r}, sources={self.sources})"
+
+
+class SearchResponse:
+    """Envelope returned by high-level Search v2.
+
+    For ``response_type='shots'`` or ``response_type='deepsearch'``, ``results`` is a :class:`SearchResult`.
+    For ``response_type='aggregate'``, ``results`` is the aggregate dict/list returned by the server.
+    """
+
+    def __init__(self, _connection, **kwargs):
+        self._connection = _connection
+        self.response_type = kwargs.get("response_type")
+        self.session_id = kwargs.get("session_id")
+        self.waiting_for = kwargs.get("waiting_for") or "none"
+        self.clarification = kwargs.get("clarification")
+        self.trace = kwargs.get("trace")
+        self.warnings = kwargs.get("warnings") or []
+        warn_response_warnings_once(self.warnings, stacklevel=6)
+        raw_results = kwargs.get("results", [])
+        if self.response_type in {"shots", "deepsearch"}:
+            self.results = SearchResult(
+                _connection,
+                results=raw_results,
+                warnings=self.warnings,
+                _emit_warnings=False,
+            )
+            self.shots = self.results.shots
+        else:
+            self.results = raw_results
+            self.shots = []
+
+    def __repr__(self) -> str:
+        if self.response_type == "deepsearch":
+            return (
+                "SearchResponse("
+                f"response_type={self.response_type}, "
+                f"session_id={self.session_id!r}, "
+                f"waiting_for={self.waiting_for!r}, "
+                f"clarification={self.clarification!r}, "
+                f"results={self.results})"
+            )
+        return f"SearchResponse(response_type={self.response_type}, results={self.results})"
+
+    def __iter__(self):
+        if self.response_type in {"shots", "deepsearch"}:
+            return iter(self.results)
+        if isinstance(self.results, list):
+            return iter(self.results)
+        return iter([self.results])
+
+    def __len__(self):
+        if self.response_type in {"shots", "deepsearch"}:
+            return len(self.results)
+        if isinstance(self.results, list):
+            return len(self.results)
+        return 1 if self.results is not None else 0
+
+    def __getitem__(self, index):
+        if self.response_type in {"shots", "deepsearch"}:
+            return self.results[index]
+        return self.results[index]
+
+    def get_shot_results(self) -> SearchResult:
+        """Return the underlying shot results for media helper methods.
+
+        ``search()`` now returns a SearchResponse envelope, but older SDK code often
+        chained ``.compile()``, ``.play()``, or ``.get_embed_code()`` directly from
+        ``search()``. Keep that flow working for response types backed by shots.
+        """
+        if self.response_type in {"shots", "deepsearch"} and isinstance(self.results, SearchResult):
+            return self.results
+        raise SearchError(
+            "This SearchResponse does not contain shot results. "
+            "compile(), play(), and get_embed_code() are only available when response_type is 'shots' or 'deepsearch'."
+        )
+
+    def compile(self) -> str:
+        """Compile shot results into a stream URL.
+
+        Available for ``response_type='shots'`` and ``response_type='deepsearch'``.
+        """
+        return self.get_shot_results().compile()
+
+    def play(self) -> str:
+        """Compile if needed, then play shot results.
+
+        Available for ``response_type='shots'`` and ``response_type='deepsearch'``.
+        """
+        return self.get_shot_results().play()
+
+    def get_embed_code(
+        self,
+        width: str = "100%",
+        height: int = 405,
+        title: str = "VideoDB Player",
+        allow_fullscreen: bool = True,
+        auto_generate: bool = True,
+    ) -> str:
+        """Generate an iframe embed code for shot results.
+
+        Available for ``response_type='shots'`` and ``response_type='deepsearch'``.
+        """
+        return self.get_shot_results().get_embed_code(
+            width=width,
+            height=height,
+            title=title,
+            allow_fullscreen=allow_fullscreen,
+            auto_generate=auto_generate,
+        )
+
+    def get_shots(self) -> List[Shot]:
+        return self.shots
+
+
 class Search(ABC):
     """Search interface inside video or collection"""
 
diff --git a/videodb/understanding.py b/videodb/understanding.py
new file mode 100644
index 0000000..150e071
--- /dev/null
+++ b/videodb/understanding.py
@@ -0,0 +1,328 @@
+import time
+from typing import Any, Dict, List, Optional
+
+from videodb._constants import ApiPath
+
+
+UNDERSTANDING_TERMINAL_STATUSES = {"done", "failed"}
+ANALYZER_TERMINAL_STATUSES = {"done", "failed", "skipped", "cancelled"}
+
+ANALYZER_TYPE_ALIASES = {
+    "spoken_words": "speech_transcription",
+}
+
+DEFAULT_ANALYZER_NAMES = {
+    "speech_transcription": "transcript",
+    "object_detection": "objects",
+    "vlm": "scene",
+    "ocr": "text",
+    "brand_detection": "brands",
+    "activity_recognition": "activity",
+    "location_detection": "location",
+}
+
+
+class UnderstandingAnalyzer:
+    """Analyzer status and output handle for one analyzer in an understanding run."""
+
+    def __init__(
+        self,
+        understanding: "Understanding",
+        id: Optional[str] = None,
+        name: Optional[str] = None,
+        type: Optional[str] = None,
+        status: Optional[str] = None,
+        **kwargs,
+    ) -> None:
+        self.understanding = understanding
+        self.id = id
+        self.name = name
+        self.type = type
+        self.status = status
+        self.extra = kwargs
+
+    def __repr__(self) -> str:
+        return (
+            f"UnderstandingAnalyzer("
+            f"id={self.id}, "
+            f"name={self.name}, "
+            f"type={self.type}, "
+            f"status={self.status})"
+        )
+
+    def __getitem__(self, key):
+        return self.__dict__[key]
+
+    @property
+    def is_complete(self) -> bool:
+        """Return True when the analyzer is in a terminal status."""
+        return self.status in ANALYZER_TERMINAL_STATUSES
+
+    @property
+    def is_successful(self) -> bool:
+        """Return True when the analyzer completed successfully."""
+        return self.status == "done"
+
+    def refresh(self) -> "UnderstandingAnalyzer":
+        """Refresh this analyzer's status from the API."""
+        analyzer = self.understanding.get_analyzer(self.name or self.id, refresh=True)
+        self.id = analyzer.id
+        self.name = analyzer.name
+        self.type = analyzer.type
+        self.status = analyzer.status
+        self.extra = analyzer.extra
+        return self
+
+    def wait_until_complete(
+        self,
+        timeout: int = 1800,
+        poll_interval: int = 10,
+    ) -> "UnderstandingAnalyzer":
+        """Poll this analyzer until it reaches a terminal status.
+
+        :param int timeout: Maximum time to wait, in seconds
+        :param int poll_interval: Seconds between status checks
+        :raises TimeoutError: If the analyzer does not complete before timeout
+        :return: This analyzer with refreshed status
+        """
+        deadline = time.time() + timeout
+        while True:
+            self.refresh()
+            if self.is_complete:
+                return self
+            if time.time() >= deadline:
+                raise TimeoutError(
+                    f"Analyzer {self.name or self.id} did not complete within {timeout}s"
+                )
+            time.sleep(poll_interval)
+
+    def get_output(self) -> Any:
+        """Return this analyzer's output.
+
+        The current API returns the analyzer's segments output.
+        """
+        identifier = self.name or self.id
+        if not identifier:
+            raise ValueError("Analyzer id or name is required")
+        return self.understanding.get_analyzer_output(identifier)
+
+    def to_index_source(self) -> Dict:
+        """Serialize this analyzer as an index ``source`` reference.
+
+        Sends only identifiers — the server re-fetches the analyzer's output from its
+        own store, so scenes never round-trip through the client:
+
+            for analyzer in understanding.list_analyzers():
+                if analyzer.is_successful:
+                    video.index(name=analyzer.name, source=analyzer)
+
+        :raises ValueError: If the analyzer has no id or its understanding id is unknown
+        """
+        understanding_id = getattr(self.understanding, "id", None)
+        if not understanding_id or not self.id:
+            raise ValueError("analyzer source requires understanding id and analyzer id")
+        # ids + type — the server's analyzer record stays the source of truth for the
+        return {
+            "understanding_id": understanding_id,
+            "analyzer_id": self.id,
+            "analyzer_type": self.type,
+        }
+
+
+class Understanding:
+    """A video understanding run.
+
+    Use :meth:`list_analyzers` or :meth:`get_analyzer` to inspect analyzers and
+    fetch analyzer outputs.
+    """
+
+    def __init__(
+        self,
+        _connection,
+        video_id: str,
+        collection_id: Optional[str] = None,
+        understanding_id: Optional[str] = None,
+        id: Optional[str] = None,
+        status: Optional[str] = None,
+        analyzers: Optional[List[Dict[str, Any]]] = None,
+        output_url: Optional[str] = None,
+        **kwargs,
+    ) -> None:
+        self._connection = _connection
+        self.video_id = video_id
+        self.collection_id = collection_id
+        self.id = understanding_id or id
+        self.status = status
+        self.output_url = output_url
+        self.extra = kwargs
+        self.analyzers = [self.create_analyzer(item) for item in (analyzers or [])]
+
+    def __repr__(self) -> str:
+        return (
+            f"Understanding("
+            f"id={self.id}, "
+            f"video_id={self.video_id}, "
+            f"status={self.status}, "
+            f"analyzers={len(self.analyzers)})"
+        )
+
+    def __getitem__(self, key):
+        return self.__dict__[key]
+
+    @property
+    def is_complete(self) -> bool:
+        """Return True when the understanding run is in a terminal status."""
+        return self.status in UNDERSTANDING_TERMINAL_STATUSES
+
+    @property
+    def is_successful(self) -> bool:
+        """Return True when the understanding run completed successfully."""
+        return self.status == "done"
+
+    def create_analyzer(self, data: Dict[str, Any]) -> UnderstandingAnalyzer:
+        return UnderstandingAnalyzer(self, **(data or {}))
+
+    def update_from_response(self, data: Dict[str, Any]) -> "Understanding":
+        data = data or {}
+        self.status = data.get("status", self.status)
+        if data.get("understanding_id") or data.get("id"):
+            self.id = data.get("understanding_id") or data.get("id")
+        if data.get("video_id"):
+            self.video_id = data.get("video_id")
+        if data.get("collection_id"):
+            self.collection_id = data.get("collection_id")
+        if data.get("output_url"):
+            self.output_url = data.get("output_url")
+        if "analyzers" in data:
+            self.analyzers = [self.create_analyzer(item) for item in data.get("analyzers") or []]
+        return self
+
+    def refresh(self) -> "Understanding":
+        """Refresh understanding and analyzer statuses from the API."""
+        data = self._connection.get(
+            path=f"{ApiPath.video}/{self.video_id}/{ApiPath.understand}/{self.id}"
+        )
+        return self.update_from_response(data)
+
+    def wait_until_complete(
+        self,
+        timeout: int = 1800,
+        poll_interval: int = 10,
+    ) -> "Understanding":
+        """Poll this understanding until it reaches a terminal status.
+
+        Terminal statuses are ``done`` and ``failed``.
+
+        :param int timeout: Maximum time to wait, in seconds
+        :param int poll_interval: Seconds between status checks
+        :raises TimeoutError: If the run does not complete before timeout
+        :return: This understanding with refreshed status
+        """
+        deadline = time.time() + timeout
+        while True:
+            self.refresh()
+            if self.is_complete:
+                return self
+            if time.time() >= deadline:
+                raise TimeoutError(f"Understanding {self.id} did not complete within {timeout}s")
+            time.sleep(poll_interval)
+
+    def list_analyzers(self) -> List[UnderstandingAnalyzer]:
+        """Return analyzers in this understanding run."""
+        return list(self.analyzers)
+
+    def get_analyzer(self, name_or_id: str, refresh: bool = False) -> UnderstandingAnalyzer:
+        """Return an analyzer by user-facing name or internal analyzer id.
+
+        :param str name_or_id: Analyzer name, e.g. ``"transcript"``, or id, e.g. ``"an_..."``
+        :param bool refresh: When True, fetch the latest analyzer status first
+        :raises ValueError: If no analyzer matches
+        :return: :class:`UnderstandingAnalyzer <UnderstandingAnalyzer>` object
+        """
+        if refresh:
+            data = self._connection.get(
+                path=f"{ApiPath.video}/{self.video_id}/{ApiPath.understand}/{self.id}",
+                params={"analyzer": name_or_id},
+            ) or {}
+            self.status = data.get("status", self.status)
+            analyzer_data = (data.get("analyzers") or [None])[0]
+            if analyzer_data:
+                refreshed = self.create_analyzer(analyzer_data)
+                for index, analyzer in enumerate(self.analyzers):
+                    if analyzer.name == refreshed.name or analyzer.id == refreshed.id:
+                        self.analyzers[index] = refreshed
+                        return refreshed
+                self.analyzers.append(refreshed)
+                return refreshed
+
+        for analyzer in self.analyzers:
+            if analyzer.name == name_or_id or analyzer.id == name_or_id:
+                return analyzer
+        raise ValueError(f"Analyzer not found: {name_or_id}")
+
+    def get_analyzer_output(self, name_or_id: str) -> Any:
+        """Return output for an analyzer by name or id."""
+        return self._connection.get(
+            path=(
+                f"{ApiPath.video}/{self.video_id}/{ApiPath.understand}/{self.id}"
+                f"/analyzers/{name_or_id}/output"
+            )
+        )
+
+    def delete(self) -> None:
+        """Delete this understanding run."""
+        self._connection.delete(
+            path=f"{ApiPath.video}/{self.video_id}/{ApiPath.understand}/{self.id}"
+        )
+
+
+def normalize_analyzer_type(analyzer_type: str) -> str:
+    return ANALYZER_TYPE_ALIASES.get(analyzer_type, analyzer_type)
+
+
+def default_analyzer_name(analyzer_type: str) -> Optional[str]:
+    return DEFAULT_ANALYZER_NAMES.get(normalize_analyzer_type(analyzer_type))
+
+
+def normalize_understanding_analyzers(analyzers: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Normalize analyzer payloads to the server contract.
+
+    The public SDK accepts friendly analyzer names like ``spoken_words`` and
+    automatically fills stable output names for built-ins when omitted.
+    """
+    if not isinstance(analyzers, list) or not analyzers:
+        raise ValueError("analyzers must be a non-empty list")
+
+    normalized = []
+    names = set()
+    generated_names = set()
+
+    for index, analyzer in enumerate(analyzers):
+        if not isinstance(analyzer, dict):
+            raise ValueError(f"analyzers[{index}] must be a dict")
+        if not analyzer.get("type"):
+            raise ValueError(f"analyzers[{index}].type is required")
+
+        item = dict(analyzer)
+        original_type = item["type"]
+        item["type"] = normalize_analyzer_type(original_type)
+
+        if not item.get("name"):
+            generated_name = default_analyzer_name(original_type)
+            if generated_name:
+                if generated_name in generated_names:
+                    raise ValueError(
+                        f"Multiple analyzers would use default name {generated_name!r}. "
+                        "Provide explicit analyzer names."
+                    )
+                item["name"] = generated_name
+                generated_names.add(generated_name)
+
+        if item.get("name"):
+            if item["name"] in names:
+                raise ValueError(f"Duplicate analyzer name: {item['name']}")
+            names.add(item["name"])
+
+        normalized.append(item)
+
+    return normalized
diff --git a/videodb/video.py b/videodb/video.py
index 37d0526..8786db6 100644
--- a/videodb/video.py
+++ b/videodb/video.py
@@ -12,8 +12,18 @@
     Workflows,
 )
 from videodb.image import Image, Frame
+from videodb.index import Index
+from videodb.understanding import Understanding, normalize_understanding_analyzers
 from videodb.scene import Scene, SceneCollection
-from videodb.search import SearchFactory, SearchResult
+from videodb.search import (
+    AskResponse,
+    SearchFactory,
+    SearchResponse,
+    SearchResult,
+    warn_explicit_legacy_search_once,
+    warn_legacy_search_once,
+    warn_response_warnings_once,
+)
 from videodb.shot import Shot
 
 _VALID_SEGMENTERS = {Segmenter.word, Segmenter.sentence, Segmenter.time}
@@ -83,6 +93,177 @@ def update(self, name: Optional[str] = None) -> None:
             self.name = response_data.get("name", name)
 
     def search(
+        self,
+        query: str,
+        *args,
+        config: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> Union[SearchResponse, SearchResult]:
+        """Search this video.
+
+        New search is used by default. Calls that use legacy-shaped parameters are
+        routed to :meth:`legacy_search` with a warning.
+        """
+        old_params = {
+            "search_type",
+            "index_type",
+            "result_threshold",
+            "dynamic_score_percentage",
+            "scene_index_id",
+            "index_id",
+            "algorithm",
+            "sort_docs_on",
+            "namespace",
+        }
+        new_params = {
+            "top_k",
+            "mode",
+            "return_fields",
+            "include_clip",
+            "session_id",
+            "config",
+        }
+        unsupported_params = {"index_name", "index_names", "index_ids"}
+
+        if config is not None:
+            kwargs["config"] = config
+
+        if args:
+            legacy_arg_names = [
+                "search_type",
+                "index_type",
+                "result_threshold",
+                "score_threshold",
+                "dynamic_score_percentage",
+                "filter",
+            ]
+            for name, value in zip(legacy_arg_names, args):
+                kwargs.setdefault(name, value)
+
+        has_old = bool(args) or any(k in kwargs and kwargs[k] is not None for k in old_params)
+        has_new = any(k in kwargs and kwargs[k] is not None for k in new_params)
+        has_unsupported = any(k in kwargs and kwargs[k] is not None for k in unsupported_params)
+
+        if kwargs.get("deepsearch_config") is not None:
+            raise ValueError(
+                "deepsearch_config is not a public search() option. Use mode='deepsearch', top_k, session_id, and return_fields for DeepSearch requests."
+            )
+        if has_old and (has_new or has_unsupported):
+            raise ValueError(
+                "Cannot mix legacy search parameters with Search V2 parameters. "
+                "Use legacy_search(...) for older indexes, or remove legacy parameters and use Search V2 search(...)."
+            )
+        if has_unsupported:
+            raise ValueError(
+                "search() chooses indexes automatically and does not accept index selectors. "
+                "Use semantic_search() for semantic index selection, query() for structured filtering, or aggregate() for counts and facets."
+            )
+
+        if has_old:
+            warn_legacy_search_once()
+            return self.legacy_search(query=query, _skip_warning=True, **kwargs)
+
+        return self._new_search(query=query, **kwargs)
+
+    def _new_search(self, query: str, **kwargs) -> SearchResponse:
+        payload = {"query": query, **{k: v for k, v in kwargs.items() if v is not None}}
+        search_data = self._connection.post(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.search}/v2",
+            data=payload,
+        )
+        return SearchResponse(self._connection, **search_data)
+
+    def ask(
+        self,
+        question: str,
+        top_k: int = 15,
+        mode: str = "default",
+        include_sources: bool = False,
+    ) -> AskResponse:
+        ask_data = self._connection.post(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.ask}",
+            data={
+                "question": question,
+                "top_k": top_k,
+                "mode": mode,
+                "include_sources": include_sources,
+            },
+        )
+        return AskResponse(self._connection, **ask_data)
+
+    def semantic_search(
+        self,
+        query: str,
+        index_names: Optional[Union[List[str], str]] = None,
+        top_k: int = 10,
+        score_threshold: Optional[float] = None,
+        filter: Optional[Union[List, Dict]] = None,
+        return_fields: Optional[Union[List, Dict, str]] = None,
+        index_ids: Optional[Union[List[str], str]] = None,
+    ) -> SearchResult:
+        search_data = self._connection.post(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.semantic_search}",
+            data={
+                "query": query,
+                "index_names": index_names,
+                "index_ids": index_ids,
+                "top_k": top_k,
+                "score_threshold": score_threshold,
+                "filter": filter,
+                "return_fields": return_fields,
+            },
+        )
+        return SearchResult(self._connection, **search_data)
+
+    def query(
+        self,
+        index_name: Optional[str] = None,
+        filter: Optional[Union[List, Dict]] = None,
+        limit: int = 100,
+        return_fields: Optional[Union[List, Dict, str]] = None,
+        sort: Optional[Union[str, List[Tuple[str, str]]]] = None,
+        index_id: Optional[str] = None,
+    ) -> SearchResult:
+        query_data = self._connection.post(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.query}",
+            data={
+                "index_name": index_name,
+                "index_id": index_id,
+                "filter": filter,
+                "limit": limit,
+                "return_fields": return_fields,
+                "sort": sort,
+            },
+        )
+        return SearchResult(self._connection, **query_data)
+
+    def aggregate(
+        self,
+        index_name: Optional[str] = None,
+        filter: Optional[Union[List, Dict]] = None,
+        group_by: Optional[str] = None,
+        metric: str = "count",
+        limit: int = 100,
+        sort: Optional[Union[str, List[Tuple[str, str]]]] = None,
+        index_id: Optional[str] = None,
+    ) -> Union[Dict, List[Dict]]:
+        aggregate_data = self._connection.post(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.aggregate}",
+            data={
+                "index_name": index_name,
+                "index_id": index_id,
+                "filter": filter,
+                "group_by": group_by,
+                "metric": metric,
+                "limit": limit,
+                "sort": sort,
+            },
+        )
+        if isinstance(aggregate_data, dict):
+            warn_response_warnings_once(aggregate_data.get("warnings") or [])
+        return aggregate_data
+
+    def legacy_search(
         self,
         query: str,
         search_type: Optional[str] = SearchType.semantic,
@@ -105,6 +286,11 @@ def search(
         :return: :class:`SearchResult <SearchResult>` object
         :rtype: :class:`videodb.search.SearchResult`
         """
+        if not kwargs.pop("_skip_warning", False):
+            warn_explicit_legacy_search_once()
+        if kwargs.get("scene_index_id") is None and kwargs.get("index_id") is not None:
+            kwargs["scene_index_id"] = kwargs.get("index_id")
+        kwargs.pop("index_id", None)
         search = SearchFactory(self._connection).get_search(search_type)
         return search.search_inside_video(
             video_id=self.id,
@@ -528,6 +714,7 @@ def index_scenes(
         name: Optional[str] = None,
         scenes: Optional[List[Scene]] = None,
         callback_url: Optional[str] = None,
+        sandbox_id: Optional[str] = None,
     ) -> Optional[str]:
         """Index the scenes of the video.
 
@@ -554,6 +741,7 @@ def index_scenes(
         :param str name: (optional) The name of the scene index
         :param list[Scene] scenes: (optional) The scenes to be indexed, List of :class:`Scene <Scene>` objects
         :param str callback_url: (optional) The callback url
+        :param str sandbox_id: (optional) ID of the sandbox to route the job to
         :raises InvalidRequestError: If the index fails or index already exists
         :return: The scene index id
         :rtype: str
@@ -570,6 +758,7 @@ def index_scenes(
                 "name": name,
                 "scenes": [scene.to_json() for scene in scenes] if scenes else None,
                 "callback_url": callback_url,
+                "sandbox_id": sandbox_id,
             },
         )
         if not scenes_data:
@@ -584,6 +773,7 @@ def index_visuals(
         model_config: Optional[Dict] = None,
         name: Optional[str] = None,
         callback_url: Optional[str] = None,
+        sandbox_id: Optional[str] = None,
     ) -> Optional[str]:
         """Index visuals (scenes) from the video.
 
@@ -597,6 +787,7 @@ def index_visuals(
         :param dict model_config: Configuration for the model
         :param str name: Name of the visual index
         :param str callback_url: URL to receive the callback (optional)
+        :param str sandbox_id: ID of the sandbox to route the job to (optional)
         :return: The scene index id
         :rtype: str
         """
@@ -629,6 +820,7 @@ def index_visuals(
                 "model_config": model_config or {},
                 "name": name,
                 "callback_url": callback_url,
+                "sandbox_id": sandbox_id,
             },
         )
         if not scenes_data:
@@ -733,6 +925,257 @@ def delete_scene_index(self, scene_index_id: str) -> None:
             path=f"{ApiPath.video}/{self.id}/{ApiPath.index}/{ApiPath.scene}/{scene_index_id}"
         )
 
+    def understand(
+        self,
+        analyzers: List[Dict[str, Any]],
+        segmentation: Optional[Dict[str, Any]] = None,
+        sampling: Optional[Dict[str, Any]] = None,
+        transform: Optional[Dict[str, Any]] = None,
+        audio_chunking: Optional[Dict[str, Any]] = None,
+        callback_url: Optional[str] = None,
+        **kwargs,
+    ) -> Understanding:
+        """Create an understanding run for this video.
+
+        :param list analyzers: Analyzer definitions. The SDK accepts friendly
+            analyzer type ``spoken_words`` and maps it to the server analyzer.
+        :param dict segmentation: Optional run-level segmentation config
+        :param dict sampling: Optional run-level sampling config
+        :param dict transform: Optional run-level transform config
+        :param dict audio_chunking: Optional run-level audio chunking config
+        :param str callback_url: Optional URL called when the run completes
+        :return: :class:`Understanding <videodb.understanding.Understanding>` object
+        """
+        normalized_analyzers = normalize_understanding_analyzers(analyzers)
+        payload = {"analyzers": normalized_analyzers}
+        optional_fields = {
+            "segmentation": segmentation,
+            "sampling": sampling,
+            "transform": transform,
+            "audio_chunking": audio_chunking,
+            "callback_url": callback_url,
+            **kwargs,
+        }
+        payload.update({key: value for key, value in optional_fields.items() if value is not None})
+
+        data = self._connection.post(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}",
+            data=payload,
+        ) or {}
+        data.setdefault(
+            "analyzers",
+            [
+                {
+                    "name": analyzer.get("name"),
+                    "type": analyzer.get("type"),
+                    "status": "pending",
+                }
+                for analyzer in normalized_analyzers
+            ],
+        )
+        data.setdefault("video_id", self.id)
+        data.setdefault("collection_id", self.collection_id)
+        return Understanding(self._connection, **data)
+
+    def get_understanding(self, understanding_id: str) -> Understanding:
+        """Get an understanding run by id.
+
+        :param str understanding_id: Understanding run id
+        :return: :class:`Understanding <videodb.understanding.Understanding>` object
+        """
+        if not understanding_id:
+            raise ValueError("understanding_id is required")
+        data = self._connection.get(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}/{understanding_id}"
+        ) or {}
+        data.setdefault("video_id", self.id)
+        data.setdefault("collection_id", self.collection_id)
+        data.setdefault("understanding_id", understanding_id)
+        return Understanding(self._connection, **data)
+
+    def list_understandings(self) -> List[Understanding]:
+        """List understanding runs for this video."""
+        data = self._connection.get(path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}")
+        results = (data or {}).get("understanding_results") or []
+        understandings = []
+        for item in results:
+            data = dict(item)
+            data.setdefault("video_id", self.id)
+            data.setdefault("collection_id", self.collection_id)
+            understandings.append(Understanding(self._connection, **data))
+        return understandings
+
+    def delete_understanding(self, understanding_id: str) -> None:
+        """Delete an understanding run."""
+        if not understanding_id:
+            raise ValueError("understanding_id is required")
+        self._connection.delete(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}/{understanding_id}"
+        )
+
+    @staticmethod
+    def _format_index_source(source: Union[object, Dict]) -> Dict:
+        """Format an index *source* into the request payload.
+
+        Exactly two source kinds are supported:
+
+          - an :class:`UnderstandingAnalyzer <videodb.understanding.UnderstandingAnalyzer>`
+            (anything exposing ``to_index_source()``) — serialized as a light reference
+            ``{understanding_id, analyzer_id, ...}``; the server re-fetches the analyzer
+            output from its own store, so scenes never round-trip through the client
+          - a dict carrying either ``scenes`` (user-provided temporal records) or an
+            ``understanding_id`` reference (optionally with ``analyzer_id`` /
+            ``analyzer_type``), passed through as-is
+          - a bare list of temporal record dicts — sugar for ``{"scenes": [...]}``
+
+        :param source: The analyzer object, source dict, or list of temporal records
+        :raises ValueError: If the source is missing or of an unsupported type
+        :return: The serialized ``source`` payload
+        :rtype: dict
+        """
+        if source is None:
+            raise ValueError("source is required")
+
+        if hasattr(source, "to_index_source"):
+            return source.to_index_source()
+
+        if isinstance(source, list):
+            return {"scenes": source}
+
+        if isinstance(source, dict):
+            if isinstance(source.get("scenes"), list) or source.get("understanding_id"):
+                return source
+            raise ValueError(
+                "source dict must carry 'scenes' (temporal records) or an "
+                "'understanding_id' reference"
+            )
+
+        raise ValueError(
+            "source must be an analyzer object, a dict with 'scenes' or "
+            "'understanding_id', or a list of temporal records — got "
+            + type(source).__name__
+        )
+
+    def _format_index(self, index_data: dict) -> Index:
+        index_data = dict(index_data)
+        video_id = index_data.pop("video_id", None) or self.id
+        collection_id = index_data.pop("collection_id", None) or self.collection_id
+        return Index(
+            self._connection,
+            video_id=video_id,
+            collection_id=collection_id,
+            **index_data,
+        )
+
+    def index(
+        self,
+        source: Union[object, Dict, List],
+        name: Optional[str] = None,
+        use_for: Optional[List[str]] = None,
+        fields: Optional[Dict[str, List[str]]] = None,
+        callback_url: Optional[str] = None,
+    ) -> Optional[Index]:
+        """Create a retrieval-ready index from an understanding artifact.
+
+        Turns an understanding artifact (or user-provided temporal records) into an
+        index that declares retrieval capabilities (``use_for``) and field-level
+        indexing configuration (``fields``).
+
+        :param source: An :class:`UnderstandingAnalyzer` object (indexed by reference —
+            scenes never leave the server), or a dict carrying ``scenes`` (temporal
+            records) or an ``understanding_id`` reference
+        :param str name: (optional) User-facing index name. Defaults to the
+            artifact/source name on the server.
+        :param list use_for: (optional) Retrieval capabilities to enable, any of
+            :attr:`IndexCapability.semantic <videodb.IndexCapability.semantic>`,
+            :attr:`IndexCapability.query <videodb.IndexCapability.query>`,
+            :attr:`IndexCapability.aggregate <videodb.IndexCapability.aggregate>`.
+            Defaults to the artifact's defaults on the server.
+        :param dict fields: (optional) Field-level indexing configuration mapping
+            field groups (``semantic``, ``filter``, ``aggregate``,
+            ``sort``) to lists of field names
+        :param str callback_url: (optional) URL called when indexing completes
+        :raises ValueError: If ``source`` is missing or of an unsupported type
+        :raises InvalidRequestError: If the index creation fails
+        :return: The created index, :class:`Index <Index>` object
+        :rtype: :class:`videodb.index.Index`
+        """
+        index_data = self._connection.post(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}",
+            data={
+                "source": self._format_index_source(source),
+                "name": name,
+                "use_for": use_for,
+                "fields": fields,
+                "callback_url": callback_url,
+            },
+        )
+        if not index_data:
+            return None
+        return self._format_index(index_data)
+
+    def get_index(
+        self, index_id: Optional[str] = None, name: Optional[str] = None
+    ) -> Optional[Index]:
+        """Get an index manifest by its ID or name.
+
+        :param str index_id: (optional) The id of the index
+        :param str name: (optional) The name of the index
+        :raises ValueError: If neither ``index_id`` nor ``name`` is provided
+        :return: The index, :class:`Index <Index>` object
+        :rtype: :class:`videodb.index.Index`
+        """
+        if not index_id and not name:
+            raise ValueError("Either index_id or name is required")
+        params = {"collection_id": self.collection_id}
+        if index_id:
+            path = f"{ApiPath.video}/{self.id}/{ApiPath.indexes}/{index_id}"
+        else:
+            path = f"{ApiPath.video}/{self.id}/{ApiPath.indexes}"
+            params["name"] = name
+        index_data = self._connection.get(path=path, params=params)
+        if not index_data:
+            return None
+        return self._format_index(index_data)
+
+    def list_indexes(self, use_for: Optional[str] = None) -> List[Index]:
+        """List all the indexes of the video.
+
+        :param str use_for: (optional) Filter by retrieval capability, any of
+            :attr:`IndexCapability.semantic <videodb.IndexCapability.semantic>`,
+            :attr:`IndexCapability.query <videodb.IndexCapability.query>`,
+            :attr:`IndexCapability.aggregate <videodb.IndexCapability.aggregate>`
+        :return: List of :class:`Index <Index>` objects
+        :rtype: list[:class:`videodb.index.Index`]
+        """
+        params = {"collection_id": self.collection_id}
+        if use_for is not None:
+            params["use_for"] = use_for
+        index_data = self._connection.get(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}",
+            params=params,
+        )
+        return [self._format_index(index) for index in index_data.get("indexes", [])]
+
+    def delete_index(self, index_id: str) -> None:
+        """Delete an index.
+
+        Removes the index's retrieval structures. It does not delete the original
+        video or stored understanding artifacts.
+
+        :param str index_id: The id of the index to be deleted
+        :raises ValueError: If ``index_id`` is not provided
+        :raises InvalidRequestError: If the delete fails
+        :return: None if the delete is successful
+        :rtype: None
+        """
+        if not index_id:
+            raise ValueError("index_id is required")
+        self._connection.delete(
+            path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}/{index_id}",
+            params={"collection_id": self.collection_id},
+        )
+
     def add_subtitle(self, style: SubtitleStyle = SubtitleStyle()) -> str:
         """Add subtitles to the video.
 
diff --git a/videodb/voice_clone.py b/videodb/voice_clone.py
new file mode 100644
index 0000000..130f116
--- /dev/null
+++ b/videodb/voice_clone.py
@@ -0,0 +1,65 @@
+class VoiceClone:
+    """Reusable cloned voice reference backed by a VideoDB audio asset."""
+
+    def __init__(
+        self,
+        _connection,
+        id=None,
+        voice_clone_id=None,
+        ref_audio_id=None,
+        ref_text=None,
+        name=None,
+        description=None,
+        language=None,
+        collection_id=None,
+        status=None,
+        created_at=None,
+        updated_at=None,
+        **kwargs,
+    ):
+        self._connection = _connection
+        self.id = voice_clone_id or id
+        self.ref_audio_id = ref_audio_id
+        self.ref_text = ref_text
+        self.name = name
+        self.description = description
+        self.language = language
+        self.collection_id = collection_id
+        self.status = status
+        self.created_at = created_at
+        self.updated_at = updated_at
+
+    def __repr__(self):
+        return (
+            f"VoiceClone(id={self.id}, name={self.name}, "
+            f"ref_audio_id={self.ref_audio_id}, status={self.status})"
+        )
+
+    def _update(self, data):
+        if not data:
+            return
+        self.id = data.get("voice_clone_id", data.get("id", self.id))
+        self.ref_audio_id = data.get("ref_audio_id", self.ref_audio_id)
+        self.ref_text = data.get("ref_text", self.ref_text)
+        self.name = data.get("name", self.name)
+        self.description = data.get("description", self.description)
+        self.language = data.get("language", self.language)
+        self.collection_id = data.get("collection_id", self.collection_id)
+        self.status = data.get("status", self.status)
+        self.created_at = data.get("created_at", self.created_at)
+        self.updated_at = data.get("updated_at", self.updated_at)
+
+    def refresh(self):
+        """Fetch latest voice clone state from the server."""
+        from videodb._constants import ApiPath
+
+        data = self._connection.get(path=f"{ApiPath.voice_clone}/{self.id}")
+        self._update(data or {})
+        return self
+
+    def delete(self):
+        """Delete this voice clone."""
+        from videodb._constants import ApiPath
+
+        self._connection.delete(path=f"{ApiPath.voice_clone}/{self.id}")
+        return None