diff --git a/backend/main.py b/backend/main.py
index bad1de2..95a7af8 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -12,7 +12,7 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from config import get_settings
-from routers import creators, health, ingest, pipeline, videos
+from routers import creators, health, ingest, pipeline, review, videos
 
 
 def _setup_logging() -> None:
@@ -81,6 +81,7 @@ app.include_router(health.router)
 app.include_router(creators.router, prefix="/api/v1")
 app.include_router(ingest.router, prefix="/api/v1")
 app.include_router(pipeline.router, prefix="/api/v1")
+app.include_router(review.router, prefix="/api/v1")
 app.include_router(videos.router, prefix="/api/v1")
 
 
diff --git a/backend/redis_client.py b/backend/redis_client.py
new file mode 100644
index 0000000..7a32103
--- /dev/null
+++ b/backend/redis_client.py
@@ -0,0 +1,15 @@
+"""Async Redis client helper for Chrysopedia."""
+
+import redis.asyncio as aioredis
+
+from config import get_settings
+
+
+async def get_redis() -> aioredis.Redis:
+    """Return an async Redis client from the configured URL.
+
+    Callers should close the connection when done, or use it
+    as a short-lived client within a request handler.
+    """
+    settings = get_settings()
+    return aioredis.from_url(settings.redis_url, decode_responses=True)
diff --git a/backend/routers/review.py b/backend/routers/review.py
new file mode 100644
index 0000000..3e32d64
--- /dev/null
+++ b/backend/routers/review.py
@@ -0,0 +1,354 @@
+"""Review queue endpoints for Chrysopedia API.
+
+Provides admin review workflow: list queue, stats, approve, reject,
+edit, split, merge key moments, and toggle review/auto mode via Redis.
+"""
+
+import logging
+import uuid
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import case, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from config import get_settings
+from database import get_session
+from models import Creator, KeyMoment, KeyMomentContentType, ReviewStatus, SourceVideo
+from redis_client import get_redis
+from schemas import (
+    KeyMomentRead,
+    MomentEditRequest,
+    MomentMergeRequest,
+    MomentSplitRequest,
+    ReviewModeResponse,
+    ReviewModeUpdate,
+    ReviewQueueItem,
+    ReviewQueueResponse,
+    ReviewStatsResponse,
+)
+
+logger = logging.getLogger("chrysopedia.review")
+
+router = APIRouter(prefix="/review", tags=["review"])
+
+REDIS_MODE_KEY = "chrysopedia:review_mode"
+
+VALID_STATUSES = {"pending", "approved", "edited", "rejected", "all"}
+
+
+# ── Helpers ──────────────────────────────────────────────────────────────────
+
+
+def _moment_to_queue_item(
+    moment: KeyMoment, video_filename: str, creator_name: str
+) -> ReviewQueueItem:
+    """Convert a KeyMoment ORM instance + joined fields to a ReviewQueueItem."""
+    data = KeyMomentRead.model_validate(moment).model_dump()
+    data["video_filename"] = video_filename
+    data["creator_name"] = creator_name
+    return ReviewQueueItem(**data)
+
+
+# ── Endpoints ────────────────────────────────────────────────────────────────
+
+
+@router.get("/queue", response_model=ReviewQueueResponse)
+async def list_queue(
+    status: Annotated[str, Query()] = "pending",
+    offset: Annotated[int, Query(ge=0)] = 0,
+    limit: Annotated[int, Query(ge=1, le=100)] = 50,
+    db: AsyncSession = Depends(get_session),
+) -> ReviewQueueResponse:
+    """List key moments in the review queue, filtered by status."""
+    if status not in VALID_STATUSES:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid status filter '{status}'. Must be one of: {', '.join(sorted(VALID_STATUSES))}",
+        )
+
+    # Base query joining KeyMoment → SourceVideo → Creator
+    base = (
+        select(
+            KeyMoment,
+            SourceVideo.filename.label("video_filename"),
+            Creator.name.label("creator_name"),
+        )
+        .join(SourceVideo, KeyMoment.source_video_id == SourceVideo.id)
+        .join(Creator, SourceVideo.creator_id == Creator.id)
+    )
+
+    if status != "all":
+        base = base.where(KeyMoment.review_status == ReviewStatus(status))
+
+    # Count total matching rows
+    count_stmt = select(func.count()).select_from(base.subquery())
+    total = (await db.execute(count_stmt)).scalar_one()
+
+    # Fetch paginated results
+    stmt = base.order_by(KeyMoment.created_at.desc()).offset(offset).limit(limit)
+    rows = (await db.execute(stmt)).all()
+
+    items = [
+        _moment_to_queue_item(row.KeyMoment, row.video_filename, row.creator_name)
+        for row in rows
+    ]
+
+    return ReviewQueueResponse(items=items, total=total, offset=offset, limit=limit)
+
+
+@router.get("/stats", response_model=ReviewStatsResponse)
+async def get_stats(
+    db: AsyncSession = Depends(get_session),
+) -> ReviewStatsResponse:
+    """Return counts of key moments grouped by review status."""
+    stmt = (
+        select(
+            KeyMoment.review_status,
+            func.count().label("cnt"),
+        )
+        .group_by(KeyMoment.review_status)
+    )
+    result = await db.execute(stmt)
+    counts = {row.review_status.value: row.cnt for row in result.all()}
+
+    return ReviewStatsResponse(
+        pending=counts.get("pending", 0),
+        approved=counts.get("approved", 0),
+        edited=counts.get("edited", 0),
+        rejected=counts.get("rejected", 0),
+    )
+
+
+@router.post("/moments/{moment_id}/approve", response_model=KeyMomentRead)
+async def approve_moment(
+    moment_id: uuid.UUID,
+    db: AsyncSession = Depends(get_session),
+) -> KeyMomentRead:
+    """Approve a key moment for publishing."""
+    moment = await db.get(KeyMoment, moment_id)
+    if moment is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Key moment {moment_id} not found",
+        )
+
+    moment.review_status = ReviewStatus.approved
+    await db.commit()
+    await db.refresh(moment)
+
+    logger.info("Approved key moment %s", moment_id)
+    return KeyMomentRead.model_validate(moment)
+
+
+@router.post("/moments/{moment_id}/reject", response_model=KeyMomentRead)
+async def reject_moment(
+    moment_id: uuid.UUID,
+    db: AsyncSession = Depends(get_session),
+) -> KeyMomentRead:
+    """Reject a key moment."""
+    moment = await db.get(KeyMoment, moment_id)
+    if moment is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Key moment {moment_id} not found",
+        )
+
+    moment.review_status = ReviewStatus.rejected
+    await db.commit()
+    await db.refresh(moment)
+
+    logger.info("Rejected key moment %s", moment_id)
+    return KeyMomentRead.model_validate(moment)
+
+
+@router.put("/moments/{moment_id}", response_model=KeyMomentRead)
+async def edit_moment(
+    moment_id: uuid.UUID,
+    body: MomentEditRequest,
+    db: AsyncSession = Depends(get_session),
+) -> KeyMomentRead:
+    """Update editable fields of a key moment and set status to edited."""
+    moment = await db.get(KeyMoment, moment_id)
+    if moment is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Key moment {moment_id} not found",
+        )
+
+    update_data = body.model_dump(exclude_unset=True)
+    # Convert content_type string to enum if provided
+    if "content_type" in update_data and update_data["content_type"] is not None:
+        try:
+            update_data["content_type"] = KeyMomentContentType(update_data["content_type"])
+        except ValueError:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Invalid content_type '{update_data['content_type']}'",
+            )
+
+    for field, value in update_data.items():
+        setattr(moment, field, value)
+
+    moment.review_status = ReviewStatus.edited
+    await db.commit()
+    await db.refresh(moment)
+
+    logger.info("Edited key moment %s (fields: %s)", moment_id, list(update_data.keys()))
+    return KeyMomentRead.model_validate(moment)
+
+
+@router.post("/moments/{moment_id}/split", response_model=list[KeyMomentRead])
+async def split_moment(
+    moment_id: uuid.UUID,
+    body: MomentSplitRequest,
+    db: AsyncSession = Depends(get_session),
+) -> list[KeyMomentRead]:
+    """Split a key moment into two at the given timestamp."""
+    moment = await db.get(KeyMoment, moment_id)
+    if moment is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Key moment {moment_id} not found",
+        )
+
+    # Validate split_time is strictly between start_time and end_time
+    if body.split_time <= moment.start_time or body.split_time >= moment.end_time:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"split_time ({body.split_time}) must be strictly between "
+                f"start_time ({moment.start_time}) and end_time ({moment.end_time})"
+            ),
+        )
+
+    # Update original moment to [start_time, split_time)
+    original_end = moment.end_time
+    moment.end_time = body.split_time
+    moment.review_status = ReviewStatus.pending
+
+    # Create new moment for [split_time, end_time]
+    new_moment = KeyMoment(
+        source_video_id=moment.source_video_id,
+        technique_page_id=moment.technique_page_id,
+        title=f"{moment.title} (split)",
+        summary=moment.summary,
+        start_time=body.split_time,
+        end_time=original_end,
+        content_type=moment.content_type,
+        plugins=moment.plugins,
+        review_status=ReviewStatus.pending,
+        raw_transcript=moment.raw_transcript,
+    )
+    db.add(new_moment)
+
+    await db.commit()
+    await db.refresh(moment)
+    await db.refresh(new_moment)
+
+    logger.info(
+        "Split key moment %s at %.2f → original [%.2f, %.2f), new [%.2f, %.2f]",
+        moment_id, body.split_time,
+        moment.start_time, moment.end_time,
+        new_moment.start_time, new_moment.end_time,
+    )
+
+    return [
+        KeyMomentRead.model_validate(moment),
+        KeyMomentRead.model_validate(new_moment),
+    ]
+
+
+@router.post("/moments/{moment_id}/merge", response_model=KeyMomentRead)
+async def merge_moments(
+    moment_id: uuid.UUID,
+    body: MomentMergeRequest,
+    db: AsyncSession = Depends(get_session),
+) -> KeyMomentRead:
+    """Merge two key moments into one."""
+    if moment_id == body.target_moment_id:
+        raise HTTPException(
+            status_code=400,
+            detail="Cannot merge a moment with itself",
+        )
+
+    source = await db.get(KeyMoment, moment_id)
+    if source is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Key moment {moment_id} not found",
+        )
+
+    target = await db.get(KeyMoment, body.target_moment_id)
+    if target is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Target key moment {body.target_moment_id} not found",
+        )
+
+    # Both must belong to the same source video
+    if source.source_video_id != target.source_video_id:
+        raise HTTPException(
+            status_code=400,
+            detail="Cannot merge moments from different source videos",
+        )
+
+    # Merge: combined summary, min start, max end
+    source.summary = f"{source.summary}\n\n{target.summary}"
+    source.start_time = min(source.start_time, target.start_time)
+    source.end_time = max(source.end_time, target.end_time)
+    source.review_status = ReviewStatus.pending
+
+    # Delete target
+    await db.delete(target)
+    await db.commit()
+    await db.refresh(source)
+
+    logger.info(
+        "Merged key moment %s with %s → [%.2f, %.2f]",
+        moment_id, body.target_moment_id,
+        source.start_time, source.end_time,
+    )
+
+    return KeyMomentRead.model_validate(source)
+
+
+@router.get("/mode", response_model=ReviewModeResponse)
+async def get_mode() -> ReviewModeResponse:
+    """Get the current review mode (review vs auto)."""
+    settings = get_settings()
+    try:
+        redis = await get_redis()
+        try:
+            value = await redis.get(REDIS_MODE_KEY)
+            if value is not None:
+                return ReviewModeResponse(review_mode=value.lower() == "true")
+        finally:
+            await redis.aclose()
+    except Exception as exc:
+        # Redis unavailable — fall back to config default
+        logger.warning("Redis unavailable for mode read, using config default: %s", exc)
+
+    return ReviewModeResponse(review_mode=settings.review_mode)
+
+
+@router.put("/mode", response_model=ReviewModeResponse)
+async def set_mode(
+    body: ReviewModeUpdate,
+) -> ReviewModeResponse:
+    """Set the review mode (review vs auto)."""
+    try:
+        redis = await get_redis()
+        try:
+            await redis.set(REDIS_MODE_KEY, str(body.review_mode))
+        finally:
+            await redis.aclose()
+    except Exception as exc:
+        logger.error("Failed to set review mode in Redis: %s", exc)
+        raise HTTPException(
+            status_code=503,
+            detail=f"Redis unavailable: {exc}",
+        )
+
+    logger.info("Review mode set to %s", body.review_mode)
+    return ReviewModeResponse(review_mode=body.review_mode)
diff --git a/backend/schemas.py b/backend/schemas.py
index ad8092f..9f8ba6e 100644
--- a/backend/schemas.py
+++ b/backend/schemas.py
@@ -194,3 +194,57 @@ class PaginatedResponse(BaseModel):
     total: int = 0
     offset: int = 0
     limit: int = 50
+
+
+# ── Review Queue ─────────────────────────────────────────────────────────────
+
+class ReviewQueueItem(KeyMomentRead):
+    """Key moment enriched with source video and creator info for review UI."""
+    video_filename: str
+    creator_name: str
+
+
+class ReviewQueueResponse(BaseModel):
+    """Paginated response for the review queue."""
+    items: list[ReviewQueueItem] = Field(default_factory=list)
+    total: int = 0
+    offset: int = 0
+    limit: int = 50
+
+
+class ReviewStatsResponse(BaseModel):
+    """Counts of key moments grouped by review status."""
+    pending: int = 0
+    approved: int = 0
+    edited: int = 0
+    rejected: int = 0
+
+
+class MomentEditRequest(BaseModel):
+    """Editable fields for a key moment."""
+    title: str | None = None
+    summary: str | None = None
+    start_time: float | None = None
+    end_time: float | None = None
+    content_type: str | None = None
+    plugins: list[str] | None = None
+
+
+class MomentSplitRequest(BaseModel):
+    """Request to split a moment at a given timestamp."""
+    split_time: float
+
+
+class MomentMergeRequest(BaseModel):
+    """Request to merge two moments."""
+    target_moment_id: uuid.UUID
+
+
+class ReviewModeResponse(BaseModel):
+    """Current review mode state."""
+    review_mode: bool
+
+
+class ReviewModeUpdate(BaseModel):
+    """Request to update the review mode."""
+    review_mode: bool
diff --git a/backend/tests/test_review.py b/backend/tests/test_review.py
new file mode 100644
index 0000000..f0b0d2a
--- /dev/null
+++ b/backend/tests/test_review.py
@@ -0,0 +1,495 @@
+"""Integration tests for the review queue endpoints.
+
+Tests run against a real PostgreSQL test database via httpx.AsyncClient.
+Redis is mocked for mode toggle tests.
+"""
+
+import uuid
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from models import (
+    ContentType,
+    Creator,
+    KeyMoment,
+    KeyMomentContentType,
+    ProcessingStatus,
+    ReviewStatus,
+    SourceVideo,
+)
+
+
+# ── Helpers ──────────────────────────────────────────────────────────────────
+
+QUEUE_URL = "/api/v1/review/queue"
+STATS_URL = "/api/v1/review/stats"
+MODE_URL = "/api/v1/review/mode"
+
+
+def _moment_url(moment_id: str, action: str = "") -> str:
+    """Build a moment action URL."""
+    base = f"/api/v1/review/moments/{moment_id}"
+    return f"{base}/{action}" if action else base
+
+
+async def _seed_creator_and_video(db_engine) -> dict:
+    """Seed a creator and source video, return their IDs."""
+    session_factory = async_sessionmaker(
+        db_engine, class_=AsyncSession, expire_on_commit=False
+    )
+    async with session_factory() as session:
+        creator = Creator(
+            name="TestCreator",
+            slug="test-creator",
+            folder_name="TestCreator",
+        )
+        session.add(creator)
+        await session.flush()
+
+        video = SourceVideo(
+            creator_id=creator.id,
+            filename="test-video.mp4",
+            file_path="TestCreator/test-video.mp4",
+            duration_seconds=600,
+            content_type=ContentType.tutorial,
+            processing_status=ProcessingStatus.extracted,
+        )
+        session.add(video)
+        await session.flush()
+
+        result = {
+            "creator_id": creator.id,
+            "creator_name": creator.name,
+            "video_id": video.id,
+            "video_filename": video.filename,
+        }
+        await session.commit()
+        return result
+
+
+async def _seed_moment(
+    db_engine,
+    video_id: uuid.UUID,
+    title: str = "Test Moment",
+    summary: str = "A test key moment",
+    start_time: float = 10.0,
+    end_time: float = 30.0,
+    review_status: ReviewStatus = ReviewStatus.pending,
+) -> uuid.UUID:
+    """Seed a single key moment and return its ID."""
+    session_factory = async_sessionmaker(
+        db_engine, class_=AsyncSession, expire_on_commit=False
+    )
+    async with session_factory() as session:
+        moment = KeyMoment(
+            source_video_id=video_id,
+            title=title,
+            summary=summary,
+            start_time=start_time,
+            end_time=end_time,
+            content_type=KeyMomentContentType.technique,
+            review_status=review_status,
+        )
+        session.add(moment)
+        await session.commit()
+        return moment.id
+
+
+async def _seed_second_video(db_engine, creator_id: uuid.UUID) -> uuid.UUID:
+    """Seed a second video for cross-video merge tests."""
+    session_factory = async_sessionmaker(
+        db_engine, class_=AsyncSession, expire_on_commit=False
+    )
+    async with session_factory() as session:
+        video = SourceVideo(
+            creator_id=creator_id,
+            filename="other-video.mp4",
+            file_path="TestCreator/other-video.mp4",
+            duration_seconds=300,
+            content_type=ContentType.tutorial,
+            processing_status=ProcessingStatus.extracted,
+        )
+        session.add(video)
+        await session.commit()
+        return video.id
+
+
+# ── Queue listing tests ─────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_list_queue_empty(client: AsyncClient):
+    """Queue returns empty list when no moments exist."""
+    resp = await client.get(QUEUE_URL)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["items"] == []
+    assert data["total"] == 0
+
+
+@pytest.mark.asyncio
+async def test_list_queue_with_moments(client: AsyncClient, db_engine):
+    """Queue returns moments enriched with video filename and creator name."""
+    seed = await _seed_creator_and_video(db_engine)
+    await _seed_moment(db_engine, seed["video_id"], title="EQ Basics")
+
+    resp = await client.get(QUEUE_URL)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["total"] == 1
+    item = data["items"][0]
+    assert item["title"] == "EQ Basics"
+    assert item["video_filename"] == seed["video_filename"]
+    assert item["creator_name"] == seed["creator_name"]
+    assert item["review_status"] == "pending"
+
+
+@pytest.mark.asyncio
+async def test_list_queue_filter_by_status(client: AsyncClient, db_engine):
+    """Queue filters correctly by status query parameter."""
+    seed = await _seed_creator_and_video(db_engine)
+    await _seed_moment(db_engine, seed["video_id"], title="Pending One")
+    await _seed_moment(
+        db_engine, seed["video_id"], title="Approved One",
+        review_status=ReviewStatus.approved,
+    )
+    await _seed_moment(
+        db_engine, seed["video_id"], title="Rejected One",
+        review_status=ReviewStatus.rejected,
+    )
+
+    # Default filter: pending
+    resp = await client.get(QUEUE_URL)
+    assert resp.json()["total"] == 1
+    assert resp.json()["items"][0]["title"] == "Pending One"
+
+    # Approved
+    resp = await client.get(QUEUE_URL, params={"status": "approved"})
+    assert resp.json()["total"] == 1
+    assert resp.json()["items"][0]["title"] == "Approved One"
+
+    # All
+    resp = await client.get(QUEUE_URL, params={"status": "all"})
+    assert resp.json()["total"] == 3
+
+
+# ── Stats tests ──────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_stats_counts(client: AsyncClient, db_engine):
+    """Stats returns correct counts per review status."""
+    seed = await _seed_creator_and_video(db_engine)
+    await _seed_moment(db_engine, seed["video_id"], review_status=ReviewStatus.pending)
+    await _seed_moment(db_engine, seed["video_id"], review_status=ReviewStatus.pending)
+    await _seed_moment(db_engine, seed["video_id"], review_status=ReviewStatus.approved)
+    await _seed_moment(db_engine, seed["video_id"], review_status=ReviewStatus.rejected)
+
+    resp = await client.get(STATS_URL)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["pending"] == 2
+    assert data["approved"] == 1
+    assert data["edited"] == 0
+    assert data["rejected"] == 1
+
+
+# ── Approve tests ────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_approve_moment(client: AsyncClient, db_engine):
+    """Approve sets review_status to approved."""
+    seed = await _seed_creator_and_video(db_engine)
+    moment_id = await _seed_moment(db_engine, seed["video_id"])
+
+    resp = await client.post(_moment_url(str(moment_id), "approve"))
+    assert resp.status_code == 200
+    assert resp.json()["review_status"] == "approved"
+
+
+@pytest.mark.asyncio
+async def test_approve_nonexistent_moment(client: AsyncClient):
+    """Approve returns 404 for nonexistent moment."""
+    fake_id = str(uuid.uuid4())
+    resp = await client.post(_moment_url(fake_id, "approve"))
+    assert resp.status_code == 404
+
+
+# ── Reject tests ─────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_reject_moment(client: AsyncClient, db_engine):
+    """Reject sets review_status to rejected."""
+    seed = await _seed_creator_and_video(db_engine)
+    moment_id = await _seed_moment(db_engine, seed["video_id"])
+
+    resp = await client.post(_moment_url(str(moment_id), "reject"))
+    assert resp.status_code == 200
+    assert resp.json()["review_status"] == "rejected"
+
+
+@pytest.mark.asyncio
+async def test_reject_nonexistent_moment(client: AsyncClient):
+    """Reject returns 404 for nonexistent moment."""
+    fake_id = str(uuid.uuid4())
+    resp = await client.post(_moment_url(fake_id, "reject"))
+    assert resp.status_code == 404
+
+
+# ── Edit tests ───────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_edit_moment(client: AsyncClient, db_engine):
+    """Edit updates fields and sets review_status to edited."""
+    seed = await _seed_creator_and_video(db_engine)
+    moment_id = await _seed_moment(db_engine, seed["video_id"], title="Original Title")
+
+    resp = await client.put(
+        _moment_url(str(moment_id)),
+        json={"title": "Updated Title", "summary": "New summary"},
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["title"] == "Updated Title"
+    assert data["summary"] == "New summary"
+    assert data["review_status"] == "edited"
+
+
+@pytest.mark.asyncio
+async def test_edit_nonexistent_moment(client: AsyncClient):
+    """Edit returns 404 for nonexistent moment."""
+    fake_id = str(uuid.uuid4())
+    resp = await client.put(
+        _moment_url(fake_id),
+        json={"title": "Won't Work"},
+    )
+    assert resp.status_code == 404
+
+
+# ── Split tests ──────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_split_moment(client: AsyncClient, db_engine):
+    """Split creates two moments with correct timestamps."""
+    seed = await _seed_creator_and_video(db_engine)
+    moment_id = await _seed_moment(
+        db_engine, seed["video_id"],
+        title="Full Moment", start_time=10.0, end_time=30.0,
+    )
+
+    resp = await client.post(
+        _moment_url(str(moment_id), "split"),
+        json={"split_time": 20.0},
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert len(data) == 2
+
+    # First (original): [10.0, 20.0)
+    assert data[0]["start_time"] == 10.0
+    assert data[0]["end_time"] == 20.0
+
+    # Second (new): [20.0, 30.0]
+    assert data[1]["start_time"] == 20.0
+    assert data[1]["end_time"] == 30.0
+    assert "(split)" in data[1]["title"]
+
+
+@pytest.mark.asyncio
+async def test_split_invalid_time_below_start(client: AsyncClient, db_engine):
+    """Split returns 400 when split_time is at or below start_time."""
+    seed = await _seed_creator_and_video(db_engine)
+    moment_id = await _seed_moment(
+        db_engine, seed["video_id"], start_time=10.0, end_time=30.0,
+    )
+
+    resp = await client.post(
+        _moment_url(str(moment_id), "split"),
+        json={"split_time": 10.0},
+    )
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_split_invalid_time_above_end(client: AsyncClient, db_engine):
+    """Split returns 400 when split_time is at or above end_time."""
+    seed = await _seed_creator_and_video(db_engine)
+    moment_id = await _seed_moment(
+        db_engine, seed["video_id"], start_time=10.0, end_time=30.0,
+    )
+
+    resp = await client.post(
+        _moment_url(str(moment_id), "split"),
+        json={"split_time": 30.0},
+    )
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_split_nonexistent_moment(client: AsyncClient):
+    """Split returns 404 for nonexistent moment."""
+    fake_id = str(uuid.uuid4())
+    resp = await client.post(
+        _moment_url(fake_id, "split"),
+        json={"split_time": 20.0},
+    )
+    assert resp.status_code == 404
+
+
+# ── Merge tests ──────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_merge_moments(client: AsyncClient, db_engine):
+    """Merge combines two moments: combined summary, min start, max end, target deleted."""
+    seed = await _seed_creator_and_video(db_engine)
+    m1_id = await _seed_moment(
+        db_engine, seed["video_id"],
+        title="First", summary="Summary A",
+        start_time=10.0, end_time=20.0,
+    )
+    m2_id = await _seed_moment(
+        db_engine, seed["video_id"],
+        title="Second", summary="Summary B",
+        start_time=25.0, end_time=35.0,
+    )
+
+    resp = await client.post(
+        _moment_url(str(m1_id), "merge"),
+        json={"target_moment_id": str(m2_id)},
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["start_time"] == 10.0
+    assert data["end_time"] == 35.0
+    assert "Summary A" in data["summary"]
+    assert "Summary B" in data["summary"]
+
+    # Target should be deleted — reject should 404
+    resp2 = await client.post(_moment_url(str(m2_id), "reject"))
+    assert resp2.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_merge_different_videos(client: AsyncClient, db_engine):
+    """Merge returns 400 when moments are from different source videos."""
+    seed = await _seed_creator_and_video(db_engine)
+    m1_id = await _seed_moment(db_engine, seed["video_id"], title="Video 1 moment")
+
+    other_video_id = await _seed_second_video(db_engine, seed["creator_id"])
+    m2_id = await _seed_moment(db_engine, other_video_id, title="Video 2 moment")
+
+    resp = await client.post(
+        _moment_url(str(m1_id), "merge"),
+        json={"target_moment_id": str(m2_id)},
+    )
+    assert resp.status_code == 400
+    assert "different source videos" in resp.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_merge_with_self(client: AsyncClient, db_engine):
+    """Merge returns 400 when trying to merge a moment with itself."""
+    seed = await _seed_creator_and_video(db_engine)
+    m_id = await _seed_moment(db_engine, seed["video_id"])
+
+    resp = await client.post(
+        _moment_url(str(m_id), "merge"),
+        json={"target_moment_id": str(m_id)},
+    )
+    assert resp.status_code == 400
+    assert "itself" in resp.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_merge_nonexistent_target(client: AsyncClient, db_engine):
+    """Merge returns 404 when target moment does not exist."""
+    seed = await _seed_creator_and_video(db_engine)
+    m_id = await _seed_moment(db_engine, seed["video_id"])
+
+    resp = await client.post(
+        _moment_url(str(m_id), "merge"),
+        json={"target_moment_id": str(uuid.uuid4())},
+    )
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_merge_nonexistent_source(client: AsyncClient):
+    """Merge returns 404 when source moment does not exist."""
+    fake_id = str(uuid.uuid4())
+    resp = await client.post(
+        _moment_url(fake_id, "merge"),
+        json={"target_moment_id": str(uuid.uuid4())},
+    )
+    assert resp.status_code == 404
+
+
+# ── Mode toggle tests ───────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_get_mode_default(client: AsyncClient):
+    """Get mode returns config default when Redis has no value."""
+    mock_redis = AsyncMock()
+    mock_redis.get = AsyncMock(return_value=None)
+    mock_redis.aclose = AsyncMock()
+
+    with patch("routers.review.get_redis", return_value=mock_redis):
+        resp = await client.get(MODE_URL)
+    assert resp.status_code == 200
+    # Default from config is True
+    assert resp.json()["review_mode"] is True
+
+
+@pytest.mark.asyncio
+async def test_set_mode(client: AsyncClient):
+    """Set mode writes to Redis and returns the new value."""
+    mock_redis = AsyncMock()
+    mock_redis.set = AsyncMock()
+    mock_redis.aclose = AsyncMock()
+
+    with patch("routers.review.get_redis", return_value=mock_redis):
+        resp = await client.put(MODE_URL, json={"review_mode": False})
+    assert resp.status_code == 200
+    assert resp.json()["review_mode"] is False
+    mock_redis.set.assert_called_once_with("chrysopedia:review_mode", "False")
+
+
+@pytest.mark.asyncio
+async def test_get_mode_from_redis(client: AsyncClient):
+    """Get mode reads the value stored in Redis."""
+    mock_redis = AsyncMock()
+    mock_redis.get = AsyncMock(return_value="False")
+    mock_redis.aclose = AsyncMock()
+
+    with patch("routers.review.get_redis", return_value=mock_redis):
+        resp = await client.get(MODE_URL)
+    assert resp.status_code == 200
+    assert resp.json()["review_mode"] is False
+
+
+@pytest.mark.asyncio
+async def test_get_mode_redis_error_fallback(client: AsyncClient):
+    """Get mode falls back to config default when Redis is unavailable."""
+    with patch("routers.review.get_redis", side_effect=ConnectionError("Redis down")):
+        resp = await client.get(MODE_URL)
+    assert resp.status_code == 200
+    # Falls back to config default (True)
+    assert resp.json()["review_mode"] is True
+
+
+@pytest.mark.asyncio
+async def test_set_mode_redis_error(client: AsyncClient):
+    """Set mode returns 503 when Redis is unavailable."""
+    with patch("routers.review.get_redis", side_effect=ConnectionError("Redis down")):
+        resp = await client.put(MODE_URL, json={"review_mode": False})
+    assert resp.status_code == 503