feat(series): chapter layer over series_page — backend (FC-6.1)

Adds an ordered chapter layer to series. Reading order becomes
(series_chapter.chapter_number, series_page.page_number); a chapter may be a
placeholder reserving a slot, and carries an optional parsed stated-page range
used to flag missing-page gaps. An image still lives in at most one series ⇒ one
chapter (image_id stays UNIQUE).

- models: series_chapter; series_page gains chapter_id (NOT NULL, cascade) +
  stated_page. Migration 0040 backfills every existing series into one
  auto-chapter holding its current flat pages — no data loss.
- SeriesService: chapter CRUD (create/update/reorder/delete/merge), page→chapter
  assignment, reorder_pages, chapter-aware set_cover; list_pages now returns
  chapters[] + gaps[] alongside a back-compat flat pages[]. Legacy series-wide
  reorder operates on the single default chapter and rejects multi-chapter series.
- API: chapter endpoints under /api/series/<tag>/chapters; POST pages accepts an
  optional chapter_id.
- TagService.merge now repoints series_chapter too, so a merged series' chapters
  (and their pages) survive the source tag's deletion instead of cascading away.
- Tests: new chapter suite; updated the 4 direct SeriesPage(...) constructions to
  supply chapter_id.

Frontend (chapter-aware manage view + reader) lands next; until then the
existing UI keeps working via the flat pages[] + single default chapter.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

alembic/versions/0040_series_chapters.py

@@ -0,0 +1,108 @@
+"""series chapters: chapter layer over series_page (FC-6.1)
+
+Revision ID: 0040
+Revises: 0039
+Create Date: 2026-06-07
+
+A series (Tag kind='series') gains an ordered chapter layer. Reading order
+becomes (series_chapter.chapter_number, series_page.page_number). Every existing
+series is backfilled into a single auto-chapter (chapter_number=1) holding its
+current flat pages, so no data is lost and the old flat ordering is preserved.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0040"
+down_revision: Union[str, None] = "0039"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "series_chapter",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column(
+            "series_tag_id",
+            sa.Integer,
+            sa.ForeignKey("tag.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("chapter_number", sa.Integer, nullable=False),
+        sa.Column("title", sa.Text, nullable=True),
+        sa.Column(
+            "is_placeholder", sa.Boolean, nullable=False, server_default="false"
+        ),
+        sa.Column("stated_page_start", sa.Integer, nullable=True),
+        sa.Column("stated_page_end", sa.Integer, nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+    )
+    op.create_index(
+        "ix_series_chapter_series_tag_id", "series_chapter", ["series_tag_id"]
+    )
+
+    # New columns on series_page; chapter_id starts nullable so we can backfill.
+    op.add_column(
+        "series_page", sa.Column("chapter_id", sa.Integer, nullable=True)
+    )
+    op.add_column(
+        "series_page", sa.Column("stated_page", sa.Integer, nullable=True)
+    )
+
+    conn = op.get_bind()
+    # One auto-chapter per existing series (any series_tag_id present in pages).
+    conn.execute(
+        sa.text(
+            "INSERT INTO series_chapter "
+            "(series_tag_id, chapter_number, is_placeholder, created_at, updated_at) "
+            "SELECT DISTINCT series_tag_id, 1, false, now(), now() "
+            "FROM series_page"
+        )
+    )
+    # Point every existing page at its series' auto-chapter.
+    conn.execute(
+        sa.text(
+            "UPDATE series_page sp "
+            "SET chapter_id = sc.id "
+            "FROM series_chapter sc "
+            "WHERE sc.series_tag_id = sp.series_tag_id"
+        )
+    )
+
+    # Now lock chapter_id down: NOT NULL + FK (cascade) + index.
+    op.alter_column("series_page", "chapter_id", nullable=False)
+    op.create_foreign_key(
+        "fk_series_page_chapter_id",
+        "series_page",
+        "series_chapter",
+        ["chapter_id"],
+        ["id"],
+        ondelete="CASCADE",
+    )
+    op.create_index(
+        "ix_series_page_chapter_id", "series_page", ["chapter_id"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_series_page_chapter_id", table_name="series_page")
+    op.drop_constraint(
+        "fk_series_page_chapter_id", "series_page", type_="foreignkey"
+    )
+    op.drop_column("series_page", "stated_page")
+    op.drop_column("series_page", "chapter_id")
+    op.drop_index("ix_series_chapter_series_tag_id", table_name="series_chapter")
+    op.drop_table("series_chapter")

backend/app/api/tags.py

@@ -368,6 +368,31 @@ def _series_err(exc: SeriesError):
     return jsonify({"error": msg}), status
 
 
+def _opt_int(body, key: str):
+    """(value, error) — value is None when absent, error is (json, status)."""
+    if not body or body.get(key) is None:
+        return None, None
+    try:
+        return int(body[key]), None
+    except (TypeError, ValueError):
+        return None, (jsonify({"error": f"{key} must be an integer"}), 400)
+
+
+def _parse_int_list(body, key: str, *, max_ids: int = 500):
+    """(list, error) for a required list of ints under `key`."""
+    if not body or key not in body:
+        return None, (jsonify({"error": f"{key} required"}), 400)
+    raw = body[key]
+    if not isinstance(raw, list) or not raw:
+        return None, (jsonify({"error": f"{key} must be a non-empty list"}), 400)
+    if len(raw) > max_ids:
+        return None, (jsonify({"error": f"too many ids (max {max_ids})"}), 400)
+    try:
+        return [int(x) for x in raw], None
+    except (TypeError, ValueError):
+        return None, (jsonify({"error": f"{key} must be integers"}), 400)
+
+
 @tags_bp.route("/series/<int:tag_id>/pages", methods=["GET"])
 async def series_pages(tag_id: int):
     async with get_session() as session:
@@ -384,9 +409,14 @@ async def series_add(tag_id: int):
     ids, err = _parse_bulk_ids(body, max_ids=500)
     if err:
         return err
+    chapter_id, cerr = _opt_int(body, "chapter_id")
+    if cerr:
+        return cerr
     async with get_session() as session:
         try:
-            n = await SeriesService(session).add_images(tag_id, ids)
+            n = await SeriesService(session).add_images(
+                tag_id, ids, chapter_id=chapter_id
+            )
         except SeriesError as exc:
             return _series_err(exc)
         await session.commit()
@@ -439,3 +469,131 @@ async def series_cover(tag_id: int):
             return _series_err(exc)
         await session.commit()
     return jsonify({"ok": True})
+
+
+# ---- chapters (FC-6.1) ----------------------------------------------------
+
+
+@tags_bp.route("/series/<int:tag_id>/chapters", methods=["POST"])
+async def series_chapter_create(tag_id: int):
+    body = await request.get_json() or {}
+    title = body.get("title")
+    if title is not None and not isinstance(title, str):
+        return jsonify({"error": "title must be a string"}), 400
+    is_placeholder = bool(body.get("is_placeholder", False))
+    start, serr = _opt_int(body, "stated_page_start")
+    if serr:
+        return serr
+    end, eerr = _opt_int(body, "stated_page_end")
+    if eerr:
+        return eerr
+    async with get_session() as session:
+        try:
+            ch = await SeriesService(session).create_chapter(
+                tag_id,
+                title=title,
+                is_placeholder=is_placeholder,
+                stated_page_start=start,
+                stated_page_end=end,
+            )
+        except SeriesError as exc:
+            return _series_err(exc)
+        await session.commit()
+    return jsonify(ch)
+
+
+@tags_bp.route("/series/<int:tag_id>/chapters/reorder", methods=["POST"])
+async def series_chapter_reorder(tag_id: int):
+    body = await request.get_json()
+    ids, err = _parse_int_list(body, "chapter_ids")
+    if err:
+        return err
+    async with get_session() as session:
+        try:
+            await SeriesService(session).reorder_chapters(tag_id, ids)
+        except SeriesError as exc:
+            return _series_err(exc)
+        await session.commit()
+    return jsonify({"ok": True})
+
+
+@tags_bp.route(
+    "/series/<int:tag_id>/chapters/<int:chapter_id>", methods=["PATCH"]
+)
+async def series_chapter_update(tag_id: int, chapter_id: int):
+    body = await request.get_json() or {}
+    kwargs: dict = {}
+    if "title" in body:
+        if body["title"] is not None and not isinstance(body["title"], str):
+            return jsonify({"error": "title must be a string"}), 400
+        kwargs.update(set_title=True, title=body["title"])
+    if "stated_page_start" in body:
+        start, serr = _opt_int(body, "stated_page_start")
+        if serr:
+            return serr
+        kwargs.update(set_start=True, stated_page_start=start)
+    if "stated_page_end" in body:
+        end, eerr = _opt_int(body, "stated_page_end")
+        if eerr:
+            return eerr
+        kwargs.update(set_end=True, stated_page_end=end)
+    async with get_session() as session:
+        try:
+            await SeriesService(session).update_chapter(
+                tag_id, chapter_id, **kwargs
+            )
+        except SeriesError as exc:
+            return _series_err(exc)
+        await session.commit()
+    return jsonify({"ok": True})
+
+
+@tags_bp.route(
+    "/series/<int:tag_id>/chapters/<int:chapter_id>", methods=["DELETE"]
+)
+async def series_chapter_delete(tag_id: int, chapter_id: int):
+    async with get_session() as session:
+        try:
+            await SeriesService(session).delete_chapter(tag_id, chapter_id)
+        except SeriesError as exc:
+            return _series_err(exc)
+        await session.commit()
+    return jsonify({"ok": True})
+
+
+@tags_bp.route(
+    "/series/<int:tag_id>/chapters/<int:chapter_id>/merge", methods=["POST"]
+)
+async def series_chapter_merge(tag_id: int, chapter_id: int):
+    body = await request.get_json()
+    target, terr = _opt_int(body, "target_chapter_id")
+    if terr:
+        return terr
+    if target is None:
+        return jsonify({"error": "target_chapter_id required"}), 400
+    async with get_session() as session:
+        try:
+            moved = await SeriesService(session).merge_chapter(
+                tag_id, chapter_id, target
+            )
+        except SeriesError as exc:
+            return _series_err(exc)
+        await session.commit()
+    return jsonify({"moved_count": moved})
+
+
+@tags_bp.route(
+    "/series/<int:tag_id>/chapters/<int:chapter_id>/reorder", methods=["POST"]
+)
+async def series_chapter_reorder_pages(tag_id: int, chapter_id: int):
+    body = await request.get_json()
+    ids, err = _parse_bulk_ids(body, max_ids=500)
+    if err:
+        return err
+    async with get_session() as session:
+        try:
+            await SeriesService(session).reorder_pages(tag_id, chapter_id, ids)
+        except SeriesError as exc:
+            return _series_err(exc)
+        await session.commit()
+    return jsonify({"ok": True})

backend/app/models/__init__.py

@@ -18,6 +18,7 @@ from .patreon_failed_media import PatreonFailedMedia
 from .patreon_seen_media import PatreonSeenMedia
 from .post import Post
 from .post_attachment import PostAttachment
+from .series_chapter import SeriesChapter
 from .series_page import SeriesPage
 from .source import Source
 from .tag import Tag, TagKind, image_tag
@@ -39,6 +40,7 @@ __all__ = [
     "PatreonSeenMedia",
     "Post",
     "PostAttachment",
+    "SeriesChapter",
     "SeriesPage",
     "ImageRecord",
     "ImageProvenance",

backend/app/models/series_chapter.py

@@ -0,0 +1,47 @@
+"""SeriesChapter — an ordered chapter/part within a series.
+
+A series IS a Tag(kind='series'); a chapter groups ordered SeriesPages under it.
+Reading order is (chapter.chapter_number, series_page.page_number): chapter_number
+sets the order of chapters, page_number orders pages within a chapter.
+
+chapter_number is an ordering key only (not unique) — reorder rewrites 1..N
+wholesale, mirroring series_page.page_number, so a reorder can't transiently
+collide on a unique index.
+
+A chapter may be a placeholder (is_placeholder=True) — a reserved empty slot for
+a section the operator doesn't have yet; it holds no pages and shows as a gap in
+the reader. stated_page_start/end carry the page range parsed from the source
+post (FC-6.2), used to flag missing-page gaps; both are nullable when unknown.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import Base
+
+
+class SeriesChapter(Base):
+    __tablename__ = "series_chapter"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    series_tag_id: Mapped[int] = mapped_column(
+        ForeignKey("tag.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    chapter_number: Mapped[int] = mapped_column(Integer, nullable=False)
+    title: Mapped[str | None] = mapped_column(Text, nullable=True)
+    is_placeholder: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, server_default="false"
+    )
+    stated_page_start: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    stated_page_end: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),
+    )

backend/app/models/series_page.py

@@ -1,9 +1,12 @@
 """SeriesPage — ordered image membership for a series-kind Tag.
 
-A series IS a Tag with kind='series'; series_page gives it ordered pages.
+A series IS a Tag with kind='series'; series_page gives it ordered pages,
-An image belongs to at most one series (UNIQUE image_id). Cover = the
+grouped into chapters (FC-6). An image belongs to at most one series
-lowest page_number. page_number is an ordering key only (not unique) —
+(UNIQUE image_id) ⇒ at most one chapter. Reading order is
-reorder rewrites 1..N wholesale.
+(chapter.chapter_number, series_page.page_number): page_number orders pages
+WITHIN a chapter and is an ordering key only (not unique) — reorder rewrites
+1..N wholesale. stated_page carries the page number parsed from the source
+post (FC-6.2), nullable when unknown.
 """
 
 from datetime import datetime
@@ -21,12 +24,18 @@ class SeriesPage(Base):
     series_tag_id: Mapped[int] = mapped_column(
         ForeignKey("tag.id", ondelete="CASCADE"), nullable=False, index=True
     )
+    chapter_id: Mapped[int] = mapped_column(
+        ForeignKey("series_chapter.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
     image_id: Mapped[int] = mapped_column(
         ForeignKey("image_record.id", ondelete="CASCADE"),
         nullable=False,
         unique=True,
     )
     page_number: Mapped[int] = mapped_column(Integer, nullable=False)
+    stated_page: Mapped[int | None] = mapped_column(Integer, nullable=True)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )

backend/app/services/series_service.py

@@ -1,4 +1,8 @@
-"""Series = a series-kind Tag + ordered series_page membership.
+"""Series = a series-kind Tag + ordered chapters, each holding ordered pages.
+
+Reading order is (series_chapter.chapter_number, series_page.page_number). A
+chapter may be a placeholder (no pages) reserving a slot. An image lives in at
+most one series ⇒ one chapter (series_page.image_id is UNIQUE).
 
 All mutations are Core/set-based and run in the request transaction.
 """
@@ -8,6 +12,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import ImageRecord, Tag, TagKind
+from ..models.series_chapter import SeriesChapter
 from ..models.series_page import SeriesPage
 from .gallery_service import thumbnail_url
 
@@ -21,6 +26,8 @@ class SeriesService:
     def __init__(self, session: AsyncSession):
         self.session = session
 
+    # ---- guards / helpers -------------------------------------------------
+
     async def _require_series(self, series_tag_id: int) -> Tag:
         tag = await self.session.get(Tag, series_tag_id)
         if tag is None:
@@ -32,7 +39,7 @@ class SeriesService:
     @staticmethod
     def _clean_ids(ids: list[int]) -> list[int]:
         if not ids:
-            raise SeriesError("image_ids must be a non-empty list")
+            raise SeriesError("ids must be a non-empty list")
         if len(ids) > 500:
             raise SeriesError("selection too large (max 500)")
         seen: set[int] = set()
@@ -43,51 +50,172 @@ class SeriesService:
                 out.append(int(x))
         return out
 
-    async def _member_order(self, series_tag_id: int) -> list[int]:
+    async def _require_chapter(self, series_tag_id: int, chapter_id: int):
+        row = (
+            await self.session.execute(
+                select(SeriesChapter).where(
+                    and_(
+                        SeriesChapter.id == chapter_id,
+                        SeriesChapter.series_tag_id == series_tag_id,
+                    )
+                )
+            )
+        ).scalar_one_or_none()
+        if row is None:
+            raise SeriesError(
+                f"chapter {chapter_id} is not in series {series_tag_id}"
+            )
+        return row
+
+    async def _ensure_default_chapter(self, series_tag_id: int) -> int:
+        """Lowest-numbered chapter of the series, creating a first chapter if
+        the series has none (a fresh series, or the legacy single-chapter path)."""
+        existing = await self.session.scalar(
+            select(SeriesChapter.id)
+            .where(SeriesChapter.series_tag_id == series_tag_id)
+            .order_by(SeriesChapter.chapter_number.asc())
+            .limit(1)
+        )
+        if existing is not None:
+            return existing
+        res = await self.session.execute(
+            pg_insert(SeriesChapter)
+            .values(series_tag_id=series_tag_id, chapter_number=1)
+            .returning(SeriesChapter.id)
+        )
+        return res.scalar_one()
+
+    async def _chapter_page_order(self, chapter_id: int) -> list[int]:
         rows = (
             await self.session.execute(
                 select(SeriesPage.image_id)
-                .where(SeriesPage.series_tag_id == series_tag_id)
+                .where(SeriesPage.chapter_id == chapter_id)
                 .order_by(SeriesPage.page_number.asc())
             )
         ).scalars().all()
         return list(rows)
 
+    async def _chapter_order(self, series_tag_id: int) -> list[int]:
+        rows = (
+            await self.session.execute(
+                select(SeriesChapter.id)
+                .where(SeriesChapter.series_tag_id == series_tag_id)
+                .order_by(SeriesChapter.chapter_number.asc())
+            )
+        ).scalars().all()
+        return list(rows)
+
+    # ---- read -------------------------------------------------------------
+
+    @staticmethod
+    def _gaps(chapters: list[dict]) -> list[dict]:
+        """Missing-page gaps between consecutive chapters with stated ranges."""
+        out: list[dict] = []
+        prev = None
+        for ch in chapters:
+            start = ch["stated_page_start"]
+            if (
+                prev is not None
+                and prev["stated_page_end"] is not None
+                and start is not None
+                and start > prev["stated_page_end"] + 1
+            ):
+                out.append(
+                    {
+                        "after_chapter_id": prev["id"],
+                        "start": prev["stated_page_end"] + 1,
+                        "end": start - 1,
+                    }
+                )
+            prev = ch
+        return out
+
     async def list_pages(self, series_tag_id: int) -> dict:
         tag = await self._require_series(series_tag_id)
         rows = (
             await self.session.execute(
                 select(
+                    SeriesChapter.id.label("chapter_id"),
+                    SeriesChapter.chapter_number,
+                    SeriesChapter.title,
+                    SeriesChapter.is_placeholder,
+                    SeriesChapter.stated_page_start,
+                    SeriesChapter.stated_page_end,
                     SeriesPage.image_id,
                     SeriesPage.page_number,
+                    SeriesPage.stated_page,
                     ImageRecord.sha256,
                     ImageRecord.mime,
                     ImageRecord.path,
                     ImageRecord.thumbnail_path,
                 )
-                .join(ImageRecord, ImageRecord.id == SeriesPage.image_id)
+                .select_from(SeriesChapter)
-                .where(SeriesPage.series_tag_id == series_tag_id)
+                .outerjoin(SeriesPage, SeriesPage.chapter_id == SeriesChapter.id)
-                .order_by(SeriesPage.page_number.asc())
+                .outerjoin(ImageRecord, ImageRecord.id == SeriesPage.image_id)
+                .where(SeriesChapter.series_tag_id == series_tag_id)
+                .order_by(
+                    SeriesChapter.chapter_number.asc(),
+                    SeriesPage.page_number.asc(),
+                )
             )
         ).all()
+
+        chapters: list[dict] = []
+        flat: list[dict] = []
+        by_id: dict[int, dict] = {}
+        for r in rows:
+            ch = by_id.get(r.chapter_id)
+            if ch is None:
+                ch = {
+                    "id": r.chapter_id,
+                    "chapter_number": r.chapter_number,
+                    "title": r.title,
+                    "is_placeholder": r.is_placeholder,
+                    "stated_page_start": r.stated_page_start,
+                    "stated_page_end": r.stated_page_end,
+                    "pages": [],
+                }
+                by_id[r.chapter_id] = ch
+                chapters.append(ch)
+            if r.image_id is None:
+                continue  # placeholder / empty chapter
+            page = {
+                "image_id": r.image_id,
+                "chapter_id": r.chapter_id,
+                "page_number": r.page_number,
+                "stated_page": r.stated_page,
+                "thumbnail_url": thumbnail_url(r.thumbnail_path, r.sha256, r.mime),
+                "image_url": f"/images/{r.path.split('/images/', 1)[-1]}",
+            }
+            ch["pages"].append(page)
+            flat.append(page)
+
         return {
             "series": {"id": tag.id, "name": tag.name},
-            "pages": [
+            "chapters": chapters,
-                {
+            "pages": flat,  # back-compat: flat reading order across chapters
-                    "image_id": r.image_id,
+            "gaps": self._gaps(chapters),
-                    "page_number": r.page_number,
-                    "thumbnail_url": thumbnail_url(r.thumbnail_path, r.sha256, r.mime),
-                    "image_url": f"/images/{r.path.split('/images/', 1)[-1]}",
-                }
-                for r in rows
-            ],
         }
 
+    # ---- pages ------------------------------------------------------------
+
     async def add_images(
-        self, series_tag_id: int, image_ids: list[int]
+        self,
+        series_tag_id: int,
+        image_ids: list[int],
+        chapter_id: int | None = None,
+        stated_pages: dict[int, int] | None = None,
     ) -> int:
+        """Append images as pages of a chapter (the series' default chapter when
+        chapter_id is None). Images already in THIS series are left untouched;
+        images in another series are moved here (image_id is UNIQUE)."""
         await self._require_series(series_tag_id)
         ids = self._clean_ids(image_ids)
+        if chapter_id is None:
+            chapter_id = await self._ensure_default_chapter(series_tag_id)
+        else:
+            await self._require_chapter(series_tag_id, chapter_id)
+
         existing = dict(
             (
                 await self.session.execute(
@@ -96,29 +224,30 @@ class SeriesService:
                 )
             ).all()
         )
-        # Already in THIS series → leave untouched (no churn).
         to_add = [i for i in ids if existing.get(i) != series_tag_id]
         if not to_add:
             return 0
-        # Remove any prior membership for the to_add images (the "move").
+        # Move: drop any prior membership for these images (UNIQUE image_id).
         await self.session.execute(
-            SeriesPage.__table__.delete().where(
+            SeriesPage.__table__.delete().where(SeriesPage.image_id.in_(to_add))
-                SeriesPage.image_id.in_(to_add)
-            )
         )
         max_pn = (
             await self.session.scalar(
-                select(func.coalesce(func.max(SeriesPage.page_number), 0))
+                select(func.coalesce(func.max(SeriesPage.page_number), 0)).where(
-                .where(SeriesPage.series_tag_id == series_tag_id)
+                    SeriesPage.chapter_id == chapter_id
+                )
             )
         ) or 0
+        sp = stated_pages or {}
         await self.session.execute(
             pg_insert(SeriesPage).values(
                 [
                     {
                         "series_tag_id": series_tag_id,
+                        "chapter_id": chapter_id,
                         "image_id": iid,
                         "page_number": max_pn + offset,
+                        "stated_page": sp.get(iid),
                     }
                     for offset, iid in enumerate(to_add, start=1)
                 ]
@@ -141,32 +270,211 @@ class SeriesService:
         )
         return res.rowcount or 0
 
-    async def reorder(
+    async def reorder_pages(
-        self, series_tag_id: int, ordered_image_ids: list[int]
+        self, series_tag_id: int, chapter_id: int, ordered_image_ids: list[int]
     ) -> None:
         await self._require_series(series_tag_id)
+        await self._require_chapter(series_tag_id, chapter_id)
         ordered = self._clean_ids(ordered_image_ids)
-        current = set(await self._member_order(series_tag_id))
+        current = set(await self._chapter_page_order(chapter_id))
         if set(ordered) != current or len(ordered) != len(current):
             raise SeriesError(
-                "ordered image_ids must exactly match series membership"
+                "ordered image_ids must exactly match the chapter's pages"
             )
         for idx, iid in enumerate(ordered, start=1):
             await self.session.execute(
                 update(SeriesPage)
                 .where(
                     and_(
-                        SeriesPage.series_tag_id == series_tag_id,
+                        SeriesPage.chapter_id == chapter_id,
                         SeriesPage.image_id == iid,
                     )
                 )
                 .values(page_number=idx)
             )
 
-    async def set_cover(self, series_tag_id: int, image_id: int) -> None:
+    async def reorder(
+        self, series_tag_id: int, ordered_image_ids: list[int]
+    ) -> None:
+        """Legacy series-wide reorder — reorders the default chapter. Valid only
+        for single-chapter series (the only shape the pre-chapter UI produced);
+        multi-chapter series must use reorder_pages with an explicit chapter."""
         await self._require_series(series_tag_id)
-        order = await self._member_order(series_tag_id)
+        chapters = await self._chapter_order(series_tag_id)
-        if image_id not in order:
+        if len(chapters) > 1:
+            raise SeriesError(
+                "series has multiple chapters; reorder a specific chapter"
+            )
+        chapter_id = (
+            chapters[0]
+            if chapters
+            else await self._ensure_default_chapter(series_tag_id)
+        )
+        await self.reorder_pages(series_tag_id, chapter_id, ordered_image_ids)
+
+    # ---- chapters ---------------------------------------------------------
+
+    async def create_chapter(
+        self,
+        series_tag_id: int,
+        *,
+        title: str | None = None,
+        is_placeholder: bool = False,
+        stated_page_start: int | None = None,
+        stated_page_end: int | None = None,
+    ) -> dict:
+        await self._require_series(series_tag_id)
+        max_cn = (
+            await self.session.scalar(
+                select(
+                    func.coalesce(func.max(SeriesChapter.chapter_number), 0)
+                ).where(SeriesChapter.series_tag_id == series_tag_id)
+            )
+        ) or 0
+        res = await self.session.execute(
+            pg_insert(SeriesChapter)
+            .values(
+                series_tag_id=series_tag_id,
+                chapter_number=max_cn + 1,
+                title=title,
+                is_placeholder=is_placeholder,
+                stated_page_start=stated_page_start,
+                stated_page_end=stated_page_end,
+            )
+            .returning(SeriesChapter.id, SeriesChapter.chapter_number)
+        )
+        row = res.one()
+        return {"id": row.id, "chapter_number": row.chapter_number}
+
+    async def update_chapter(
+        self,
+        series_tag_id: int,
+        chapter_id: int,
+        *,
+        title: str | None = None,
+        stated_page_start: int | None = None,
+        stated_page_end: int | None = None,
+        set_title: bool = False,
+        set_start: bool = False,
+        set_end: bool = False,
+    ) -> None:
+        """Partial chapter edit. The set_* flags say which fields to write (so
+        None can be written explicitly, e.g. clearing a stated page)."""
+        await self._require_series(series_tag_id)
+        await self._require_chapter(series_tag_id, chapter_id)
+        values: dict = {}
+        if set_title:
+            values["title"] = title
+        if set_start:
+            values["stated_page_start"] = stated_page_start
+        if set_end:
+            values["stated_page_end"] = stated_page_end
+        if not values:
+            return
+        await self.session.execute(
+            update(SeriesChapter)
+            .where(SeriesChapter.id == chapter_id)
+            .values(**values)
+        )
+
+    async def _renumber_chapters(self, series_tag_id: int) -> None:
+        for idx, cid in enumerate(
+            await self._chapter_order(series_tag_id), start=1
+        ):
+            await self.session.execute(
+                update(SeriesChapter)
+                .where(SeriesChapter.id == cid)
+                .values(chapter_number=idx)
+            )
+
+    async def reorder_chapters(
+        self, series_tag_id: int, ordered_chapter_ids: list[int]
+    ) -> None:
+        await self._require_series(series_tag_id)
+        ordered = self._clean_ids(ordered_chapter_ids)
+        current = set(await self._chapter_order(series_tag_id))
+        if set(ordered) != current or len(ordered) != len(current):
+            raise SeriesError(
+                "ordered chapter_ids must exactly match the series' chapters"
+            )
+        for idx, cid in enumerate(ordered, start=1):
+            await self.session.execute(
+                update(SeriesChapter)
+                .where(SeriesChapter.id == cid)
+                .values(chapter_number=idx)
+            )
+
+    async def delete_chapter(self, series_tag_id: int, chapter_id: int) -> None:
+        await self._require_series(series_tag_id)
+        await self._require_chapter(series_tag_id, chapter_id)
+        # Pages cascade-delete with the chapter (FK ondelete=CASCADE).
+        await self.session.execute(
+            SeriesChapter.__table__.delete().where(
+                SeriesChapter.id == chapter_id
+            )
+        )
+        await self._renumber_chapters(series_tag_id)
+
+    async def merge_chapter(
+        self, series_tag_id: int, source_chapter_id: int, target_chapter_id: int
+    ) -> int:
+        """Move source chapter's pages onto the end of the target chapter, then
+        delete the now-empty source chapter. Returns the number of pages moved."""
+        await self._require_series(series_tag_id)
+        if source_chapter_id == target_chapter_id:
+            raise SeriesError("cannot merge a chapter into itself")
+        await self._require_chapter(series_tag_id, source_chapter_id)
+        await self._require_chapter(series_tag_id, target_chapter_id)
+        moving = await self._chapter_page_order(source_chapter_id)
+        max_pn = (
+            await self.session.scalar(
+                select(func.coalesce(func.max(SeriesPage.page_number), 0)).where(
+                    SeriesPage.chapter_id == target_chapter_id
+                )
+            )
+        ) or 0
+        for offset, iid in enumerate(moving, start=1):
+            await self.session.execute(
+                update(SeriesPage)
+                .where(
+                    and_(
+                        SeriesPage.chapter_id == source_chapter_id,
+                        SeriesPage.image_id == iid,
+                    )
+                )
+                .values(chapter_id=target_chapter_id, page_number=max_pn + offset)
+            )
+        await self.session.execute(
+            SeriesChapter.__table__.delete().where(
+                SeriesChapter.id == source_chapter_id
+            )
+        )
+        await self._renumber_chapters(series_tag_id)
+        return len(moving)
+
+    async def set_cover(self, series_tag_id: int, image_id: int) -> None:
+        """Cover = the (chapter_number=1, page_number=1) image. Bring the image's
+        chapter to the front and the image to the front of its chapter."""
+        await self._require_series(series_tag_id)
+        row = (
+            await self.session.execute(
+                select(SeriesPage.chapter_id)
+                .where(
+                    and_(
+                        SeriesPage.series_tag_id == series_tag_id,
+                        SeriesPage.image_id == image_id,
+                    )
+                )
+            )
+        ).scalar_one_or_none()
+        if row is None:
             raise SeriesError(f"image {image_id} is not in this series")
-        new_order = [image_id] + [i for i in order if i != image_id]
+        chapter_id = row
-        await self.reorder(series_tag_id, new_order)
+        chapters = await self._chapter_order(series_tag_id)
+        if chapters and chapters[0] != chapter_id:
+            new_chapters = [chapter_id] + [c for c in chapters if c != chapter_id]
+            await self.reorder_chapters(series_tag_id, new_chapters)
+        pages = await self._chapter_page_order(chapter_id)
+        if pages and pages[0] != image_id:
+            new_pages = [image_id] + [p for p in pages if p != image_id]
+            await self.reorder_pages(series_tag_id, chapter_id, new_pages)

backend/app/services/tag_service.py

@@ -515,8 +515,18 @@ class TagService:
         )
 
     async def _repoint_series_pages(self, src: int, tgt: int) -> None:
+        from ..models.series_chapter import SeriesChapter
         from ..models.series_page import SeriesPage
 
+        # Move the chapters first so the pages' chapter_id FK stays valid: a
+        # chapter left pointing at src would cascade-delete (with its pages) when
+        # src is removed. chapter_number may now collide across the merged set —
+        # acceptable (it's an ordering key, not unique).
+        await self.session.execute(
+            update(SeriesChapter)
+            .where(SeriesChapter.series_tag_id == src)
+            .values(series_tag_id=tgt)
+        )
         # image_id is UNIQUE across series_page and src != tgt, so an
         # image in src's series cannot already be in tgt's — no collision.
         await self.session.execute(

tests/test_cleanup_service.py

@@ -9,6 +9,7 @@ import pytest
 from sqlalchemy import func, select
 
 from backend.app.models import Artist, ImageRecord, Tag, TagKind
+from backend.app.models.series_chapter import SeriesChapter
 from backend.app.models.series_page import SeriesPage
 from backend.app.models.tag import image_tag
 from backend.app.services import cleanup_service
@@ -376,7 +377,12 @@ def test_reset_content_tagging_deletes_content_keeps_fandom_series(db_sync, tmp_
         {"image_record_id": img.id, "tag_id": c.id},
         {"image_record_id": img.id, "tag_id": s.id},  # series membership
     ]))
-    db_sync.add(SeriesPage(series_tag_id=s.id, image_id=img.id, page_number=1))
+    ch = SeriesChapter(series_tag_id=s.id, chapter_number=1)
+    db_sync.add(ch)
+    db_sync.flush()
+    db_sync.add(SeriesPage(
+        series_tag_id=s.id, chapter_id=ch.id, image_id=img.id, page_number=1
+    ))
     db_sync.commit()
 
     result = cleanup_service.reset_content_tagging(db_sync, dry_run=False)

tests/test_series_chapters.py

@@ -0,0 +1,193 @@
+"""FC-6.1 — chapter layer over series_page."""
+
+import pytest
+from sqlalchemy import func, select
+
+from backend.app.models import ImageRecord, TagKind
+from backend.app.models.series_chapter import SeriesChapter
+from backend.app.models.series_page import SeriesPage
+from backend.app.services.series_service import SeriesError, SeriesService
+from backend.app.services.tag_service import TagService
+
+pytestmark = pytest.mark.integration
+
+_SEQ = 0
+
+
+async def _img(db):
+    global _SEQ
+    _SEQ += 1
+    rec = ImageRecord(
+        path=f"/tmp/fc_sc_{_SEQ}.png", sha256=f"{_SEQ:064d}",
+        size_bytes=1, mime="image/png", origin="uploaded",
+    )
+    db.add(rec)
+    await db.flush()
+    return rec.id
+
+
+async def _series(db, name):
+    return (await TagService(db).find_or_create(name, TagKind.series)).id
+
+
+async def _chapter_numbers(db, sid):
+    rows = (
+        await db.execute(
+            select(SeriesChapter.id, SeriesChapter.chapter_number)
+            .where(SeriesChapter.series_tag_id == sid)
+            .order_by(SeriesChapter.chapter_number)
+        )
+    ).all()
+    return [(r.id, r.chapter_number) for r in rows]
+
+
+async def _page_order(db, chapter_id):
+    rows = (
+        await db.execute(
+            select(SeriesPage.image_id)
+            .where(SeriesPage.chapter_id == chapter_id)
+            .order_by(SeriesPage.page_number)
+        )
+    ).scalars().all()
+    return list(rows)
+
+
+@pytest.mark.asyncio
+async def test_add_images_creates_default_chapter(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C1")
+    i1, i2 = await _img(db), await _img(db)
+    await svc.add_images(sid, [i1, i2])
+    chapters = await _chapter_numbers(db, sid)
+    assert len(chapters) == 1
+    assert chapters[0][1] == 1
+    assert await _page_order(db, chapters[0][0]) == [i1, i2]
+
+
+@pytest.mark.asyncio
+async def test_create_and_target_specific_chapter(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C2")
+    i1, i2 = await _img(db), await _img(db)
+    await svc.add_images(sid, [i1])  # default chapter 1
+    ch2 = await svc.create_chapter(sid, title="Two")
+    assert ch2["chapter_number"] == 2
+    await svc.add_images(sid, [i2], chapter_id=ch2["id"])
+    assert await _page_order(db, ch2["id"]) == [i2]
+    # i1 stayed in chapter 1.
+    chapters = await _chapter_numbers(db, sid)
+    assert len(chapters) == 2
+
+
+@pytest.mark.asyncio
+async def test_reorder_chapters(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C3")
+    c1 = (await svc.create_chapter(sid))["id"]
+    c2 = (await svc.create_chapter(sid))["id"]
+    c3 = (await svc.create_chapter(sid))["id"]
+    await svc.reorder_chapters(sid, [c3, c1, c2])
+    assert await _chapter_numbers(db, sid) == [(c3, 1), (c1, 2), (c2, 3)]
+    with pytest.raises(SeriesError):
+        await svc.reorder_chapters(sid, [c1, c2])
+
+
+@pytest.mark.asyncio
+async def test_reorder_pages_within_chapter(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C4")
+    i1, i2, i3 = await _img(db), await _img(db), await _img(db)
+    await svc.add_images(sid, [i1, i2, i3])
+    cid = (await _chapter_numbers(db, sid))[0][0]
+    await svc.reorder_pages(sid, cid, [i3, i1, i2])
+    assert await _page_order(db, cid) == [i3, i1, i2]
+
+
+@pytest.mark.asyncio
+async def test_legacy_reorder_rejects_multi_chapter(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C5")
+    i1 = await _img(db)
+    await svc.add_images(sid, [i1])
+    await svc.create_chapter(sid)  # now 2 chapters
+    with pytest.raises(SeriesError):
+        await svc.reorder(sid, [i1])
+
+
+@pytest.mark.asyncio
+async def test_merge_chapter_moves_pages_and_renumbers(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C6")
+    i1, i2, i3 = await _img(db), await _img(db), await _img(db)
+    await svc.add_images(sid, [i1])  # chapter 1
+    c2 = (await svc.create_chapter(sid))["id"]
+    await svc.add_images(sid, [i2, i3], chapter_id=c2)
+    c1 = (await _chapter_numbers(db, sid))[0][0]
+    moved = await svc.merge_chapter(sid, c2, c1)
+    assert moved == 2
+    assert await _page_order(db, c1) == [i1, i2, i3]
+    # source chapter gone; remaining renumbered 1..N.
+    assert await _chapter_numbers(db, sid) == [(c1, 1)]
+
+
+@pytest.mark.asyncio
+async def test_delete_chapter_cascades_pages_and_renumbers(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C7")
+    i1, i2 = await _img(db), await _img(db)
+    await svc.add_images(sid, [i1])
+    c2 = (await svc.create_chapter(sid))["id"]
+    await svc.add_images(sid, [i2], chapter_id=c2)
+    await svc.delete_chapter(sid, c2)
+    assert [c for _, c in await _chapter_numbers(db, sid)] == [1]
+    # i2's page cascaded away with its chapter.
+    cnt = await db.scalar(
+        select(func.count()).select_from(SeriesPage)
+        .where(SeriesPage.image_id == i2)
+    )
+    assert cnt == 0
+
+
+@pytest.mark.asyncio
+async def test_set_cover_brings_chapter_and_page_to_front(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C8")
+    i1, i2, i3 = await _img(db), await _img(db), await _img(db)
+    await svc.add_images(sid, [i1])  # chapter 1
+    c2 = (await svc.create_chapter(sid))["id"]
+    await svc.add_images(sid, [i2, i3], chapter_id=c2)
+    await svc.set_cover(sid, i3)
+    chapters = await _chapter_numbers(db, sid)
+    assert chapters[0][0] == c2  # i3's chapter is now first
+    assert (await _page_order(db, c2))[0] == i3  # i3 first in its chapter
+
+
+@pytest.mark.asyncio
+async def test_list_pages_flags_gaps_from_stated_ranges(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C9")
+    await svc.create_chapter(sid, stated_page_start=1, stated_page_end=4)
+    await svc.create_chapter(sid, stated_page_start=9, stated_page_end=12)
+    out = await svc.list_pages(sid)
+    assert len(out["chapters"]) == 2
+    assert len(out["gaps"]) == 1
+    assert out["gaps"][0]["start"] == 5
+    assert out["gaps"][0]["end"] == 8
+
+
+@pytest.mark.asyncio
+async def test_add_with_stated_pages_records_them(db):
+    svc = SeriesService(db)
+    sid = await _series(db, "C10")
+    i1, i2 = await _img(db), await _img(db)
+    await svc.add_images(sid, [i1, i2], stated_pages={i1: 9, i2: 10})
+    rows = dict(
+        (
+            await db.execute(
+                select(SeriesPage.image_id, SeriesPage.stated_page)
+                .where(SeriesPage.series_tag_id == sid)
+            )
+        ).all()
+    )
+    assert rows[i1] == 9
+    assert rows[i2] == 10

tests/test_series_model.py

@@ -3,6 +3,7 @@ from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 
 from backend.app.models import ImageRecord, TagKind
+from backend.app.models.series_chapter import SeriesChapter
 from backend.app.models.series_page import SeriesPage
 from backend.app.services.tag_service import TagService
 
@@ -24,10 +25,20 @@ async def _img(db):
 
 
 @pytest.mark.asyncio
+async def _chapter(db, series_tag_id):
+    ch = SeriesChapter(series_tag_id=series_tag_id, chapter_number=1)
+    db.add(ch)
+    await db.flush()
+    return ch.id
+
+
 async def test_series_page_roundtrip_and_unique_image(db):
     s = await TagService(db).find_or_create("Vol1", TagKind.series)
     i1 = await _img(db)
-    db.add(SeriesPage(series_tag_id=s.id, image_id=i1, page_number=1))
+    ch = await _chapter(db, s.id)
+    db.add(SeriesPage(
+        series_tag_id=s.id, chapter_id=ch, image_id=i1, page_number=1
+    ))
     await db.flush()
     got = await db.scalar(
         select(SeriesPage.page_number).where(SeriesPage.image_id == i1)
@@ -35,6 +46,9 @@ async def test_series_page_roundtrip_and_unique_image(db):
     assert got == 1
     # UNIQUE image_id: same image can't be a page twice (any series).
     s2 = await TagService(db).find_or_create("Vol2", TagKind.series)
-    db.add(SeriesPage(series_tag_id=s2.id, image_id=i1, page_number=1))
+    ch2 = await _chapter(db, s2.id)
+    db.add(SeriesPage(
+        series_tag_id=s2.id, chapter_id=ch2, image_id=i1, page_number=1
+    ))
     with pytest.raises(IntegrityError):
         await db.flush()

tests/test_tag_merge.py

@@ -414,16 +414,27 @@ async def test_alias_create_does_not_clobber_existing(db):
 
 @pytest.mark.asyncio
 async def test_merge_repoints_series_pages(db):
+    from backend.app.models.series_chapter import SeriesChapter
     from backend.app.models.series_page import SeriesPage
 
     svc = TagService(db)
     a = await svc.find_or_create("SerA", TagKind.series)
     b = await svc.find_or_create("SerB", TagKind.series)
     img = await _img(db)
-    db.add(SeriesPage(series_tag_id=a.id, image_id=img, page_number=1))
+    cha = SeriesChapter(series_tag_id=a.id, chapter_number=1)
+    db.add(cha)
+    await db.flush()
+    db.add(SeriesPage(
+        series_tag_id=a.id, chapter_id=cha.id, image_id=img, page_number=1
+    ))
     await db.flush()
     await svc.merge(a.id, b.id)
     owner = await db.scalar(
         select(SeriesPage.series_tag_id).where(SeriesPage.image_id == img)
     )
     assert owner == b.id
+    # The chapter moved to the survivor too, so the page's FK stayed valid.
+    ch_owner = await db.scalar(
+        select(SeriesChapter.series_tag_id).where(SeriesChapter.id == cha.id)
+    )
+    assert ch_owner == b.id