feat(series): flat series sequence + cosmetic chapter dividers (#789 Phase 1)

Reframe a series from "ordered chapters that own pages" to ONE flat,
series-global ordered run of pages with optional cosmetic chapter DIVIDERS
over it. A chapter no longer wraps content — it's a labeled divider anchored
to the page that begins it; a page's chapter is derived as the nearest
preceding divider. This is what lets installments assembled from multiple
sources sit in one continuous, correctly-numbered sequence (operator's
Goblin Juice case).

- migration 0047: flatten each series to a series-global page_number
  (preserving today's reading order); convert each existing chapter to a
  divider anchored at its first page (keeping title/stated_part); drop
  series_page.chapter_id; reshape series_chapter (anchor_page_id UNIQUE FK,
  drop chapter_number/is_placeholder/stated_page_start/end). Loss-safe for
  content; drops empty placeholder chapters + a redundant page-1 divider.
- series_page: page_number is now the series-global order; no chapter_id.
- series_chapter: anchored divider (anchor_page_id, title, stated_part).
- series_service: flat list_pages (one run + derived dividers + per-page
  source_post + part_gaps), series-wide reorder/renumber, divider CRUD
  (create/update/move/delete); retired per-chapter reorder/merge/placement.
- api/tags: drop chapter_id from add; /chapters endpoints are divider
  create/update/delete (removed chapter reorder/merge/page-reorder).
- series_match_service: series "end" reads max(series_page.stated_page);
  accept appends via add_post. tag_service series-merge appends src's pages
  after tgt's max so the merged series stays one clean run.
- frontend: seriesManage store + SeriesManageView → one continuous
  drag-reorder grid with inline divider bars + series-global page numbers;
  reader walks the flat run, headings from dividers; PostSeriesMenu copy.
- tests reworked across the series suite for the divider model.

Phase 2 (pending staging for add-from-post) is separate.

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

backend/app/api/tags.py

@@ -391,14 +391,9 @@ 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, chapter_id=chapter_id
-            )
+            n = await SeriesService(session).add_images(tag_id, ids)
         except SeriesError as exc:
             return _series_err(exc)
         await session.commit()
@@ -453,30 +448,30 @@ async def series_cover(tag_id: int):
     return jsonify({"ok": True})
 
 
-# ---- chapters (FC-6.1) ----------------------------------------------------
+# ---- chapter dividers (FC-6.x) -------------------------------------------
+# A chapter is a cosmetic divider anchored to the page that begins it; it owns
+# no pages. Page ordering is series-wide (the /reorder endpoint above), so there
+# is no per-chapter reorder/merge/chapter-reorder — those are gone.
 
 
 @tags_bp.route("/series/<int:tag_id>/chapters", methods=["POST"])
 async def series_chapter_create(tag_id: int):
     body = await request.get_json() or {}
+    anchor, aerr = _opt_int(body, "anchor_image_id")
+    if aerr:
+        return aerr
+    if anchor is None:
+        return jsonify({"error": "anchor_image_id required"}), 400
     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
+    part, perr = _opt_int(body, "stated_part")
+    if perr:
+        return perr
     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,
+            ch = await SeriesService(session).create_divider(
+                tag_id, anchor, title=title, stated_part=part,
             )
         except SeriesError as exc:
             return _series_err(exc)
@@ -484,21 +479,6 @@ async def series_chapter_create(tag_id: int):
     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"]
 )
@@ -514,19 +494,18 @@ async def series_chapter_update(tag_id: int, chapter_id: int):
         if perr:
             return perr
         kwargs.update(set_part=True, stated_part=part)
-    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)
+    if "anchor_image_id" in body:
+        anchor, aerr = _opt_int(body, "anchor_image_id")
+        if aerr:
+            return aerr
+        if anchor is None:
+            return jsonify(
+                {"error": "anchor_image_id must be an integer"}
+            ), 400
+        kwargs.update(set_anchor=True, anchor_image_id=anchor)
     async with get_session() as session:
         try:
-            await SeriesService(session).update_chapter(
+            await SeriesService(session).update_divider(
                 tag_id, chapter_id, **kwargs
             )
         except SeriesError as exc:
@@ -541,45 +520,7 @@ async def series_chapter_update(tag_id: int, chapter_id: int):
 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)
+            await SeriesService(session).delete_divider(tag_id, chapter_id)
         except SeriesError as exc:
             return _series_err(exc)
         await session.commit()
@@ -635,7 +576,7 @@ async def series_add_post(tag_id: int):
         return jsonify({"error": "post_id required"}), 400
     async with get_session() as session:
         try:
-            out = await SeriesService(session).add_post_as_chapter(tag_id, post_id)
+            out = await SeriesService(session).add_post(tag_id, post_id)
         except SeriesError as exc:
             return _series_err(exc)
         await session.commit()

backend/app/models/series_chapter.py

@@ -1,28 +1,22 @@
-"""SeriesChapter — an ordered chapter/part within a series.
+"""SeriesChapter — a cosmetic chapter DIVIDER within a series (FC-6.x reframe).
 
-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.
+A series is ONE flat, series-global ordered run of SeriesPages. A chapter is NOT
+a container — it owns no pages. It is a labeled divider anchored to the page that
+BEGINS the chapter (anchor_page_id → series_page): "a new chapter starts here."
+A page's chapter is derived at read time as the nearest preceding divider.
 
-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.
+Dividers never affect page ordering or the series-global page numbers; they stay
+pinned to their anchor page across reorders. anchor_page_id is UNIQUE — at most
+one chapter begins at a given page — and FK-cascades, so removing the anchor page
+from the series drops the divider (the chapter merges into the preceding run).
 
-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.
-
-stated_part is the operator-facing "Part N" label (FC-6.4), separate from the
-positional chapter_number: chapter_number is auto-managed ordering (rewritten
-1..N on reorder/delete), while stated_part is the real installment number the
-operator types — e.g. a series authored from a post that is Part 2 of a story.
-Nullable when unset (the UI then falls back to showing chapter_number).
+title is the optional chapter name; stated_part is the optional operator-facing
+"Part N" label (shown instead of a derived ordinal when set).
 """
 
 from datetime import datetime
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, Text, func
+from sqlalchemy import DateTime, ForeignKey, Integer, Text, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from .base import Base
@@ -35,14 +29,13 @@ class SeriesChapter(Base):
     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)
-    stated_part: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    title: Mapped[str | None] = mapped_column(Text, nullable=True)
-    is_placeholder: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, server_default="false"
+    anchor_page_id: Mapped[int] = mapped_column(
+        ForeignKey("series_page.id", ondelete="CASCADE"),
+        nullable=False,
+        unique=True,
     )
-    stated_page_start: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    stated_page_end: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    title: Mapped[str | None] = mapped_column(Text, nullable=True)
+    stated_part: 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/models/series_page.py

@@ -1,12 +1,15 @@
 """SeriesPage — ordered image membership for a series-kind Tag.
 
-A series IS a Tag with kind='series'; series_page gives it ordered pages,
-grouped into chapters (FC-6). An image belongs to at most one series
-(UNIQUE image_id) ⇒ at most one chapter. Reading order is
-(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.
+A series IS a Tag with kind='series'; series_page gives it a SINGLE flat,
+series-global ordered run of pages (FC-6.x divider reframe). An image belongs to
+at most one series (UNIQUE image_id). Reading order is `page_number` alone — a
+series-wide ordering key (not unique), rewritten 1..N wholesale on reorder so a
+reorder can't transiently collide on an index.
+
+Chapters are cosmetic DIVIDERS anchored to a page (see SeriesChapter); they do
+NOT own pages, so there is no chapter_id here — a page's chapter is derived at
+read time as the nearest preceding divider. stated_page carries the printed page
+number parsed from the source post, nullable when unknown.
 """
 
 from datetime import datetime
@@ -24,11 +27,6 @@ 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,

backend/app/services/series_match_service.py

@@ -18,7 +18,6 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import ImageRecord, Post, Tag, TagKind
-from ..models.series_chapter import SeriesChapter
 from ..models.series_page import SeriesPage
 from ..models.series_suggestion import SeriesSuggestion
 from ..models.tag import image_tag
@@ -144,9 +143,11 @@ class SeriesMatchService:
         return [t for t in rows if t]
 
     async def _series_max_stated_end(self, series_tag_id: int) -> int | None:
+        # Highest printed page across the series' pages (FC-6.x: stated pages
+        # live on series_page now; chapters are cosmetic dividers).
         return await self.session.scalar(
-            select(func.max(SeriesChapter.stated_page_end)).where(
-                SeriesChapter.series_tag_id == series_tag_id
+            select(func.max(SeriesPage.stated_page)).where(
+                SeriesPage.series_tag_id == series_tag_id
             )
         )
 
@@ -288,7 +289,7 @@ class SeriesMatchService:
             raise SeriesError(f"suggestion {suggestion_id} not found")
         if s.status != "pending":
             raise SeriesError(f"suggestion {suggestion_id} is already {s.status}")
-        out = await SeriesService(self.session).add_post_as_chapter(
+        out = await SeriesService(self.session).add_post(
             s.series_tag_id, s.post_id
         )
         s.status = "added"

backend/app/services/series_service.py

@@ -1,14 +1,19 @@
-"""Series = a series-kind Tag + ordered chapters, each holding ordered pages.
+"""Series = a series-kind Tag + ONE flat, series-global ordered run of pages,
+with optional cosmetic chapter DIVIDERS over it (FC-6.x reframe, #789).
 
-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).
+Reading order is series_page.page_number alone (rewritten 1..N wholesale on any
+reorder; page_number is not unique, so a reorder never transiently collides). A
+chapter is NOT a container — it is a labeled divider anchored to the page that
+begins it (series_chapter.anchor_page_id); a page's chapter is derived as the
+nearest preceding divider. An image lives in at most one series
+(series_page.image_id is UNIQUE).
 
 All mutations are Core/set-based and run in the request transaction.
 """
 
-from sqlalchemy import and_, func, select, update
+from sqlalchemy import and_, func, select, text, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import Artist, ImageRecord, Post, Tag, TagKind
@@ -51,12 +56,56 @@ class SeriesService:
                 out.append(int(x))
         return out
 
-    async def _require_chapter(self, series_tag_id: int, chapter_id: int):
+    async def _page_order(self, series_tag_id: int) -> list[int]:
+        """The series' image_ids in reading order (series-global page_number)."""
+        rows = (
+            await self.session.execute(
+                select(SeriesPage.image_id)
+                .where(SeriesPage.series_tag_id == series_tag_id)
+                .order_by(SeriesPage.page_number.asc(), SeriesPage.id.asc())
+            )
+        ).scalars().all()
+        return list(rows)
+
+    async def _page_id_for(self, series_tag_id: int, image_id: int) -> int:
+        """series_page.id for an image in the series (raises if not a member)."""
+        pid = await self.session.scalar(
+            select(SeriesPage.id).where(
+                and_(
+                    SeriesPage.series_tag_id == series_tag_id,
+                    SeriesPage.image_id == image_id,
+                )
+            )
+        )
+        if pid is None:
+            raise SeriesError(f"image {image_id} is not in this series")
+        return pid
+
+    async def _renumber(self, series_tag_id: int) -> None:
+        """Compact page_number to 1..N preserving current order (set-based)."""
+        await self.session.execute(
+            text(
+                """
+                WITH ordered AS (
+                    SELECT id, ROW_NUMBER() OVER (
+                        ORDER BY page_number, id
+                    ) AS rn
+                    FROM series_page WHERE series_tag_id = :sid
+                )
+                UPDATE series_page sp SET page_number = ordered.rn
+                FROM ordered
+                WHERE sp.id = ordered.id AND sp.page_number <> ordered.rn
+                """
+            ),
+            {"sid": series_tag_id},
+        )
+
+    async def _require_divider(self, series_tag_id: int, divider_id: int):
         row = (
             await self.session.execute(
                 select(SeriesChapter).where(
                     and_(
-                        SeriesChapter.id == chapter_id,
+                        SeriesChapter.id == divider_id,
                         SeriesChapter.series_tag_id == series_tag_id,
                     )
                 )
@@ -64,94 +113,32 @@ class SeriesService:
         ).scalar_one_or_none()
         if row is None:
             raise SeriesError(
-                f"chapter {chapter_id} is not in series {series_tag_id}"
+                f"chapter {divider_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.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."""
+    def _part_gaps(dividers: list[dict]) -> list[dict]:
+        """Missing-Part hints between consecutive dividers whose stated_part
+        numbers jump by more than 1 (e.g. Part 1 then Part 3). Only dividers
+        that carry a stated_part participate."""
         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
-
-    @staticmethod
-    def _part_gaps(chapters: list[dict]) -> list[dict]:
-        """Missing-Part gaps between consecutive chapters whose stated_part
-        numbers jump by more than 1 (e.g. a series with Part 1 and Part 3, or one
-        authored straight from a Part 2 post). Mirrors _gaps but on stated_part —
-        only chapters that actually carry a stated_part participate."""
-        out: list[dict] = []
-        prev = None
-        for ch in chapters:
-            cur = ch["stated_part"]
+        for d in dividers:
+            cur = d["stated_part"]
             if cur is None:
                 continue
             if prev is not None and cur > prev["stated_part"] + 1:
                 out.append(
                     {
-                        "after_chapter_id": prev["id"],
+                        "after_divider_id": prev["id"],
                         "start": prev["stated_part"] + 1,
                         "end": cur - 1,
                     }
                 )
-            prev = ch
+            prev = d
         return out
 
     async def list_pages(self, series_tag_id: int) -> dict:
@@ -159,13 +146,6 @@ class SeriesService:
         rows = (
             await self.session.execute(
                 select(
-                    SeriesChapter.id.label("chapter_id"),
-                    SeriesChapter.chapter_number,
-                    SeriesChapter.stated_part,
-                    SeriesChapter.title,
-                    SeriesChapter.is_placeholder,
-                    SeriesChapter.stated_page_start,
-                    SeriesChapter.stated_page_end,
                     SeriesPage.image_id,
                     SeriesPage.page_number,
                     SeriesPage.stated_page,
@@ -176,71 +156,62 @@ class SeriesService:
                     ImageRecord.primary_post_id,
                     Post.post_title,
                 )
-                .select_from(SeriesChapter)
-                .outerjoin(SeriesPage, SeriesPage.chapter_id == SeriesChapter.id)
-                .outerjoin(ImageRecord, ImageRecord.id == SeriesPage.image_id)
+                .select_from(SeriesPage)
+                .join(ImageRecord, ImageRecord.id == SeriesPage.image_id)
                 .outerjoin(Post, Post.id == ImageRecord.primary_post_id)
-                .where(SeriesChapter.series_tag_id == series_tag_id)
-                .order_by(
-                    SeriesChapter.chapter_number.asc(),
-                    SeriesPage.page_number.asc(),
-                )
+                .where(SeriesPage.series_tag_id == series_tag_id)
+                .order_by(SeriesPage.page_number.asc(), SeriesPage.id.asc())
             )
         ).all()
-
-        chapters: list[dict] = []
-        flat: list[dict] = []
-        by_id: dict[int, dict] = {}
-        # chapter_id -> {post_id: title} seen across its pages, so we can label a
-        # chapter with its source post when all its pages come from one post.
-        posts_seen: dict[int, dict[int, str | None]] = {}
-        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,
-                    "stated_part": r.stated_part,
-                    "title": r.title,
-                    "is_placeholder": r.is_placeholder,
-                    "stated_page_start": r.stated_page_start,
-                    "stated_page_end": r.stated_page_end,
-                    "source_post": None,
-                    "pages": [],
-                }
-                by_id[r.chapter_id] = ch
-                posts_seen[r.chapter_id] = {}
-                chapters.append(ch)
-            if r.image_id is None:
-                continue  # placeholder / empty chapter
-            if r.primary_post_id is not None:
-                posts_seen[r.chapter_id][r.primary_post_id] = r.post_title
-            page = {
+        pages = [
+            {
                 "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),
+                "thumbnail_url": thumbnail_url(
+                    r.thumbnail_path, r.sha256, r.mime
+                ),
                 "image_url": f"/images/{r.path.split('/images/', 1)[-1]}",
+                "source_post": (
+                    {"id": r.primary_post_id, "title": r.post_title}
+                    if r.primary_post_id is not None
+                    else None
+                ),
             }
-            ch["pages"].append(page)
-            flat.append(page)
+            for r in rows
+        ]
 
-        # A chapter's source_post is set only when every page shares one post —
-        # the common case (a series authored from a post). Mixed chapters stay
-        # null rather than guessing.
-        for ch in chapters:
-            seen = posts_seen.get(ch["id"], {})
-            if len(seen) == 1:
-                pid, title = next(iter(seen.items()))
-                ch["source_post"] = {"id": pid, "title": title}
+        # Dividers, ordered by the global position of their anchor page.
+        drows = (
+            await self.session.execute(
+                select(
+                    SeriesChapter.id,
+                    SeriesChapter.title,
+                    SeriesChapter.stated_part,
+                    SeriesPage.image_id.label("anchor_image_id"),
+                    SeriesPage.page_number,
+                )
+                .join(SeriesPage, SeriesPage.id == SeriesChapter.anchor_page_id)
+                .where(SeriesChapter.series_tag_id == series_tag_id)
+                .order_by(SeriesPage.page_number.asc())
+            )
+        ).all()
+        dividers = [
+            {
+                "id": r.id,
+                "anchor_image_id": r.anchor_image_id,
+                "page_number": r.page_number,
+                "title": r.title,
+                "stated_part": r.stated_part,
+            }
+            for r in drows
+        ]
 
         return {
             "series": {"id": tag.id, "name": tag.name},
-            "chapters": chapters,
-            "pages": flat,  # back-compat: flat reading order across chapters
-            "gaps": self._gaps(chapters),
-            "part_gaps": self._part_gaps(chapters),
+            "pages": pages,
+            "dividers": dividers,
+            "part_gaps": self._part_gaps(dividers),
         }
 
     # ---- pages ------------------------------------------------------------
@@ -249,19 +220,13 @@ class SeriesService:
         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)."""
+        """Append images as pages at the END of the series' flat run. 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(
@@ -273,15 +238,16 @@ class SeriesService:
         to_add = [i for i in ids if existing.get(i) != series_tag_id]
         if not to_add:
             return 0
-        # Move: drop any prior membership for these images (UNIQUE image_id).
+        # Move: drop any prior membership for these images (UNIQUE image_id). A
+        # dropped page that anchored a divider in another series cascades it away.
         await self.session.execute(
             SeriesPage.__table__.delete().where(SeriesPage.image_id.in_(to_add))
         )
         max_pn = (
             await self.session.scalar(
-                select(func.coalesce(func.max(SeriesPage.page_number), 0)).where(
-                    SeriesPage.chapter_id == chapter_id
-                )
+                select(
+                    func.coalesce(func.max(SeriesPage.page_number), 0)
+                ).where(SeriesPage.series_tag_id == series_tag_id)
             )
         ) or 0
         sp = stated_pages or {}
@@ -290,7 +256,6 @@ class SeriesService:
                 [
                     {
                         "series_tag_id": series_tag_id,
-                        "chapter_id": chapter_id,
                         "image_id": iid,
                         "page_number": max_pn + offset,
                         "stated_page": sp.get(iid),
@@ -314,220 +279,134 @@ class SeriesService:
                 )
             )
         )
+        # Dividers anchored to a removed page cascade away (chapter merges into
+        # the preceding run); compact the remaining pages back to 1..N.
+        await self._renumber(series_tag_id)
         return res.rowcount or 0
 
-    async def reorder_pages(
-        self, series_tag_id: int, chapter_id: int, ordered_image_ids: list[int]
+    async def reorder(
+        self, series_tag_id: int, ordered_image_ids: list[int]
     ) -> None:
+        """Series-wide reorder. ordered_image_ids must be exactly the series'
+        pages; page_number is rewritten 1..N in that order."""
         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._chapter_page_order(chapter_id))
+        current = set(await self._page_order(series_tag_id))
         if set(ordered) != current or len(ordered) != len(current):
             raise SeriesError(
-                "ordered image_ids must exactly match the chapter's pages"
+                "ordered image_ids must exactly match the series' pages"
             )
         for idx, iid in enumerate(ordered, start=1):
             await self.session.execute(
                 update(SeriesPage)
                 .where(
                     and_(
-                        SeriesPage.chapter_id == chapter_id,
+                        SeriesPage.series_tag_id == series_tag_id,
                         SeriesPage.image_id == iid,
                     )
                 )
                 .values(page_number=idx)
             )
 
-    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."""
+    async def set_cover(self, series_tag_id: int, image_id: int) -> None:
+        """Cover = the first page. Move the image to the front of the series."""
         await self._require_series(series_tag_id)
-        chapters = await self._chapter_order(series_tag_id)
-        if len(chapters) > 1:
-            raise SeriesError(
-                "series has multiple chapters; reorder a specific chapter"
+        pages = await self._page_order(series_tag_id)
+        if image_id not in pages:
+            raise SeriesError(f"image {image_id} is not in this series")
+        if pages and pages[0] != image_id:
+            await self.reorder(
+                series_tag_id,
+                [image_id] + [p for p in pages if p != image_id],
             )
-        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 ---------------------------------------------------------
+    # ---- chapter dividers -------------------------------------------------
 
-    async def create_chapter(
+    async def create_divider(
         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,
+        anchor_image_id: int,
         *,
         title: str | None = None,
         stated_part: int | None = None,
-        stated_page_start: int | None = None,
-        stated_page_end: int | None = None,
+    ) -> dict:
+        """Place a chapter divider before the page for anchor_image_id ("a new
+        chapter starts here"). One divider per page (UNIQUE anchor)."""
+        await self._require_series(series_tag_id)
+        page_id = await self._page_id_for(series_tag_id, anchor_image_id)
+        try:
+            async with self.session.begin_nested():
+                res = await self.session.execute(
+                    pg_insert(SeriesChapter)
+                    .values(
+                        series_tag_id=series_tag_id,
+                        anchor_page_id=page_id,
+                        title=title,
+                        stated_part=stated_part,
+                    )
+                    .returning(SeriesChapter.id)
+                )
+                divider_id = res.scalar_one()
+        except IntegrityError as exc:
+            raise SeriesError(
+                "a chapter divider already starts at that page"
+            ) from exc
+        return {"id": divider_id, "anchor_image_id": anchor_image_id}
+
+    async def update_divider(
+        self,
+        series_tag_id: int,
+        divider_id: int,
+        *,
+        title: str | None = None,
+        stated_part: int | None = None,
+        anchor_image_id: int | None = None,
         set_title: bool = False,
         set_part: bool = False,
-        set_start: bool = False,
-        set_end: bool = False,
+        set_anchor: 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 or part)."""
+        """Partial divider edit. set_* flags say which fields to write (so None
+        can be written explicitly, e.g. clearing a stated_part). set_anchor moves
+        the divider to sit before a different page."""
         await self._require_series(series_tag_id)
-        await self._require_chapter(series_tag_id, chapter_id)
+        await self._require_divider(series_tag_id, divider_id)
         values: dict = {}
         if set_title:
             values["title"] = title
         if set_part:
             values["stated_part"] = stated_part
-        if set_start:
-            values["stated_page_start"] = stated_page_start
-        if set_end:
-            values["stated_page_end"] = stated_page_end
+        if set_anchor:
+            if anchor_image_id is None:
+                raise SeriesError("anchor_image_id required to move a divider")
+            values["anchor_page_id"] = await self._page_id_for(
+                series_tag_id, anchor_image_id
+            )
         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):
+        try:
+            async with self.session.begin_nested():
+                await self.session.execute(
+                    update(SeriesChapter)
+                    .where(SeriesChapter.id == divider_id)
+                    .values(**values)
+                )
+        except IntegrityError as exc:
             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)
-            )
+                "a chapter divider already starts at that page"
+            ) from exc
 
-    async def delete_chapter(self, series_tag_id: int, chapter_id: int) -> None:
+    async def delete_divider(
+        self, series_tag_id: int, divider_id: int
+    ) -> None:
+        """Remove a divider — the chapter merges into the run above it. Pages
+        are untouched (dividers own nothing)."""
         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._require_divider(series_tag_id, divider_id)
         await self.session.execute(
             SeriesChapter.__table__.delete().where(
-                SeriesChapter.id == chapter_id
+                SeriesChapter.id == divider_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")
-        chapter_id = row
-        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)
 
     # ---- post → series (FC-6.2) ------------------------------------------
 
@@ -553,8 +432,8 @@ class SeriesService:
 
     async def promote_post_to_series(self, post_id: int) -> dict:
         """Case 1: a self-contained multi-image post becomes its own series.
-        Creates a series tag named after the post, one chapter, the post's
-        images as its pages (stated pages parsed from the post when present)."""
+        Creates a series tag named after the post and the post's images as its
+        flat pages (stated pages parsed from the post when present)."""
         from .tag_service import TagService
 
         post = await self.session.get(Post, post_id)
@@ -566,25 +445,19 @@ class SeriesService:
         name = (post.post_title or f"Series from post {post_id}").strip()[:200]
         tag = await TagService(self.session).find_or_create(name, TagKind.series)
         rng = parse_page_range(f"{post.post_title or ''} {post.description or ''}")
-        start, end = rng if rng else (None, None)
-        ch = await self.create_chapter(
-            tag.id, stated_page_start=start, stated_page_end=end,
-        )
+        start = rng[0] if rng else None
         added = await self.add_images(
-            tag.id, image_ids, chapter_id=ch["id"],
-            stated_pages=self._stated_map(image_ids, start),
+            tag.id, image_ids, stated_pages=self._stated_map(image_ids, start),
         )
         return {
-            "series_tag_id": tag.id, "name": tag.name,
-            "chapter_id": ch["id"], "added": added,
+            "series_tag_id": tag.id, "name": tag.name, "added": added,
         }
 
-    async def add_post_as_chapter(
-        self, series_tag_id: int, post_id: int
-    ) -> dict:
-        """Case 2: append a post as the next chapter of an existing series. The
-        chapter is titled after the post and slotted by parsed page number when
-        present (else appended at the end)."""
+    async def add_post(self, series_tag_id: int, post_id: int) -> dict:
+        """Case 2: append a post's images to the end of an existing series' flat
+        run (stated pages parsed from the post when present). No chapter wrapping
+        — the operator orders/labels afterward. (Phase 2 will stage these as
+        pending instead of appending directly.)"""
         await self._require_series(series_tag_id)
         post = await self.session.get(Post, post_id)
         if post is None:
@@ -593,89 +466,55 @@ class SeriesService:
         if not image_ids:
             raise SeriesError("post has no images to add")
         rng = parse_page_range(f"{post.post_title or ''} {post.description or ''}")
-        start, end = rng if rng else (None, None)
-        ch = await self.create_chapter(
-            series_tag_id, title=post.post_title or None,
-            stated_page_start=start, stated_page_end=end,
-        )
+        start = rng[0] if rng else None
         added = await self.add_images(
-            series_tag_id, image_ids, chapter_id=ch["id"],
+            series_tag_id, image_ids,
             stated_pages=self._stated_map(image_ids, start),
         )
-        if start is not None:
-            await self._place_chapter_by_stated(series_tag_id, ch["id"], start)
-        return {
-            "series_tag_id": series_tag_id, "chapter_id": ch["id"],
-            "added": added,
-        }
-
-    async def _place_chapter_by_stated(
-        self, series_tag_id: int, chapter_id: int, start: int
-    ) -> None:
-        """Move `chapter_id` to sit before the first chapter whose stated start
-        is higher — so a post stating pages 5-8 lands ahead of the 9-12 chapter."""
-        rows = (
-            await self.session.execute(
-                select(SeriesChapter.id, SeriesChapter.stated_page_start)
-                .where(SeriesChapter.series_tag_id == series_tag_id)
-                .order_by(SeriesChapter.chapter_number.asc())
-            )
-        ).all()
-        current = [r.id for r in rows]
-        others = [r for r in rows if r.id != chapter_id]
-        insert_at = len(others)
-        for idx, r in enumerate(others):
-            if r.stated_page_start is not None and r.stated_page_start > start:
-                insert_at = idx
-                break
-        new_order = [r.id for r in others]
-        new_order.insert(insert_at, chapter_id)
-        if new_order != current:
-            await self.reorder_chapters(series_tag_id, new_order)
+        return {"series_tag_id": series_tag_id, "added": added}
 
     # ---- browse list (FC-6.2) --------------------------------------------
 
     async def list_series(
         self, *, sort: str = "recent", artist_id: int | None = None
     ) -> list[dict]:
-        """Series cards for the browse view: cover thumb, name, artist, chapter
-        + page counts, a gap flag, and last-updated. sort ∈ recent|name|size."""
-        # Page counts + most-recent activity per series.
+        """Series cards for the browse view: cover thumb, name, artist, divider
+        + page counts, a missing-Part flag, and last-updated. sort ∈
+        recent|name|size."""
+        # Page counts + most-recent page activity per series.
         page_rows = (
             await self.session.execute(
                 select(
                     SeriesPage.series_tag_id,
                     func.count().label("pages"),
+                    func.max(SeriesPage.updated_at).label("updated_at"),
                 ).group_by(SeriesPage.series_tag_id)
             )
         ).all()
         page_count = {r.series_tag_id: r.pages for r in page_rows}
+        updated_by_series = {r.series_tag_id: r.updated_at for r in page_rows}
 
-        ch_rows = (
+        # Dividers per series (for count + missing-Part gap), ordered by the
+        # global position of their anchor page.
+        drows = (
             await self.session.execute(
                 select(
                     SeriesChapter.series_tag_id,
                     SeriesChapter.id,
-                    SeriesChapter.chapter_number,
-                    SeriesChapter.stated_page_start,
-                    SeriesChapter.stated_page_end,
-                    SeriesChapter.updated_at,
+                    SeriesChapter.stated_part,
+                    SeriesPage.page_number,
                 )
+                .join(SeriesPage, SeriesPage.id == SeriesChapter.anchor_page_id)
                 .order_by(
-                    SeriesChapter.series_tag_id,
-                    SeriesChapter.chapter_number.asc(),
+                    SeriesChapter.series_tag_id, SeriesPage.page_number.asc()
                 )
             )
         ).all()
-        chapters_by_series: dict[int, list] = {}
-        updated_by_series: dict[int, object] = {}
-        for r in ch_rows:
-            chapters_by_series.setdefault(r.series_tag_id, []).append(r)
-            prev = updated_by_series.get(r.series_tag_id)
-            if prev is None or (r.updated_at and r.updated_at > prev):
-                updated_by_series[r.series_tag_id] = r.updated_at
+        dividers_by_series: dict[int, list] = {}
+        for r in drows:
+            dividers_by_series.setdefault(r.series_tag_id, []).append(r)
 
-        # Cover = the (chapter_number, page_number)-minimum page per series.
+        # Cover = the first page (min page_number) of each series.
         cover_q = (
             select(
                 SeriesPage.series_tag_id,
@@ -687,13 +526,11 @@ class SeriesService:
                 .over(
                     partition_by=SeriesPage.series_tag_id,
                     order_by=(
-                        SeriesChapter.chapter_number.asc(),
-                        SeriesPage.page_number.asc(),
+                        SeriesPage.page_number.asc(), SeriesPage.id.asc()
                     ),
                 )
                 .label("rn"),
             )
-            .join(SeriesChapter, SeriesChapter.id == SeriesPage.chapter_id)
             .join(ImageRecord, ImageRecord.id == SeriesPage.image_id)
             .subquery()
         )
@@ -737,20 +574,14 @@ class SeriesService:
                 cover is None or cover.artist_id != artist_id
             ):
                 continue
-            chs = chapters_by_series.get(t.id, [])
+            divs = dividers_by_series.get(t.id, [])
             gap = bool(
-                self._gaps(
-                    [
-                        {
-                            "id": c.id,
-                            "stated_page_start": c.stated_page_start,
-                            "stated_page_end": c.stated_page_end,
-                        }
-                        for c in chs
-                    ]
+                self._part_gaps(
+                    [{"id": d.id, "stated_part": d.stated_part} for d in divs]
                 )
             )
             artist = artist_by_id.get(cover.artist_id) if cover else None
+            updated = updated_by_series.get(t.id)
             out.append(
                 {
                     "id": t.id,
@@ -764,14 +595,10 @@ class SeriesService:
                     ),
                     "artist_name": artist.name if artist else None,
                     "artist_slug": artist.slug if artist else None,
-                    "chapter_count": len(chs),
+                    "chapter_count": len(divs),
                     "page_count": page_count.get(t.id, 0),
                     "has_gap": gap,
-                    "updated_at": (
-                        updated_by_series.get(t.id).isoformat()
-                        if updated_by_series.get(t.id)
-                        else None
-                    ),
+                    "updated_at": updated.isoformat() if updated else None,
                 }
             )
 

backend/app/services/tag_service.py

@@ -518,21 +518,30 @@ class TagService:
         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).
+        # Move src's dividers onto tgt. Their anchor_page_id stays valid — the
+        # anchored pages move too — and anchor is globally UNIQUE so no collision.
         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.
+        # Append src's pages after tgt's so the merged series stays ONE clean
+        # series-global run (image_id is UNIQUE, so the two page sets are
+        # disjoint). Offset src's page_number past tgt's current max.
+        tgt_max = (
+            await self.session.scalar(
+                select(
+                    func.coalesce(func.max(SeriesPage.page_number), 0)
+                ).where(SeriesPage.series_tag_id == tgt)
+            )
+        ) or 0
         await self.session.execute(
             update(SeriesPage)
             .where(SeriesPage.series_tag_id == src)
-            .values(series_tag_id=tgt)
+            .values(
+                series_tag_id=tgt,
+                page_number=SeriesPage.page_number + tgt_max,
+            )
         )
 
     async def _repoint_fandom_children(