diff --git a/src/evalwire/cli.py b/src/evalwire/cli.py
index d6c6f19..bb1491c 100644
--- a/src/evalwire/cli.py
+++ b/src/evalwire/cli.py
@@ -1,4 +1,4 @@
-"""evalwire CLI — ``evalwire upload``, ``evalwire run``, ``evalwire export``, ``evalwire compare``, and ``evalwire report`` commands."""
+"""evalwire CLI — ``evalwire upload``, ``evalwire run``, ``evalwire validate``, ``evalwire export``, ``evalwire compare``, and ``evalwire report`` commands."""
 
 import sys
 from typing import Literal, cast
@@ -58,6 +58,12 @@ def main() -> None:
     default=None,
     help="Pipe-split delimiter.",
 )
+@click.option(
+    "--strict",
+    is_flag=True,
+    default=False,
+    help="Abort upload if validation issues are found.",
+)
 @click.option(
     "--config",
     "config_path",
@@ -71,6 +77,7 @@ def upload_cmd(
     output_keys: str | None,
     tag_column: str | None,
     delimiter: str | None,
+    strict: bool,
     config_path: str | None,
 ) -> None:
     """Upload a CSV testset to Arize Phoenix as one or more named datasets."""
@@ -102,6 +109,26 @@ def upload_cmd(
                 "No CSV path provided. Use --csv or set csv_path in evalwire.toml."
             )
 
+        if strict:
+            from evalwire.validator import DatasetValidator
+
+            validator = DatasetValidator()
+            result = validator.validate(
+                csv_path=resolved_csv,
+                input_keys=resolved_input_keys,
+                output_keys=resolved_output_keys,
+                tag_column=resolved_tag_column,
+            )
+            if not result.is_valid:
+                for issue in result.issues:
+                    row_info = f"row {issue.row}: " if issue.row is not None else ""
+                    click.echo(f"  {row_info}{issue.message}", err=True)
+                click.echo(
+                    f"Validation failed: {len(result.issues)} issue(s) found. Upload aborted.",
+                    err=True,
+                )
+                sys.exit(2)
+
         from evalwire.uploader import DatasetUploader
 
         client = _make_client()
@@ -302,3 +329,60 @@ def report_cmd(experiment_id: str | None) -> None:
     except Exception as exc:
         click.echo(f"Error: {exc}", err=True)
         sys.exit(2)
+
+
+@main.command("validate")
+@click.option("--csv", "csv_path", default=None, help="Path to the CSV file.")
+@click.option(
+    "--input-keys",
+    default="user_query",
+    show_default=True,
+    help="Comma-separated input column names.",
+)
+@click.option(
+    "--output-keys",
+    default="expected_output",
+    show_default=True,
+    help="Comma-separated output column names.",
+)
+@click.option(
+    "--tag-column",
+    default="tags",
+    show_default=True,
+    help="Column used for dataset splitting.",
+)
+def validate_cmd(
+    csv_path: str | None,
+    input_keys: str,
+    output_keys: str,
+    tag_column: str,
+) -> None:
+    """Validate a CSV testset for structural and content correctness."""
+    if not csv_path:
+        raise click.UsageError("No CSV path provided. Use --csv.")
+    try:
+        from evalwire.validator import DatasetValidator
+
+        resolved_input_keys = [k.strip() for k in input_keys.split(",")]
+        resolved_output_keys = [k.strip() for k in output_keys.split(",")]
+
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=csv_path,
+            input_keys=resolved_input_keys,
+            output_keys=resolved_output_keys,
+            tag_column=tag_column,
+        )
+        if result.is_valid:
+            click.echo("Validation passed: testset is valid.")
+        else:
+            for issue in result.issues:
+                row_info = f"row {issue.row}: " if issue.row is not None else ""
+                click.echo(f"  {row_info}{issue.message}")
+            click.echo(f"Validation failed: {len(result.issues)} issue(s) found.")
+            sys.exit(1)
+    except click.UsageError:
+        raise
+    except Exception as exc:
+        click.echo(f"Error: {exc}", err=True)
+        sys.exit(2)
diff --git a/src/evalwire/validator.py b/src/evalwire/validator.py
new file mode 100644
index 0000000..99626c6
--- /dev/null
+++ b/src/evalwire/validator.py
@@ -0,0 +1,109 @@
+"""Testset validation for evalwire CSV uploads."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import pandas as pd
+
+
+@dataclass
+class ValidationIssue:
+    """A single validation problem found in a testset."""
+
+    row: int | None
+    message: str
+
+
+@dataclass
+class ValidationResult:
+    """Aggregated result of a validation run."""
+
+    issues: list[ValidationIssue] = field(default_factory=list)
+
+    @property
+    def is_valid(self) -> bool:
+        return len(self.issues) == 0
+
+    def __repr__(self) -> str:
+        return f"ValidationResult(issues={len(self.issues)})"
+
+
+class DatasetValidator:
+    """Validate a CSV testset before uploading to Phoenix.
+
+    Checks performed:
+    - Required columns (input keys, output keys, tag column) are present.
+    - No row has an empty tag value.
+    - No row has an empty expected output value.
+    """
+
+    def validate(
+        self,
+        csv_path: Path | str,
+        input_keys: list[str],
+        output_keys: list[str],
+        tag_column: str = "tags",
+    ) -> ValidationResult:
+        """Validate *csv_path* against the given schema.
+
+        Parameters
+        ----------
+        csv_path:
+            Path to the CSV file to validate.
+        input_keys:
+            Expected input column names.
+        output_keys:
+            Expected output column names.
+        tag_column:
+            Name of the column used for dataset splitting.
+
+        Returns
+        -------
+        ValidationResult
+            Contains all discovered issues (structural and row-level).
+
+        Raises
+        ------
+        FileNotFoundError
+            If *csv_path* does not exist.
+        """
+        csv_path = Path(csv_path)
+        if not csv_path.exists():
+            raise FileNotFoundError(f"CSV file not found: {csv_path}")
+
+        df = pd.read_csv(csv_path)
+        issues: list[ValidationIssue] = []
+
+        required_columns = list(input_keys) + list(output_keys) + [tag_column]
+        for col in required_columns:
+            if col not in df.columns:
+                issues.append(
+                    ValidationIssue(row=None, message=f"missing column: {col}")
+                )
+
+        missing_cols = {col for col in required_columns if col not in df.columns}
+
+        if tag_column not in missing_cols:
+            for idx, value in enumerate(df[tag_column], start=1):
+                if pd.isna(value) or str(value).strip() == "":
+                    issues.append(
+                        ValidationIssue(
+                            row=idx, message=f"empty tag in column '{tag_column}'"
+                        )
+                    )
+
+        for output_key in output_keys:
+            if output_key in missing_cols:
+                continue
+            for idx, value in enumerate(df[output_key], start=1):
+                if pd.isna(value) or str(value).strip() == "":
+                    issues.append(
+                        ValidationIssue(
+                            row=idx,
+                            message=f"empty expected output in column '{output_key}'",
+                        )
+                    )
+
+        return ValidationResult(issues=issues)
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 074469d..9616327 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -342,3 +342,77 @@ def test_report_exits_zero(self):
     def test_report_missing_experiment_flag_exits_nonzero(self):
         result = _runner().invoke(main, ["report"])
         assert result.exit_code != 0
+
+
+class TestValidateCommand:
+    def test_valid_csv_exits_zero(self, sample_csv: Path):
+        result = _runner().invoke(main, ["validate", "--csv", str(sample_csv)])
+        assert result.exit_code == 0
+        assert "valid" in result.output.lower()
+
+    def test_missing_csv_exits_nonzero(self):
+        result = _runner().invoke(main, ["validate"])
+        assert result.exit_code != 0
+
+    def test_invalid_csv_exits_nonzero(self, tmp_path: Path):
+        bad = tmp_path / "bad.csv"
+        bad.write_text("wrong_col\nval\n")
+        result = _runner().invoke(main, ["validate", "--csv", str(bad)])
+        assert result.exit_code != 0
+        assert "issue" in result.output.lower() or "missing" in result.output.lower()
+
+    def test_invalid_csv_reports_all_issues(self, tmp_path: Path):
+        bad = tmp_path / "bad.csv"
+        bad.write_text("wrong,also_wrong\nval,val\n")
+        result = _runner().invoke(
+            main,
+            [
+                "validate",
+                "--csv",
+                str(bad),
+                "--input-keys",
+                "user_query",
+                "--output-keys",
+                "expected_output",
+            ],
+        )
+        assert result.exit_code != 0
+        assert "user_query" in result.output or "expected_output" in result.output
+
+    def test_custom_input_output_tag_columns(self, tmp_path: Path):
+        f = tmp_path / "custom.csv"
+        f.write_text("q,ans,grp\nhello,world,g1\n")
+        result = _runner().invoke(
+            main,
+            [
+                "validate",
+                "--csv",
+                str(f),
+                "--input-keys",
+                "q",
+                "--output-keys",
+                "ans",
+                "--tag-column",
+                "grp",
+            ],
+        )
+        assert result.exit_code == 0
+
+
+class TestUploadStrictFlag:
+    def test_upload_strict_passes_for_valid_csv(self, sample_csv: Path):
+        client = _mock_client()
+        with patch("evalwire.cli._make_client", return_value=client):
+            result = _runner().invoke(
+                main, ["upload", "--csv", str(sample_csv), "--strict"]
+            )
+        assert result.exit_code == 0
+
+    def test_upload_strict_fails_for_invalid_csv(self, tmp_path: Path):
+        bad = tmp_path / "bad.csv"
+        bad.write_text("wrong\nval\n")
+        client = _mock_client()
+        with patch("evalwire.cli._make_client", return_value=client):
+            result = _runner().invoke(main, ["upload", "--csv", str(bad), "--strict"])
+        assert result.exit_code != 0
+        assert client.datasets.create_dataset.call_count == 0
diff --git a/tests/test_validator.py b/tests/test_validator.py
new file mode 100644
index 0000000..ac7743d
--- /dev/null
+++ b/tests/test_validator.py
@@ -0,0 +1,191 @@
+"""Tests for evalwire.validator — DatasetValidator."""
+
+import textwrap
+from pathlib import Path
+
+import pytest
+
+from evalwire.validator import DatasetValidator, ValidationIssue, ValidationResult
+
+
+@pytest.fixture()
+def valid_csv(tmp_path: Path) -> Path:
+    content = textwrap.dedent("""\
+        user_query,expected_output,tags
+        "what is python","A language","es_search"
+        "find cycling paths","url-a","es_search | source_router"
+    """)
+    f = tmp_path / "valid.csv"
+    f.write_text(content)
+    return f
+
+
+@pytest.fixture()
+def missing_tag_col_csv(tmp_path: Path) -> Path:
+    content = textwrap.dedent("""\
+        user_query,expected_output
+        "what is python","A language"
+    """)
+    f = tmp_path / "no_tags.csv"
+    f.write_text(content)
+    return f
+
+
+@pytest.fixture()
+def empty_tag_csv(tmp_path: Path) -> Path:
+    content = textwrap.dedent("""\
+        user_query,expected_output,tags
+        "what is python","A language",""
+        "find paths","url","es_search"
+    """)
+    f = tmp_path / "empty_tag.csv"
+    f.write_text(content)
+    return f
+
+
+@pytest.fixture()
+def empty_expected_output_csv(tmp_path: Path) -> Path:
+    content = textwrap.dedent("""\
+        user_query,expected_output,tags
+        "what is python","","es_search"
+        "find paths","url","es_search"
+    """)
+    f = tmp_path / "empty_output.csv"
+    f.write_text(content)
+    return f
+
+
+@pytest.fixture()
+def missing_input_col_csv(tmp_path: Path) -> Path:
+    content = textwrap.dedent("""\
+        wrong_col,expected_output,tags
+        "what is python","A language","es_search"
+    """)
+    f = tmp_path / "missing_input.csv"
+    f.write_text(content)
+    return f
+
+
+class TestValidationResult:
+    def test_is_valid_when_no_issues(self):
+        result = ValidationResult(issues=[])
+        assert result.is_valid is True
+
+    def test_is_invalid_when_has_issues(self):
+        result = ValidationResult(issues=[ValidationIssue(row=1, message="bad")])
+        assert result.is_valid is False
+
+    def test_repr_shows_issue_count(self):
+        result = ValidationResult(issues=[ValidationIssue(row=1, message="bad")])
+        assert "1" in repr(result)
+
+
+class TestValidationIssue:
+    def test_has_row_and_message(self):
+        issue = ValidationIssue(row=3, message="empty tag")
+        assert issue.row == 3
+        assert issue.message == "empty tag"
+
+    def test_row_none_for_structural_issues(self):
+        issue = ValidationIssue(row=None, message="missing column: tags")
+        assert issue.row is None
+
+
+class TestDatasetValidator:
+    def test_valid_csv_returns_no_issues(self, valid_csv: Path):
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=valid_csv,
+            input_keys=["user_query"],
+            output_keys=["expected_output"],
+        )
+        assert result.is_valid is True
+        assert result.issues == []
+
+    def test_missing_tag_column_is_an_issue(self, missing_tag_col_csv: Path):
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=missing_tag_col_csv,
+            input_keys=["user_query"],
+            output_keys=["expected_output"],
+        )
+        assert not result.is_valid
+        messages = [i.message for i in result.issues]
+        assert any("tags" in m for m in messages)
+
+    def test_missing_input_column_is_an_issue(self, missing_input_col_csv: Path):
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=missing_input_col_csv,
+            input_keys=["user_query"],
+            output_keys=["expected_output"],
+        )
+        assert not result.is_valid
+        messages = [i.message for i in result.issues]
+        assert any("user_query" in m for m in messages)
+
+    def test_empty_tag_reported_with_row_number(self, empty_tag_csv: Path):
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=empty_tag_csv,
+            input_keys=["user_query"],
+            output_keys=["expected_output"],
+        )
+        assert not result.is_valid
+        rows_with_issues = [i.row for i in result.issues if i.row is not None]
+        assert 1 in rows_with_issues
+
+    def test_empty_expected_output_reported_with_row_number(
+        self, empty_expected_output_csv: Path
+    ):
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=empty_expected_output_csv,
+            input_keys=["user_query"],
+            output_keys=["expected_output"],
+        )
+        assert not result.is_valid
+        rows_with_issues = [i.row for i in result.issues if i.row is not None]
+        assert 1 in rows_with_issues
+
+    def test_custom_tag_column_name(self, tmp_path: Path):
+        f = tmp_path / "custom.csv"
+        f.write_text("q,ans,group\nhi,hello,grp1\n")
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=f,
+            input_keys=["q"],
+            output_keys=["ans"],
+            tag_column="group",
+        )
+        assert result.is_valid
+
+    def test_nonexistent_file_raises(self, tmp_path: Path):
+        validator = DatasetValidator()
+        with pytest.raises(FileNotFoundError):
+            validator.validate(
+                csv_path=tmp_path / "nonexistent.csv",
+                input_keys=["q"],
+                output_keys=["ans"],
+            )
+
+    def test_multiple_issues_all_reported(self, tmp_path: Path):
+        f = tmp_path / "bad.csv"
+        f.write_text("wrong,also_wrong\nval,val\n")
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=f,
+            input_keys=["user_query"],
+            output_keys=["expected_output"],
+        )
+        assert len(result.issues) >= 3
+
+    def test_structural_issues_have_no_row(self, missing_tag_col_csv: Path):
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=missing_tag_col_csv,
+            input_keys=["user_query"],
+            output_keys=["expected_output"],
+        )
+        structural = [i for i in result.issues if i.row is None]
+        assert len(structural) >= 1