diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md
index 2a7667af..ccf463a4 100644
--- a/CLI-COMMANDS.md
+++ b/CLI-COMMANDS.md
@@ -471,8 +471,42 @@ roboflow workspace stats --start-date 2026-01-01 --end-date 2026-03-31
 roboflow universe search "hard hats" --type dataset --limit 5
 ```
 
+### Native video upload and status
+
+Action Recognition projects take whole videos as Sources. `video upload` streams the
+original MP4/MOV bytes without re-encoding, then reports the **canonical video ID** to use
+for every later action. That ID can differ from the ID reserved at the start of the upload,
+because identical content is deduplicated onto the existing Source.
+
+```bash
+# 1. Upload original bytes and wait for a terminal state (the default).
+roboflow video upload -p my-ar-project -f clip.mov --json
+# { "videoId": "aBcD1234", "status": "uploaded", "resolvedBatch": { ... } }
+```
+
+Pass `--no-wait` to return as soon as the bytes are stored, then poll separately.
+`video upload-status` reports ingestion state — it is distinct from `video status`, which
+checks a legacy video *inference* job.
+
+```bash
+# 2. Poll ingestion yourself.
+roboflow video upload -p my-ar-project -f clip.mov --no-wait --json
+roboflow video upload-status aBcD1234 -p my-ar-project --json
+roboflow video upload-status aBcD1234 -p my-ar-project --wait --poll-timeout 120
+```
+
+Upload accepts `-b/--batch`, `-t/--tag` (comma-separated), `--metadata` (JSON object) and
+`-s/--split`.
+
+Exit codes follow the CLI contract: `0` success, `1` error, `2` auth, `3` not found. A
+`failed` ingestion state and a `--wait` timeout both exit nonzero; the timeout message names
+the video ID so you can re-check it with `video upload-status`.
+
 ### Video inference
 
+Separate from native upload: this submits a legacy asynchronous inference job for a trained
+model version.
+
 ```bash
 roboflow video infer -p my-project -v 3 -f video.mp4 --fps 10
 roboflow video status <job-id>
@@ -566,7 +600,7 @@ Version numbers are always numeric — that's how `x/y` is disambiguated between
 | `asynctasks` | Inspect async background tasks (e.g. project forks) |
 | `trash` | List items in Trash |
 | `universe` | Search Roboflow Universe |
-| `video` | Video inference |
+| `video` | Native video upload/status and video inference |
 | `batch` | Batch processing jobs *(coming soon)* |
 | `completion` | Install or generate shell completion scripts (bash, zsh, fish) |
 
diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py
index 6a4b1cd4..40ca178b 100644
--- a/roboflow/adapters/rfapi.py
+++ b/roboflow/adapters/rfapi.py
@@ -54,7 +54,7 @@ def get_project(api_key, workspace_url, project_url):
     url = f"{API_URL}/{workspace_url}/{project_url}?api_key={api_key}"
     response = requests.get(url)
     if response.status_code != 200:
-        raise RoboflowError(response.text)
+        raise RoboflowError(response.text, status_code=response.status_code)
     result = response.json()
     return result
 
diff --git a/roboflow/cli/handlers/video.py b/roboflow/cli/handlers/video.py
index 1fdc9905..02b909ae 100644
--- a/roboflow/cli/handlers/video.py
+++ b/roboflow/cli/handlers/video.py
@@ -1,14 +1,18 @@
-"""Video inference commands."""
+"""Video commands: native video Source ingestion and legacy video inference."""
 
 from __future__ import annotations
 
-from typing import Annotated
+from typing import Annotated, Optional
 
 import typer
 
 from roboflow.cli._compat import SortedGroup, ctx_to_args
 
-video_app = typer.Typer(cls=SortedGroup, help="Video inference operations", no_args_is_help=True)
+video_app = typer.Typer(
+    cls=SortedGroup,
+    help="Native video upload and video inference operations",
+    no_args_is_help=True,
+)
 
 
 @video_app.command("infer")
@@ -29,11 +33,74 @@ def status(
     ctx: typer.Context,
     job_id: Annotated[str, typer.Argument(help="Job ID to check")],
 ) -> None:
-    """Check video inference job status."""
+    """Check a legacy video inference job's status (not native upload ingestion)."""
     args = ctx_to_args(ctx, job_id=job_id)
     _video_status(args)
 
 
+@video_app.command("upload")
+def upload(
+    ctx: typer.Context,
+    project: Annotated[str, typer.Option("-p", "--project", help="Project ID, or workspace/project")],
+    video_file: Annotated[str, typer.Option("-f", "--file", help="Path to an original .mp4 or .mov file")],
+    batch: Annotated[Optional[str], typer.Option("-b", "--batch", help="Annotation batch name")] = None,
+    metadata: Annotated[
+        Optional[str], typer.Option("--metadata", help='JSON object of metadata, e.g. \'{"camera": "one"}\'')
+    ] = None,
+    poll_interval: Annotated[
+        float, typer.Option("--poll-interval", help="Seconds between status polls while waiting")
+    ] = 2.0,
+    poll_timeout: Annotated[float, typer.Option("--poll-timeout", help="Seconds to wait for a terminal state")] = 300.0,
+    split: Annotated[Optional[str], typer.Option("-s", "--split", help="Dataset split: train, valid or test")] = None,
+    tag: Annotated[Optional[str], typer.Option("-t", "--tag", help="Comma-separated tag names")] = None,
+    wait: Annotated[
+        bool, typer.Option("--wait/--no-wait", help="Poll until the upload reaches a terminal state")
+    ] = True,
+) -> None:
+    """Upload original video bytes as a native video Source.
+
+    Streams the file unchanged and reports the canonical video ID.
+    """
+    args = ctx_to_args(
+        ctx,
+        project=project,
+        video_file=video_file,
+        batch=batch,
+        metadata=metadata,
+        poll_interval=poll_interval,
+        poll_timeout=poll_timeout,
+        split=split,
+        tag=tag,
+        wait=wait,
+    )
+    _video_upload(args)
+
+
+@video_app.command("upload-status")
+def upload_status(
+    ctx: typer.Context,
+    video_id: Annotated[str, typer.Argument(help="Video ID reported by 'roboflow video upload'")],
+    project: Annotated[str, typer.Option("-p", "--project", help="Project ID, or workspace/project")],
+    poll_interval: Annotated[
+        float, typer.Option("--poll-interval", help="Seconds between status polls while waiting")
+    ] = 2.0,
+    poll_timeout: Annotated[float, typer.Option("--poll-timeout", help="Seconds to wait for a terminal state")] = 300.0,
+    wait: Annotated[
+        bool, typer.Option("--wait/--no-wait", help="Poll until a terminal state instead of reading once")
+    ] = False,
+) -> None:
+    """Check a native video upload's ingestion status and canonical video ID."""
+    args = ctx_to_args(
+        ctx,
+        video_id=video_id,
+        project=project,
+        poll_interval=poll_interval,
+        poll_timeout=poll_timeout,
+        wait=wait,
+    )
+    _video_upload_status(args)
+
+
 # ---------------------------------------------------------------------------
 # Business logic (unchanged from argparse version)
 # ---------------------------------------------------------------------------
@@ -113,3 +180,194 @@ def _video_status(args) -> None:  # noqa: ANN001
     if progress:
         text_lines.append(f"Progress: {progress}")
     output(args, data, text="\n".join(text_lines))
+
+
+# ---------------------------------------------------------------------------
+# Native video Source business logic
+# ---------------------------------------------------------------------------
+
+_TERMINAL_UPLOAD_STATES = frozenset({"uploaded", "failed"})
+
+
+def _load_project(args):  # noqa: ANN001
+    """Load the project for a native video command, honoring CLI credential precedence."""
+    from roboflow.adapters import rfapi
+    from roboflow.cli._output import output_api_error
+    from roboflow.cli._resolver import resolve_project_context
+
+    resolved = resolve_project_context(args)
+    if resolved is None:
+        return None
+    api_key, workspace, project_slug = resolved
+
+    try:
+        data = rfapi.get_project(api_key, workspace, project_slug)
+    except rfapi.RoboflowError as exc:
+        output_api_error(
+            args,
+            exc,
+            hint=f"Check that project '{workspace}/{project_slug}' exists and your API key can read it.",
+        )
+        return None
+
+    from roboflow.core.project import Project
+
+    return Project(api_key, data["project"])
+
+
+def _unknown_video_hint(args) -> str:  # noqa: ANN001
+    return f"Check the video ID reported by 'roboflow video upload -p {args.project}'."
+
+
+def _emit_upload_status(args, status) -> None:  # noqa: ANN001
+    """Render an ingestion status, exiting nonzero when the upload failed."""
+    from roboflow.cli._output import output, output_error
+
+    state = status.get("status", "unknown")
+    video_id = status.get("videoId", "")
+
+    if state == "failed":
+        output_error(
+            args,
+            f"Native video upload {video_id} failed during processing.",
+            hint=status.get("error") or "Re-upload the original file and check that it is a valid MP4/MOV.",
+        )
+        return
+
+    lines = [f"Video ID: {video_id}", f"Status:   {state}"]
+    if status.get("duplicate"):
+        lines.append("Duplicate: yes (deduplicated onto an existing Source)")
+    resolved_batch = status.get("resolvedBatch")
+    if isinstance(resolved_batch, dict):
+        lines.append(f"Batch:    {resolved_batch.get('name', '')} ({resolved_batch.get('id', '')})")
+    elif resolved_batch:
+        lines.append(f"Batch:    {resolved_batch}")
+    if state not in _TERMINAL_UPLOAD_STATES:
+        lines.append(f"Still processing. Re-check with 'roboflow video upload-status {video_id} -p {args.project}'.")
+
+    # The server status document is the JSON payload, so --json stays a stable
+    # passthrough of videoId/status/duplicate/resolvedBatch.
+    output(args, status, text="\n".join(lines))
+
+
+def _poll_bounds_are_valid(args) -> bool:  # noqa: ANN001
+    """Reject bad wait bounds before any network call, so an upload never starts and then fails."""
+    from roboflow.cli._output import output_error
+
+    if args.wait and (args.poll_interval <= 0 or args.poll_timeout < 0):
+        output_error(
+            args,
+            f"Invalid wait bounds: --poll-interval {args.poll_interval}, --poll-timeout {args.poll_timeout}.",
+            hint="Use a positive --poll-interval and a nonnegative --poll-timeout.",
+        )
+        return False
+    return True
+
+
+def _wait_for_upload(args, project, video_id):  # noqa: ANN001
+    """Bounded wait, reporting the video ID so a timeout stays actionable."""
+    from roboflow.adapters import rfapi
+    from roboflow.cli._output import output_api_error
+
+    try:
+        return project.wait_for_video_upload(
+            video_id,
+            poll_interval=args.poll_interval,
+            poll_timeout=args.poll_timeout,
+        )
+    except rfapi.RoboflowError as exc:
+        output_api_error(
+            args,
+            exc,
+            hint=f"Re-check with 'roboflow video upload-status {video_id} -p {args.project}'.",
+            not_found_hint=_unknown_video_hint(args),
+        )
+        return None
+
+
+def _video_upload(args) -> None:  # noqa: ANN001
+    import json as json_mod
+    import os
+
+    from roboflow.adapters import rfapi
+    from roboflow.cli._output import output_api_error, output_error
+
+    if not os.path.isfile(args.video_file):
+        output_error(args, f"Video file not found: {args.video_file}", hint="Check the path to the video file.")
+        return
+    if not _poll_bounds_are_valid(args):
+        return
+
+    metadata = None
+    if args.metadata:
+        try:
+            metadata = json_mod.loads(args.metadata)
+        except json_mod.JSONDecodeError as exc:
+            output_error(args, f"Invalid metadata JSON: {exc}", hint='Example: \'{"camera": "one"}\'')
+            return
+        if not isinstance(metadata, dict):
+            output_error(args, "Metadata must be a JSON object.", hint='Example: \'{"camera": "one"}\'')
+            return
+
+    tags = [t.strip() for t in args.tag.split(",") if t.strip()] if args.tag else None
+
+    project = _load_project(args)
+    if project is None:
+        return
+
+    try:
+        # Upload without waiting so the reserved video ID is known even if a
+        # later bounded wait times out; `wait_for_video_upload` continues it.
+        status = project.upload_video(
+            args.video_file,
+            batch_name=args.batch,
+            tag_names=tags,
+            metadata=metadata,
+            split=args.split,
+            wait=False,
+        )
+    except ValueError as exc:
+        output_error(args, str(exc), hint="Native video upload accepts original .mp4 and .mov files.")
+        return
+    except rfapi.RoboflowError as exc:
+        # The failing call may come after the bytes were stored; a re-upload then
+        # deduplicates onto that Source instead of creating a second one.
+        output_api_error(
+            args,
+            exc,
+            hint="Check the project type and plan limits. If the bytes were already stored, "
+            "re-uploading the same file reuses that Source.",
+        )
+        return
+
+    video_id = status.get("videoId")
+    if args.wait and video_id and status.get("status") not in _TERMINAL_UPLOAD_STATES:
+        status = _wait_for_upload(args, project, video_id)
+        if status is None:
+            return
+
+    _emit_upload_status(args, status)
+
+
+def _video_upload_status(args) -> None:  # noqa: ANN001
+    from roboflow.adapters import rfapi
+    from roboflow.cli._output import output_api_error
+
+    if not _poll_bounds_are_valid(args):
+        return
+    project = _load_project(args)
+    if project is None:
+        return
+
+    if args.wait:
+        status = _wait_for_upload(args, project, args.video_id)
+        if status is None:
+            return
+    else:
+        try:
+            status = project.get_video_upload_status(args.video_id)
+        except rfapi.RoboflowError as exc:
+            output_api_error(args, exc, not_found_hint=_unknown_video_hint(args))
+            return
+
+    _emit_upload_status(args, status)
diff --git a/tests/cli/test_video_handler.py b/tests/cli/test_video_handler.py
index 6deb1bdf..9995d681 100644
--- a/tests/cli/test_video_handler.py
+++ b/tests/cli/test_video_handler.py
@@ -1,6 +1,7 @@
 """Tests for the video CLI handler."""
 
 import json
+import os
 import unittest
 from unittest.mock import patch
 
@@ -60,5 +61,354 @@ def test_status_passes_job_id_to_api(self, _mock_key, mock_api) -> None:
         mock_api.assert_called_once_with("fake-key", "my-unique-job-777")
 
 
+AR_PROJECT_PAYLOAD = {
+    "project": {
+        "annotation": "actions",
+        "classes": {"walking": 2},
+        "colors": {"walking": "#FF00FF"},
+        "created": 1759000000.0,
+        "id": "model-evaluation-workspace/penguin-actions",
+        "images": 1,
+        "name": "Penguin Actions",
+        "public": False,
+        "splits": {"train": 1, "test": 0, "valid": 0},
+        "type": "action-recognition",
+        "unannotated": 0,
+        "updated": 1759000001.0,
+    }
+}
+
+
+class NativeVideoCliTest(unittest.TestCase):
+    """Shared fixtures that let real command dispatch build a real Project."""
+
+    def setUp(self) -> None:
+        import tempfile
+
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.video_path = os.path.join(self.tmp.name, "clip.mp4")
+        with open(self.video_path, "wb") as handle:
+            handle.write(b"original video bytes")
+
+        key_patch = patch("roboflow.config.load_roboflow_api_key", return_value="fake-key")
+        key_patch.start()
+        self.addCleanup(key_patch.stop)
+
+        project_patch = patch("roboflow.adapters.rfapi.get_project", return_value=AR_PROJECT_PAYLOAD)
+        self.mock_get_project = project_patch.start()
+        self.addCleanup(project_patch.stop)
+
+    @property
+    def project_ref(self) -> str:
+        return "model-evaluation-workspace/penguin-actions"
+
+
+class TestNativeVideoRegistration(NativeVideoCliTest):
+    """The real CLI exposes the native video commands alongside inference."""
+
+    def test_native_commands_are_registered(self) -> None:
+        for command in ("upload", "upload-status"):
+            with self.subTest(command=command):
+                result = runner.invoke(app, ["video", command, "--help"])
+                self.assertEqual(result.exit_code, 0)
+
+
+class TestVideoUpload(NativeVideoCliTest):
+    """`roboflow video upload` streams original bytes and reports canonical IDs."""
+
+    @patch("roboflow.core.project.Project.wait_for_video_upload")
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_forwards_all_options_and_waits_by_default(self, mock_upload, mock_wait) -> None:
+        mock_upload.return_value = {"videoId": "upload-1", "status": "pending"}
+        mock_wait.return_value = {
+            "videoId": "source-9",
+            "status": "uploaded",
+            "resolvedBatch": {"id": "b1", "name": "clips"},
+        }
+
+        result = runner.invoke(
+            app,
+            [
+                "video",
+                "upload",
+                "-p",
+                self.project_ref,
+                "-f",
+                self.video_path,
+                "-b",
+                "clips",
+                "-t",
+                "indoor, penguin",
+                "--metadata",
+                '{"camera": "one"}',
+                "-s",
+                "valid",
+                "--poll-interval",
+                "0.5",
+                "--poll-timeout",
+                "30",
+            ],
+        )
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        mock_upload.assert_called_once_with(
+            self.video_path,
+            batch_name="clips",
+            tag_names=["indoor", "penguin"],
+            metadata={"camera": "one"},
+            split="valid",
+            wait=False,
+        )
+        # The bounded wait continues on the ID the first status reported.
+        mock_wait.assert_called_once_with("upload-1", poll_interval=0.5, poll_timeout=30.0)
+        self.assertIn("source-9", result.output)
+        self.assertIn("uploaded", result.output)
+
+    @patch("roboflow.core.project.Project.wait_for_video_upload")
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_no_wait_reports_reservation_without_polling(self, mock_upload, mock_wait) -> None:
+        mock_upload.return_value = {"videoId": "upload-1", "status": "pending"}
+
+        result = runner.invoke(
+            app, ["--json", "video", "upload", "-p", self.project_ref, "-f", self.video_path, "--no-wait"]
+        )
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        mock_wait.assert_not_called()
+        data = json.loads(result.output)
+        self.assertEqual(data, {"videoId": "upload-1", "status": "pending"})
+
+    @patch("roboflow.core.project.Project.wait_for_video_upload")
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_terminal_dedup_status_skips_the_wait(self, mock_upload, mock_wait) -> None:
+        mock_upload.return_value = {
+            "videoId": "source-2",
+            "status": "uploaded",
+            "duplicate": True,
+            "resolvedBatch": None,
+        }
+
+        result = runner.invoke(app, ["--json", "video", "upload", "-p", self.project_ref, "-f", self.video_path])
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        mock_wait.assert_not_called()
+        data = json.loads(result.output)
+        self.assertEqual(data["videoId"], "source-2")
+        self.assertIs(data["duplicate"], True)
+        self.assertIsNone(data["resolvedBatch"])
+
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_failed_processing_exits_nonzero(self, mock_upload) -> None:
+        mock_upload.return_value = {"videoId": "upload-1", "status": "failed"}
+
+        result = runner.invoke(app, ["--json", "video", "upload", "-p", self.project_ref, "-f", self.video_path])
+
+        self.assertNotEqual(result.exit_code, 0)
+        payload = json.loads(result.output)
+        self.assertIn("failed", payload["error"]["message"])
+
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_wait_timeout_names_the_video_id_to_recheck(self, mock_upload) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+
+        mock_upload.return_value = {"videoId": "upload-7", "status": "pending"}
+        with patch(
+            "roboflow.core.project.Project.wait_for_video_upload",
+            side_effect=RoboflowError("Video upload upload-7 is still pending after 30s"),
+        ):
+            result = runner.invoke(
+                app,
+                ["--json", "video", "upload", "-p", self.project_ref, "-f", self.video_path, "--poll-timeout", "30"],
+            )
+
+        self.assertNotEqual(result.exit_code, 0)
+        payload = json.loads(result.output)
+        self.assertIn("upload-status upload-7", payload["error"]["hint"])
+
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_local_rejections_never_reach_the_api(self, mock_upload) -> None:
+        absent = os.path.join(self.tmp.name, "absent.mp4")
+        cases = [
+            (["-f", absent], "Video file not found", "Check the path"),
+            # A bad bound must fail before the upload stores bytes it then cannot report.
+            (["-f", self.video_path, "--poll-interval", "0"], "Invalid wait bounds", "positive --poll-interval"),
+            (["-f", self.video_path, "--poll-timeout", "-1"], "Invalid wait bounds", "nonnegative --poll-timeout"),
+        ]
+        for flags, message, hint in cases:
+            with self.subTest(flags=flags):
+                result = runner.invoke(app, ["--json", "video", "upload", "-p", self.project_ref, *flags])
+
+                self.assertEqual(result.exit_code, 1)
+                error = json.loads(result.output)["error"]
+                self.assertIn(message, error["message"])
+                self.assertIn(hint, error["hint"])
+        mock_upload.assert_not_called()
+        self.mock_get_project.assert_not_called()
+
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_unsupported_container_gets_its_own_hint(self, mock_upload) -> None:
+        mock_upload.side_effect = ValueError("Native video upload accepts .mp4 and .mov files")
+
+        result = runner.invoke(app, ["--json", "video", "upload", "-p", self.project_ref, "-f", self.video_path])
+
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn("accepts original .mp4 and .mov", json.loads(result.output)["error"]["hint"])
+
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_invalid_metadata_never_reaches_the_api(self, mock_upload) -> None:
+        cases = [("{not json", "Invalid metadata JSON"), ("[1, 2]", "Metadata must be a JSON object")]
+        for metadata, message in cases:
+            with self.subTest(metadata=metadata):
+                result = runner.invoke(
+                    app,
+                    [
+                        "--json",
+                        "video",
+                        "upload",
+                        "-p",
+                        self.project_ref,
+                        "-f",
+                        self.video_path,
+                        "--metadata",
+                        metadata,
+                    ],
+                )
+
+                self.assertEqual(result.exit_code, 1)
+                self.assertIn(message, json.loads(result.output)["error"]["message"])
+        mock_upload.assert_not_called()
+        self.mock_get_project.assert_not_called()
+
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_upload_api_errors_follow_exit_code_contract(self, mock_upload) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+
+        for status_code, exit_code in ((401, 2), (404, 3), (503, 1), (None, 1)):
+            with self.subTest(status_code=status_code):
+                mock_upload.side_effect = RoboflowError("upload failed", status_code=status_code)
+
+                result = runner.invoke(
+                    app, ["--json", "video", "upload", "-p", self.project_ref, "-f", self.video_path]
+                )
+
+                self.assertEqual(result.exit_code, exit_code)
+                error = json.loads(result.output)["error"]
+                self.assertEqual(error["message"], "upload failed")
+                if status_code != 401:
+                    # A failure after the PUT must not send the user hunting for a file problem.
+                    self.assertIn("re-uploading the same file reuses that Source", error["hint"])
+
+    def test_project_lookup_failure_follows_exit_code_contract(self) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+
+        for status_code, exit_code in ((401, 2), (404, 3), (500, 1)):
+            with self.subTest(status_code=status_code):
+                self.mock_get_project.side_effect = RoboflowError("project lookup failed", status_code=status_code)
+
+                result = runner.invoke(
+                    app, ["--json", "video", "upload", "-p", self.project_ref, "-f", self.video_path]
+                )
+
+                self.assertEqual(result.exit_code, exit_code)
+                self.assertIn("project lookup failed", json.loads(result.output)["error"]["message"])
+
+    def test_missing_api_key_exits_with_auth_code(self) -> None:
+        with patch("roboflow.config.load_roboflow_api_key", return_value=None):
+            result = runner.invoke(app, ["--json", "video", "upload", "-p", self.project_ref, "-f", self.video_path])
+        self.assertEqual(result.exit_code, 2)
+
+    @patch("roboflow.core.project.Project.upload_video")
+    def test_explicit_api_key_takes_precedence(self, mock_upload) -> None:
+        mock_upload.return_value = {"videoId": "v1", "status": "uploaded"}
+        result = runner.invoke(
+            app,
+            ["--api-key", "explicit-key", "video", "upload", "-p", self.project_ref, "-f", self.video_path],
+        )
+        self.assertEqual(result.exit_code, 0, result.output)
+        self.mock_get_project.assert_called_once_with("explicit-key", "model-evaluation-workspace", "penguin-actions")
+
+
+class TestVideoUploadStatus(NativeVideoCliTest):
+    """`roboflow video upload-status` reads native ingestion state."""
+
+    @patch("roboflow.core.project.Project.get_video_upload_status")
+    def test_single_read_by_default(self, mock_status) -> None:
+        mock_status.return_value = {"videoId": "source-3", "status": "uploaded"}
+
+        result = runner.invoke(app, ["--json", "video", "upload-status", "source-3", "-p", self.project_ref])
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        mock_status.assert_called_once_with("source-3")
+        self.assertEqual(json.loads(result.output)["status"], "uploaded")
+
+    @patch("roboflow.core.project.Project.wait_for_video_upload")
+    @patch("roboflow.core.project.Project.get_video_upload_status")
+    def test_wait_uses_the_bounded_poll(self, mock_status, mock_wait) -> None:
+        mock_wait.return_value = {"videoId": "source-3", "status": "uploaded"}
+
+        result = runner.invoke(
+            app,
+            ["video", "upload-status", "source-3", "-p", self.project_ref, "--wait", "--poll-timeout", "12"],
+        )
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        mock_status.assert_not_called()
+        mock_wait.assert_called_once_with("source-3", poll_interval=2.0, poll_timeout=12.0)
+
+    @patch("roboflow.core.project.Project.get_video_upload_status")
+    def test_pending_state_points_at_the_recheck_command(self, mock_status) -> None:
+        mock_status.return_value = {"videoId": "source-3", "status": "pending"}
+
+        result = runner.invoke(app, ["video", "upload-status", "source-3", "-p", self.project_ref])
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        self.assertIn("upload-status source-3", result.output)
+
+    @patch("roboflow.core.project.Project.wait_for_video_upload")
+    def test_invalid_wait_bounds_never_reach_the_api(self, mock_wait) -> None:
+        result = runner.invoke(
+            app, ["--json", "video", "upload-status", "v1", "-p", self.project_ref, "--wait", "--poll-interval", "0"]
+        )
+
+        self.assertEqual(result.exit_code, 1)
+        mock_wait.assert_not_called()
+        self.mock_get_project.assert_not_called()
+
+    def test_api_errors_follow_exit_code_contract(self) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+
+        reads = [("get_video_upload_status", []), ("wait_for_video_upload", ["--wait"])]
+        for method, flags in reads:
+            for status_code, exit_code in ((401, 2), (404, 3), (500, 1)):
+                with self.subTest(method=method, status_code=status_code):
+                    with patch(
+                        f"roboflow.core.project.Project.{method}",
+                        side_effect=RoboflowError("status read failed", status_code=status_code),
+                    ):
+                        result = runner.invoke(
+                            app, ["--json", "video", "upload-status", "nope", "-p", self.project_ref, *flags]
+                        )
+
+                    self.assertEqual(result.exit_code, exit_code)
+                    if status_code == 404:
+                        self.assertIn("Check the video ID", json.loads(result.output)["error"]["hint"])
+
+    @patch("roboflow.core.project.Project.get_video_upload_status")
+    def test_failed_state_exits_nonzero(self, mock_status) -> None:
+        mock_status.return_value = {"videoId": "source-3", "status": "failed"}
+        result = runner.invoke(app, ["--json", "video", "upload-status", "source-3", "-p", self.project_ref])
+        self.assertNotEqual(result.exit_code, 0)
+
+
+class TestLegacyVideoContractsIntact(unittest.TestCase):
+    """The native commands must not disturb legacy video inference."""
+
+    def test_infer_still_takes_a_version_number(self) -> None:
+        result = runner.invoke(app, ["video", "infer", "--help"])
+        self.assertEqual(result.exit_code, 0)
+        self.assertIn("--version", result.output)
+
+
 if __name__ == "__main__":
     unittest.main()