diff --git a/docs/superpowers/plans/2026-06-27-schema-registry-and-versioning.md b/docs/superpowers/plans/2026-06-27-schema-registry-and-versioning.md
new file mode 100644
index 000000000..5df7ad48a
--- /dev/null
+++ b/docs/superpowers/plans/2026-06-27-schema-registry-and-versioning.md
@@ -0,0 +1,1461 @@
+# Schema Registry & Versioning Implementation Plan (Phase 1 of 6)
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Stand up the `Schema` entity (the v2 "Dataset") and its object-store-backed, versioned Pandera body — schema CRUD, version publishing, derived column cache, and server-side validation — as an isolated `/api/v2` module alongside untouched v1.
+
+**Architecture:** New isolated v2 module set (`models/v2/`, `contexts/v2/`, `api/v2/`, `api/schemas/v2/`) sharing only `workspaces`/`users`/auth with v1. A `Schema` row is the core entity (1:1 with the UI's "Dataset"); each `SchemaVersion` row points at a Pandera `DataFrameSchema` JSON body stored in the workspace's object-store bucket (versioned via S3 `version_id`/`etag`), plus a denormalized `columns_cache` derived from that body for fast validation/UI. Postgres is the source of truth; the object store holds schema bodies.
+
+**Tech Stack:** FastAPI, SQLAlchemy (async) + Alembic, Pandera (+ pandas), aioboto3 S3 (existing `contexts/files.py`), pytest-asyncio + factory-boy.
+
+## Global Constraints
+
+- Python 3.10+ (server). Use `uv` exclusively for deps: `uv add <pkg>`, `uv run <tool>`. Never edit `pyproject.toml` by hand for deps.
+- All schema changes go through Alembic (`uv run alembic -c src/extralit_server/alembic.ini ...`).
+- v1 back-compat is NOT a goal; do not modify v1 handlers/models in this phase (v2 is purely additive). v1 is retired in Phase 6.
+- New tables/models use distinct names safe to run beside v1: tables `schemas`, `schema_versions`; ORM classes `Schema`, `SchemaVersion`.
+- Postgres is source of truth; the object store holds Pandera bodies, versioned via native S3 `version_id`/`etag`.
+- Object-store layout: bucket = workspace name (existing per-workspace bucket convention); schema body key = `schemas/{schema_id}/v{version}.json`.
+- All work happens in the git branch `feat/schema-centric-data-model` (already created). Commit after every task.
+- Run commands from `extralit-server/` unless noted. Test command: `uv run pytest <path> -v --disable-warnings`.
+
+---
+
+## File Structure
+
+**Create:**
+- `src/extralit_server/models/v2/__init__.py` — re-exports v2 ORM models for Alembic discovery.
+- `src/extralit_server/models/v2/schemas.py` — `Schema`, `SchemaVersion` ORM models.
+- `src/extralit_server/api/schemas/v2/__init__.py` — package marker.
+- `src/extralit_server/api/schemas/v2/schemas.py` — Pydantic request/response models.
+- `src/extralit_server/contexts/v2/__init__.py` — package marker.
+- `src/extralit_server/contexts/v2/schema_bodies.py` — pure Pandera body helpers (column-cache derivation + record validation).
+- `src/extralit_server/contexts/v2/schemas.py` — schema/version business logic (DB + object store).
+- `src/extralit_server/api/v2/__init__.py` — `create_api_v2()` factory.
+- `src/extralit_server/api/v2/schemas.py` — `/api/v2` schema router.
+- `src/extralit_server/alembic/versions/<rev>_create_schema_and_schema_version_tables.py` — migration.
+- `tests/unit/contexts/v2/__init__.py`, `tests/unit/contexts/v2/test_schema_bodies.py` — pure-logic tests.
+- `tests/integration/api/v2/__init__.py`, `tests/integration/api/v2/test_schemas.py` — API tests.
+
+**Modify:**
+- `src/extralit_server/models/__init__.py` — import v2 models so `DatabaseModel.metadata` sees them.
+- `src/extralit_server/enums.py` — add `SchemaKind`, `SchemaStatus`.
+- `src/extralit_server/_app.py:212` area — mount `create_api_v2()` at `/api/v2`.
+- `tests/factories.py` — add `SchemaFactory`, `SchemaVersionFactory`.
+
+---
+
+## Task 1: Add Pandera dependency and v2 enums
+
+**Files:**
+- Modify: `pyproject.toml` (via `uv add`), `src/extralit_server/enums.py`
+- Test: `tests/unit/test_enums_v2.py` (Create)
+
+**Interfaces:**
+- Produces: `SchemaKind` (`singleton`|`table`), `SchemaStatus` (`draft`|`published`) in `extralit_server.enums`; `pandera` + `pandas` importable in the server venv.
+
+- [ ] **Step 1: Add pandera (pulls pandas)**
+
+Run: `uv add 'pandera[io]>=0.20'`
+Expected: resolves and installs `pandera` (with the `io` extra for `to_json`/`from_json`) and `pandas`; `uv.lock` updated. The version floor and `io` extra matter because Task 2 relies on `DataFrameSchema.to_json()`/`from_json()` round-tripping per-column `metadata` (the `review` widget). Task 2's metadata round-trip test is the gate — if it fails on the installed version, follow the fallback note in Task 2.
+
+- [ ] **Step 2: Verify pandera imports**
+
+Run: `uv run python -c "import pandera as pa, pandas as pd; print(pa.__version__, pd.__version__)"`
+Expected: prints two version numbers, no ImportError.
+
+- [ ] **Step 3: Write the failing test**
+
+Create `tests/unit/test_enums_v2.py`:
+
+```python
+from extralit_server.enums import SchemaKind, SchemaStatus
+
+
+def test_schema_kind_values():
+    assert SchemaKind.singleton == "singleton"
+    assert SchemaKind.table == "table"
+    assert {k.value for k in SchemaKind} == {"singleton", "table"}
+
+
+def test_schema_status_values():
+    assert SchemaStatus.draft == "draft"
+    assert SchemaStatus.published == "published"
+    assert {s.value for s in SchemaStatus} == {"draft", "published"}
+```
+
+- [ ] **Step 4: Run test to verify it fails**
+
+Run: `uv run pytest tests/unit/test_enums_v2.py -v`
+Expected: FAIL with `ImportError: cannot import name 'SchemaKind'`.
+
+- [ ] **Step 5: Add the enums**
+
+Append to `src/extralit_server/enums.py`:
+
+```python
+class SchemaKind(StrEnum):
+    singleton = "singleton"  # exactly one row per `reference` (document-level extraction)
+    table = "table"  # many rows per `reference` (table extraction)
+
+
+class SchemaStatus(StrEnum):
+    draft = "draft"
+    published = "published"
+```
+
+- [ ] **Step 6: Run test to verify it passes**
+
+Run: `uv run pytest tests/unit/test_enums_v2.py -v`
+Expected: PASS (2 passed).
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add pyproject.toml uv.lock src/extralit_server/enums.py tests/unit/test_enums_v2.py
+git commit -m "feat(v2): add pandera dep and SchemaKind/SchemaStatus enums"
+```
+
+---
+
+## Task 2: Pandera body helpers (column cache + validation)
+
+Pure functions, no DB/IO — the validation core. TDD-friendly.
+
+**Files:**
+- Create: `src/extralit_server/contexts/v2/__init__.py` (empty), `src/extralit_server/contexts/v2/schema_bodies.py`
+- Test: `tests/unit/contexts/v2/__init__.py` (empty), `tests/unit/contexts/v2/test_schema_bodies.py`
+
+**Interfaces:**
+- Produces:
+  - `derive_columns_cache(body_json: str) -> list[dict]` — each dict `{"name": str, "dtype": str, "nullable": bool, "review": dict | None}`.
+  - `validate_record_fields(body_json: str, fields: dict) -> dict` — returns coerced fields; raises `SchemaValidationError(errors: list[dict])` on failure.
+  - `SchemaValidationError(Exception)` with `.errors: list[dict]`.
+- Consumes: `pandera`, `pandas`.
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `tests/unit/contexts/v2/__init__.py` (empty) and `tests/unit/contexts/v2/test_schema_bodies.py`:
+
+```python
+import json
+
+import pandera as pa
+import pytest
+
+from extralit_server.contexts.v2.schema_bodies import (
+    SchemaValidationError,
+    derive_columns_cache,
+    validate_record_fields,
+)
+
+
+def _body() -> str:
+    schema = pa.DataFrameSchema(
+        columns={
+            "name": pa.Column(pa.String, nullable=False),
+            "age": pa.Column(pa.Int, nullable=True),
+        }
+    )
+    return schema.to_json()
+
+
+def test_derive_columns_cache_lists_columns_with_dtype_and_nullable():
+    cache = derive_columns_cache(_body())
+    by_name = {c["name"]: c for c in cache}
+    assert set(by_name) == {"name", "age"}
+    assert by_name["name"]["nullable"] is False
+    assert by_name["age"]["nullable"] is True
+    assert "int" in by_name["age"]["dtype"].lower()
+
+
+def test_derive_columns_cache_round_trips_review_metadata():
+    # Guards the whole review-widget pipeline: metadata must survive to_json/from_json.
+    schema = pa.DataFrameSchema(
+        columns={"rating": pa.Column(pa.Int, nullable=True, metadata={"review": {"type": "rating"}})}
+    )
+    cache = derive_columns_cache(schema.to_json())
+    assert cache[0]["review"] == {"type": "rating"}
+
+
+def test_validate_record_fields_returns_native_json_types():
+    coerced = validate_record_fields(_body(), {"name": "Ada", "age": 36})
+    assert coerced["name"] == "Ada"
+    assert coerced["age"] == 36
+    # Must be native python types (not numpy scalars) and JSON-serializable for the
+    # record.fields JSONB column in Phase 2.
+    assert type(coerced["age"]) is int
+    json.dumps(coerced)  # raises if numpy scalars / NaN leaked through
+
+
+def test_validate_record_fields_converts_nulls_to_none():
+    coerced = validate_record_fields(_body(), {"name": "Ada", "age": None})
+    assert coerced["age"] is None
+    json.dumps(coerced)
+
+
+def test_validate_record_fields_raises_on_type_error():
+    with pytest.raises(SchemaValidationError) as exc:
+        validate_record_fields(_body(), {"name": "Ada", "age": "not-a-number"})
+    assert isinstance(exc.value.errors, list)
+    assert len(exc.value.errors) >= 1
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `uv run pytest tests/unit/contexts/v2/test_schema_bodies.py -v`
+Expected: FAIL with `ModuleNotFoundError: extralit_server.contexts.v2.schema_bodies`.
+
+- [ ] **Step 3: Implement the helpers**
+
+Create `src/extralit_server/contexts/v2/__init__.py` (empty file) and `src/extralit_server/contexts/v2/schema_bodies.py`:
+
+```python
+"""Pure helpers for working with a Pandera DataFrameSchema body (JSON).
+
+No DB or object-store access — given a schema body string, derive a denormalized
+column cache and validate a single record's `fields` dict against it.
+"""
+
+import json
+from typing import Any
+
+import pandas as pd
+import pandera as pa
+
+
+class SchemaValidationError(Exception):
+    """Raised when a record's fields fail Pandera validation."""
+
+    def __init__(self, errors: list[dict[str, Any]]) -> None:
+        self.errors = errors
+        super().__init__(f"Record failed schema validation with {len(errors)} error(s)")
+
+
+def _load(body_json: str) -> pa.DataFrameSchema:
+    return pa.DataFrameSchema.from_json(body_json)
+
+
+def derive_columns_cache(body_json: str) -> list[dict[str, Any]]:
+    """Return one entry per column: name, dtype, nullable, and optional review widget.
+
+    The `review` widget is read from the Pandera column's `metadata` under the
+    `review` key when present (e.g. {"review": {"type": "rating"}}).
+    """
+    schema = _load(body_json)
+    cache: list[dict[str, Any]] = []
+    for name, column in schema.columns.items():
+        metadata = column.metadata or {}
+        cache.append(
+            {
+                "name": name,
+                "dtype": str(column.dtype),
+                "nullable": bool(column.nullable),
+                "review": metadata.get("review"),
+            }
+        )
+    return cache
+
+
+def validate_record_fields(body_json: str, fields: dict[str, Any]) -> dict[str, Any]:
+    """Validate+coerce a single record's fields against the schema body.
+
+    Returns the coerced single-row mapping. Raises SchemaValidationError with a
+    list of {column, check, error} dicts on failure.
+    """
+    schema = _load(body_json)
+    frame = pd.DataFrame([fields])
+    try:
+        validated = schema.validate(frame, lazy=True)
+    except pa.errors.SchemaErrors as exc:
+        failures = exc.failure_cases
+        errors = [
+            {
+                "column": row.get("column"),
+                "check": row.get("check"),
+                "error": str(row.get("failure_case")),
+            }
+            for row in failures.to_dict(orient="records")
+        ]
+        raise SchemaValidationError(errors) from exc
+    # Round-trip through pandas JSON so numpy scalars (int64/bool_/float64) become native
+    # python types and NaN/NaT nulls become None — required for the record.fields JSONB column.
+    return json.loads(validated.iloc[[0]].to_json(orient="records"))[0]
+```
+
+> **Fallback (if the metadata round-trip test in Step 1 fails on the installed Pandera):**
+> Pandera's `to_json`/`from_json` does not reliably preserve per-`Column.metadata` on
+> all versions. If `test_derive_columns_cache_round_trips_review_metadata` fails, do NOT
+> store the `review` widget inside the Pandera body. Instead carry it in a side map:
+> change `SchemaVersionCreate` (Task 6) to accept an optional `review_widgets: dict[str, dict]`
+> (column name → widget config), persist it on `SchemaVersion` (add a `review_widgets` JSONB
+> column + migration), and have `derive_columns_cache` accept that map and merge it into each
+> column's `review`. Record this decision in spec §12 before proceeding.
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `uv run pytest tests/unit/contexts/v2/test_schema_bodies.py -v`
+Expected: PASS (3 passed).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/extralit_server/contexts/v2/__init__.py src/extralit_server/contexts/v2/schema_bodies.py tests/unit/contexts/v2/
+git commit -m "feat(v2): pandera column-cache derivation and record validation helpers"
+```
+
+---
+
+## Task 3: ORM models — Schema and SchemaVersion
+
+**Files:**
+- Create: `src/extralit_server/models/v2/__init__.py`, `src/extralit_server/models/v2/schemas.py`
+- Modify: `src/extralit_server/models/__init__.py`
+- Test: `tests/integration/models/v2/__init__.py` (Create), `tests/integration/models/v2/test_schema_models.py` (Create)
+
+**Interfaces:**
+- Produces ORM classes (in `extralit_server.models.v2.schemas`, re-exported from `extralit_server.models.v2`):
+  - `Schema`: `id`, `workspace_id`, `name`, `kind` (`SchemaKind`), `status` (`SchemaStatus`), `current_version_id: UUID | None`, `settings: dict`, `inserted_at`, `updated_at`; relationship `versions: list[SchemaVersion]`. Uniq(`workspace_id`,`name`).
+  - `SchemaVersion`: `id`, `schema_id`, `version: int`, `object_key`, `object_version_id: str | None`, `etag`, `checksum`, `parent_version_id: UUID | None`, `columns_cache: list`, `created_by: UUID | None`; relationship `schema: Schema`. Uniq(`schema_id`,`version`).
+
+- [ ] **Step 1: Write the failing test**
+
+Create `tests/integration/models/v2/__init__.py` (empty) and `tests/integration/models/v2/test_schema_models.py`:
+
+```python
+import pytest
+
+from extralit_server.enums import SchemaKind, SchemaStatus
+from extralit_server.models.v2 import Schema, SchemaVersion
+from tests.factories import WorkspaceFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_create_schema_and_version(db):
+    workspace = await WorkspaceFactory.create()
+    schema = await Schema.create(
+        db,
+        name="population",
+        kind=SchemaKind.table,
+        status=SchemaStatus.draft,
+        workspace_id=workspace.id,
+    )
+    version = await SchemaVersion.create(
+        db,
+        schema_id=schema.id,
+        version=1,
+        object_key=f"schemas/{schema.id}/v1.json",
+        etag="abc123",
+        checksum="def456",
+        columns_cache=[{"name": "n", "dtype": "str", "nullable": False, "review": None}],
+    )
+    assert schema.id is not None
+    assert version.schema_id == schema.id
+    assert version.version == 1
+
+    loaded = await Schema.get(db, schema.id)
+    assert loaded.name == "population"
+    assert loaded.kind == SchemaKind.table
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest tests/integration/models/v2/test_schema_models.py -v`
+Expected: FAIL with `ModuleNotFoundError: extralit_server.models.v2`.
+
+- [ ] **Step 3: Implement the models**
+
+Create `src/extralit_server/models/v2/schemas.py`:
+
+```python
+from uuid import UUID
+
+from sqlalchemy import JSON, ForeignKey, String, Text, UniqueConstraint
+from sqlalchemy import Enum as SAEnum
+from sqlalchemy.ext.mutable import MutableDict, MutableList
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from extralit_server.enums import SchemaKind, SchemaStatus
+from extralit_server.models.base import DatabaseModel
+
+SchemaKindEnum = SAEnum(SchemaKind, name="schema_kind_enum")
+SchemaStatusEnum = SAEnum(SchemaStatus, name="schema_status_enum")
+
+
+class Schema(DatabaseModel):
+    __tablename__ = "schemas"
+
+    name: Mapped[str] = mapped_column(String, index=True)
+    kind: Mapped[SchemaKind] = mapped_column(SchemaKindEnum, default=SchemaKind.table)
+    status: Mapped[SchemaStatus] = mapped_column(SchemaStatusEnum, default=SchemaStatus.draft, index=True)
+    current_version_id: Mapped[UUID | None] = mapped_column(
+        ForeignKey("schema_versions.id", ondelete="SET NULL", use_alter=True), nullable=True
+    )
+    settings: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSON), default=dict)
+    workspace_id: Mapped[UUID] = mapped_column(ForeignKey("workspaces.id", ondelete="CASCADE"), index=True)
+
+    versions: Mapped[list["SchemaVersion"]] = relationship(
+        back_populates="schema",
+        order_by="SchemaVersion.version",
+        cascade="all, delete-orphan",
+        foreign_keys="SchemaVersion.schema_id",
+    )
+
+    __table_args__ = (UniqueConstraint("workspace_id", "name", name="schema_workspace_id_name_uq"),)
+
+    def __repr__(self) -> str:
+        return f"Schema(id={self.id!s}, name={self.name!r}, kind={self.kind!r}, status={self.status!r})"
+
+
+class SchemaVersion(DatabaseModel):
+    __tablename__ = "schema_versions"
+
+    schema_id: Mapped[UUID] = mapped_column(ForeignKey("schemas.id", ondelete="CASCADE"), index=True)
+    version: Mapped[int] = mapped_column(index=True)
+    object_key: Mapped[str] = mapped_column(Text)
+    object_version_id: Mapped[str | None] = mapped_column(Text, nullable=True)
+    etag: Mapped[str] = mapped_column(String)
+    checksum: Mapped[str] = mapped_column(String)
+    parent_version_id: Mapped[UUID | None] = mapped_column(
+        ForeignKey("schema_versions.id", ondelete="SET NULL"), nullable=True
+    )
+    columns_cache: Mapped[list] = mapped_column(MutableList.as_mutable(JSON), default=list)
+    created_by: Mapped[UUID | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
+
+    schema: Mapped["Schema"] = relationship(back_populates="versions", foreign_keys=[schema_id])
+
+    __table_args__ = (UniqueConstraint("schema_id", "version", name="schema_version_schema_id_version_uq"),)
+
+    def __repr__(self) -> str:
+        return f"SchemaVersion(id={self.id!s}, schema_id={self.schema_id!s}, version={self.version!r})"
+```
+
+Create `src/extralit_server/models/v2/__init__.py`:
+
+```python
+from extralit_server.models.v2.schemas import Schema, SchemaVersion
+
+__all__ = ["Schema", "SchemaVersion"]
+```
+
+- [ ] **Step 4: Register v2 models for metadata discovery**
+
+In `src/extralit_server/models/__init__.py`, add an import line so `DatabaseModel.metadata` includes the v2 tables (append after the existing model imports; keep alphabetical neighbours intact):
+
+```python
+from extralit_server.models.v2 import Schema, SchemaVersion  # noqa: F401
+```
+
+- [ ] **Step 5: Run test to verify it passes**
+
+Run: `uv run pytest tests/integration/models/v2/test_schema_models.py -v`
+Expected: PASS (the `db` fixture runs migrations to `head`; since the migration is not yet created this will FAIL with "relation schemas does not exist"). If so, proceed to Task 4 and re-run this test at Task 4 Step 5.
+
+> Note: this model test depends on Task 4's migration. Commit the models now; the green run happens after Task 4.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/extralit_server/models/v2/ src/extralit_server/models/__init__.py tests/integration/models/v2/
+git commit -m "feat(v2): Schema and SchemaVersion ORM models"
+```
+
+---
+
+## Task 4: Alembic migration for schemas + schema_versions
+
+**Files:**
+- Create: `src/extralit_server/alembic/versions/<rev>_create_schema_and_schema_version_tables.py`
+
+**Interfaces:**
+- Produces: `schemas` and `schema_versions` tables matching Task 3 models.
+
+- [ ] **Step 1: Generate a migration skeleton (auto-fills revision + down_revision)**
+
+Run: `uv run alembic -c src/extralit_server/alembic.ini revision --autogenerate -m "create schema and schema_version tables"`
+Expected: a new file under `alembic/versions/`. Note its path. Autogenerate may produce partial ops; you will replace the `upgrade`/`downgrade` bodies in the next step with the exact content below (keep the generated `revision`/`down_revision` header lines).
+
+- [ ] **Step 2: Replace the upgrade/downgrade bodies**
+
+In the generated file, keep the top `revision`/`down_revision`/`branch_labels`/`depends_on` lines as generated; replace the bodies with:
+
+```python
+import sqlalchemy as sa
+from alembic import op
+
+# (keep the auto-generated revision/down_revision header lines above)
+
+
+def upgrade() -> None:
+    op.create_table(
+        "schemas",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("kind", sa.Enum("singleton", "table", name="schema_kind_enum"), nullable=False),
+        sa.Column("status", sa.Enum("draft", "published", name="schema_status_enum"), nullable=False),
+        sa.Column("current_version_id", sa.Uuid(), nullable=True),
+        sa.Column("settings", sa.JSON(), nullable=False),
+        sa.Column("workspace_id", sa.Uuid(), nullable=False),
+        sa.Column("inserted_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["workspace_id"], ["workspaces.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("workspace_id", "name", name="schema_workspace_id_name_uq"),
+    )
+    op.create_index(op.f("ix_schemas_name"), "schemas", ["name"], unique=False)
+    op.create_index(op.f("ix_schemas_status"), "schemas", ["status"], unique=False)
+    op.create_index(op.f("ix_schemas_workspace_id"), "schemas", ["workspace_id"], unique=False)
+
+    op.create_table(
+        "schema_versions",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("schema_id", sa.Uuid(), nullable=False),
+        sa.Column("version", sa.Integer(), nullable=False),
+        sa.Column("object_key", sa.Text(), nullable=False),
+        sa.Column("object_version_id", sa.Text(), nullable=True),
+        sa.Column("etag", sa.String(), nullable=False),
+        sa.Column("checksum", sa.String(), nullable=False),
+        sa.Column("parent_version_id", sa.Uuid(), nullable=True),
+        sa.Column("columns_cache", sa.JSON(), nullable=False),
+        sa.Column("created_by", sa.Uuid(), nullable=True),
+        sa.Column("inserted_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["schema_id"], ["schemas.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["parent_version_id"], ["schema_versions.id"], ondelete="SET NULL"),
+        sa.ForeignKeyConstraint(["created_by"], ["users.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("schema_id", "version", name="schema_version_schema_id_version_uq"),
+    )
+    op.create_index(op.f("ix_schema_versions_schema_id"), "schema_versions", ["schema_id"], unique=False)
+    op.create_index(op.f("ix_schema_versions_version"), "schema_versions", ["version"], unique=False)
+
+    # Deferred FK: schemas.current_version_id -> schema_versions.id (created after both tables exist)
+    op.create_foreign_key(
+        "schema_current_version_id_fk",
+        "schemas",
+        "schema_versions",
+        ["current_version_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("schema_current_version_id_fk", "schemas", type_="foreignkey")
+    op.drop_index(op.f("ix_schema_versions_version"), table_name="schema_versions")
+    op.drop_index(op.f("ix_schema_versions_schema_id"), table_name="schema_versions")
+    op.drop_table("schema_versions")
+    op.drop_index(op.f("ix_schemas_workspace_id"), table_name="schemas")
+    op.drop_index(op.f("ix_schemas_status"), table_name="schemas")
+    op.drop_index(op.f("ix_schemas_name"), table_name="schemas")
+    op.drop_table("schemas")
+    sa.Enum(name="schema_kind_enum").drop(op.get_bind(), checkfirst=True)
+    sa.Enum(name="schema_status_enum").drop(op.get_bind(), checkfirst=True)
+```
+
+- [ ] **Step 3: Apply the migration**
+
+Run: `uv run alembic -c src/extralit_server/alembic.ini upgrade head`
+Expected: applies cleanly, creating `schemas` and `schema_versions`.
+
+- [ ] **Step 4: Verify downgrade then re-upgrade (round-trip)**
+
+Run: `uv run alembic -c src/extralit_server/alembic.ini downgrade -1 && uv run alembic -c src/extralit_server/alembic.ini upgrade head`
+Expected: both succeed with no errors.
+
+- [ ] **Step 5: Run the Task 3 model test (now green)**
+
+Run: `uv run pytest tests/integration/models/v2/test_schema_models.py -v`
+Expected: PASS (1 passed).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/extralit_server/alembic/versions/
+git commit -m "feat(v2): alembic migration for schemas and schema_versions tables"
+```
+
+---
+
+## Task 5: Test factories for Schema and SchemaVersion
+
+**Files:**
+- Modify: `tests/factories.py`
+
+**Interfaces:**
+- Produces: `SchemaFactory` (SubFactory `WorkspaceFactory`), `SchemaVersionFactory` (SubFactory `SchemaFactory`) usable as `await SchemaFactory.create(...)`.
+
+- [ ] **Step 1: Write the failing test**
+
+Create `tests/integration/models/v2/test_schema_factories.py`:
+
+```python
+import pytest
+
+from extralit_server.models.v2 import Schema, SchemaVersion
+from tests.factories import SchemaFactory, SchemaVersionFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_schema_factory_creates_row(db):
+    schema = await SchemaFactory.create(name="outcomes")
+    assert isinstance(schema, Schema)
+    assert schema.workspace_id is not None
+
+
+async def test_schema_version_factory_links_schema(db):
+    version = await SchemaVersionFactory.create()
+    assert isinstance(version, SchemaVersion)
+    assert version.schema_id is not None
+    assert version.version >= 1
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest tests/integration/models/v2/test_schema_factories.py -v`
+Expected: FAIL with `ImportError: cannot import name 'SchemaFactory'`.
+
+- [ ] **Step 3: Add the factories**
+
+In `tests/factories.py`, add the v2 import to the model import block:
+
+```python
+from extralit_server.models.v2 import Schema as SchemaModel
+from extralit_server.models.v2 import SchemaVersion as SchemaVersionModel
+```
+
+and append (after `WorkspaceFactory` is defined, anywhere later in the file):
+
+```python
+class SchemaFactory(BaseFactory):
+    class Meta:
+        model = SchemaModel
+
+    name = factory.Sequence(lambda n: f"schema-{n}")
+    kind = SchemaKind.table
+    status = SchemaStatus.draft
+    workspace = factory.SubFactory(WorkspaceFactory)
+
+
+class SchemaVersionFactory(BaseFactory):
+    class Meta:
+        model = SchemaVersionModel
+
+    schema = factory.SubFactory(SchemaFactory)
+    version = factory.Sequence(lambda n: n + 1)
+    object_key = factory.LazyAttribute(lambda o: f"schemas/{o.schema.id}/v{o.version}.json")
+    etag = factory.Sequence(lambda n: f"etag-{n}")
+    checksum = factory.Sequence(lambda n: f"checksum-{n}")
+    columns_cache = factory.LazyFunction(list)  # fresh list per row, not a shared mutable default
+```
+
+Add the enum import near the top of `tests/factories.py` (alongside the existing `from extralit_server.enums import ...` line):
+
+```python
+from extralit_server.enums import SchemaKind, SchemaStatus
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run pytest tests/integration/models/v2/test_schema_factories.py -v`
+Expected: PASS (2 passed).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add tests/factories.py tests/integration/models/v2/test_schema_factories.py
+git commit -m "test(v2): SchemaFactory and SchemaVersionFactory"
+```
+
+---
+
+## Task 6: Pydantic API schemas (request/response)
+
+**Files:**
+- Create: `src/extralit_server/api/schemas/v2/__init__.py` (empty), `src/extralit_server/api/schemas/v2/schemas.py`
+- Test: `tests/unit/api/schemas/v2/__init__.py` (empty), `tests/unit/api/schemas/v2/test_schema_models.py`
+
+**Interfaces:**
+- Produces Pydantic models in `extralit_server.api.schemas.v2.schemas`:
+  - `SchemaCreate { name: str, kind: SchemaKind, workspace_id: UUID, settings: dict = {} }`
+  - `SchemaUpdate { name: str | None, settings: dict | None }`
+  - `SchemaVersionCreate { body: str }` (the Pandera `DataFrameSchema.to_json()` string)
+  - `SchemaVersionRead` (from_attributes): `id, schema_id, version, object_key, object_version_id, etag, checksum, parent_version_id, columns_cache, inserted_at`
+  - `SchemaRead` (from_attributes): `id, name, kind, status, current_version_id, settings, workspace_id, inserted_at, updated_at`
+  - `Schemas { items: list[SchemaRead] }`
+
+- [ ] **Step 1: Write the failing test**
+
+Create `tests/unit/api/schemas/v2/__init__.py` (empty) and `tests/unit/api/schemas/v2/test_schema_models.py`:
+
+```python
+from uuid import uuid4
+
+from extralit_server.api.schemas.v2.schemas import SchemaCreate, SchemaVersionCreate
+from extralit_server.enums import SchemaKind
+
+
+def test_schema_create_defaults_settings_to_empty_dict():
+    payload = SchemaCreate(name="population", kind=SchemaKind.table, workspace_id=uuid4())
+    assert payload.settings == {}
+
+
+def test_schema_version_create_requires_body():
+    v = SchemaVersionCreate(body='{"columns": {}}')
+    assert v.body.startswith("{")
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest tests/unit/api/schemas/v2/test_schema_models.py -v`
+Expected: FAIL with `ModuleNotFoundError: extralit_server.api.schemas.v2.schemas`.
+
+- [ ] **Step 3: Implement the Pydantic models**
+
+Create `src/extralit_server/api/schemas/v2/__init__.py` (empty) and `src/extralit_server/api/schemas/v2/schemas.py`:
+
+```python
+from datetime import datetime
+from typing import Any
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field, constr
+
+from extralit_server.enums import SchemaKind, SchemaStatus
+
+SchemaName = constr(min_length=1, max_length=200)
+
+
+class SchemaCreate(BaseModel):
+    name: SchemaName
+    kind: SchemaKind = SchemaKind.table
+    workspace_id: UUID
+    settings: dict[str, Any] = Field(default_factory=dict)
+
+
+class SchemaUpdate(BaseModel):
+    name: SchemaName | None = None
+    settings: dict[str, Any] | None = None
+
+
+class SchemaVersionCreate(BaseModel):
+    body: str = Field(..., description="Pandera DataFrameSchema serialized via .to_json()")
+
+
+class SchemaVersionRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: UUID
+    schema_id: UUID
+    version: int
+    object_key: str
+    object_version_id: str | None
+    etag: str
+    checksum: str
+    parent_version_id: UUID | None
+    columns_cache: list[dict[str, Any]]
+    inserted_at: datetime
+
+
+class SchemaRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: UUID
+    name: str
+    kind: SchemaKind
+    status: SchemaStatus
+    current_version_id: UUID | None
+    settings: dict[str, Any]
+    workspace_id: UUID
+    inserted_at: datetime
+    updated_at: datetime
+
+
+class Schemas(BaseModel):
+    items: list[SchemaRead]
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run pytest tests/unit/api/schemas/v2/test_schema_models.py -v`
+Expected: PASS (2 passed).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/extralit_server/api/schemas/v2/ tests/unit/api/schemas/v2/
+git commit -m "feat(v2): pydantic request/response schemas for Schema and SchemaVersion"
+```
+
+---
+
+## Task 7: Schema context (DB + object store, version publishing)
+
+**Files:**
+- Create: `src/extralit_server/contexts/v2/schemas.py`
+- Test: `tests/integration/contexts/v2/__init__.py` (empty), `tests/integration/contexts/v2/test_schemas_context.py`
+
+**Interfaces:**
+- Consumes: `Schema`/`SchemaVersion` models; `contexts.files.put_object`; `contexts.v2.schema_bodies.derive_columns_cache`; `contexts.files.compute_hash`.
+- Produces (all `async`, in `extralit_server.contexts.v2.schemas`):
+  - `create_schema(db, *, name, kind, workspace_id, settings=None) -> Schema`
+  - `get_schema(db, schema_id) -> Schema | None`
+  - `list_schemas(db, *, workspace_id=None) -> list[Schema]`
+  - `update_schema(db, schema, *, name=None, settings=None) -> Schema`
+  - `delete_schema(db, schema) -> Schema`
+  - `publish_version(db, s3_client, schema, *, body: str, bucket: str, created_by=None) -> SchemaVersion` — uploads body to `schemas/{schema_id}/v{n}.json`, derives `columns_cache`, creates the version, advances `schema.current_version_id`, sets `status=published`.
+  - `object_key_for(schema_id, version) -> str`
+
+- [ ] **Step 1: Write the failing test**
+
+Create `tests/integration/contexts/v2/__init__.py` (empty) and `tests/integration/contexts/v2/test_schemas_context.py`:
+
+```python
+from unittest.mock import AsyncMock
+
+import pandera as pa
+import pytest
+
+from extralit_server.contexts.files import ObjectMetadata
+from extralit_server.contexts.v2 import schemas as schemas_ctx
+from extralit_server.enums import SchemaKind, SchemaStatus
+from extralit_server.models.v2 import Schema, SchemaVersion
+from tests.factories import WorkspaceFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+def _body() -> str:
+    return pa.DataFrameSchema(columns={"name": pa.Column(pa.String, nullable=False)}).to_json()
+
+
+async def test_create_and_list_schema(db):
+    ws = await WorkspaceFactory.create()
+    schema = await schemas_ctx.create_schema(db, name="population", kind=SchemaKind.table, workspace_id=ws.id)
+    assert isinstance(schema, Schema)
+
+    listed = await schemas_ctx.list_schemas(db, workspace_id=ws.id)
+    assert [s.id for s in listed] == [schema.id]
+
+
+async def test_publish_version_uploads_body_and_advances_pointer(db):
+    ws = await WorkspaceFactory.create()
+    schema = await schemas_ctx.create_schema(db, name="population", kind=SchemaKind.table, workspace_id=ws.id)
+
+    s3 = AsyncMock()
+    # put_object returns ObjectMetadata; emulate via patching contexts.files.put_object
+    version = await schemas_ctx.publish_version(
+        db, s3, schema, body=_body(), bucket=ws.name, created_by=None
+    )
+
+    assert isinstance(version, SchemaVersion)
+    assert version.version == 1
+    assert version.object_key == f"schemas/{schema.id}/v1.json"
+    assert any(c["name"] == "name" for c in version.columns_cache)
+
+    refreshed = await Schema.get(db, schema.id)
+    assert refreshed.current_version_id == version.id
+    assert refreshed.status == SchemaStatus.published
+
+    # Second publish increments version and links lineage
+    v2 = await schemas_ctx.publish_version(db, s3, refreshed, body=_body(), bucket=ws.name)
+    assert v2.version == 2
+    assert v2.parent_version_id == version.id
+```
+
+> The test passes a bare `AsyncMock` as the S3 client. Implement `publish_version` to call `contexts.files.put_object`, which we patch in Step 3 via a thin wrapper so the mock works without real S3. See implementation note below.
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest tests/integration/contexts/v2/test_schemas_context.py -v`
+Expected: FAIL with `ModuleNotFoundError` / `AttributeError: module ... has no attribute 'create_schema'`.
+
+- [ ] **Step 3: Implement the context**
+
+Create `src/extralit_server/contexts/v2/schemas.py`:
+
+```python
+"""Business logic for v2 Schemas and their object-store-backed versions."""
+
+from typing import TYPE_CHECKING, Any
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from extralit_server.contexts import files as files_ctx
+from extralit_server.contexts.v2.schema_bodies import derive_columns_cache
+from extralit_server.enums import SchemaKind, SchemaStatus
+from extralit_server.models.v2 import Schema, SchemaVersion
+
+if TYPE_CHECKING:
+    from types_aiobotocore_s3.client import S3Client
+
+
+def object_key_for(schema_id: UUID, version: int) -> str:
+    return f"schemas/{schema_id}/v{version}.json"
+
+
+async def create_schema(
+    db: AsyncSession,
+    *,
+    name: str,
+    kind: SchemaKind,
+    workspace_id: UUID,
+    settings: dict[str, Any] | None = None,
+) -> Schema:
+    return await Schema.create(
+        db,
+        name=name,
+        kind=kind,
+        workspace_id=workspace_id,
+        settings=settings or {},
+        status=SchemaStatus.draft,
+    )
+
+
+async def get_schema(db: AsyncSession, schema_id: UUID) -> Schema | None:
+    return await Schema.get(db, schema_id)
+
+
+async def list_schemas(db: AsyncSession, *, workspace_id: UUID | None = None) -> list[Schema]:
+    stmt = select(Schema)
+    if workspace_id is not None:
+        stmt = stmt.filter_by(workspace_id=workspace_id)
+    stmt = stmt.order_by(Schema.inserted_at)
+    return (await db.execute(stmt)).scalars().all()
+
+
+async def update_schema(
+    db: AsyncSession,
+    schema: Schema,
+    *,
+    name: str | None = None,
+    settings: dict[str, Any] | None = None,
+) -> Schema:
+    values: dict[str, Any] = {}
+    if name is not None:
+        values["name"] = name
+    if settings is not None:
+        values["settings"] = settings
+    if not values:
+        return schema
+    # replace_dict=True gives PUT semantics: a provided `settings` payload replaces the
+    # stored dict wholesale. CRUDMixin.fill() otherwise merges dicts (mixins.py:47-50),
+    # which would make removing a settings key impossible.
+    return await schema.update(db, replace_dict=True, **values)
+
+
+async def delete_schema(db: AsyncSession, schema: Schema) -> Schema:
+    return await schema.delete(db)
+
+
+async def _next_version_number(db: AsyncSession, schema_id: UUID) -> int:
+    versions = (
+        (await db.execute(select(SchemaVersion).filter_by(schema_id=schema_id))).scalars().all()
+    )
+    return (max((v.version for v in versions), default=0)) + 1
+
+
+async def publish_version(
+    db: AsyncSession,
+    s3_client: "S3Client",
+    schema: Schema,
+    *,
+    body: str,
+    bucket: str,
+    created_by: UUID | None = None,
+) -> SchemaVersion:
+    """Upload a Pandera body to the object store and register a new SchemaVersion."""
+    next_version = await _next_version_number(db, schema.id)
+    key = object_key_for(schema.id, next_version)
+
+    metadata = await files_ctx.put_object(
+        s3_client, bucket, key, body, content_type="application/json"
+    )
+
+    parent = await db.get(SchemaVersion, schema.current_version_id) if schema.current_version_id else None
+
+    version = await SchemaVersion.create(
+        db,
+        schema_id=schema.id,
+        version=next_version,
+        object_key=key,
+        object_version_id=getattr(metadata, "version_id", None),
+        etag=metadata.etag,
+        checksum=files_ctx.compute_hash(body.encode("utf-8")),
+        parent_version_id=parent.id if parent else None,
+        columns_cache=derive_columns_cache(body),
+        created_by=created_by,
+        autocommit=False,
+    )
+    await schema.update(
+        db, current_version_id=version.id, status=SchemaStatus.published, autocommit=False
+    )
+    await db.commit()
+    return version
+```
+
+> Implementation note for the test mock: `files_ctx.put_object` calls `s3_client.put_object(...)` then `s3_client.head_object(...)`. With a bare `AsyncMock`, `head_object` returns an `AsyncMock`, and `ObjectMetadata` construction would fail. To keep the context test hermetic, patch `put_object` in the test instead of using the real one. Update the test's `test_publish_version_*` to add at the top of the test body:
+>
+> ```python
+> from extralit_server.contexts import files as files_ctx
+> files_ctx_put = AsyncMock(return_value=ObjectMetadata(
+>     bucket_name=ws.name, object_name="k", etag="etag-1", size=1, last_modified=None,
+>     content_type="application/json", version_id="ver-1", metadata={},
+> ))
+> monkeypatch.setattr("extralit_server.contexts.v2.schemas.files_ctx.put_object", files_ctx_put)
+> ```
+>
+> and add `monkeypatch` to the test signature: `async def test_publish_version_uploads_body_and_advances_pointer(db, monkeypatch):`. (`ObjectMetadata.last_modified` accepts `None`; if its type rejects None, pass a fixed `datetime(2026, 1, 1)`.)
+
+- [ ] **Step 4: Update the test per the mock note, then run**
+
+Apply the monkeypatch note to the publish test. Run: `uv run pytest tests/integration/contexts/v2/test_schemas_context.py -v`
+Expected: PASS (2 passed).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/extralit_server/contexts/v2/schemas.py tests/integration/contexts/v2/
+git commit -m "feat(v2): schema context with object-store version publishing"
+```
+
+---
+
+## Task 8: API router and `/api/v2` mount
+
+**Files:**
+- Create: `src/extralit_server/api/policies/v1/schema_policy.py`, `src/extralit_server/api/v2/__init__.py`, `src/extralit_server/api/v2/schemas.py`
+- Modify: `src/extralit_server/api/policies/v1/__init__.py` (export `SchemaPolicy`), `src/extralit_server/_app.py`
+- Test: `tests/integration/api/v2/__init__.py` (empty), `tests/integration/api/v2/test_schemas.py`
+
+**Interfaces:**
+- Consumes: `contexts.v2.schemas`, `contexts.files.get_s3_client`, `security.auth.get_current_user`, `database.get_async_db`, `api.policies.v1.{authorize, SchemaPolicy}`.
+- Produces: `SchemaPolicy` with `list(workspace_id)`, `create(workspace_id)`, `get(schema)`, `update(schema)`, `delete(schema)`, `publish(schema)` — each returning a `PolicyAction`, mirroring `DatasetPolicy`.
+- Produces endpoints under `/api/v2`:
+  - `POST /schemas` → `SchemaRead` (201)
+  - `GET /schemas?workspace_id=` → `Schemas`
+  - `GET /schemas/{schema_id}` → `SchemaRead`
+  - `PUT /schemas/{schema_id}` → `SchemaRead`
+  - `DELETE /schemas/{schema_id}` → `SchemaRead`
+  - `POST /schemas/{schema_id}/versions` → `SchemaVersionRead` (201)
+  - `GET /schemas/{schema_id}/versions` → `list[SchemaVersionRead]`
+  - `GET /schemas/{schema_id}/columns` → `list[dict]` (current version's `columns_cache`)
+
+- [ ] **Step 1: Write the failing test**
+
+Create `tests/integration/api/v2/__init__.py` (empty) and `tests/integration/api/v2/test_schemas.py`:
+
+```python
+import pandera as pa
+import pytest
+
+from extralit_server.enums import SchemaKind
+from tests.factories import OwnerFactory, WorkspaceFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+def _body() -> str:
+    return pa.DataFrameSchema(columns={"name": pa.Column(pa.String, nullable=False)}).to_json()
+
+
+async def test_create_get_list_schema(async_client, owner_auth_header):
+    ws = await WorkspaceFactory.create()
+    resp = await async_client.post(
+        "/api/v2/schemas",
+        headers=owner_auth_header,
+        json={"name": "population", "kind": SchemaKind.table.value, "workspace_id": str(ws.id)},
+    )
+    assert resp.status_code == 201, resp.text
+    schema_id = resp.json()["id"]
+
+    resp = await async_client.get(f"/api/v2/schemas/{schema_id}", headers=owner_auth_header)
+    assert resp.status_code == 200
+    assert resp.json()["name"] == "population"
+
+    resp = await async_client.get(
+        f"/api/v2/schemas?workspace_id={ws.id}", headers=owner_auth_header
+    )
+    assert resp.status_code == 200
+    assert [s["id"] for s in resp.json()["items"]] == [schema_id]
+
+
+async def test_publish_version_and_columns(async_client, owner_auth_header, monkeypatch):
+    from unittest.mock import AsyncMock
+    from datetime import datetime
+    from extralit_server.contexts.files import ObjectMetadata
+
+    monkeypatch.setattr(
+        "extralit_server.contexts.v2.schemas.files_ctx.put_object",
+        AsyncMock(return_value=ObjectMetadata(
+            bucket_name="b", object_name="k", etag="etag-1", size=1,
+            last_modified=datetime(2026, 1, 1), content_type="application/json",
+            version_id="ver-1", metadata={},
+        )),
+    )
+
+    ws = await WorkspaceFactory.create()
+    resp = await async_client.post(
+        "/api/v2/schemas",
+        headers=owner_auth_header,
+        json={"name": "outcomes", "kind": SchemaKind.table.value, "workspace_id": str(ws.id)},
+    )
+    schema_id = resp.json()["id"]
+
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema_id}/versions",
+        headers=owner_auth_header,
+        json={"body": _body()},
+    )
+    assert resp.status_code == 201, resp.text
+    assert resp.json()["version"] == 1
+
+    resp = await async_client.get(f"/api/v2/schemas/{schema_id}/columns", headers=owner_auth_header)
+    assert resp.status_code == 200
+    assert any(c["name"] == "name" for c in resp.json())
+
+
+async def test_non_member_cannot_create_or_read_schema(async_client, annotator_auth_header):
+    # The annotator behind annotator_auth_header is NOT a member of this workspace.
+    ws = await WorkspaceFactory.create()
+    resp = await async_client.post(
+        "/api/v2/schemas",
+        headers=annotator_auth_header,
+        json={"name": "secret", "kind": SchemaKind.table.value, "workspace_id": str(ws.id)},
+    )
+    assert resp.status_code == 403, resp.text
+```
+
+> If your test suite's fixtures use different names than `async_client` / `owner_auth_header` /
+> `annotator_auth_header` / `OwnerFactory`, match the existing v1 API tests under
+> `tests/integration/api/handlers/v1/`. Grep there first:
+> `grep -rn "owner_auth_header\|annotator_auth_header\|async_client" tests/integration/api | head`.
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest tests/integration/api/v2/test_schemas.py -v`
+Expected: FAIL with 404s (router not mounted) or fixture import error.
+
+- [ ] **Step 3: Implement the authorization policy and router**
+
+First create `src/extralit_server/api/policies/v1/schema_policy.py` (mirrors `DatasetPolicy`; owner always allowed, otherwise workspace membership — admin for writes):
+
+```python
+from uuid import UUID
+
+from extralit_server.api.policies.v1.commons import PolicyAction
+from extralit_server.models import User
+from extralit_server.models.v2 import Schema
+
+
+class SchemaPolicy:
+    @classmethod
+    def list(cls, workspace_id: UUID) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or await actor.is_member(workspace_id)
+
+        return is_allowed
+
+    @classmethod
+    def create(cls, workspace_id: UUID) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or (actor.is_admin and await actor.is_member(workspace_id))
+
+        return is_allowed
+
+    @classmethod
+    def get(cls, schema: Schema) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or await actor.is_member(schema.workspace_id)
+
+        return is_allowed
+
+    @classmethod
+    def update(cls, schema: Schema) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id))
+
+        return is_allowed
+
+    @classmethod
+    def delete(cls, schema: Schema) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id))
+
+        return is_allowed
+
+    @classmethod
+    def publish(cls, schema: Schema) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id))
+
+        return is_allowed
+```
+
+Export it from `src/extralit_server/api/policies/v1/__init__.py` (append alongside the existing policy exports):
+
+```python
+from extralit_server.api.policies.v1.schema_policy import SchemaPolicy  # noqa: F401
+```
+
+Then create `src/extralit_server/api/v2/schemas.py`:
+
+```python
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, Query, Security, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from extralit_server.api.policies.v1 import SchemaPolicy, authorize
+from extralit_server.api.schemas.v2.schemas import (
+    Schemas,
+    SchemaCreate,
+    SchemaRead,
+    SchemaUpdate,
+    SchemaVersionCreate,
+    SchemaVersionRead,
+)
+from extralit_server.contexts import files as files_ctx
+from extralit_server.contexts.v2 import schemas as schemas_ctx
+from extralit_server.database import get_async_db
+from extralit_server.errors.future import NotFoundError
+from extralit_server.models import User, Workspace
+from extralit_server.security import auth
+
+router = APIRouter(tags=["v2: schemas"])
+
+
+async def _get_schema_or_404(db: AsyncSession, schema_id: UUID):
+    schema = await schemas_ctx.get_schema(db, schema_id)
+    if schema is None:
+        raise NotFoundError(f"Schema with id `{schema_id}` not found")
+    return schema
+
+
+@router.post("/schemas", response_model=SchemaRead, status_code=status.HTTP_201_CREATED)
+async def create_schema(
+    *,
+    payload: SchemaCreate,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    await authorize(current_user, SchemaPolicy.create(payload.workspace_id))
+    schema = await schemas_ctx.create_schema(
+        db,
+        name=payload.name,
+        kind=payload.kind,
+        workspace_id=payload.workspace_id,
+        settings=payload.settings,
+    )
+    return schema
+
+
+@router.get("/schemas", response_model=Schemas)
+async def list_schemas(
+    *,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+    workspace_id: Annotated[UUID, Query(description="Workspace to list schemas for (required)")],
+):
+    # workspace_id is required so every list is scoped + authorized (no cross-workspace listing).
+    await authorize(current_user, SchemaPolicy.list(workspace_id))
+    items = await schemas_ctx.list_schemas(db, workspace_id=workspace_id)
+    return Schemas(items=items)
+
+
+@router.get("/schemas/{schema_id}", response_model=SchemaRead)
+async def get_schema(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.get(schema))
+    return schema
+
+
+@router.put("/schemas/{schema_id}", response_model=SchemaRead)
+async def update_schema(
+    *,
+    schema_id: UUID,
+    payload: SchemaUpdate,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.update(schema))
+    return await schemas_ctx.update_schema(db, schema, name=payload.name, settings=payload.settings)
+
+
+@router.delete("/schemas/{schema_id}", response_model=SchemaRead)
+async def delete_schema(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.delete(schema))
+    return await schemas_ctx.delete_schema(db, schema)
+
+
+@router.post(
+    "/schemas/{schema_id}/versions",
+    response_model=SchemaVersionRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def publish_schema_version(
+    *,
+    schema_id: UUID,
+    payload: SchemaVersionCreate,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    s3_client=Depends(files_ctx.get_s3_client),
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.publish(schema))
+    workspace = await Workspace.get_or_raise(db, schema.workspace_id)
+    return await schemas_ctx.publish_version(
+        db, s3_client, schema, body=payload.body, bucket=workspace.name, created_by=current_user.id
+    )
+
+
+@router.get("/schemas/{schema_id}/versions", response_model=list[SchemaVersionRead])
+async def list_schema_versions(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.get(schema))
+    return sorted(await schema.awaitable_attrs.versions, key=lambda v: v.version)
+
+
+@router.get("/schemas/{schema_id}/columns", response_model=list[dict])
+async def get_schema_columns(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.get(schema))
+    if schema.current_version_id is None:
+        return []
+    from extralit_server.models.v2 import SchemaVersion
+
+    version = await SchemaVersion.get(db, schema.current_version_id)
+    return version.columns_cache if version else []
+```
+
+Create `src/extralit_server/api/v2/__init__.py`:
+
+```python
+from fastapi import FastAPI
+
+from extralit_server._version import __version__ as extralit_version
+from extralit_server.api.errors.v1.exception_handlers import add_exception_handlers as add_exception_handlers_v1
+from extralit_server.api.handlers.v1 import authentication as authentication_v1
+from extralit_server.api.v2 import schemas as schemas_v2
+from extralit_server.errors.base_errors import __ALL__
+from extralit_server.errors.error_handler import APIErrorHandler
+
+
+def create_api_v2() -> FastAPI:
+    api_v2 = FastAPI(
+        title="Extralit v2",
+        description="Extralit Server API v2 (schema-centric)",
+        version=str(extralit_version),
+        responses={error.HTTP_STATUS: error.api_documentation() for error in __ALL__},
+    )
+    APIErrorHandler.configure_app(api_v2)
+    add_exception_handlers_v1(api_v2)
+
+    # Auth endpoints are reused from v1 so v2 tokens work identically.
+    api_v2.include_router(authentication_v1.router)
+    api_v2.include_router(schemas_v2.router)
+    return api_v2
+
+
+api_v2 = create_api_v2()
+```
+
+- [ ] **Step 4: Mount the v2 app**
+
+In `src/extralit_server/_app.py`, add the import near the existing `from extralit_server.api.routes import api_v1` (line ~26):
+
+```python
+from extralit_server.api.v2 import api_v2
+```
+
+and immediately after the existing `app.mount("/api/v1", api_v1)` (line ~212) add:
+
+```python
+    app.mount("/api/v2", api_v2)
+```
+
+- [ ] **Step 5: Run test to verify it passes**
+
+Run: `uv run pytest tests/integration/api/v2/test_schemas.py -v`
+Expected: PASS (2 passed). If fixture names differ, adjust per the Step-1 note and re-run.
+
+- [ ] **Step 6: Run the full v2 test set + lint**
+
+Run:
+```bash
+uv run pytest tests/unit/test_enums_v2.py tests/unit/contexts/v2 tests/unit/api/schemas/v2 tests/integration/models/v2 tests/integration/contexts/v2 tests/integration/api/v2 -v
+uv run ruff check src/extralit_server/api/v2 src/extralit_server/contexts/v2 src/extralit_server/models/v2 src/extralit_server/api/schemas/v2 src/extralit_server/api/policies/v1/schema_policy.py
+```
+Expected: all pass (including `test_non_member_cannot_create_or_read_schema`); ruff clean (fix any reported issues).
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add src/extralit_server/api/v2/ src/extralit_server/api/policies/v1/ src/extralit_server/_app.py tests/integration/api/v2/
+git commit -m "feat(v2): /api/v2 schema router with CRUD, version publishing, and per-workspace authz"
+```
+
+---
+
+## Self-Review (completed during authoring)
+
+**Spec coverage (Phase 1 scope only):** Schema-as-entity ✓ (Task 3); object-store-backed
+versioned body + thin DB registry + lineage ✓ (Tasks 3,4,7); `columns_cache` derived from
+Pandera body ✓ (Task 2,7); server-side validation primitive ✓ (Task 2, consumed by records in
+Phase 2); additive Alembic tables beside v1 ✓ (Task 4); isolated `/api/v2` module with
+per-workspace authorization mirroring v1 `DatasetPolicy` ✓ (Task 8). Out-of-phase items
+(records, LanceDB, annotation, queue, migrator, retirements) are explicitly deferred to
+Phases 2–6 below — not gaps.
+
+**Review findings addressed (roborev job 52):** HIGH missing authorization → `SchemaPolicy` +
+`authorize()` on every route, `workspace_id` made required on list (Task 8). MEDIUM Pandera
+metadata round-trip → dedicated test + version floor + documented fallback (Tasks 1,2). MEDIUM
+numpy-typed coerced values → native-type JSON round-trip + tests (Task 2). LOW PUT merge
+semantics → `replace_dict=True` (Task 7). LOW shared mutable defaults → `default=dict` /
+`factory.LazyFunction(list)` (Tasks 3,5).
+
+**Placeholder scan:** none — every code/test step contains full content; the only deferred
+value is the Alembic `revision`/`down_revision` header, which Alembic generates in Task 4 Step 1.
+
+**Type consistency:** `derive_columns_cache`/`validate_record_fields`/`SchemaValidationError`
+(Task 2) match their consumers in Task 7 and the `columns_cache` shape used in Tasks 3/5/8.
+`publish_version` signature matches its router call in Task 8. Model field names match the
+migration columns and the Pydantic `from_attributes` readers.
+
+## Downstream phases (separate plans, written after this lands)
+
+2. **Records** — `record` table (`schema_version_id` pin, `reference`, `fields` JSONB),
+   validated bulk-upsert (consumes Task 2 helpers), `GET /references/{reference}` cross-schema view.
+3. **LanceDB index** — `index/` engine (one table per schema, schema evolution), inline sync on
+   record write, `:search`/similarity; delete `search_engine/` (ES/OpenSearch).
+4. **Annotation v2** — `question` (column binding; table=N columns), `suggestion`, `response`.
+5. **Queue** — `queue`/`queue_item`/`queue_assignment`, distribution/overlap, `GET /queues/{id}/next`.
+6. **Migrator + retirement** — v1→v2 per-dataset migrator; delete v1 tables/handlers + frontend
+   retirement ledger items from the spec §9.
diff --git a/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md b/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md
new file mode 100644
index 000000000..203681f3e
--- /dev/null
+++ b/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md
@@ -0,0 +1,263 @@
+# Schema-Centric Data Model — Design Spec
+
+**Date:** 2026-06-27
+**Status:** Approved design (server data model + API; SDK/frontend follow)
+**Author:** brainstorming session (Jonny + Claude)
+
+## 1. Problem
+
+Extralit inherited Argilla's general-purpose-yet-static model. A dataset's "schema"
+is only an implicit bag of `Field` rows; records store extracted data as a flat
+`records.fields` JSON blob with **no formal binding to a schema or its version**. The
+Pandera `DataFrameSchema` that actually describes the extraction shape lives only in
+the SDK and as JSON files in workspace object storage — the server never validates
+against it. Search runs on an Elasticsearch/OpenSearch abstraction.
+
+We want the **data record to be a first-class citizen that maps to a group of columns
+defined by a user-defined, versioned Schema**, building toward a **project-level
+extraction table that every record maps toward**. Annotation (Questions / Responses /
+Suggestions) becomes a thin per-cell review layer on top of schema columns, not a
+parallel primitive. LanceDB (with native schema evolution + vector/FTS indexes)
+replaces Elasticsearch.
+
+## 2. Fixed constraints (decided)
+
+1. **Pivot depth — Schema-first, annotation as a thin layer.** Schema + its columns
+   are *the* core model. Questions / Responses / Suggestions are an optional per-cell
+   review/validation layer. Records are LLM-generated, so span / ranking / rating
+   surfaces and distribution/overlap remain important for human-in-the-loop.
+2. **Schema home — object-store body + thin DB registry pointer.** The Pandera
+   `DataFrameSchema` body lives in object storage with its native versioning
+   (`etag` / `version_id` / `last_modified`). A DB registry row carries schema
+   identity + version pointer + lineage. (Object store vs DB source-of-truth dynamics
+   stay flexible while LanceDB↔annotation↔schema dynamics settle.)
+3. **Delivery — parallel v2, migrated gradually.** New schema-centric tables/endpoints
+   are built alongside v1 using Alembic. Datasets migrate over gradually; **v1 is then
+   deleted — back-compat is explicitly NOT a goal** (keep v1 only as a migration source
+   and where a piece is independently useful).
+4. **First target — server data model + API.** Fix Schema-as-entity and the
+   `Record → schema_version` binding first. SDK and frontend follow.
+
+## 3. Core model decisions (from interview)
+
+- **Postgres is the source of truth** for record cell-values; **LanceDB is a derived
+  index** (rebuildable from Postgres), replacing ES entirely.
+- **Cell storage = structured JSONB + schema FK.** `record.fields` stays a JSONB map
+  keyed by column name, gains a formal `schema_version_id` FK, and is **validated
+  server-side against the Pandera schema on write**. No per-schema DDL churn.
+- **Schema ↔ Dataset is 1:1, collapsed into one primitive.** The `schema` row *is* the
+  extraction table. UI calls it a "Dataset"; the model calls it `Schema`. This removes
+  a redundant entity. (If schema-reuse-across-datasets is needed later, split into two
+  FK-linked rows then.)
+- **A Record pins its own `schema_version_id`.** The Dataset/Schema carries a
+  `current_version_id` pointer that advances; older records keep the shape they were
+  extracted under. The backing LanceDB table evolves to the column **superset** via
+  Lance schema evolution.
+- **`reference` (the document) is the cross-schema join key.** A document's full
+  "project-level extraction" = the union of its records across every schema that share
+  the `reference` (one document-level/singleton row + N table rows per schema).
+- **Question is the review-config primitive**, reframed as a binding to schema
+  **column(s)**: one column normally, **multiple columns when `type=table`** (a
+  sub-table). Suggestion/Response hang off a Question, which resolves to cell(s).
+- **The Queue is the first-class UI citizen**, not the Dataset. It is a **persisted**
+  entity: ordered walk over `reference`s, cross-Dataset scope, distribution/overlap and
+  assignment for human-in-the-loop on LLM output.
+
+## 4. Architecture (Approach A — isolated v2 module)
+
+New isolated module set sharing only `workspaces`, `users`, `documents`, and
+auth/security with v1:
+
+- `models/v2/` — SQLAlchemy models for the tables in §5.
+- `contexts/v2/` — business logic (schemas, records, validation, index sync, queues).
+- `api/v2/` — FastAPI routers mounted at `/api/v2`.
+- `index/` (new) — LanceDB index engine + sync; replaces `search_engine/`.
+
+Three stores, clear roles:
+
+- **Postgres** — source of truth: registry, record cell-values, annotation, queue.
+- **Object store (S3/MinIO)** — Pandera `DataFrameSchema` bodies, versioned via native
+  `version_id`/`etag` (reuse `contexts/files.py`).
+- **LanceDB** — one table per schema, derived from Postgres; owns vector + full-text +
+  scalar filtering. The `search_engine/` ES/OpenSearch stack is **deleted, not
+  reimplemented behind the old ABC.**
+
+## 5. Relational schema (new v2 tables, additive Alembic)
+
+Distinct/`v2_`-prefixed names during the parallel phase; renamed to canonical names on
+v1 retirement.
+
+| Table | Purpose | Key columns |
+|---|---|---|
+| `schema` | **Core entity** = extraction table (≙ "Dataset" in UI). Registry pointer to the Pandera body. | `id`, `workspace_id`→workspaces, `name`, `kind` (`singleton`\|`table`), `status` (`draft`\|`published`), `current_version_id`→schema_version (nullable until first publish), `settings` JSONB (guidelines, etc.), `inserted_at`, `updated_at`; uniq(`workspace_id`,`name`) |
+| `schema_version` | Immutable version → object-store body + lineage + denormalized column cache. | `id`, `schema_id`, `version`, `object_key`, `object_version_id`, `etag`, `checksum`, `parent_version_id` (lineage, nullable), `columns_cache` JSONB (per-column name/dtype/nullable/review-widget, derived from the S3 body for fast validation + UI), `created_by`→users, `inserted_at`; uniq(`schema_id`,`version`) |
+| `record` | One typed row; pins its version; carries the cross-schema join key. | `id`, `schema_id`, `schema_version_id` (pinned), `reference` (indexed), `external_id` (nullable), `fields` JSONB (cells keyed by column), `metadata` JSONB, `status` (`pending`\|`completed`\|`discarded`), `inserted_at`, `updated_at`; idx(`schema_id`,`reference`), idx(`reference`); uniq(`schema_id`,`external_id`) |
+| `question` | Review config bound to column(s): 1 normally, N if `type=table`. | `id`, `schema_id`, `name`, `title`, `description`, `type` (`text`\|`label`\|`multi_label`\|`rating`\|`ranking`\|`span`\|`table`), `columns` JSONB (bound column names), `settings` JSONB, `required`, `inserted_at`, `updated_at`; uniq(`schema_id`,`name`) |
+| `suggestion` | LLM output per (record, question). | `id`, `record_id`, `question_id`, `value` JSONB, `score` JSONB (float \| float[]), `agent`, `type`, `inserted_at`, `updated_at`; uniq(`record_id`,`question_id`) |
+| `response` | Human review per (record, user); `values` keyed by question/column → resolves to cells; multiple users per record = overlap. | `id`, `record_id`, `user_id`, `values` JSONB, `status` (`draft`\|`submitted`\|`discarded`), `inserted_at`, `updated_at`; uniq(`record_id`,`user_id`) |
+| `queue` | First-class review surface; spans schemas; owns distribution. | `id`, `workspace_id`, `name`, `scope` JSONB (schema ids + filters), `ordering` JSONB, `distribution` JSONB (annotators/reference, completion rule), `status`, `inserted_at`, `updated_at` |
+| `queue_item` | One **reference** (document) to review, ordered. | `id`, `queue_id`, `reference`, `position`, `status` (`pending`\|`in_progress`\|`completed`); uniq(`queue_id`,`reference`) |
+| `queue_assignment` | Distribution/overlap: who reviews which reference. | `id`, `queue_item_id`, `user_id`, `status`, `inserted_at`; uniq(`queue_item_id`,`user_id`) |
+
+**Reused as-is:** `workspaces`, `users`, `documents` (`documents.reference`/`pmid`/`doi`
+become the queue join key).
+**Dropped vs v1:** `vectors` + `vectors_settings` (LanceDB owns vectors inline);
+`fields` + `metadata_properties` as standalone entities (folded into the Pandera
+schema body + `columns_cache`).
+
+## 6. Storage, indexing & validation flow
+
+**Schema publish.** Client/SDK uploads a Pandera `DataFrameSchema` body → object store
+(versioned). Server creates a `schema_version` row capturing `object_key` +
+`object_version_id` + `etag` + `checksum`, derives `columns_cache` from the body, links
+`parent_version_id`, and advances `schema.current_version_id`. The schema's LanceDB
+table is created or **evolved** to the new column superset.
+
+**Record write (bulk upsert).** Resolve the record's `schema_version_id` (default =
+`current_version_id`). Validate `fields` against that version's Pandera schema
+(coerce + check) using `columns_cache` (fall back to fetching the S3 body when needed).
+Persist to Postgres (truth). Then **sync** the row into the schema's LanceDB table
+(record_id, reference, schema_version_id, status, column values, embeddings, metadata).
+
+**Index engine.** New `index/` package: a LanceDB engine exposing `upsert_records`,
+`delete_records`, `search` (scalar filter + FTS), `similarity_search` (vector), and
+`rebuild(schema_id)`. Postgres is authoritative; the Lance table is always rebuildable.
+Sync is inline on commit for the first cut (job-queue offload is a later optimization).
+
+**Reference grouping.** `GET /api/v2/references/{reference}` returns all records across
+every schema in the workspace sharing that `reference` — the document's project-level
+extraction view that the Queue UI renders.
+
+## 7. API surface (`/api/v2`)
+
+- **Schemas (= Datasets):** `POST /schemas`, `GET /schemas`, `GET /schemas/{id}`,
+  `PUT /schemas/{id}`, `DELETE /schemas/{id}`; `GET /schemas/{id}/columns`.
+- **Schema versions:** `POST /schemas/{id}/versions` (register body + evolve Lance
+  table), `GET /schemas/{id}/versions`, `GET /schemas/{id}/versions/{version}`.
+- **Records:** `POST /schemas/{id}/records:bulk-upsert`, `GET /schemas/{id}/records`
+  (filter/paginate), `DELETE /schemas/{id}/records`,
+  `POST /schemas/{id}/records:search` (LanceDB: text / vector / scalar filter).
+- **References:** `GET /references/{reference}` (cross-schema document view).
+- **Questions:** `POST|GET|PUT|DELETE /schemas/{id}/questions`.
+- **Annotation:** `PUT /records/{id}/suggestions`, `PUT /records/{id}/responses`.
+- **Queues:** `POST|GET|PUT|DELETE /queues`; `GET /queues/{id}/next` (next reference
+  for current user per distribution); `POST /queues/{id}/items` (build/refresh from
+  scope); `GET /queues/{id}/progress`; assignment endpoints.
+
+## 8. Migration (v1 → v2, gradual)
+
+A per-dataset migrator: for each v1 `Dataset` → synthesize a Pandera `DataFrameSchema`
+from its `Field` rows, upload as the first `schema_version`, create the `schema` row;
+copy `records` (`fields` JSON → v2 `record.fields`, derive `reference` from the
+document link / `external_id` / metadata); copy `questions` (map to column bindings),
+`responses`, `suggestions`; build the LanceDB table. Idempotent and re-runnable per
+dataset. Once a workspace's datasets are migrated and verified, delete the
+corresponding v1 rows/handlers.
+
+## 9. Retirement ledger (lean-up; v1 back-compat not a goal)
+
+### Server (`extralit-server`)
+
+**Delete (replaced by LanceDB / schema-centric model):**
+- `search_engine/elasticsearch.py`, `search_engine/opensearch.py`,
+  `search_engine/commons.py` (`BaseElasticAndOpenSearchEngine` + ES DSL builders),
+  ES-coupled methods in `search_engine/base.py`.
+- `cli/search_engine/reindex.py`, `cli/search_engine/__main__.py` (ES reindex CLI).
+- `models` `Vector` + `VectorSettings`; `api/handlers/v1/vectors_settings.py`;
+  `contexts/records_bulk.py::_upsert_records_vectors`.
+
+**Fold into schema/record v2 (then drop v1 handler):**
+- `api/handlers/v1/fields.py`, `metadata_properties.py` → schema `columns_cache`.
+- `api/handlers/v1/questions.py`, `suggestions.py`, `responses.py` → thin v2 annotation.
+- `contexts/datasets.py` (674 LOC) → split into `contexts/v2/{schemas,records,
+  annotation,index}.py`.
+- `contexts/distribution.py` → fold into queue distribution logic.
+
+**Audit / likely peripheral (decide per use case):** `api/handlers/v1/chat.py`
+(RAG → rewrite on LanceDB or externalize), `models.py` proxy, `webhooks/` +
+`webhook_jobs.py`, `imports.py`/`contexts/imports.py`, `contexts/hub.py`.
+**Keep:** auth/security, `workspaces`, `users`, `documents`, `files.py` (object store),
+`alembic/`, `validators/`, `utils/`, `errors/`.
+
+### Frontend (`extralit-frontend`) — later phase, recorded now
+
+**High (architectural):** `components/features/dataset-creation/*` (creation wizard +
+question/field builders), `pages/dataset/[id]/*` routing, `pages/index.vue` (dataset
+list home), `pages/new/*`; `v1/domain/entities/{dataset,record,page}`;
+`v1/infrastructure/repositories/{DatasetRepository,RecordRepository}.ts`;
+the criteria chain (`SortCriteria`, `MetadataCriteria`, `SimilarityCriteria`,
+`ResponseCriteria`, `SearchTextCriteria`).
+**Medium (Argilla annotation widgets → schema-driven forms):**
+`components/features/annotation/container/questions/form/*`
+(span, rating, ranking, single/multi-label, text-area), `header/filters/*`,
+`header/responses-filter/*`, `container/similarity/*`, `progress/*`,
+`v1/domain/entities/distribution`.
+**Low (naming/branding/test debt):** `*FeedbackTask*` naming, `e2e/` Argilla baselines,
+"dataset"-heavy translation keys, `nuxt.config.ts` argilla doc link.
+*(Replaced by: a Queue-first navigation, a schema editor, and schema-driven cell
+forms — designed in the frontend phase.)*
+
+## 10. Testing strategy
+
+- **Unit:** Pandera validation (coerce/reject), `columns_cache` derivation, version
+  lineage, reference grouping, queue distribution/assignment selection.
+- **Contexts (async pytest):** schema publish + version evolution, record bulk-upsert
+  with validation, annotation upsert resolving to cells.
+- **Index:** LanceDB engine against an ephemeral table — upsert/search/similarity/
+  rebuild; schema-evolution to superset; Postgres↔Lance parity after `rebuild`.
+- **API:** `/api/v2` happy-path + validation-failure + authz per router.
+- **Migration:** v1 fixture dataset → v2; record/annotation parity; idempotent re-run.
+
+## 11. Scope boundary (this build)
+
+**In:** v2 Alembic tables (§5); `models/v2` + `contexts/v2` + `api/v2`; Pandera
+server-side validation; LanceDB `index/` engine + inline sync + rebuild; reference
+grouping; persisted Queue with distribution; v1→v2 migrator; server-side retirements
+in §9 that are unblocked by the above.
+**Out (later phases):** SDK threading; frontend Queue UI + schema editor + frontend
+retirements; job-queue offload of index sync; RAG/chat rewrite.
+
+## 12. Open items to resolve during planning
+
+- Exact `version` format (monotonic int vs semver) and whether a draft version can be
+  edited in place before publish.
+- `reference` derivation rules during migration when no `documents` link exists.
+- Whether `response.values` stays keyed-by-question (v1-style) or moves per-question
+  rows — current design keeps the v1-style keyed map for minimal churn.
+
+## 13. Resolved during Phase 1 implementation
+
+- **Review-widget storage — side map, not Pandera body metadata (RESOLVED 2026-06-27).**
+  The installed Pandera (0.32.0, with pandas 3.0.1) does **not** preserve per-`Column.metadata`
+  through `DataFrameSchema.to_json()` / `from_json()` — the metadata is silently dropped. Per the
+  Phase-1 plan's documented fallback, the per-column review widget is therefore **not** stored
+  inside the Pandera body. Instead it is carried in a side map `review_widgets: dict[str, dict]`
+  (column name → widget config): accepted on `SchemaVersionCreate`, persisted as a
+  `schema_versions.review_widgets` JSONB column, and merged into each column's `review` entry by
+  `derive_columns_cache(body_json, review_widgets=...)`. The Pandera body remains the source of
+  truth for column name/dtype/nullable; `review_widgets` is an out-of-band annotation overlay.
+- **Single-row record validation tolerates nulls in nullable numpy-int columns (RESOLVED
+  2026-06-27).** `pa.Int` maps to numpy `int64`, which cannot hold `None`; coercing a null in a
+  nullable Int column to `int64` fails. `validate_record_fields` therefore separates null-valued
+  fields out, raises only when a null lands in a non-nullable column, validates+coerces the
+  remaining non-null fields against a reduced sub-schema, and re-attaches nulls as `None` in the
+  returned mapping. It also rejects a required (non-nullable) column that is entirely omitted
+  from `fields` (a `missing` error), and converts coerced values to native JSON types without a
+  lossy `DataFrame.to_json` round-trip (numpy scalars via `.item()`, Timestamps as ISO strings)
+  so high-precision floats and datetimes survive into `record.fields`.
+
+### Review findings — accepted/deferred (roborev jobs 55–61)
+
+- **`put_object` now propagates the S3 `VersionId` (FIXED).** `contexts/files.py::put_object`
+  previously dropped `VersionId`, so `schema_versions.object_version_id` was always NULL on
+  versioned buckets; it now mirrors `get_object`.
+- **Concurrent same-schema publish race (ACCEPTED for Phase 1, Low).** `publish_version` computes
+  `max(version)+1` in Python; two simultaneous publishes can collide on the `(schema_id, version)`
+  unique constraint (uncaught → 500) and orphan an uploaded object. Same-schema concurrency is
+  rare; inline sync is already slated for job-queue offload later, where this is hardened.
+- **Fetch-then-authorize existence signal (ACCEPTED, Low).** v2 read/write handlers return 404 for
+  a missing id and 403 for an existing-but-unauthorized one, mirroring the v1 pattern. IDs are
+  UUIDs; the disclosure surface is negligible and kept consistent with v1.
+- **Models + their migration live in adjacent commits (NOTED).** The roborev per-commit review of
+  the models commit flagged the absence of the table migration in that same commit; the migration
+  lands in the immediately following commit, and the branch is internally consistent and green.
diff --git a/extralit-server/pyproject.toml b/extralit-server/pyproject.toml
index e5ad2d7e3..a8a8d765b 100644
--- a/extralit-server/pyproject.toml
+++ b/extralit-server/pyproject.toml
@@ -76,6 +76,7 @@ dependencies = [
     "ocrmypdf>=16.11.0",
     "pdf2image>=1.17.0",
     "opencv-python-headless>=4.11.0.86",
+    "pandera[io]>=0.20",
 ]
 
 [project.optional-dependencies]
diff --git a/extralit-server/src/extralit_server/_app.py b/extralit-server/src/extralit_server/_app.py
index 5ab6d335b..a743c1104 100644
--- a/extralit-server/src/extralit_server/_app.py
+++ b/extralit-server/src/extralit_server/_app.py
@@ -24,6 +24,7 @@
 from extralit_server import helpers
 from extralit_server._version import __version__ as extralit_version
 from extralit_server.api.routes import api_v1
+from extralit_server.api.v2 import api_v2
 from extralit_server.constants import DEFAULT_API_KEY, DEFAULT_PASSWORD, DEFAULT_USERNAME
 from extralit_server.contexts import accounts, files
 from extralit_server.database import get_async_db
@@ -210,6 +211,7 @@ async def add_server_timing_header(request: Request, call_next):
 def configure_api_router(app: FastAPI):
     """Configures and set the api router to app"""
     app.mount("/api/v1", api_v1)
+    app.mount("/api/v2", api_v2)
 
 
 def configure_telemetry(app: FastAPI):
diff --git a/extralit-server/src/extralit_server/alembic/versions/9f3010c649c8_create_schema_and_schema_version_tables.py b/extralit-server/src/extralit_server/alembic/versions/9f3010c649c8_create_schema_and_schema_version_tables.py
new file mode 100644
index 000000000..288c650ce
--- /dev/null
+++ b/extralit-server/src/extralit_server/alembic/versions/9f3010c649c8_create_schema_and_schema_version_tables.py
@@ -0,0 +1,88 @@
+"""create schema and schema_version tables
+
+Revision ID: 9f3010c649c8
+Revises: 54d65879a68e
+Create Date: 2026-06-27 17:00:36.438902
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "9f3010c649c8"
+down_revision = "54d65879a68e"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "schemas",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("kind", sa.Enum("singleton", "table", name="schema_kind_enum"), nullable=False),
+        sa.Column("status", sa.Enum("draft", "published", name="schema_status_enum"), nullable=False),
+        sa.Column("current_version_id", sa.Uuid(), nullable=True),
+        sa.Column("settings", sa.JSON(), nullable=False),
+        sa.Column("workspace_id", sa.Uuid(), nullable=False),
+        sa.Column("inserted_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["workspace_id"], ["workspaces.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("workspace_id", "name", name="schema_workspace_id_name_uq"),
+    )
+    op.create_index(op.f("ix_schemas_name"), "schemas", ["name"], unique=False)
+    op.create_index(op.f("ix_schemas_status"), "schemas", ["status"], unique=False)
+    op.create_index(op.f("ix_schemas_workspace_id"), "schemas", ["workspace_id"], unique=False)
+
+    op.create_table(
+        "schema_versions",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("schema_id", sa.Uuid(), nullable=False),
+        sa.Column("version", sa.Integer(), nullable=False),
+        sa.Column("object_key", sa.Text(), nullable=False),
+        sa.Column("object_version_id", sa.Text(), nullable=True),
+        sa.Column("etag", sa.String(), nullable=False),
+        sa.Column("checksum", sa.String(), nullable=False),
+        sa.Column("parent_version_id", sa.Uuid(), nullable=True),
+        sa.Column("columns_cache", sa.JSON(), nullable=False),
+        sa.Column("review_widgets", sa.JSON(), nullable=False),
+        sa.Column("created_by", sa.Uuid(), nullable=True),
+        sa.Column("inserted_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["schema_id"], ["schemas.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["parent_version_id"], ["schema_versions.id"], ondelete="SET NULL"),
+        sa.ForeignKeyConstraint(["created_by"], ["users.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("schema_id", "version", name="schema_version_schema_id_version_uq"),
+    )
+    op.create_index(op.f("ix_schema_versions_schema_id"), "schema_versions", ["schema_id"], unique=False)
+    op.create_index(op.f("ix_schema_versions_version"), "schema_versions", ["version"], unique=False)
+
+    # Deferred FK: schemas.current_version_id -> schema_versions.id (created after both tables exist).
+    # SQLite cannot ALTER-add a constraint; the column carries no DB-level FK there (model behaviour
+    # is unaffected and the test suite runs on SQLite). Postgres gets the real constraint.
+    if op.get_bind().dialect.name != "sqlite":
+        op.create_foreign_key(
+            "schema_current_version_id_fk",
+            "schemas",
+            "schema_versions",
+            ["current_version_id"],
+            ["id"],
+            ondelete="SET NULL",
+        )
+
+
+def downgrade() -> None:
+    if op.get_bind().dialect.name != "sqlite":
+        op.drop_constraint("schema_current_version_id_fk", "schemas", type_="foreignkey")
+    op.drop_index(op.f("ix_schema_versions_version"), table_name="schema_versions")
+    op.drop_index(op.f("ix_schema_versions_schema_id"), table_name="schema_versions")
+    op.drop_table("schema_versions")
+    op.drop_index(op.f("ix_schemas_workspace_id"), table_name="schemas")
+    op.drop_index(op.f("ix_schemas_status"), table_name="schemas")
+    op.drop_index(op.f("ix_schemas_name"), table_name="schemas")
+    op.drop_table("schemas")
+    sa.Enum(name="schema_kind_enum").drop(op.get_bind(), checkfirst=True)
+    sa.Enum(name="schema_status_enum").drop(op.get_bind(), checkfirst=True)
diff --git a/extralit-server/src/extralit_server/api/policies/v1/__init__.py b/extralit-server/src/extralit_server/api/policies/v1/__init__.py
index 1e7ab92e2..ef07848f5 100644
--- a/extralit-server/src/extralit_server/api/policies/v1/__init__.py
+++ b/extralit-server/src/extralit_server/api/policies/v1/__init__.py
@@ -8,6 +8,7 @@
 from extralit_server.api.policies.v1.question_policy import QuestionPolicy
 from extralit_server.api.policies.v1.record_policy import RecordPolicy
 from extralit_server.api.policies.v1.response_policy import ResponsePolicy
+from extralit_server.api.policies.v1.schema_policy import SchemaPolicy
 from extralit_server.api.policies.v1.suggestion_policy import SuggestionPolicy
 from extralit_server.api.policies.v1.user_policy import UserPolicy
 from extralit_server.api.policies.v1.vector_settings_policy import VectorSettingsPolicy
@@ -25,6 +26,7 @@
     "QuestionPolicy",
     "RecordPolicy",
     "ResponsePolicy",
+    "SchemaPolicy",
     "SuggestionPolicy",
     "UserPolicy",
     "VectorSettingsPolicy",
diff --git a/extralit-server/src/extralit_server/api/policies/v1/schema_policy.py b/extralit-server/src/extralit_server/api/policies/v1/schema_policy.py
new file mode 100644
index 000000000..15c7ba87e
--- /dev/null
+++ b/extralit-server/src/extralit_server/api/policies/v1/schema_policy.py
@@ -0,0 +1,49 @@
+from uuid import UUID
+
+from extralit_server.api.policies.v1.commons import PolicyAction
+from extralit_server.models import User
+from extralit_server.models.v2 import Schema
+
+
+class SchemaPolicy:
+    @classmethod
+    def list(cls, workspace_id: UUID) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or await actor.is_member(workspace_id)
+
+        return is_allowed
+
+    @classmethod
+    def create(cls, workspace_id: UUID) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or (actor.is_admin and await actor.is_member(workspace_id))
+
+        return is_allowed
+
+    @classmethod
+    def get(cls, schema: Schema) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or await actor.is_member(schema.workspace_id)
+
+        return is_allowed
+
+    @classmethod
+    def update(cls, schema: Schema) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id))
+
+        return is_allowed
+
+    @classmethod
+    def delete(cls, schema: Schema) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id))
+
+        return is_allowed
+
+    @classmethod
+    def publish(cls, schema: Schema) -> PolicyAction:
+        async def is_allowed(actor: User) -> bool:
+            return actor.is_owner or (actor.is_admin and await actor.is_member(schema.workspace_id))
+
+        return is_allowed
diff --git a/extralit-server/src/extralit_server/api/schemas/v2/__init__.py b/extralit-server/src/extralit_server/api/schemas/v2/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/src/extralit_server/api/schemas/v2/schemas.py b/extralit-server/src/extralit_server/api/schemas/v2/schemas.py
new file mode 100644
index 000000000..e9a77a699
--- /dev/null
+++ b/extralit-server/src/extralit_server/api/schemas/v2/schemas.py
@@ -0,0 +1,62 @@
+from datetime import datetime
+from typing import Any
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field, constr
+
+from extralit_server.enums import SchemaKind, SchemaStatus
+
+SchemaName = constr(min_length=1, max_length=200)
+
+
+class SchemaCreate(BaseModel):
+    name: SchemaName
+    kind: SchemaKind = SchemaKind.table
+    workspace_id: UUID
+    settings: dict[str, Any] = Field(default_factory=dict)
+
+
+class SchemaUpdate(BaseModel):
+    name: SchemaName | None = None
+    settings: dict[str, Any] | None = None
+
+
+class SchemaVersionCreate(BaseModel):
+    body: str = Field(..., description="Pandera DataFrameSchema serialized via .to_json()")
+    # Per-column review widgets carried out-of-band (column name -> widget config); see spec §13.
+    # Pandera's to_json drops Column.metadata, so the review widget cannot live in `body`.
+    review_widgets: dict[str, dict[str, Any]] = Field(default_factory=dict)
+
+
+class SchemaVersionRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: UUID
+    schema_id: UUID
+    version: int
+    object_key: str
+    object_version_id: str | None
+    etag: str
+    checksum: str
+    parent_version_id: UUID | None
+    columns_cache: list[dict[str, Any]]
+    review_widgets: dict[str, dict[str, Any]]
+    inserted_at: datetime
+
+
+class SchemaRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: UUID
+    name: str
+    kind: SchemaKind
+    status: SchemaStatus
+    current_version_id: UUID | None
+    settings: dict[str, Any]
+    workspace_id: UUID
+    inserted_at: datetime
+    updated_at: datetime
+
+
+class Schemas(BaseModel):
+    items: list[SchemaRead]
diff --git a/extralit-server/src/extralit_server/api/v2/__init__.py b/extralit-server/src/extralit_server/api/v2/__init__.py
new file mode 100644
index 000000000..a3972b217
--- /dev/null
+++ b/extralit-server/src/extralit_server/api/v2/__init__.py
@@ -0,0 +1,27 @@
+from fastapi import FastAPI
+
+from extralit_server._version import __version__ as extralit_version
+from extralit_server.api.errors.v1.exception_handlers import add_exception_handlers as add_exception_handlers_v1
+from extralit_server.api.handlers.v1 import authentication as authentication_v1
+from extralit_server.api.v2 import schemas as schemas_v2
+from extralit_server.errors.base_errors import __ALL__
+from extralit_server.errors.error_handler import APIErrorHandler
+
+
+def create_api_v2() -> FastAPI:
+    api_v2 = FastAPI(
+        title="Extralit v2",
+        description="Extralit Server API v2 (schema-centric)",
+        version=str(extralit_version),
+        responses={error.HTTP_STATUS: error.api_documentation() for error in __ALL__},
+    )
+    APIErrorHandler.configure_app(api_v2)
+    add_exception_handlers_v1(api_v2)
+
+    # Auth endpoints are reused from v1 so v2 tokens work identically.
+    api_v2.include_router(authentication_v1.router)
+    api_v2.include_router(schemas_v2.router)
+    return api_v2
+
+
+api_v2 = create_api_v2()
diff --git a/extralit-server/src/extralit_server/api/v2/schemas.py b/extralit-server/src/extralit_server/api/v2/schemas.py
new file mode 100644
index 000000000..a2a291484
--- /dev/null
+++ b/extralit-server/src/extralit_server/api/v2/schemas.py
@@ -0,0 +1,152 @@
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, Query, Security, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from extralit_server.api.policies.v1 import SchemaPolicy, authorize
+from extralit_server.api.schemas.v2.schemas import (
+    SchemaCreate,
+    SchemaRead,
+    Schemas,
+    SchemaUpdate,
+    SchemaVersionCreate,
+    SchemaVersionRead,
+)
+from extralit_server.contexts import files as files_ctx
+from extralit_server.contexts.v2 import schemas as schemas_ctx
+from extralit_server.database import get_async_db
+from extralit_server.errors.future import NotFoundError
+from extralit_server.models import User, Workspace
+from extralit_server.models.v2 import Schema, SchemaVersion
+from extralit_server.security import auth
+
+router = APIRouter(tags=["v2: schemas"])
+
+
+async def _get_schema_or_404(db: AsyncSession, schema_id: UUID) -> Schema:
+    schema = await schemas_ctx.get_schema(db, schema_id)
+    if schema is None:
+        raise NotFoundError(f"Schema with id `{schema_id}` not found")
+    return schema
+
+
+@router.post("/schemas", response_model=SchemaRead, status_code=status.HTTP_201_CREATED)
+async def create_schema(
+    *,
+    payload: SchemaCreate,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    await authorize(current_user, SchemaPolicy.create(payload.workspace_id))
+    return await schemas_ctx.create_schema(
+        db,
+        name=payload.name,
+        kind=payload.kind,
+        workspace_id=payload.workspace_id,
+        settings=payload.settings,
+    )
+
+
+@router.get("/schemas", response_model=Schemas)
+async def list_schemas(
+    *,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+    workspace_id: Annotated[UUID, Query(description="Workspace to list schemas for (required)")],
+):
+    # workspace_id is required so every list is scoped + authorized (no cross-workspace listing).
+    await authorize(current_user, SchemaPolicy.list(workspace_id))
+    items = await schemas_ctx.list_schemas(db, workspace_id=workspace_id)
+    return Schemas(items=items)
+
+
+@router.get("/schemas/{schema_id}", response_model=SchemaRead)
+async def get_schema(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.get(schema))
+    return schema
+
+
+@router.put("/schemas/{schema_id}", response_model=SchemaRead)
+async def update_schema(
+    *,
+    schema_id: UUID,
+    payload: SchemaUpdate,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.update(schema))
+    return await schemas_ctx.update_schema(db, schema, name=payload.name, settings=payload.settings)
+
+
+@router.delete("/schemas/{schema_id}", response_model=SchemaRead)
+async def delete_schema(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.delete(schema))
+    return await schemas_ctx.delete_schema(db, schema)
+
+
+@router.post(
+    "/schemas/{schema_id}/versions",
+    response_model=SchemaVersionRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def publish_schema_version(
+    *,
+    schema_id: UUID,
+    payload: SchemaVersionCreate,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    s3_client=Depends(files_ctx.get_s3_client),
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.publish(schema))
+    workspace = await Workspace.get_or_raise(db, schema.workspace_id)
+    return await schemas_ctx.publish_version(
+        db,
+        s3_client,
+        schema,
+        body=payload.body,
+        bucket=workspace.name,
+        review_widgets=payload.review_widgets,
+        created_by=current_user.id,
+    )
+
+
+@router.get("/schemas/{schema_id}/versions", response_model=list[SchemaVersionRead])
+async def list_schema_versions(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.get(schema))
+    return sorted(await schema.awaitable_attrs.versions, key=lambda v: v.version)
+
+
+@router.get("/schemas/{schema_id}/columns", response_model=list[dict])
+async def get_schema_columns(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.get(schema))
+    if schema.current_version_id is None:
+        return []
+    version = await SchemaVersion.get(db, schema.current_version_id)
+    return version.columns_cache if version else []
diff --git a/extralit-server/src/extralit_server/contexts/files.py b/extralit-server/src/extralit_server/contexts/files.py
index 9df2b319b..2d2d4fe88 100644
--- a/extralit-server/src/extralit_server/contexts/files.py
+++ b/extralit-server/src/extralit_server/contexts/files.py
@@ -317,6 +317,9 @@ async def put_object(
             size=head_response["ContentLength"],
             last_modified=head_response["LastModified"],
             content_type=head_response.get("ContentType", content_type),
+            # Propagate the S3 object version (mirrors get_object) so callers can pin the
+            # immutable version; otherwise schema_versions.object_version_id is always NULL.
+            version_id=head_response.get("VersionId"),
             metadata=head_response.get("Metadata", {}),
         )
 
diff --git a/extralit-server/src/extralit_server/contexts/v2/__init__.py b/extralit-server/src/extralit_server/contexts/v2/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/src/extralit_server/contexts/v2/schema_bodies.py b/extralit-server/src/extralit_server/contexts/v2/schema_bodies.py
new file mode 100644
index 000000000..b8e4c25f0
--- /dev/null
+++ b/extralit-server/src/extralit_server/contexts/v2/schema_bodies.py
@@ -0,0 +1,141 @@
+"""Pure helpers for working with a Pandera DataFrameSchema body (JSON).
+
+No DB or object-store access — given a schema body string, derive a denormalized
+column cache and validate a single record's `fields` dict against it.
+
+Two installed-version realities shape this module (see spec §13):
+
+* Pandera 0.32 drops per-`Column.metadata` through ``to_json``/``from_json``, so the
+  per-column review widget cannot live inside the body. It is carried out-of-band in a
+  ``review_widgets`` side map (column name -> widget config) and merged here.
+* ``pa.Int`` maps to numpy ``int64``, which cannot hold ``None``. A null in a nullable
+  Int column therefore fails dtype coercion, so ``validate_record_fields`` validates only
+  the non-null fields and re-attaches nulls as ``None``.
+"""
+
+import math
+from typing import Any
+
+import pandas as pd
+import pandera.pandas as pa
+
+
+class SchemaValidationError(Exception):
+    """Raised when a record's fields fail Pandera validation."""
+
+    def __init__(self, errors: list[dict[str, Any]]) -> None:
+        self.errors = errors
+        super().__init__(f"Record failed schema validation with {len(errors)} error(s)")
+
+
+def _load(body_json: str) -> pa.DataFrameSchema:
+    return pa.DataFrameSchema.from_json(body_json)
+
+
+def _is_null(value: Any) -> bool:
+    return value is None or (isinstance(value, float) and math.isnan(value))
+
+
+def derive_columns_cache(
+    body_json: str, review_widgets: dict[str, dict[str, Any]] | None = None
+) -> list[dict[str, Any]]:
+    """Return one entry per column: name, dtype, nullable, and optional review widget.
+
+    The Pandera body is the source of truth for name/dtype/nullable. The per-column
+    ``review`` widget is taken from the ``review_widgets`` side map (column name ->
+    widget config); if a column is absent from the map its ``review`` is ``None``.
+    """
+    review_widgets = review_widgets or {}
+    schema = _load(body_json)
+    cache: list[dict[str, Any]] = []
+    for name, column in schema.columns.items():
+        cache.append(
+            {
+                "name": name,
+                "dtype": str(column.dtype),
+                "nullable": bool(column.nullable),
+                "review": review_widgets.get(name),
+            }
+        )
+    return cache
+
+
+def validate_record_fields(body_json: str, fields: dict[str, Any]) -> dict[str, Any]:
+    """Validate+coerce a single record's fields against the schema body.
+
+    Returns the coerced single-row mapping with native JSON types (no numpy scalars),
+    nulls preserved as ``None``. Raises ``SchemaValidationError`` with a list of
+    ``{column, check, error}`` dicts on failure.
+    """
+    schema = _load(body_json)
+    errors: list[dict[str, Any]] = []
+
+    null_fields: dict[str, None] = {}
+    non_null: dict[str, Any] = {}
+    for name, value in fields.items():
+        if _is_null(value):
+            column = schema.columns.get(name)
+            if column is not None and not column.nullable:
+                errors.append({"column": name, "check": "not_nullable", "error": "null value not allowed"})
+            null_fields[name] = None
+        else:
+            non_null[name] = value
+
+    # A required (non-nullable) column entirely omitted from `fields` is a violation too —
+    # not just one explicitly set to null.
+    for name, column in schema.columns.items():
+        if not column.nullable and name not in fields:
+            errors.append({"column": name, "check": "missing", "error": "required column missing"})
+
+    coerced: dict[str, Any] = {}
+    present_columns = {name: col for name, col in schema.columns.items() if name in non_null}
+    if present_columns:
+        sub_schema = pa.DataFrameSchema(present_columns, coerce=True)
+        frame = pd.DataFrame([{name: non_null[name] for name in present_columns}])
+        try:
+            validated = sub_schema.validate(frame, lazy=True)
+        except pa.errors.SchemaErrors as exc:
+            for row in exc.failure_cases.to_dict(orient="records"):
+                errors.append(
+                    {
+                        "column": row.get("column"),
+                        "check": row.get("check"),
+                        "error": str(row.get("failure_case")),
+                    }
+                )
+            raise SchemaValidationError(errors) from exc
+        # Convert numpy scalars to native python types for the record.fields JSONB column.
+        coerced = _row_to_native(validated)
+
+    if errors:
+        raise SchemaValidationError(errors)
+
+    # Non-schema fields that were non-null but absent from the schema fall through as-is.
+    extras = {name: value for name, value in non_null.items() if name not in present_columns}
+    return {**coerced, **extras, **null_fields}
+
+
+def _row_to_native(frame: pd.DataFrame) -> dict[str, Any]:
+    """Convert the single validated row to native, JSON-serializable python types.
+
+    Avoids the lossy ``DataFrame.to_json`` detour, which truncates floats to 10 decimal
+    places and serializes datetimes as deprecated epoch integers. numpy scalars become
+    native python via ``.item()`` (exact for floats), Timestamps become ISO strings, and
+    NaN/NaT become ``None``.
+
+    Each cell is read per-column (``frame[col].iloc[0]``) rather than via ``frame.iloc[0]``:
+    a row Series upcasts to the columns' common dtype, so an ``int64`` cell in a frame that
+    also has a ``float64`` column would silently become a python ``float``.
+    """
+    out: dict[str, Any] = {}
+    for col in frame.columns:
+        value = frame[col].iloc[0]
+        if pd.isna(value):
+            out[col] = None
+        elif isinstance(value, pd.Timestamp):
+            out[col] = value.isoformat()
+        elif hasattr(value, "item"):
+            out[col] = value.item()
+        else:
+            out[col] = value
+    return out
diff --git a/extralit-server/src/extralit_server/contexts/v2/schemas.py b/extralit-server/src/extralit_server/contexts/v2/schemas.py
new file mode 100644
index 000000000..cb451b8b0
--- /dev/null
+++ b/extralit-server/src/extralit_server/contexts/v2/schemas.py
@@ -0,0 +1,126 @@
+"""Business logic for v2 Schemas and their object-store-backed versions."""
+
+from typing import TYPE_CHECKING, Any
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from extralit_server.contexts import files as files_ctx
+from extralit_server.contexts.v2.schema_bodies import derive_columns_cache
+from extralit_server.enums import SchemaStatus
+from extralit_server.models.v2 import Schema, SchemaVersion
+
+if TYPE_CHECKING:
+    from types_aiobotocore_s3.client import S3Client
+
+    from extralit_server.enums import SchemaKind
+
+
+def object_key_for(schema_id: UUID, version: int) -> str:
+    return f"schemas/{schema_id}/v{version}.json"
+
+
+async def create_schema(
+    db: AsyncSession,
+    *,
+    name: str,
+    kind: "SchemaKind",
+    workspace_id: UUID,
+    settings: dict[str, Any] | None = None,
+) -> Schema:
+    return await Schema.create(
+        db,
+        name=name,
+        kind=kind,
+        workspace_id=workspace_id,
+        settings=settings or {},
+        status=SchemaStatus.draft,
+    )
+
+
+async def get_schema(db: AsyncSession, schema_id: UUID) -> Schema | None:
+    return await Schema.get(db, schema_id)
+
+
+async def list_schemas(db: AsyncSession, *, workspace_id: UUID | None = None) -> list[Schema]:
+    stmt = select(Schema)
+    if workspace_id is not None:
+        stmt = stmt.filter_by(workspace_id=workspace_id)
+    stmt = stmt.order_by(Schema.inserted_at)
+    return (await db.execute(stmt)).scalars().all()
+
+
+async def update_schema(
+    db: AsyncSession,
+    schema: Schema,
+    *,
+    name: str | None = None,
+    settings: dict[str, Any] | None = None,
+) -> Schema:
+    values: dict[str, Any] = {}
+    if name is not None:
+        values["name"] = name
+    if settings is not None:
+        values["settings"] = settings
+    if not values:
+        return schema
+    # replace_dict=True gives PUT semantics: a provided `settings` payload replaces the
+    # stored dict wholesale. CRUDMixin.fill() otherwise merges dicts, which would make
+    # removing a settings key impossible.
+    return await schema.update(db, replace_dict=True, **values)
+
+
+async def delete_schema(db: AsyncSession, schema: Schema) -> Schema:
+    return await schema.delete(db)
+
+
+async def _next_version_number(db: AsyncSession, schema_id: UUID) -> int:
+    versions = (await db.execute(select(SchemaVersion).filter_by(schema_id=schema_id))).scalars().all()
+    return (max((v.version for v in versions), default=0)) + 1
+
+
+async def publish_version(
+    db: AsyncSession,
+    s3_client: "S3Client",
+    schema: Schema,
+    *,
+    body: str,
+    bucket: str,
+    review_widgets: dict[str, dict[str, Any]] | None = None,
+    created_by: UUID | None = None,
+) -> SchemaVersion:
+    """Upload a Pandera body to the object store and register a new SchemaVersion.
+
+    `review_widgets` is the out-of-band per-column widget overlay (spec §13); it is
+    persisted on the version and merged into the derived `columns_cache`.
+    """
+    review_widgets = review_widgets or {}
+    next_version = await _next_version_number(db, schema.id)
+    key = object_key_for(schema.id, next_version)
+
+    metadata = await files_ctx.put_object(s3_client, bucket, key, body, content_type="application/json")
+
+    parent = await db.get(SchemaVersion, schema.current_version_id) if schema.current_version_id else None
+
+    version = await SchemaVersion.create(
+        db,
+        schema_id=schema.id,
+        version=next_version,
+        object_key=key,
+        object_version_id=getattr(metadata, "version_id", None),
+        etag=metadata.etag,
+        checksum=files_ctx.compute_hash(body.encode("utf-8")),
+        parent_version_id=parent.id if parent else None,
+        columns_cache=derive_columns_cache(body, review_widgets),
+        review_widgets=review_widgets,
+        created_by=created_by,
+        autocommit=False,
+    )
+    # Flush so the version row (and its uuid `id`, a flush-time default) is persisted before we
+    # point `schema.current_version_id` at it. Doing both in one flush would form a schemas<->
+    # schema_versions FK cycle and leave version.id unset.
+    await db.flush()
+    await schema.update(db, current_version_id=version.id, status=SchemaStatus.published, autocommit=False)
+    await db.commit()
+    return version
diff --git a/extralit-server/src/extralit_server/enums.py b/extralit-server/src/extralit_server/enums.py
index 41129c144..9a82f6508 100644
--- a/extralit-server/src/extralit_server/enums.py
+++ b/extralit-server/src/extralit_server/enums.py
@@ -95,3 +95,13 @@ class SimilarityOrder(StrEnum):
 class OptionsOrder(StrEnum):
     natural = "natural"
     suggestion = "suggestion"
+
+
+class SchemaKind(StrEnum):
+    singleton = "singleton"  # exactly one row per `reference` (document-level extraction)
+    table = "table"  # many rows per `reference` (table extraction)
+
+
+class SchemaStatus(StrEnum):
+    draft = "draft"
+    published = "published"
diff --git a/extralit-server/src/extralit_server/models/__init__.py b/extralit-server/src/extralit_server/models/__init__.py
index 55780cdd5..5d092989b 100644
--- a/extralit-server/src/extralit_server/models/__init__.py
+++ b/extralit-server/src/extralit_server/models/__init__.py
@@ -5,3 +5,4 @@
 
 from .database import *
 from .metadata_properties import *
+from .v2 import Schema, SchemaVersion  # register v2 tables on DatabaseModel.metadata
diff --git a/extralit-server/src/extralit_server/models/v2/__init__.py b/extralit-server/src/extralit_server/models/v2/__init__.py
new file mode 100644
index 000000000..62fca1211
--- /dev/null
+++ b/extralit-server/src/extralit_server/models/v2/__init__.py
@@ -0,0 +1,3 @@
+from extralit_server.models.v2.schemas import Schema, SchemaVersion
+
+__all__ = ["Schema", "SchemaVersion"]
diff --git a/extralit-server/src/extralit_server/models/v2/schemas.py b/extralit-server/src/extralit_server/models/v2/schemas.py
new file mode 100644
index 000000000..9378a50fb
--- /dev/null
+++ b/extralit-server/src/extralit_server/models/v2/schemas.py
@@ -0,0 +1,70 @@
+from typing import TYPE_CHECKING
+from uuid import UUID
+
+from sqlalchemy import JSON, ForeignKey, String, Text, UniqueConstraint
+from sqlalchemy import Enum as SAEnum
+from sqlalchemy.ext.mutable import MutableDict, MutableList
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from extralit_server.enums import SchemaKind, SchemaStatus
+from extralit_server.models.base import DatabaseModel
+
+if TYPE_CHECKING:
+    from extralit_server.models.database import Workspace
+
+SchemaKindEnum = SAEnum(SchemaKind, name="schema_kind_enum")
+SchemaStatusEnum = SAEnum(SchemaStatus, name="schema_status_enum")
+
+
+class Schema(DatabaseModel):
+    __tablename__ = "schemas"
+
+    name: Mapped[str] = mapped_column(String, index=True)
+    kind: Mapped[SchemaKind] = mapped_column(SchemaKindEnum, default=SchemaKind.table)
+    status: Mapped[SchemaStatus] = mapped_column(SchemaStatusEnum, default=SchemaStatus.draft, index=True)
+    current_version_id: Mapped[UUID | None] = mapped_column(
+        ForeignKey("schema_versions.id", ondelete="SET NULL", use_alter=True), nullable=True
+    )
+    settings: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSON), default=dict)
+    workspace_id: Mapped[UUID] = mapped_column(ForeignKey("workspaces.id", ondelete="CASCADE"), index=True)
+
+    # One-directional: no reverse `Workspace.schemas` collection in this phase.
+    workspace: Mapped["Workspace"] = relationship("Workspace")
+
+    versions: Mapped[list["SchemaVersion"]] = relationship(
+        back_populates="schema",
+        order_by="SchemaVersion.version",
+        cascade="all, delete-orphan",
+        foreign_keys="SchemaVersion.schema_id",
+    )
+
+    __table_args__ = (UniqueConstraint("workspace_id", "name", name="schema_workspace_id_name_uq"),)
+
+    def __repr__(self) -> str:
+        return f"Schema(id={self.id!s}, name={self.name!r}, kind={self.kind!r}, status={self.status!r})"
+
+
+class SchemaVersion(DatabaseModel):
+    __tablename__ = "schema_versions"
+
+    schema_id: Mapped[UUID] = mapped_column(ForeignKey("schemas.id", ondelete="CASCADE"), index=True)
+    version: Mapped[int] = mapped_column(index=True)
+    object_key: Mapped[str] = mapped_column(Text)
+    object_version_id: Mapped[str | None] = mapped_column(Text, nullable=True)
+    etag: Mapped[str] = mapped_column(String)
+    checksum: Mapped[str] = mapped_column(String)
+    parent_version_id: Mapped[UUID | None] = mapped_column(
+        ForeignKey("schema_versions.id", ondelete="SET NULL"), nullable=True
+    )
+    columns_cache: Mapped[list] = mapped_column(MutableList.as_mutable(JSON), default=list)
+    # Out-of-band per-column review widgets (column name -> widget config); see spec §13.
+    # Pandera's to_json drops Column.metadata, so this overlay is the source for columns_cache.review.
+    review_widgets: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSON), default=dict)
+    created_by: Mapped[UUID | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
+
+    schema: Mapped["Schema"] = relationship(back_populates="versions", foreign_keys=[schema_id])
+
+    __table_args__ = (UniqueConstraint("schema_id", "version", name="schema_version_schema_id_version_uq"),)
+
+    def __repr__(self) -> str:
+        return f"SchemaVersion(id={self.id!s}, schema_id={self.schema_id!s}, version={self.version!r})"
diff --git a/extralit-server/tests/factories.py b/extralit-server/tests/factories.py
index 76c3317a5..d8452802c 100644
--- a/extralit-server/tests/factories.py
+++ b/extralit-server/tests/factories.py
@@ -9,7 +9,14 @@
 from sqlalchemy.ext.asyncio import async_object_session
 
 from extralit_server.contexts.files import ObjectMetadata, get_s3_client
-from extralit_server.enums import DatasetDistributionStrategy, FieldType, MetadataPropertyType, OptionsOrder
+from extralit_server.enums import (
+    DatasetDistributionStrategy,
+    FieldType,
+    MetadataPropertyType,
+    OptionsOrder,
+    SchemaKind,
+    SchemaStatus,
+)
 from extralit_server.models import (
     Dataset,
     DatasetUser,
@@ -31,6 +38,8 @@
     WorkspaceUser,
 )
 from extralit_server.models.base import DatabaseModel
+from extralit_server.models.v2 import Schema as SchemaModel
+from extralit_server.models.v2 import SchemaVersion as SchemaVersionModel
 from extralit_server.webhooks.v1.enums import WebhookEvent
 from tests.database import SyncTestSession, TestSession
 
@@ -637,3 +646,28 @@ def mock_get_object(bucket_name, object_name, version_id=None):
         client.get_object = mock_get_object
 
         return file
+
+
+class SchemaFactory(BaseFactory):
+    class Meta:
+        model = SchemaModel
+
+    name = factory.Sequence(lambda n: f"schema-{n}")
+    kind = SchemaKind.table
+    status = SchemaStatus.draft
+    workspace = factory.SubFactory(WorkspaceFactory)
+
+
+class SchemaVersionFactory(BaseFactory):
+    class Meta:
+        model = SchemaVersionModel
+
+    schema = factory.SubFactory(SchemaFactory)
+    version = factory.Sequence(lambda n: n + 1)
+    # The SubFactory result is a coroutine during attribute evaluation, so the object key
+    # is derived from `version` only here; the real `schemas/{id}/v{n}.json` key is computed
+    # by publish_version (the context owns persistence, not the factory).
+    object_key = factory.LazyAttribute(lambda o: f"schemas/v{o.version}.json")
+    etag = factory.Sequence(lambda n: f"etag-{n}")
+    checksum = factory.Sequence(lambda n: f"checksum-{n}")
+    columns_cache = factory.LazyFunction(list)  # fresh list per row, not a shared mutable default
diff --git a/extralit-server/tests/integration/__init__.py b/extralit-server/tests/integration/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/api/__init__.py b/extralit-server/tests/integration/api/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/api/schemas/__init__.py b/extralit-server/tests/integration/api/schemas/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/api/schemas/v2/__init__.py b/extralit-server/tests/integration/api/schemas/v2/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/api/schemas/v2/test_schema_models.py b/extralit-server/tests/integration/api/schemas/v2/test_schema_models.py
new file mode 100644
index 000000000..8440edf68
--- /dev/null
+++ b/extralit-server/tests/integration/api/schemas/v2/test_schema_models.py
@@ -0,0 +1,19 @@
+from uuid import uuid4
+
+from extralit_server.api.schemas.v2.schemas import SchemaCreate, SchemaVersionCreate
+from extralit_server.enums import SchemaKind
+
+
+def test_schema_create_defaults_settings_to_empty_dict():
+    payload = SchemaCreate(name="population", kind=SchemaKind.table, workspace_id=uuid4())
+    assert payload.settings == {}
+
+
+def test_schema_version_create_requires_body():
+    v = SchemaVersionCreate(body='{"columns": {}}')
+    assert v.body.startswith("{")
+
+
+def test_schema_version_create_defaults_review_widgets_to_empty_dict():
+    v = SchemaVersionCreate(body='{"columns": {}}')
+    assert v.review_widgets == {}
diff --git a/extralit-server/tests/integration/api/v2/__init__.py b/extralit-server/tests/integration/api/v2/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/api/v2/test_schemas.py b/extralit-server/tests/integration/api/v2/test_schemas.py
new file mode 100644
index 000000000..f85a1f253
--- /dev/null
+++ b/extralit-server/tests/integration/api/v2/test_schemas.py
@@ -0,0 +1,84 @@
+import pandera.pandas as pa
+import pytest
+
+from extralit_server.enums import SchemaKind
+from tests.factories import WorkspaceFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+def _body() -> str:
+    return pa.DataFrameSchema(columns={"name": pa.Column(pa.String, nullable=False)}).to_json()
+
+
+async def test_create_get_list_schema(async_client, owner_auth_header):
+    ws = await WorkspaceFactory.create()
+    resp = await async_client.post(
+        "/api/v2/schemas",
+        headers=owner_auth_header,
+        json={"name": "population", "kind": SchemaKind.table.value, "workspace_id": str(ws.id)},
+    )
+    assert resp.status_code == 201, resp.text
+    schema_id = resp.json()["id"]
+
+    resp = await async_client.get(f"/api/v2/schemas/{schema_id}", headers=owner_auth_header)
+    assert resp.status_code == 200
+    assert resp.json()["name"] == "population"
+
+    resp = await async_client.get(f"/api/v2/schemas?workspace_id={ws.id}", headers=owner_auth_header)
+    assert resp.status_code == 200
+    assert [s["id"] for s in resp.json()["items"]] == [schema_id]
+
+
+async def test_publish_version_and_columns(async_client, owner_auth_header, monkeypatch):
+    from datetime import datetime
+    from unittest.mock import AsyncMock
+
+    from extralit_server.contexts.files import ObjectMetadata
+
+    monkeypatch.setattr(
+        "extralit_server.contexts.v2.schemas.files_ctx.put_object",
+        AsyncMock(
+            return_value=ObjectMetadata(
+                bucket_name="b",
+                object_name="k",
+                etag="etag-1",
+                size=1,
+                last_modified=datetime(2026, 1, 1),
+                content_type="application/json",
+                version_id="ver-1",
+                metadata={},
+            )
+        ),
+    )
+
+    ws = await WorkspaceFactory.create()
+    resp = await async_client.post(
+        "/api/v2/schemas",
+        headers=owner_auth_header,
+        json={"name": "outcomes", "kind": SchemaKind.table.value, "workspace_id": str(ws.id)},
+    )
+    schema_id = resp.json()["id"]
+
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema_id}/versions",
+        headers=owner_auth_header,
+        json={"body": _body()},
+    )
+    assert resp.status_code == 201, resp.text
+    assert resp.json()["version"] == 1
+
+    resp = await async_client.get(f"/api/v2/schemas/{schema_id}/columns", headers=owner_auth_header)
+    assert resp.status_code == 200
+    assert any(c["name"] == "name" for c in resp.json())
+
+
+async def test_non_member_cannot_create_or_read_schema(async_client, annotator_auth_header):
+    # The annotator behind annotator_auth_header is NOT a member of this workspace.
+    ws = await WorkspaceFactory.create()
+    resp = await async_client.post(
+        "/api/v2/schemas",
+        headers=annotator_auth_header,
+        json={"name": "secret", "kind": SchemaKind.table.value, "workspace_id": str(ws.id)},
+    )
+    assert resp.status_code == 403, resp.text
diff --git a/extralit-server/tests/integration/conftest.py b/extralit-server/tests/integration/conftest.py
new file mode 100644
index 000000000..dd1d44889
--- /dev/null
+++ b/extralit-server/tests/integration/conftest.py
@@ -0,0 +1,62 @@
+"""Fixtures for the isolated v2 (`/api/v2`) test suite.
+
+These mirror the v1 fixtures in `tests/unit/conftest.py` but deliberately omit the
+session-scoped OpenSearch fixture (v2 does not use Elasticsearch/OpenSearch) and wire
+the test database + a mocked S3 client onto the separately-mounted `api_v2` sub-app.
+"""
+
+from collections.abc import AsyncGenerator
+from unittest.mock import AsyncMock
+
+import pytest
+import pytest_asyncio
+from httpx import AsyncClient
+
+from extralit_server.constants import API_KEY_HEADER_NAME
+from extralit_server.contexts import files as files_ctx
+from extralit_server.database import get_async_db
+from extralit_server.models import User
+from tests.database import TestSession
+from tests.factories import AnnotatorFactory, OwnerFactory
+
+
+@pytest_asyncio.fixture
+async def owner() -> User:
+    return await OwnerFactory.create(first_name="Owner", username="owner", api_key="owner.apikey")
+
+
+@pytest_asyncio.fixture
+async def annotator() -> User:
+    return await AnnotatorFactory.create(first_name="Annotator", username="annotator", api_key="annotator.apikey")
+
+
+@pytest.fixture
+def owner_auth_header(owner: User) -> dict[str, str]:
+    return {API_KEY_HEADER_NAME: owner.api_key}
+
+
+@pytest.fixture
+def annotator_auth_header(annotator: User) -> dict[str, str]:
+    return {API_KEY_HEADER_NAME: annotator.api_key}
+
+
+@pytest_asyncio.fixture
+async def async_client() -> AsyncGenerator[AsyncClient, None]:
+    from extralit_server import app
+    from extralit_server.api.v2 import api_v2
+
+    async def override_get_async_db():
+        yield TestSession()
+
+    async def override_get_s3_client():
+        # publish_version uploads via contexts.files.put_object, which tests monkeypatch; the
+        # yielded client is never used for real I/O, so a bare AsyncMock is sufficient.
+        yield AsyncMock()
+
+    api_v2.dependency_overrides[get_async_db] = override_get_async_db
+    api_v2.dependency_overrides[files_ctx.get_s3_client] = override_get_s3_client
+
+    async with AsyncClient(app=app, base_url="http://testserver") as client:
+        yield client
+
+    api_v2.dependency_overrides.clear()
diff --git a/extralit-server/tests/integration/contexts/__init__.py b/extralit-server/tests/integration/contexts/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/contexts/v2/__init__.py b/extralit-server/tests/integration/contexts/v2/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/contexts/v2/test_schema_bodies.py b/extralit-server/tests/integration/contexts/v2/test_schema_bodies.py
new file mode 100644
index 000000000..7fe6676ab
--- /dev/null
+++ b/extralit-server/tests/integration/contexts/v2/test_schema_bodies.py
@@ -0,0 +1,105 @@
+import json
+
+import pandera as pa
+import pytest
+
+from extralit_server.contexts.v2.schema_bodies import (
+    SchemaValidationError,
+    derive_columns_cache,
+    validate_record_fields,
+)
+
+
+def _body() -> str:
+    schema = pa.DataFrameSchema(
+        columns={
+            "name": pa.Column(pa.String, nullable=False),
+            "age": pa.Column(pa.Int, nullable=True),
+        }
+    )
+    return schema.to_json()
+
+
+def test_derive_columns_cache_lists_columns_with_dtype_and_nullable():
+    cache = derive_columns_cache(_body())
+    by_name = {c["name"]: c for c in cache}
+    assert set(by_name) == {"name", "age"}
+    assert by_name["name"]["nullable"] is False
+    assert by_name["age"]["nullable"] is True
+    assert "int" in by_name["age"]["dtype"].lower()
+
+
+def test_derive_columns_cache_defaults_review_to_none():
+    # Pandera 0.32 drops per-Column.metadata through to_json/from_json, so the body alone
+    # carries no review widget — `review` is None unless supplied via the side map.
+    cache = derive_columns_cache(_body())
+    assert all(c["review"] is None for c in cache)
+
+
+def test_derive_columns_cache_merges_review_widgets_side_map():
+    # The review widget is carried out-of-band (see spec §13) and merged per column name.
+    cache = derive_columns_cache(_body(), review_widgets={"age": {"type": "rating"}})
+    by_name = {c["name"]: c for c in cache}
+    assert by_name["age"]["review"] == {"type": "rating"}
+    assert by_name["name"]["review"] is None
+
+
+def test_validate_record_fields_returns_native_json_types():
+    coerced = validate_record_fields(_body(), {"name": "Ada", "age": 36})
+    assert coerced["name"] == "Ada"
+    assert coerced["age"] == 36
+    # Must be native python types (not numpy scalars) and JSON-serializable for the
+    # record.fields JSONB column in Phase 2.
+    assert type(coerced["age"]) is int
+    json.dumps(coerced)  # raises if numpy scalars / NaN leaked through
+
+
+def test_validate_record_fields_converts_nulls_to_none():
+    coerced = validate_record_fields(_body(), {"name": "Ada", "age": None})
+    assert coerced["age"] is None
+    json.dumps(coerced)
+
+
+def test_validate_record_fields_raises_on_type_error():
+    with pytest.raises(SchemaValidationError) as exc:
+        validate_record_fields(_body(), {"name": "Ada", "age": "not-a-number"})
+    assert isinstance(exc.value.errors, list)
+    assert len(exc.value.errors) >= 1
+
+
+def test_validate_record_fields_preserves_high_precision_float():
+    body = pa.DataFrameSchema(columns={"ratio": pa.Column(pa.Float, nullable=False)}).to_json()
+    coerced = validate_record_fields(body, {"ratio": 1.123456789012345})
+    # The lossy to_json detour truncated to 10 decimals; native conversion must not.
+    assert coerced["ratio"] == 1.123456789012345
+    json.dumps(coerced)
+
+
+def test_validate_record_fields_serializes_datetime_as_iso_string():
+    body = pa.DataFrameSchema(columns={"observed_at": pa.Column("datetime64[ns]", nullable=False)}).to_json()
+    coerced = validate_record_fields(body, {"observed_at": "2024-01-02T03:04:05"})
+    assert coerced["observed_at"].startswith("2024-01-02T03:04:05")
+    json.dumps(coerced)  # ISO string is JSON-serializable (epoch-int default would also be, but wrong)
+
+
+def test_validate_record_fields_rejects_missing_required_column():
+    # `name` is non-nullable; omitting it entirely (not just null) must be rejected.
+    with pytest.raises(SchemaValidationError) as exc:
+        validate_record_fields(_body(), {"age": 5})
+    assert any(e["check"] == "missing" and e["column"] == "name" for e in exc.value.errors)
+
+
+def test_validate_record_fields_preserves_int_in_all_numeric_schema():
+    # No string/object column, so a row Series would upcast Int->float64. Per-column
+    # extraction must keep the int64 cell as a python int, not 3.0.
+    body = pa.DataFrameSchema(
+        columns={
+            "count": pa.Column(pa.Int, nullable=False),
+            "ratio": pa.Column(pa.Float, nullable=False),
+        }
+    ).to_json()
+    coerced = validate_record_fields(body, {"count": 3, "ratio": 0.5})
+    assert type(coerced["count"]) is int
+    assert coerced["count"] == 3
+    assert coerced["ratio"] == 0.5
+    json.dumps(coerced)
diff --git a/extralit-server/tests/integration/contexts/v2/test_schemas_context.py b/extralit-server/tests/integration/contexts/v2/test_schemas_context.py
new file mode 100644
index 000000000..de6748f1a
--- /dev/null
+++ b/extralit-server/tests/integration/contexts/v2/test_schemas_context.py
@@ -0,0 +1,82 @@
+from datetime import datetime
+from unittest.mock import AsyncMock
+
+import pandera.pandas as pa
+import pytest
+
+from extralit_server.contexts.files import ObjectMetadata
+from extralit_server.contexts.v2 import schemas as schemas_ctx
+from extralit_server.enums import SchemaKind, SchemaStatus
+from extralit_server.models.v2 import Schema, SchemaVersion
+from tests.factories import WorkspaceFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+def _body() -> str:
+    return pa.DataFrameSchema(columns={"name": pa.Column(pa.String, nullable=False)}).to_json()
+
+
+def _patch_put_object(monkeypatch, bucket: str) -> AsyncMock:
+    put = AsyncMock(
+        return_value=ObjectMetadata(
+            bucket_name=bucket,
+            object_name="k",
+            etag="etag-1",
+            size=1,
+            last_modified=datetime(2026, 1, 1),
+            content_type="application/json",
+            version_id="ver-1",
+            metadata={},
+        )
+    )
+    monkeypatch.setattr("extralit_server.contexts.v2.schemas.files_ctx.put_object", put)
+    return put
+
+
+async def test_create_and_list_schema(db):
+    ws = await WorkspaceFactory.create()
+    schema = await schemas_ctx.create_schema(db, name="population", kind=SchemaKind.table, workspace_id=ws.id)
+    assert isinstance(schema, Schema)
+
+    listed = await schemas_ctx.list_schemas(db, workspace_id=ws.id)
+    assert [s.id for s in listed] == [schema.id]
+
+
+async def test_publish_version_uploads_body_and_advances_pointer(db, monkeypatch):
+    ws = await WorkspaceFactory.create()
+    _patch_put_object(monkeypatch, ws.name)
+    schema = await schemas_ctx.create_schema(db, name="population", kind=SchemaKind.table, workspace_id=ws.id)
+
+    s3 = AsyncMock()
+    version = await schemas_ctx.publish_version(db, s3, schema, body=_body(), bucket=ws.name, created_by=None)
+
+    assert isinstance(version, SchemaVersion)
+    assert version.version == 1
+    assert version.object_key == f"schemas/{schema.id}/v1.json"
+    # The S3 object version returned by put_object is pinned on the row.
+    assert version.object_version_id == "ver-1"
+    assert any(c["name"] == "name" for c in version.columns_cache)
+
+    refreshed = await Schema.get(db, schema.id)
+    assert refreshed.current_version_id == version.id
+    assert refreshed.status == SchemaStatus.published
+
+    # Second publish increments version and links lineage
+    v2 = await schemas_ctx.publish_version(db, s3, refreshed, body=_body(), bucket=ws.name)
+    assert v2.version == 2
+    assert v2.parent_version_id == version.id
+
+
+async def test_publish_version_merges_review_widgets_into_columns_cache(db, monkeypatch):
+    ws = await WorkspaceFactory.create()
+    _patch_put_object(monkeypatch, ws.name)
+    schema = await schemas_ctx.create_schema(db, name="ratings", kind=SchemaKind.table, workspace_id=ws.id)
+
+    s3 = AsyncMock()
+    version = await schemas_ctx.publish_version(
+        db, s3, schema, body=_body(), bucket=ws.name, review_widgets={"name": {"type": "text"}}
+    )
+    by_name = {c["name"]: c for c in version.columns_cache}
+    assert by_name["name"]["review"] == {"type": "text"}
+    assert version.review_widgets == {"name": {"type": "text"}}
diff --git a/extralit-server/tests/integration/models/__init__.py b/extralit-server/tests/integration/models/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/models/v2/__init__.py b/extralit-server/tests/integration/models/v2/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/models/v2/test_schema_factories.py b/extralit-server/tests/integration/models/v2/test_schema_factories.py
new file mode 100644
index 000000000..2a9befc92
--- /dev/null
+++ b/extralit-server/tests/integration/models/v2/test_schema_factories.py
@@ -0,0 +1,19 @@
+import pytest
+
+from extralit_server.models.v2 import Schema, SchemaVersion
+from tests.factories import SchemaFactory, SchemaVersionFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_schema_factory_creates_row(db):
+    schema = await SchemaFactory.create(name="outcomes")
+    assert isinstance(schema, Schema)
+    assert schema.workspace_id is not None
+
+
+async def test_schema_version_factory_links_schema(db):
+    version = await SchemaVersionFactory.create()
+    assert isinstance(version, SchemaVersion)
+    assert version.schema_id is not None
+    assert version.version >= 1
diff --git a/extralit-server/tests/integration/models/v2/test_schema_models.py b/extralit-server/tests/integration/models/v2/test_schema_models.py
new file mode 100644
index 000000000..6f736b647
--- /dev/null
+++ b/extralit-server/tests/integration/models/v2/test_schema_models.py
@@ -0,0 +1,75 @@
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from extralit_server.enums import SchemaKind, SchemaStatus
+from extralit_server.models.v2 import Schema, SchemaVersion
+from tests.factories import SchemaFactory, WorkspaceFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_create_schema_and_version(db):
+    workspace = await WorkspaceFactory.create()
+    schema = await Schema.create(
+        db,
+        name="population",
+        kind=SchemaKind.table,
+        status=SchemaStatus.draft,
+        workspace_id=workspace.id,
+    )
+    version = await SchemaVersion.create(
+        db,
+        schema_id=schema.id,
+        version=1,
+        object_key=f"schemas/{schema.id}/v1.json",
+        etag="abc123",
+        checksum="def456",
+        columns_cache=[{"name": "n", "dtype": "str", "nullable": False, "review": None}],
+    )
+    assert schema.id is not None
+    assert version.schema_id == schema.id
+    assert version.version == 1
+
+    loaded = await Schema.get(db, schema.id)
+    assert loaded.name == "population"
+    assert loaded.kind == SchemaKind.table
+
+
+async def test_schema_and_version_apply_column_defaults(db):
+    workspace = await WorkspaceFactory.create()
+    # Schema created with only the required fields applies kind/status/settings defaults.
+    schema = await Schema.create(db, name="defaults", workspace_id=workspace.id)
+    assert schema.kind == SchemaKind.table
+    assert schema.status == SchemaStatus.draft
+    assert schema.settings == {}
+    assert schema.current_version_id is None
+
+    # SchemaVersion created without columns_cache/review_widgets applies []/{} defaults.
+    version = await SchemaVersion.create(db, schema_id=schema.id, version=1, object_key="k", etag="e", checksum="c")
+    assert version.columns_cache == []
+    assert version.review_widgets == {}
+    assert version.parent_version_id is None
+
+
+async def test_duplicate_schema_name_in_workspace_raises(db):
+    workspace = await WorkspaceFactory.create()
+    db.add_all(
+        [
+            Schema(name="dup", workspace_id=workspace.id),
+            Schema(name="dup", workspace_id=workspace.id),
+        ]
+    )
+    with pytest.raises(IntegrityError, match="schema_workspace_id_name_uq|UNIQUE"):
+        await db.commit()
+
+
+async def test_duplicate_schema_version_raises(db):
+    schema = await SchemaFactory.create()
+    db.add_all(
+        [
+            SchemaVersion(schema_id=schema.id, version=1, object_key="a", etag="e", checksum="c"),
+            SchemaVersion(schema_id=schema.id, version=1, object_key="b", etag="e", checksum="c"),
+        ]
+    )
+    with pytest.raises(IntegrityError, match="schema_version_schema_id_version_uq|UNIQUE"):
+        await db.commit()
diff --git a/extralit-server/tests/integration/test_enums_v2.py b/extralit-server/tests/integration/test_enums_v2.py
new file mode 100644
index 000000000..529bb2901
--- /dev/null
+++ b/extralit-server/tests/integration/test_enums_v2.py
@@ -0,0 +1,13 @@
+from extralit_server.enums import SchemaKind, SchemaStatus
+
+
+def test_schema_kind_values():
+    assert SchemaKind.singleton == "singleton"
+    assert SchemaKind.table == "table"
+    assert {k.value for k in SchemaKind} == {"singleton", "table"}
+
+
+def test_schema_status_values():
+    assert SchemaStatus.draft == "draft"
+    assert SchemaStatus.published == "published"
+    assert {s.value for s in SchemaStatus} == {"draft", "published"}
diff --git a/extralit-server/uv.lock b/extralit-server/uv.lock
index 0dd242c83..20ee48233 100644
--- a/extralit-server/uv.lock
+++ b/extralit-server/uv.lock
@@ -1180,6 +1180,7 @@ dependencies = [
     { name = "opencv-python-headless" },
     { name = "opensearch-py" },
     { name = "packaging" },
+    { name = "pandera", extra = ["io"] },
     { name = "pdf2image" },
     { name = "pillow" },
     { name = "psutil" },
@@ -1254,6 +1255,7 @@ requires-dist = [
     { name = "opencv-python-headless", specifier = ">=4.11.0.86" },
     { name = "opensearch-py", specifier = "~=2.0.0" },
     { name = "packaging", specifier = ">=23.2" },
+    { name = "pandera", extras = ["io"], specifier = ">=0.20" },
     { name = "pdf2image", specifier = ">=1.17.0" },
     { name = "pillow", specifier = ">=10.1.0" },
     { name = "psutil", specifier = "~=5.8,<5.10" },
@@ -2479,6 +2481,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/da/d9/f7f9379981e39b8c2511c9e0326d212accacb82f12fbfdc1aa2ce2a7b2b6/multiprocess-0.70.16-py39-none-any.whl", hash = "sha256:a0bafd3ae1b732eac64be2e72038231c1ba97724b60b09400d68f229fcc2fbf3", size = 133351, upload-time = "2024-01-28T18:52:31.981Z" },
 ]
 
+[[package]]
+name = "mypy-extensions"
+version = "1.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/a2/6e/371856a3fb9d31ca8dac321cda606860fa4548858c0cc45d9d1d4ca2628b/mypy_extensions-1.1.0.tar.gz", hash = "sha256:52e68efc3284861e772bbcd66823fde5ae21fd2fdb51c62a211403730b916558", size = 6343, upload-time = "2025-04-22T14:54:24.164Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/79/7b/2c79738432f5c924bef5071f933bcc9efd0473bac3b4aa584a6f7c1c8df8/mypy_extensions-1.1.0-py3-none-any.whl", hash = "sha256:1be4cccdb0f2482337c4743e60421de3a356cd97508abadd57d47403e94f5505", size = 4963, upload-time = "2025-04-22T14:54:22.983Z" },
+]
+
 [[package]]
 name = "networkx"
 version = "3.4.2"
@@ -3087,6 +3098,27 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/68/b0/34937815889fa982613775e4b97fddd13250f11012d769949c5465af2150/pandas-3.0.1-cp314-cp314t-win_arm64.whl", hash = "sha256:108dd1790337a494aa80e38def654ca3f0968cf4f362c85f44c15e471667102d", size = 9452085, upload-time = "2026-02-17T22:20:14.331Z" },
 ]
 
+[[package]]
+name = "pandera"
+version = "0.32.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "packaging" },
+    { name = "pydantic" },
+    { name = "typeguard" },
+    { name = "typing-extensions" },
+    { name = "typing-inspect" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/0f/e1/b605cc7b203deb015e3fbb3a9ef90cd83050af4c61941f14bed956f890ca/pandera-0.32.0.tar.gz", hash = "sha256:070c705eafbcfb4dd2ce494154d4bc0ce6d2b8e02a95ae15234040683bbab1f5", size = 867262, upload-time = "2026-06-19T02:01:07.22Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/76/ea/6e6af2a8118c51bf3f1a6958688b77b0a3a652242a0b383f1e3cdf6cd2b4/pandera-0.32.0-py3-none-any.whl", hash = "sha256:b8488704d80ed7fc0a812b0a8123e2028903e90e8ec5c31f880894c8e94786b3", size = 442527, upload-time = "2026-06-19T02:01:05.66Z" },
+]
+
+[package.optional-dependencies]
+io = [
+    { name = "pyyaml" },
+]
+
 [[package]]
 name = "pdf2image"
 version = "1.17.0"
@@ -5149,6 +5181,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e5/2e/300174fca375a27a7c28dd80e990d857d7b3e3b25980c65063f980aa2f17/ty-0.0.46-py3-none-win_arm64.whl", hash = "sha256:ebd320d82605079b901a095dc4711037a0c488b4ace79a602fef4df0d3f4cf74", size = 11439595, upload-time = "2026-06-09T03:28:15.355Z" },
 ]
 
+[[package]]
+name = "typeguard"
+version = "4.5.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/67/1c/dfba5c4633cafc4c701f237d2ba63b416805047fd6d96aab4cfc40969f98/typeguard-4.5.2.tar.gz", hash = "sha256:5a16dcac23502039299c97c8941651bc33d7ea8cc4b2f7d6bbb1b528f6eea423", size = 80240, upload-time = "2026-05-14T12:59:40.857Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/5b/29/74eeb4d3f3ae61ca096b018ad486b3b3c74b17bec09ab4edab721cbefec3/typeguard-4.5.2-py3-none-any.whl", hash = "sha256:fcf9de18bd945cdb4c7b996e12b4c51ce83f92f191314a6d7cf1739586ec98cf", size = 36748, upload-time = "2026-05-14T12:59:39.473Z" },
+]
+
 [[package]]
 name = "typer"
 version = "0.24.1"
@@ -5203,6 +5247,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/18/67/36e9267722cc04a6b9f15c7f3441c2363321a3ea07da7ae0c0707beb2a9c/typing_extensions-4.15.0-py3-none-any.whl", hash = "sha256:f0fa19c6845758ab08074a0cfa8b7aecb71c999ca73d62883bc25cc018c4e548", size = 44614, upload-time = "2025-08-25T13:49:24.86Z" },
 ]
 
+[[package]]
+name = "typing-inspect"
+version = "0.9.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "mypy-extensions" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/dc/74/1789779d91f1961fa9438e9a8710cdae6bd138c80d7303996933d117264a/typing_inspect-0.9.0.tar.gz", hash = "sha256:b23fc42ff6f6ef6954e4852c1fb512cdd18dbea03134f91f856a95ccc9461f78", size = 13825, upload-time = "2023-05-24T20:25:47.612Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/65/f3/107a22063bf27bdccf2024833d3445f4eea42b2e598abfbd46f6a63b6cb0/typing_inspect-0.9.0-py3-none-any.whl", hash = "sha256:9ee6fc59062311ef8547596ab6b955e1b8aa46242d854bfc78f4f6b0eff35f9f", size = 8827, upload-time = "2023-05-24T20:25:45.287Z" },
+]
+
 [[package]]
 name = "tzdata"
 version = "2025.3"