feat(pixiv): ledger models + migration 0076 + PixivIngester adapter (#129 step 3)

pixiv_seen_media / pixiv_failed_media mirror the Patreon/SubscribeStar
ledgers (keys are always synthesized <illust_id>:p<num> / <illust_id>:ugoira
— pximg URLs carry no content hash). PixivIngester wires client/downloader/
ledgers into ingest_core with drift label 'Pixiv app API' and the new
body_canary=False opt-out: caption-less pixiv artists are common, so the
zero-bodies #862 alarm would false-positive here — the client's
response-shape drift checks cover that failure class instead. auth_token
joins the uniform adapter constructor (pixiv is the first token-auth native
platform). verify_pixiv_credential = one OAuth refresh, no feed walk.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM

alembic/versions/0076_pixiv_ledgers.py

@@ -0,0 +1,82 @@
+"""pixiv_seen_media + pixiv_failed_media: per-source ledgers
+
+Revision ID: 0076
+Revises: 0075
+Create Date: 2026-07-03
+
+Pixiv native ingester (milestone #129, gallery-dl → native-core migration).
+Mirrors the Patreon (0037/0038) and SubscribeStar (0054) ledger tables: a
+seen-ledger so routine walks skip already-ingested media (recovery bypasses
+it) and a dead-letter ledger so persistently-failing media stops re-burning
+backfill chunks. Pixiv URLs carry no content hash, so `filehash` is always the
+synthesized ``<illust_id>:p<num>`` / ``<illust_id>:ugoira`` key — String(128)
+matches the siblings. UNIQUE (source_id, filehash) is the upsert key on each.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0076"
+down_revision: Union[str, None] = "0075"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "pixiv_seen_media",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column(
+            "source_id",
+            sa.Integer,
+            sa.ForeignKey("source.id", ondelete="CASCADE"),
+            nullable=False,
+            index=True,
+        ),
+        sa.Column("filehash", sa.String(128), nullable=False),
+        sa.Column("post_id", sa.String(64), nullable=True),
+        sa.Column(
+            "seen_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+        sa.UniqueConstraint(
+            "source_id", "filehash", name="uq_pixiv_seen_media_source_id"
+        ),
+    )
+    op.create_table(
+        "pixiv_failed_media",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column(
+            "source_id",
+            sa.Integer,
+            sa.ForeignKey("source.id", ondelete="CASCADE"),
+            nullable=False,
+            index=True,
+        ),
+        sa.Column("filehash", sa.String(128), nullable=False),
+        sa.Column("attempts", sa.Integer, nullable=False, server_default="1"),
+        sa.Column("last_error", sa.Text, nullable=True),
+        sa.Column(
+            "first_failed_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+        sa.Column(
+            "last_failed_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+        sa.UniqueConstraint(
+            "source_id", "filehash", name="uq_pixiv_failed_media_source_id"
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("pixiv_failed_media")
+    op.drop_table("pixiv_seen_media")

backend/app/models/__init__.py

@@ -23,6 +23,8 @@ from .library_audit_run import LibraryAuditRun
 from .ml_settings import MLSettings
 from .patreon_failed_media import PatreonFailedMedia
 from .patreon_seen_media import PatreonSeenMedia
+from .pixiv_failed_media import PixivFailedMedia
+from .pixiv_seen_media import PixivSeenMedia
 from .post import Post
 from .post_attachment import PostAttachment
 from .series_chapter import SeriesChapter
@@ -48,6 +50,8 @@ __all__ = [
     "Credential",
     "PatreonFailedMedia",
     "PatreonSeenMedia",
+    "PixivFailedMedia",
+    "PixivSeenMedia",
     "SubscribeStarFailedMedia",
     "SubscribeStarSeenMedia",
     "Post",

backend/app/models/pixiv_failed_media.py

@@ -0,0 +1,45 @@
+"""PixivFailedMedia — per-source dead-letter ledger of Pixiv media that keeps
+failing to download/validate.
+
+Mirror of PatreonFailedMedia/SubscribeStarFailedMedia. Media that fails every
+walk (404'd pximg URL, deleted work, persistently-corrupt bytes) would
+otherwise re-error forever and re-burn backfill chunks. After ``attempts``
+reaches the dead-letter threshold the ingester skips it on routine
+tick/backfill walks (recovery still re-attempts). A later clean download
+clears the row.
+
+`filehash` is the same synthesized ``<illust_id>:p<num>`` /
+``<illust_id>:ugoira`` key the seen-ledger uses. UNIQUE (source_id, filehash)
+is the upsert key.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import ForeignKey, Integer, String, Text, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.types import DateTime
+
+from .base import Base
+
+
+class PixivFailedMedia(Base):
+    __tablename__ = "pixiv_failed_media"
+    __table_args__ = (
+        UniqueConstraint(
+            "source_id", "filehash", name="uq_pixiv_failed_media_source_id"
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    source_id: Mapped[int] = mapped_column(
+        ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    filehash: Mapped[str] = mapped_column(String(128), nullable=False)
+    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    first_failed_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    last_failed_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )

backend/app/models/pixiv_seen_media.py

@@ -0,0 +1,42 @@
+"""PixivSeenMedia — per-source ledger of Pixiv media already
+downloaded+processed.
+
+Mirror of PatreonSeenMedia/SubscribeStarSeenMedia for the Pixiv native
+ingester (replacing gallery-dl). One queryable row per (source, media) so
+routine walks skip media we've already ingested; recovery mode bypasses the
+ledger to re-walk.
+
+Pixiv original URLs carry no content hash, so `filehash` is always the
+synthesized ``<illust_id>:p<num>`` (page) / ``<illust_id>:ugoira`` (frame
+zip) key — stable across any URL-shape drift. String(128) matches the sibling
+ledgers.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.types import DateTime
+
+from .base import Base
+
+
+class PixivSeenMedia(Base):
+    __tablename__ = "pixiv_seen_media"
+    __table_args__ = (
+        # Dedup key the downloader upserts against: one ledger row per
+        # (source, media). A second sighting of the same media is a no-op.
+        UniqueConstraint(
+            "source_id", "filehash", name="uq_pixiv_seen_media_source_id"
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    source_id: Mapped[int] = mapped_column(
+        ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    filehash: Mapped[str] = mapped_column(String(128), nullable=False)
+    post_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    seen_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )

backend/app/services/ingest_core.py

@@ -90,6 +90,7 @@ class Ingester:
         platform: str,
         error_base: type[Exception],
         drift_label: str | None = None,
+        body_canary: bool = True,
     ):
         self.client = client
         self.downloader = downloader
@@ -105,6 +106,12 @@ class Ingester:
         # update"). Defaults to the platform name; adapters pass a richer phrase
         # (e.g. "Patreon API", "SubscribeStar markup").
         self._drift_label = drift_label or platform
+        # #862 canary opt-out: platforms whose posts legitimately have empty
+        # bodies across large samples (pixiv — caption-less artists are common)
+        # would false-positive the zero-bodies-means-drift alarm; their clients
+        # catch drift structurally (response-shape checks) instead. The
+        # "bodies X/N" summary line still surfaces the ratio either way.
+        self._body_canary = body_canary
 
     # -- public ------------------------------------------------------------
 
@@ -536,7 +543,11 @@ class Ingester:
         # creds") so the breakage screams instead of silently archiving empties.
         # Only reached on an otherwise-clean walk (timeout/stop/error returned
         # above), so it never masks a more specific failure.
-        if posts_recorded >= _CANARY_MIN_SAMPLE and posts_with_body == 0:
+        if (
+            self._body_canary
+            and posts_recorded >= _CANARY_MIN_SAMPLE
+            and posts_with_body == 0
+        ):
             msg = (
                 f"Post-body canary: extracted a body from 0 of {posts_recorded} "
                 "posts — Patreon's body field shape likely changed; the ingester "

backend/app/services/pixiv_ingester.py

@@ -0,0 +1,117 @@
+"""Native Pixiv ingester — the Pixiv ADAPTER over the platform-agnostic core
+(`ingest_core.Ingester`).
+
+Thin counterpart to patreon_ingester / subscribestar_ingester: wires the Pixiv
+client/downloader/ledger models/constraints/key into the core and supplies the
+Pixiv failure mapping. The modes (tick / backfill / recovery / recapture), the
+seen + dead-letter ledgers, cursor checkpointing, and the post-first capture
+all live in the core. `download_service.download_source` drives
+`PixivIngester.run` exactly as it drives the other two.
+
+`campaign_id` is the numeric pixiv user id (download_backends extracts it from
+the source URL — no network resolver). Auth is the operator's OAuth refresh
+token (the token-type Credential), passed as `auth_token` — pixiv is the first
+native platform authenticating by token rather than cookies, so the uniform
+constructor accepts both and ignores what it doesn't need.
+
+FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Callable
+from pathlib import Path
+
+from ..models import PixivFailedMedia, PixivSeenMedia
+from .ingest_core import DEAD_LETTER_THRESHOLD, Ingester
+from .pixiv_client import MediaItem, PixivAPIError, PixivClient
+from .pixiv_downloader import PixivDownloader
+
+__all__ = [
+    "DEAD_LETTER_THRESHOLD",
+    "PixivIngester",
+    "_ledger_key",
+    "verify_pixiv_credential",
+]
+
+log = logging.getLogger(__name__)
+
+_LEDGER_KEY_MAX = 128
+
+
+def _ledger_key(media: MediaItem) -> str:
+    """Stable per-media identity for the cross-run seen-ledger. Pixiv original
+    URLs carry no content hash, so the key is the page/zip identity scoped to
+    its work: `<illust_id>:p<num>` / `<illust_id>:ugoira`. Bounded to the
+    column width."""
+    if media.filehash:
+        return media.filehash
+    return f"{media.post_id}:{media.media_id}"[:_LEDGER_KEY_MAX]
+
+
+class PixivIngester(Ingester):
+    """Walk a pixiv user's works, download unseen originals, return a
+    `DownloadResult`. A thin adapter over `ingest_core.Ingester`; `client` /
+    `downloader` are injectable seams so unit tests run without network."""
+
+    def __init__(
+        self,
+        images_root: Path,
+        cookies_path: str | None,
+        session_factory: Callable[[], object],
+        *,
+        validate: bool = True,
+        rate_limit: float = 0.0,
+        request_sleep: float = 0.0,
+        auth_token: str | None = None,
+        client: PixivClient | None = None,
+        downloader: PixivDownloader | None = None,
+    ):
+        self.images_root = Path(images_root)
+        self.cookies_path = str(cookies_path) if cookies_path else None
+        resolved_client = (
+            client
+            if client is not None
+            else PixivClient(auth_token, request_sleep=request_sleep)
+        )
+        resolved_downloader = (
+            downloader
+            if downloader is not None
+            else PixivDownloader(
+                self.images_root, cookies_path, validate=validate, rate_limit=rate_limit,
+            )
+        )
+        super().__init__(
+            client=resolved_client,
+            downloader=resolved_downloader,
+            session_factory=session_factory,
+            seen_model=PixivSeenMedia,
+            failed_model=PixivFailedMedia,
+            seen_constraint="uq_pixiv_seen_media_source_id",
+            failed_constraint="uq_pixiv_failed_media_source_id",
+            ledger_key=_ledger_key,
+            platform="pixiv",
+            error_base=PixivAPIError,
+            # API_DRIFT message phrasing; the base Ingester._failure_result owns
+            # the auth/drift/HTTP→error_type mapping (shared across platforms).
+            drift_label="Pixiv app API",
+            # Captions are legitimately empty for many pixiv artists, so the
+            # zero-bodies #862 canary would false-positive here; the client's
+            # response-shape checks (missing `illusts` → drift) cover the same
+            # failure class structurally.
+            body_canary=False,
+        )
+
+
+async def verify_pixiv_credential(
+    auth_token: str | None,
+) -> tuple[bool | None, str]:
+    """Native Pixiv credential probe — one OAuth refresh via
+    PixivClient.verify_auth (the exact call that fails when the token is
+    bad; no feed walk). Returns the uniform `(ok, message)` contract so
+    download_backends.verify_source_credential treats it like the others."""
+    client = PixivClient(auth_token)
+    loop = asyncio.get_running_loop()
+    return await loop.run_in_executor(None, client.verify_auth)

tests/test_pixiv_ingester.py

@@ -0,0 +1,305 @@
+"""PixivIngester tests — the adapter's wiring over the shared core.
+
+The core's walk/skip/cursor/budget behavior is exercised exhaustively by
+test_patreon_ingester / test_subscribestar_native; these cover what is
+pixiv-SPECIFIC: the synthesized ledger key, the real Postgres pixiv ledgers
+(the upserts' constraint names must match migration 0076), the failure
+mapping including the rate-limit retry_after carry, and the #862 body-canary
+opt-out (caption-less pixiv feeds must not fail API_DRIFT).
+"""
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.orm import sessionmaker
+
+from backend.app.models import Artist, PixivFailedMedia, PixivSeenMedia, Source
+from backend.app.services.gallery_dl import ErrorType
+from backend.app.services.ingest_core import _CANARY_MIN_SAMPLE
+from backend.app.services.native_ingest_common import MediaOutcome, PostRecordOutcome
+from backend.app.services.pixiv_client import (
+    MediaItem,
+    PixivAPIError,
+    PixivAuthError,
+    PixivDriftError,
+)
+from backend.app.services.pixiv_ingester import PixivIngester, _ledger_key
+
+pytestmark = pytest.mark.integration
+
+
+# --- fakes ----------------------------------------------------------------
+
+
+def _media(post_id, num, *, filehash=None, media_id=None):
+    mid = media_id if media_id is not None else f"p{num}"
+    return MediaItem(
+        url=f"https://i.pximg.net/img-original/img/2026/07/01/00/00/00/{post_id}_{mid}.png",
+        filename=f"{post_id}_Work_{num:02d}.png",
+        kind="image",
+        filehash=filehash,
+        post_id=str(post_id),
+        media_id=mid,
+    )
+
+
+class _FakeClient:
+    """Stub PixivClient. `pages` is a list of (page_cursor, [posts]); each post
+    is (post_id, [MediaItem]). `raise_exc` trips a client-level failure;
+    `captions` toggles per-post body text (the canary input)."""
+
+    def __init__(self, pages, raise_exc=None, captions=True):
+        self._pages = pages
+        self._raise_exc = raise_exc
+        self._captions = captions
+
+    def iter_posts(self, campaign_id, cursor=None):
+        if self._raise_exc is not None:
+            raise self._raise_exc
+        for page_cursor, posts in self._pages:
+            for post_id, media in posts:
+                caption = f"<p>caption {post_id}</p>" if self._captions else ""
+                yield (
+                    {
+                        "id": post_id,
+                        "_media": media,
+                        "attributes": {
+                            "title": f"Work {post_id}",
+                            "post_type": "illust",
+                            "content": caption,
+                            "published_at": "2026-07-01T00:00:00+09:00",
+                        },
+                        "_work": {},
+                    },
+                    {},
+                    page_cursor,
+                )
+
+    def extract_media(self, post, included_index):
+        return post["_media"]
+
+    def post_meta(self, post):
+        return {"title": post.get("id"), "date": None}
+
+    @staticmethod
+    def post_is_gated(post):
+        return False
+
+    @staticmethod
+    def post_record_key(post):
+        pid = str(post.get("id") or "")
+        return (f"post:{pid}", pid) if pid else None
+
+
+class _FakeDownloader:
+    """Stub PixivDownloader. Honors the injected is_seen; media whose ledger
+    key is in `error` report a download failure."""
+
+    def __init__(self, tmp_path, error=None):
+        self.tmp_path = tmp_path
+        self.error = set(error or ())
+        self.download_calls = 0
+
+    def download_post(self, post, media_items, artist_slug, *, is_seen,
+                      should_stop=lambda: False, recapture=False):
+        outcomes = []
+        for m in media_items:
+            if should_stop():
+                break
+            if is_seen(m) and not recapture:
+                outcomes.append(
+                    MediaOutcome(media=m, status="skipped_seen", path=None, error=None)
+                )
+            elif _ledger_key(m) in self.error:
+                self.download_calls += 1
+                outcomes.append(
+                    MediaOutcome(media=m, status="error", path=None, error="403 pximg")
+                )
+            else:
+                self.download_calls += 1
+                p = self.tmp_path / m.filename
+                p.write_bytes(b"x")
+                outcomes.append(
+                    MediaOutcome(media=m, status="downloaded", path=p, error=None)
+                )
+        return outcomes
+
+    def write_post_record(self, post, artist_slug):
+        p = self.tmp_path / f"_post_{post.get('id')}.json"
+        p.write_text("{}")
+        attrs = post.get("attributes") or {}
+        body = attrs.get("content")
+        return PostRecordOutcome(
+            path=p,
+            post_type=attrs.get("post_type"),
+            title=attrs.get("title"),
+            body_chars=len(body) if isinstance(body, str) else 0,
+        )
+
+
+@pytest.fixture
+async def source_id(db):
+    artist = Artist(name="Pixland", slug="pixland")
+    db.add(artist)
+    await db.flush()
+    source = Source(
+        artist_id=artist.id, platform="pixiv",
+        url="https://www.pixiv.net/users/99", enabled=True, config_overrides={},
+    )
+    db.add(source)
+    await db.commit()
+    return source.id
+
+
+def _ingester(sync_engine, tmp_path, client, downloader):
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    return PixivIngester(
+        images_root=tmp_path, cookies_path=None, session_factory=factory,
+        client=client, downloader=downloader,
+    )
+
+
+def _run(ing, source_id, mode="tick"):
+    return ing.run(
+        source_id=source_id, campaign_id="99", artist_slug="pixland",
+        url="https://www.pixiv.net/users/99", mode=mode,
+    )
+
+
+def _seen_keys(sync_engine, source_id):
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    with factory() as s:
+        return set(s.execute(
+            select(PixivSeenMedia.filehash).where(
+                PixivSeenMedia.source_id == source_id
+            )
+        ).scalars().all())
+
+
+# --- ledger key -------------------------------------------------------------
+
+
+def test_ledger_key_shapes():
+    assert _ledger_key(_media(111, 0)) == "111:p0"
+    assert _ledger_key(_media(111, 4)) == "111:p4"
+    assert _ledger_key(_media(333, 0, media_id="ugoira")) == "333:ugoira"
+    assert _ledger_key(_media(1, 0, filehash="f" * 32)) == "f" * 32
+    long_key = _ledger_key(_media("9" * 200, 0))
+    assert len(long_key) <= 128
+
+
+# --- walk + ledgers ---------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_tick_downloads_and_marks_seen(source_id, sync_engine, tmp_path):
+    m0, m1 = _media(111, 0), _media(111, 1)
+    client = _FakeClient([(None, [(111, [m0, m1])])])
+    downloader = _FakeDownloader(tmp_path)
+    ing = _ingester(sync_engine, tmp_path, client, downloader)
+
+    result = _run(ing, source_id)
+
+    assert result.success is True
+    assert result.files_downloaded == 2
+    assert len(result.post_record_paths) == 1
+    # 2 media keys + the synthetic post key, in the pixiv-shaped key format.
+    assert _seen_keys(sync_engine, source_id) == {"111:p0", "111:p1", "post:111"}
+
+
+@pytest.mark.asyncio
+async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path):
+    m0 = _media(111, 0)
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    with factory() as s:
+        s.add(PixivSeenMedia(
+            source_id=source_id, filehash=_ledger_key(m0), post_id="111",
+        ))
+        s.commit()
+
+    client = _FakeClient([(None, [(111, [m0])])])
+    downloader = _FakeDownloader(tmp_path)
+    ing = _ingester(sync_engine, tmp_path, client, downloader)
+
+    result = _run(ing, source_id)
+    assert result.files_downloaded == 0
+    assert downloader.download_calls == 0
+
+
+@pytest.mark.asyncio
+async def test_failed_media_lands_in_dead_letter_ledger(
+    source_id, sync_engine, tmp_path
+):
+    m0 = _media(111, 0)
+    client = _FakeClient([(None, [(111, [m0])])])
+    downloader = _FakeDownloader(tmp_path, error={_ledger_key(m0)})
+    ing = _ingester(sync_engine, tmp_path, client, downloader)
+
+    result = _run(ing, source_id)
+    assert result.run_stats["per_item_failures"] == 1
+
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    with factory() as s:
+        row = s.execute(
+            select(PixivFailedMedia).where(
+                PixivFailedMedia.source_id == source_id
+            )
+        ).scalar_one()
+    assert row.filehash == "111:p0"
+    assert row.attempts == 1
+    assert "403" in row.last_error
+
+
+# --- failure mapping ----------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_auth_error_maps_to_auth_error(source_id, sync_engine, tmp_path):
+    client = _FakeClient([], raise_exc=PixivAuthError(
+        "rotate the token", status_code=400,
+    ))
+    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
+    result = _run(ing, source_id)
+    assert result.success is False
+    assert result.error_type == ErrorType.AUTH_ERROR
+
+
+@pytest.mark.asyncio
+async def test_drift_error_maps_to_api_drift(source_id, sync_engine, tmp_path):
+    client = _FakeClient([], raise_exc=PixivDriftError("no illusts list"))
+    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
+    result = _run(ing, source_id)
+    assert result.error_type == ErrorType.API_DRIFT
+    assert "Pixiv app API changed" in result.error_message
+
+
+@pytest.mark.asyncio
+async def test_rate_limit_carries_retry_after(source_id, sync_engine, tmp_path):
+    client = _FakeClient([], raise_exc=PixivAPIError(
+        "rate limited", status_code=429, retry_after=300.0,
+    ))
+    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
+    result = _run(ing, source_id)
+    assert result.error_type == ErrorType.RATE_LIMITED
+    assert result.retry_after_seconds == 300.0
+
+
+# --- body canary opt-out --------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_captionless_feed_does_not_trip_body_canary(
+    source_id, sync_engine, tmp_path
+):
+    """#862 canary opt-out: a pixiv backfill recording ≥ the canary sample of
+    caption-less works is NORMAL (many artists never write captions) — it must
+    complete, not fail API_DRIFT the way a zero-body Patreon walk does."""
+    n = _CANARY_MIN_SAMPLE + 5
+    posts = [(1000 + i, [_media(1000 + i, 0)]) for i in range(n)]
+    client = _FakeClient([(None, posts)], captions=False)
+    downloader = _FakeDownloader(tmp_path)
+    ing = _ingester(sync_engine, tmp_path, client, downloader)
+
+    result = _run(ing, source_id, mode="backfill")
+    assert result.success is True
+    assert result.error_type is None
+    assert result.files_downloaded == n