2026-07-03 16:36:56 -04:00
27 changed files with 2434 additions and 50 deletions
@@ -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")
@@ -238,6 +238,7 @@ async def preview_source_endpoint(source_id: int):
            )
        cred = CredentialService(session, _get_crypto())
        cookies_path = await cred.get_cookies_path(rec.platform)
+        auth_token = await cred.get_token(rec.platform)

    # The walk + ledger reads are sync (run off the request loop); the process
    # sync engine is the same one the download task uses.
@@ -249,6 +250,7 @@ async def preview_source_endpoint(source_id: int):
        cookies_path=str(cookies_path) if cookies_path else None,
        images_root=Path("/images"),
        sync_session_factory=sync_session_factory(),
+        auth_token=auth_token,
    )
    if "error" in result:
        return _bad("preview_failed", detail=result["error"], status=409)
@@ -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",
@@ -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()
+    )
@@ -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()
+    )
@@ -27,12 +27,15 @@ from .gallery_dl import DownloadResult, ErrorType
 from .native_ingest_common import NativeIngestError
 from .patreon_ingester import PatreonIngester
 from .patreon_resolver import extract_vanity, resolve_campaign_id_for_source
+from .pixiv_client import user_id_from_url
+from .pixiv_ingester import PixivIngester
 from .subscribestar_ingester import SubscribeStarIngester

 # Platforms whose download + verify go through the native ingester rather than
-# gallery-dl. gallery-dl still serves the rest (hentaifoundry, discord, pixiv,
-# deviantart) until they migrate too.
-NATIVE_INGESTER_PLATFORMS = frozenset({"patreon", "subscribestar"})
+# gallery-dl. gallery-dl still serves the rest (hentaifoundry, discord,
+# deviantart — the latter slated for retirement, not migration) until they
+# migrate too.
+NATIVE_INGESTER_PLATFORMS = frozenset({"patreon", "subscribestar", "pixiv"})

 # Mirrors patreon_resolver._CAMPAIGNS_URL — surfaced in resolution-failure
 # messages so the operator sees the exact lookup endpoint that was hit.
@@ -46,6 +49,7 @@ def _native_ingester_cls(platform: str):
    dispatch pick up the replacement."""
    return {
        "patreon": PatreonIngester,
+        "pixiv": PixivIngester,
        "subscribestar": SubscribeStarIngester,
    }[platform]

@@ -98,14 +102,34 @@ async def _resolve_native_campaign_id(
    platform: str, url: str, cookies_path: str | None, overrides: dict,
 ) -> tuple[str | None, str | None]:
    """`(campaign_id, resolved_campaign_id)` for a native source. SubscribeStar's
-    feed id IS the creator URL (no lookup → resolved None). Patreon resolves the
+    feed id IS the creator URL; Pixiv's is the numeric user id parsed straight
+    from it (no lookup → resolved None either way). Patreon resolves the
    campaign id from the vanity URL (resolved non-None when a lookup actually ran,
    so phase 3 caches it)."""
    if platform == "subscribestar":
        return url, None
+    if platform == "pixiv":
+        return user_id_from_url(url), None
    return await resolve_campaign_id_for_source(url, cookies_path, overrides)


+def _campaign_resolution_error(platform: str, url: str) -> str:
+    """Operator-facing message for a native source whose campaign id could not
+    be resolved — names the platform's own lookup mechanism."""
+    if platform == "pixiv":
+        return (
+            f"Could not extract a pixiv user id. source_url={url!r} — expected "
+            "a URL like https://www.pixiv.net/users/<id>."
+        )
+    vanity = extract_vanity(url)
+    return (
+        f"Could not resolve Patreon campaign id. source_url={url!r}; "
+        f"vanity={vanity!r}; "
+        f"lookup=GET {_CAMPAIGNS_API}?filter[vanity]={vanity or ''} "
+        "(vanity lookup failed — cookies expired or creator moved?)"
+    )
+
+
 async def _run_native_ingester(
    ctx: dict, source_config, mode: str | None, gdl, sync_session_factory,
 ) -> tuple[DownloadResult, str | None]:
@@ -123,9 +147,9 @@ async def _run_native_ingester(
        platform, ctx["url"], ctx["cookies_path"], overrides
    )
    if not campaign_id:
-        # Only reachable for Patreon (SubscribeStar's campaign id is the URL).
+        # Patreon: vanity lookup failed. Pixiv: no numeric user id in the URL.
+        # (SubscribeStar's campaign id is the URL itself — never lands here.)
        url = ctx["url"]
-        vanity = extract_vanity(url)
        return (
            DownloadResult(
                success=False,
@@ -133,12 +157,7 @@ async def _run_native_ingester(
                artist_slug=ctx["artist_slug"],
                platform=platform,
                error_type=ErrorType.NOT_FOUND,
-                error_message=(
-                    f"Could not resolve Patreon campaign id. source_url={url!r}; "
-                    f"vanity={vanity!r}; "
-                    f"lookup=GET {_CAMPAIGNS_API}?filter[vanity]={vanity or ''} "
-                    "(vanity lookup failed — cookies expired or creator moved?)"
-                ),
+                error_message=_campaign_resolution_error(platform, url),
            ),
            None,
        )
@@ -161,6 +180,10 @@ async def _run_native_ingester(
        validate=gdl._validate_files,
        rate_limit=rate_limit,
        request_sleep=request_sleep,
+        # Uniform across adapters: token platforms (pixiv) authenticate with
+        # it, cookie platforms accept-and-ignore — so this construction stays
+        # platform-agnostic.
+        auth_token=ctx["auth_token"],
    )
    loop = asyncio.get_running_loop()
    dl_result = await loop.run_in_executor(
@@ -190,6 +213,7 @@ async def preview_source(
    cookies_path: str | None,
    images_root: Path,
    sync_session_factory,
+    auth_token: str | None = None,
    page_limit: int = 3,
 ) -> dict:
    """Dry-run preview for a native platform (plan #708 B4): resolve the campaign
@@ -205,18 +229,12 @@ async def preview_source(
        platform, url, cookies_path, config_overrides or {}
    )
    if not campaign_id:
-        vanity = extract_vanity(url)
-        return {
-            "error": (
-                f"Couldn't resolve the campaign id. source_url={url!r}; "
-                f"vanity={vanity!r}; lookup=GET {_CAMPAIGNS_API}?filter[vanity]={vanity or ''} "
-                "(cookies expired, or the creator moved/renamed?)."
-            )
-        }
+        return {"error": _campaign_resolution_error(platform, url)}
    ingester = _native_ingester_cls(platform)(
        images_root=images_root,
        cookies_path=cookies_path,
        session_factory=sync_session_factory,
+        auth_token=auth_token,
    )
    loop = asyncio.get_running_loop()
    try:
@@ -245,13 +263,18 @@ async def verify_source_credential(
    this and render the result.
    """
    if uses_native_ingester(platform):
-        # Native ingester platforms verify via their own lightweight auth probe
-        # (one authenticated feed fetch). SubscribeStar's probe takes the creator
-        # URL directly; Patreon's resolves the campaign id first.
+        # Native ingester platforms verify via their own lightweight auth probe.
+        # SubscribeStar's probe takes the creator URL directly; Patreon's
+        # resolves the campaign id first; Pixiv's is one OAuth refresh (the
+        # exact call that fails when the token is bad — no feed walk).
        if platform == "subscribestar":
            from .subscribestar_ingester import verify_subscribestar_credential

            return await verify_subscribestar_credential(url, cookies_path, config_overrides)
+        if platform == "pixiv":
+            from .pixiv_ingester import verify_pixiv_credential
+
+            return await verify_pixiv_credential(auth_token)
        from .patreon_ingester import verify_patreon_credential

        return await verify_patreon_credential(url, cookies_path, config_overrides)
@@ -298,8 +298,9 @@ class GalleryDLService:
    # removed at the plan-#697 cutover — it now uses the native ingester
    # (services/patreon_ingester.py), not gallery-dl.
    PLATFORM_DEFAULTS = {
-        # subscribestar removed — it's a native-ingester platform now (#71); the
-        # remaining entries are the gallery-dl platforms not yet migrated.
+        # subscribestar removed — native-ingester platform now (#71); pixiv
+        # removed likewise (#129). The remaining entries are the gallery-dl
+        # platforms not yet migrated.
        "hentaifoundry": {
            "content_types": ["all"],
            "directory": [],
@@ -315,12 +316,6 @@ class GalleryDLService:
            "reactions": False,
            "threads": True,
        },
-        "pixiv": {
-            "content_types": ["all"],
-            "directory": ["{category}"],
-            "filename": "{id}_{title[:50]}_{num:>02}.{extension}",
-            "ugoira": True,
-        },
        "deviantart": {
            "content_types": ["all"],
            "directory": [],
@@ -694,9 +689,6 @@ class GalleryDLService:
        if auth_token and platform == "discord":
            config["extractor"].setdefault("discord", {})
            config["extractor"]["discord"]["token"] = auth_token
-        if auth_token and platform == "pixiv":
-            config["extractor"].setdefault("pixiv", {})
-            config["extractor"]["pixiv"]["refresh-token"] = auth_token

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, dir=str(self._config_dir),
@@ -882,8 +874,6 @@ class GalleryDLService:
            config["extractor"]["cookies"] = cookies_path
        if auth_token and platform == "discord":
            config["extractor"].setdefault("discord", {})["token"] = auth_token
-        if auth_token and platform == "pixiv":
-            config["extractor"].setdefault("pixiv", {})["refresh-token"] = auth_token

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, dir=str(self._config_dir),
@@ -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 "
@@ -63,6 +63,17 @@ _HEAD_KINDS = (TagKind.general, TagKind.character)
 # tag.kind -> the suggestion category the rail groups under.
 _CATEGORY = {TagKind.general: "general", TagKind.character: "character"}

+# System-tag (wip/banner/editor screenshot) heads surface as suggestions at
+# this FLAT confidence floor instead of their auto-derived (precision-tuned)
+# suggest threshold. The auto threshold is high, so it hides the borderline /
+# false-positive guesses — which are exactly the cases the operator wants to
+# SEE and REJECT to sharpen these heads (hard-negative mining: "negatively
+# reinforce what isn't a system tag"). Operator-set 0.65 (2026-07-03): high
+# enough not to spam near-zero scores, low enough to surface real mistakes.
+# Content-tag heads keep their own thresholds; the typed-dropdown's
+# threshold_override still overrides everything (show-all mode).
+_SYSTEM_TAG_SUGGEST_FLOOR = 0.65
+

 class HeadTrainingAlreadyRunning(Exception):
    """Raised by start_head_training_run when a run is already in flight."""
@@ -297,7 +308,7 @@ async def _current_heads(session: AsyncSession, embedding_version: str):
    rows = (
        await session.execute(
            select(
-                TagHead.tag_id, Tag.name, Tag.kind,
+                TagHead.tag_id, Tag.name, Tag.kind, Tag.is_system,
                TagHead.weights, TagHead.bias,
                TagHead.suggest_threshold, TagHead.auto_apply_threshold,
            )
@@ -317,6 +328,7 @@ async def _current_heads(session: AsyncSession, embedding_version: str):
                "name": r.name,
                "category": _CATEGORY.get(r.kind, "general"),
                "auto_apply_threshold": r.auto_apply_threshold,
+                "is_system": bool(r.is_system),
            }
            for r in rows
        ]
@@ -333,7 +345,10 @@ async def score_image(
    category, score}], ranked. A concept surfaces when its score clears the
    head's own suggest_threshold — or, when threshold_override is given (the
    typed-dropdown "show everything" mode), that flat floor instead (0 → every
-    head). Empty if the image has no embedding or no heads exist yet.
+    head). System-tag heads (wip/banner/editor) instead use a flat
+    _SYSTEM_TAG_SUGGEST_FLOOR so their false positives surface for rejection
+    (still overridden by threshold_override). Empty if the image has no
+    embedding or no heads exist yet.

    MAX-OVER-BAG: the image is scored as a BAG of embeddings — the whole-image
    vector PLUS every concept-region crop the agent embedded (same model
@@ -383,7 +398,14 @@ async def score_image(
    probs = (1.0 / (1.0 + np.exp(-Z))).max(axis=0)      # (H,) best over the bag
    out = []
    for i, p in enumerate(probs):
-        cut = threshold_override if threshold_override is not None else heads["thr"][i]
+        if threshold_override is not None:
+            cut = threshold_override
+        elif heads["meta"][i]["is_system"]:
+            # System tags surface at the flat floor (see _SYSTEM_TAG_SUGGEST_FLOOR)
+            # so their false positives show up for the operator to reject.
+            cut = _SYSTEM_TAG_SUGGEST_FLOOR
+        else:
+            cut = heads["thr"][i]
        if p >= cut:
            m = heads["meta"][i]
            out.append({
@@ -87,9 +87,14 @@ class PatreonIngester(Ingester):
        validate: bool = True,
        rate_limit: float = 0.0,
        request_sleep: float = 0.0,
+        auth_token: str | None = None,
        client: PatreonClient | None = None,
        downloader: PatreonDownloader | None = None,
    ):
+        # auth_token: accepted for the uniform native-ingester construction
+        # (download_backends passes it to every adapter); Patreon
+        # authenticates by cookies, so it's unused here.
+        del auth_token
        self.images_root = Path(images_root)
        self.cookies_path = str(cookies_path) if cookies_path else None
        # Pacing (plan #703): request_sleep paces the API page fetches,
@@ -0,0 +1,531 @@
+"""Native Pixiv client — the Pixiv adapter's read path.
+
+Pixiv has a real (if unofficial) API: the mobile app API gallery-dl drives
+(`PixivAppAPI`). Per the downloader ground rule — gallery-dl is the
+known-working base — this client mirrors gallery-dl 1.32.5's request profile
+EXACTLY: the same iOS app headers on every request, the same OAuth
+refresh-token dance against oauth.secure.pixiv.net (X-Client-Time +
+X-Client-Hash), and the same `/v1/user/illusts` walk paginated by `next_url`.
+Deviating from that profile is how the SubscribeStar/Patreon spikes broke, so
+any change here should be diffed against gallery-dl's extractor first.
+
+Feed shape (characterized from gallery-dl 1.32.5, extractor/pixiv.py):
+  - `GET /v1/user/illusts?user_id=<id>` returns `{"illusts": [work...],
+    "next_url": "https://app-api...?user_id=..&offset=30" | null}`.
+  - Pagination: re-issue the SAME endpoint with `next_url`'s query params. The
+    query string doubles as our resumable page cursor (re-fetching it re-serves
+    the same page — the ingest-core resume contract).
+  - A work carries id/title/type(illust|manga|ugoira)/caption(HTML)/
+    create_date(ISO+09:00)/tags[{name,translated_name}]/user/page_count/
+    x_restrict/series/total_view/total_bookmarks/meta_single_page/meta_pages.
+  - Files: multi-page → meta_pages[].image_urls.original; single page →
+    meta_single_page.original_image_url; ugoira → `/v1/ugoira/metadata` zip
+    (600x600 → 1920x1080 URL swap, gallery-dl's default non-original mode).
+
+`campaign_id` for Pixiv is the numeric user id (extracted from the source URL
+by `user_id_from_url` — no network resolver needed).
+
+Gated works: pixiv serves a `https://s.pximg.net/common/images/limit_*.png`
+placeholder as the "original" when a work is blocked for this account
+(sanity-level filter, my-pixiv lock, deleted). gallery-dl's fallback for those
+is a web-AJAX scrape that needs PHPSESSID browser cookies — FC stores only the
+OAuth refresh token, so (exactly like our previous gallery-dl configuration,
+which warned "No PHPSESSID cookie set") those works are skipped, via the
+post_is_gated seam. Auth failures are loud (rotate the refresh token); a
+response missing the fields we depend on is DRIFT (update this client).
+
+FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import time
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from urllib.parse import parse_qsl, urlsplit
+
+import requests
+
+from ..utils.paths import safe_ext
+from .native_ingest_common import (
+    _MAX_429_RETRIES,
+    NativeAuthError,
+    NativeDriftError,
+    NativeIngestError,
+    make_session,
+    retry_after_seconds,
+)
+
+log = logging.getLogger(__name__)
+
+_TIMEOUT_SECONDS = 30.0
+_API_ROOT = "https://app-api.pixiv.net"
+_OAUTH_URL = "https://oauth.secure.pixiv.net/auth/token"
+
+# gallery-dl's public Pixiv-app credentials (PixivAppAPI, also pixivpy's) —
+# these identify the official iOS app to the API, NOT the operator; the
+# operator's identity is the OAuth refresh token.
+_CLIENT_ID = "MOBrBDS8blbauoSck0ZfDbtuzpyT"
+_CLIENT_SECRET = "lsACyCD94FhDUtGTXi3QzcFE2uU1hqtDaKeqrdwj"
+_HASH_SECRET = (
+    "28c1fdd170a5204386cb1313c7077b34"
+    "f83e4aaf4aa829ce78c231e05b0bae2c"
+)
+
+# The exact header set gallery-dl 1.32.5 installs on its session — the proven
+# app-API request profile. The Referer also unlocks i.pximg.net media GETs
+# (403 without it), so the downloader reuses this constant.
+PIXIV_APP_HEADERS = {
+    "App-OS": "ios",
+    "App-OS-Version": "16.7.2",
+    "App-Version": "7.19.1",
+    "User-Agent": "PixivIOSApp/7.19.1 (iOS 16.7.2; iPhone12,8)",
+    "Referer": "https://app-api.pixiv.net/",
+}
+
+# Placeholder image prefix pixiv serves instead of a blocked work's original
+# (limit_sanity_level / limit_mypixiv / limit_unknown variants).
+_LIMIT_URL = "https://s.pximg.net/common/images/limit_"
+
+# The app API reports rate-limiting as an error MESSAGE (often on HTTP 403),
+# not only as HTTP 429. gallery-dl sleeps 300s in-walk; sleeping that long
+# inside our time-boxed chunk would eat the whole budget, so we surface it as
+# a typed 429 and let download_service's cooldown machinery honor the wait.
+_RATE_LIMIT_RETRY_AFTER = 300.0
+
+_TITLE_MAX = 50  # gallery-dl pixiv filename template: {title[:50]}
+
+_RATINGS = {0: "General", 1: "R-18", 2: "R-18G"}
+
+
+class PixivAPIError(NativeIngestError):
+    """Base for native Pixiv client failures. status_code / retry_after are
+    inherited from NativeIngestError."""
+
+
+class PixivAuthError(PixivAPIError, NativeAuthError):
+    """Auth failure — missing/expired/revoked OAuth refresh token. Fix =
+    rotate the credential (Settings → Credentials → Pixiv), not update the
+    client. Maps to error_type 'auth_error'."""
+
+
+class PixivDriftError(PixivAPIError, NativeDriftError):
+    """A response did not match the shape this client depends on (missing
+    `illusts`, un-parseable JSON where JSON was promised). Fail loud so the
+    run flags 'the Pixiv app API changed' instead of silently importing
+    nothing. Maps to API_DRIFT."""
+
+
+@dataclass
+class MediaItem:
+    """One resolved downloadable file belonging to a Pixiv work.
+
+    Fields mirror the other native clients' MediaItem so the downloader and
+    ledger are structurally the same. Pixiv original URLs carry no content
+    hash, so `filehash` is always None and the ledger keys on
+    `<post_id>:<media_id>` where media_id is `p<num>` (page) or `ugoira`
+    (the frame zip) — stable across URL-shape drift.
+    """
+
+    url: str
+    filename: str
+    kind: str
+    filehash: str | None
+    post_id: str
+    media_id: str
+
+
+def user_id_from_url(url: str) -> str | None:
+    """The numeric pixiv user id from a source URL, or None.
+
+    Handles the modern forms FC accepts as sources
+    (https://www.pixiv.net/users/<id>, /en/users/<id>) plus the legacy
+    member.php?id=<id>. This IS the campaign id — no network resolver.
+    """
+    parts = urlsplit(url or "")
+    if "pixiv.net" not in parts.netloc:
+        return None
+    segs = [s for s in parts.path.split("/") if s]
+    if segs and segs[0] == "en":
+        segs = segs[1:]
+    if len(segs) >= 2 and segs[0] == "users" and segs[1].isdigit():
+        return segs[1]
+    if segs and segs[0] == "member.php":
+        qid = dict(parse_qsl(parts.query)).get("id", "")
+        if qid.isdigit():
+            return qid
+    return None
+
+
+def _work_filename(work: dict, num: int, url: str) -> str:
+    """gallery-dl layout parity: `{id}_{title[:50]}_{num:>02}.{extension}`
+    (the downloader sanitizes the final segment)."""
+    title = work.get("title")
+    title50 = (title if isinstance(title, str) else "")[:_TITLE_MAX]
+    ext = safe_ext(urlsplit(url).path.rsplit("/", 1)[-1])
+    return f"{work.get('id')}_{title50}_{num:02d}{ext}"
+
+
+class PixivClient:
+    """Synchronous Pixiv app-API read client. Construct with the operator's
+    OAuth refresh token (the same token-type Credential the gallery-dl path
+    consumed as `extractor.pixiv.refresh-token`)."""
+
+    def __init__(
+        self,
+        refresh_token: str | None,
+        *,
+        request_sleep: float = 0.0,
+        max_retries: int = _MAX_429_RETRIES,
+        session: requests.Session | None = None,
+    ):
+        self.refresh_token = refresh_token
+        self._request_sleep = request_sleep or 0.0
+        self._max_retries = max_retries
+        # No cookies — the app API authenticates via the Bearer token _login
+        # installs. make_session still supplies the retry/UA plumbing; the
+        # extra_headers overwrite its browser UA with the app profile.
+        self._session = (
+            session if session is not None
+            else make_session(None, extra_headers=PIXIV_APP_HEADERS)
+        )
+        self._authed_user: dict = {}
+        # Monotonic deadline after which the access token must be refreshed;
+        # 0 forces a refresh on first use.
+        self._token_deadline = 0.0
+
+    # -- auth ----------------------------------------------------------------
+
+    def _login(self) -> None:
+        """Exchange the refresh token for a Bearer access token (gallery-dl's
+        `_login_impl`, including the X-Client-Time/X-Client-Hash pair the
+        endpoint validates). No-op while the current token is still fresh."""
+        if time.monotonic() < self._token_deadline:
+            return
+        if not self.refresh_token:
+            raise PixivAuthError(
+                "No Pixiv refresh token configured — add the OAuth refresh "
+                "token as the Pixiv credential (token type)."
+            )
+        # gallery-dl stamps naive-UTC with a literal +00:00 suffix.
+        now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S+00:00")
+        headers = {
+            "X-Client-Time": now,
+            "X-Client-Hash": hashlib.md5(
+                (now + _HASH_SECRET).encode()
+            ).hexdigest(),
+        }
+        data = {
+            "client_id": _CLIENT_ID,
+            "client_secret": _CLIENT_SECRET,
+            "grant_type": "refresh_token",
+            "refresh_token": self.refresh_token,
+            "get_secure_url": "1",
+        }
+        try:
+            resp = self._session.post(
+                _OAUTH_URL, data=data, headers=headers,
+                timeout=_TIMEOUT_SECONDS,
+            )
+        except requests.RequestException as exc:
+            raise PixivAPIError(f"Pixiv OAuth request failed: {exc}") from exc
+        if resp.status_code >= 400:
+            raise PixivAuthError(
+                "Pixiv rejected the refresh token (HTTP "
+                f"{resp.status_code}) — rotate the Pixiv credential.",
+                status_code=resp.status_code,
+            )
+        try:
+            payload = resp.json()["response"]
+            access = payload["access_token"]
+        except (ValueError, KeyError, TypeError) as exc:
+            raise PixivDriftError(
+                f"Pixiv OAuth response shape changed: {exc}"
+            ) from exc
+        self._authed_user = payload.get("user") or {}
+        self._session.headers["Authorization"] = f"Bearer {access}"
+        # expires_in is 3600 today; refresh 60s early so a long walk never
+        # rides an expiring token into a spurious 400.
+        expires_in = payload.get("expires_in")
+        lifetime = float(expires_in) if isinstance(expires_in, (int, float)) else 3600.0
+        self._token_deadline = time.monotonic() + max(60.0, lifetime - 60.0)
+
+    # -- request -------------------------------------------------------------
+
+    def _call(self, endpoint: str, params: dict) -> dict:
+        """Authenticated app-API GET → parsed JSON body, with the shared 429
+        backoff and the loud auth/drift/rate-limit mapping."""
+        self._login()
+        if self._request_sleep > 0:
+            time.sleep(self._request_sleep)
+        url = _API_ROOT + endpoint
+        attempt = 0
+        while True:
+            try:
+                resp = self._session.get(
+                    url, params=params, timeout=_TIMEOUT_SECONDS
+                )
+            except requests.RequestException as exc:
+                raise PixivAPIError(
+                    f"Pixiv request failed ({endpoint}): {exc}"
+                ) from exc
+            if resp.status_code == 429 and attempt < self._max_retries:
+                attempt += 1
+                delay = retry_after_seconds(resp, attempt)
+                log.warning(
+                    "Pixiv 429 (%s) — backing off %.1fs (retry %d/%d)",
+                    endpoint, delay, attempt, self._max_retries,
+                )
+                time.sleep(delay)
+                continue
+            break
+
+        try:
+            body = resp.json()
+        except ValueError as exc:
+            raise PixivDriftError(
+                f"Pixiv returned non-JSON for {endpoint} "
+                f"(HTTP {resp.status_code})"
+            ) from exc
+
+        error = body.get("error") if isinstance(body, dict) else None
+        message = ""
+        if isinstance(error, dict):
+            message = str(
+                error.get("user_message") or error.get("message") or ""
+            )
+        # Rate limiting first: the app API reports it as an error MESSAGE
+        # (often on HTTP 403), which must not be mistaken for an auth failure.
+        if resp.status_code == 429 or "rate limit" in message.lower():
+            raise PixivAPIError(
+                f"Pixiv rate limit hit ({endpoint}): {message or 'HTTP 429'}",
+                status_code=429,
+                retry_after=_RATE_LIMIT_RETRY_AFTER,
+            )
+        if resp.status_code in (400, 401, 403):
+            # Invalid/expired access token surfaces as 400 invalid_grant-style
+            # errors on the app API; 401/403 are straight auth rejections.
+            raise PixivAuthError(
+                f"Pixiv rejected the request ({endpoint}, HTTP "
+                f"{resp.status_code}): {message or 'auth rejected'} — "
+                "rotate the Pixiv refresh token.",
+                status_code=resp.status_code,
+            )
+        if resp.status_code >= 400:
+            raise PixivAPIError(
+                f"Pixiv API error ({endpoint}, HTTP {resp.status_code}): "
+                f"{message or 'unknown error'}",
+                status_code=resp.status_code,
+            )
+        if error:
+            # HTTP 200 carrying an error object — unexpected, but never
+            # silently treat it as data.
+            raise PixivAPIError(
+                f"Pixiv API error ({endpoint}): {message or error}",
+                status_code=resp.status_code,
+            )
+        return body
+
+    # -- normalization -------------------------------------------------------
+
+    @staticmethod
+    def _normalize(work: dict) -> dict:
+        """Wrap an app-API work in the `{"id", "attributes", ...}` post shape
+        the platform-agnostic core and shared helpers read. The raw work rides
+        along under `_work` for extract_media / the post record."""
+        title = work.get("title")
+        caption = work.get("caption")
+        wtype = work.get("type")
+        return {
+            "id": work.get("id"),
+            "attributes": {
+                "title": title if isinstance(title, str) else "",
+                "content": caption if isinstance(caption, str) else "",
+                "published_at": work.get("create_date"),
+                "post_type": wtype if isinstance(wtype, str) else "illust",
+            },
+            "_work": work,
+        }
+
+    # -- post-first seams ----------------------------------------------------
+
+    @staticmethod
+    def post_record_key(post: dict) -> tuple[str, str] | None:
+        """`(ledger_key, post_id)` gating post-record capture through the seen
+        ledger (`post:<id>` synthetic key), or None when the work has no id."""
+        pid = post.get("id")
+        pid = str(pid) if pid is not None else ""
+        if not pid:
+            return None
+        return (f"post:{pid}", pid)
+
+    @staticmethod
+    def post_meta(post: dict) -> dict:
+        attrs = post.get("attributes") or {}
+        return {"title": attrs.get("title") or None, "date": attrs.get("published_at")}
+
+    @staticmethod
+    def post_is_gated(post: dict) -> bool:
+        """True when this account cannot fetch the work's real files: pixiv
+        substitutes a `limit_*` placeholder for the original (sanity-level
+        filter / my-pixiv lock / deleted), or zeroes the author (deleted
+        account). Mirrors #874 semantics: gated content leaves NO trace — a
+        placeholder thumbnail and an empty stub would only pollute the
+        archive. (gallery-dl's PHPSESSID web-scrape fallback for these is out
+        of scope: FC holds no pixiv browser cookies — module docstring.)"""
+        work = post.get("_work") or {}
+        user = work.get("user") or {}
+        if not user.get("id"):
+            return True
+        if work.get("meta_pages"):
+            return False
+        single = work.get("meta_single_page") or {}
+        original = single.get("original_image_url")
+        return isinstance(original, str) and original.startswith(_LIMIT_URL)
+
+    # -- media ---------------------------------------------------------------
+
+    def extract_media(self, post: dict, included_index: dict) -> list[MediaItem]:
+        """Resolve a work's downloadable files (gallery-dl's `_extract_files`):
+        multi-page originals, the single-page original, or the ugoira frame
+        zip. `included_index` is unused (pixiv works are self-contained)."""
+        work = post.get("_work") or {}
+        pid = str(post.get("id") or "")
+        if not pid or self.post_is_gated(post):
+            return []
+
+        if work.get("type") == "ugoira":
+            return self._ugoira_media(work, pid)
+
+        meta_pages = work.get("meta_pages") or []
+        if meta_pages:
+            items = []
+            for num, page in enumerate(meta_pages):
+                urls = page.get("image_urls") or {}
+                url = urls.get("original")
+                if not isinstance(url, str) or not url:
+                    continue
+                items.append(
+                    MediaItem(
+                        url=url,
+                        filename=_work_filename(work, num, url),
+                        kind="image",
+                        filehash=None,
+                        post_id=pid,
+                        media_id=f"p{num}",
+                    )
+                )
+            return items
+
+        single = work.get("meta_single_page") or {}
+        url = single.get("original_image_url")
+        if not isinstance(url, str) or not url or url.startswith(_LIMIT_URL):
+            return []
+        return [
+            MediaItem(
+                url=url,
+                filename=_work_filename(work, 0, url),
+                kind="image",
+                filehash=None,
+                post_id=pid,
+                media_id="p0",
+            )
+        ]
+
+    def _ugoira_media(self, work: dict, pid: str) -> list[MediaItem]:
+        """The ugoira frame zip (gallery-dl's default non-original mode):
+        `/v1/ugoira/metadata` → zip_urls.medium with the 600x600→1920x1080
+        swap. Frame delays are memoized onto the work so the post record
+        captures them (a future ugoira→video conversion needs the timings —
+        the zip alone has none). A metadata failure downgrades to 'no media'
+        with a warning (matching gallery-dl) instead of failing the walk —
+        except auth failures, which stay loud."""
+        try:
+            body = self._call("/v1/ugoira/metadata", {"illust_id": pid})
+            meta = body["ugoira_metadata"]
+            zip_url = meta["zip_urls"]["medium"]
+        except PixivAuthError:
+            raise
+        except (PixivAPIError, KeyError, TypeError) as exc:
+            log.warning("Pixiv ugoira metadata failed for %s: %s", pid, exc)
+            return []
+        work["_ugoira_frames"] = meta.get("frames") or []
+        url = zip_url.replace("_ugoira600x600", "_ugoira1920x1080", 1)
+        return [
+            MediaItem(
+                url=url,
+                filename=_work_filename(work, 0, url),
+                kind="ugoira",
+                filehash=None,
+                post_id=pid,
+                media_id="ugoira",
+            )
+        ]
+
+    # -- iteration -----------------------------------------------------------
+
+    def iter_posts(
+        self, campaign_id: str, cursor: str | None = None
+    ) -> Iterator[tuple[dict, dict, str | None]]:
+        """Yield (post, {}, page_cursor) for every work in the user's feed.
+
+        `campaign_id` is the numeric pixiv user id. `cursor` is the query
+        string of the app API's `next_url` (offset pagination); None fetches
+        page 1. The yielded `page_cursor` is the cursor that FETCHED this
+        work's page, so the core checkpoints a value that re-serves the same
+        page on resume (the shared cursor contract)."""
+        if not str(campaign_id or "").isdigit():
+            raise PixivDriftError(
+                f"Pixiv campaign id must be a numeric user id, got "
+                f"{campaign_id!r}"
+            )
+        current = cursor
+        while True:
+            page_cursor = current
+            if current is None:
+                params: dict = {"user_id": campaign_id}
+            else:
+                params = dict(parse_qsl(current))
+            data = self._call("/v1/user/illusts", params)
+            works = data.get("illusts")
+            if not isinstance(works, list):
+                raise PixivDriftError(
+                    "Pixiv user-illusts response had no 'illusts' list "
+                    f"(keys: {sorted(data)[:8]})"
+                )
+            for work in works:
+                if not isinstance(work, dict):
+                    continue
+                yield self._normalize(work), {}, page_cursor
+            next_url = data.get("next_url")
+            if not next_url:
+                return
+            current = str(next_url).rpartition("?")[2]
+
+    # -- verify --------------------------------------------------------------
+
+    def verify_auth(self) -> tuple[bool | None, str]:
+        """Cheap credential probe: run the OAuth refresh (the thing that fails
+        when the token is bad) without walking any feed."""
+        try:
+            self._token_deadline = 0.0  # force a real refresh
+            self._login()
+        except PixivAuthError as exc:
+            return False, f"Pixiv rejected the credential — {exc}"
+        except PixivAPIError as exc:
+            return None, f"Couldn't verify (network/HTTP issue): {exc}"
+        account = self._authed_user.get("account") or self._authed_user.get("name")
+        suffix = f" as {account}" if account else ""
+        return True, f"Credentials valid — Pixiv OAuth refresh succeeded{suffix}."
+
+
+def rating_label(x_restrict) -> str | None:
+    """Human rating from pixiv's x_restrict (0/1/2) — written into the post
+    record so the archive keeps the R-18 flag without the reader needing to
+    know pixiv's numeric scheme."""
+    if isinstance(x_restrict, bool) or not isinstance(x_restrict, int):
+        return None
+    return _RATINGS.get(x_restrict)
@@ -0,0 +1,262 @@
+"""Native Pixiv media downloader — the Pixiv counterpart to
+patreon_downloader / subscribestar_downloader.
+
+Given a normalized Pixiv work and its resolved `MediaItem`s
+(pixiv_client.extract_media), download the originals to gallery-dl's on-disk
+layout (so pre-cutover gallery-dl downloads are recognized on disk and not
+re-fetched), write the post-first sidecars the importer consumes, and report
+per-media outcomes.
+
+On-disk layout (matches FC's gallery-dl pixiv config, PLATFORM_DEFAULTS:
+base-directory `<images_root>/<artist_slug>/pixiv` + `directory:
+["{category}"]` + filename `{id}_{title[:50]}_{num:>02}.{extension}`):
+
+    <images_root>/<artist_slug>/pixiv/pixiv/<id>_<title50>_<NN>.<ext>
+
+— note the intentional DOUBLE `pixiv` segment: gallery-dl appended
+`{category}` under a base-directory that already ended in the platform name,
+and tier-2 disk-skip parity requires reproducing that exactly. The layout is
+FLAT (no per-post directory), so the post-first record is `_post_<id>.json`
+in the same directory (the id suffix prevents the collisions a bare
+`_post.json` would have here; phase 3 receives explicit post_record_paths, so
+the name is a convention, not a discovery key).
+
+Simpler than Patreon (no Mux/yt-dlp video branch) — the one special file is
+the ugoira frame zip, downloaded as-is; FC's archive-containment import
+extracts the frames, and the frame DELAYS ride the post record (the zip
+carries none — a future ugoira→video conversion needs them).
+
+PURE: no DB; the seen-skip is an injected predicate. FC runs on a plain-HTTP
+homelab; nothing here uses a secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import time
+from collections.abc import Callable
+from pathlib import Path
+
+import requests
+
+from .native_ingest_common import (
+    BaseNativeDownloader,
+    MediaOutcome,
+    PostRecordOutcome,
+)
+from .pixiv_client import PIXIV_APP_HEADERS, rating_label
+
+log = logging.getLogger(__name__)
+
+# Control chars (0x00–0x1f + 0x7f DEL) — gallery-dl's default `path-remove`.
+_GDL_PATH_REMOVE_RE = re.compile(r"[\x00-\x1f\x7f]")
+
+
+def gdl_clean_filename(name: str) -> str:
+    """Reproduce gallery-dl's on-disk filename EXACTLY as it wrote it on this
+    Linux host, so the tier-2 disk-skip recognizes pre-cutover files instead of
+    re-downloading them.
+
+    gallery-dl's PathFormat.build_filename is `clean_path(clean_segment(name))`.
+    On Linux (verified against gallery-dl 1.32.5 path.py) the defaults resolve to:
+      - path-restrict "auto" → "/"          → clean_segment replaces ONLY "/" → "_"
+      - path-remove   "\\x00-\\x1f\\x7f"       → clean_path DELETES control chars
+      - path-strip    "auto" → ""           → NO trailing dot/space stripping
+    Crucially it does NOT touch the Windows-forbidden set (<>:"|?*) — those stay
+    raw in titles on disk. A stricter sanitizer here would rename any such title,
+    miss the on-disk match, and re-pull the whole work. Order mirrors gallery-dl
+    (segment inner, path outer); for these disjoint char sets it's commutative.
+    """
+    return _GDL_PATH_REMOVE_RE.sub("", name.replace("/", "_"))
+
+# Enrichment keys copied verbatim from the app-API work dict into the post
+# record (they're already JSON scalars/objects). Everything lands in
+# Post.raw_metadata via the importer, so the archive keeps pixiv's stats and
+# structure without a schema change.
+_WORK_PASSTHROUGH_KEYS = (
+    "type",
+    "page_count",
+    "width",
+    "height",
+    "total_view",
+    "total_bookmarks",
+    "total_comments",
+    "is_bookmarked",
+    "illust_ai_type",
+    "series",
+)
+
+
+class PixivDownloader(BaseNativeDownloader):
+    """Download resolved Pixiv media to gallery-dl's on-disk layout.
+    Subclasses BaseNativeDownloader for the shared streaming GET
+    (transient-retry + Range-resume) and validation/quarantine. PURE: no DB."""
+
+    def __init__(
+        self,
+        images_root: Path,
+        cookies_path: str | None = None,
+        *,
+        validate: bool = True,
+        rate_limit: float = 0.0,
+        session: requests.Session | None = None,
+    ):
+        super().__init__(
+            images_root, cookies_path, platform="pixiv",
+            validate=validate, rate_limit=rate_limit, session=session,
+        )
+        if session is None:
+            # i.pximg.net 403s any GET without the app Referer; mirror the
+            # client's full app-header profile (gallery-dl serves media off
+            # the same session it drives the API with). An injected session
+            # (tests) owns its own headers.
+            self.session.headers.update(PIXIV_APP_HEADERS)
+
+    # -- public ------------------------------------------------------------
+
+    def download_post(
+        self,
+        post: dict,
+        media_items: list,
+        artist_slug: str,
+        *,
+        is_seen: Callable[[object], bool] = lambda m: False,
+        should_stop: Callable[[], bool] = lambda: False,
+        recapture: bool = False,
+    ) -> list[MediaOutcome]:
+        """Download every media item of one work; return per-item outcomes.
+        Mirrors SubscribeStarDownloader.download_post (two-tier skip, mid-post
+        time-box, recapture surfacing)."""
+        flat_dir = self._flat_dir(artist_slug)
+        outcomes: list[MediaOutcome] = []
+        for media in media_items:
+            if should_stop():
+                break
+            try:
+                outcomes.append(
+                    self._download_one(
+                        post, media, flat_dir, artist_slug, is_seen,
+                        recapture=recapture,
+                    )
+                )
+            except Exception as exc:  # resilient: isolate one item's failure
+                log.warning(
+                    "Pixiv media failed (work %s, %s): %s",
+                    post.get("id"), getattr(media, "media_id", "?"), exc,
+                )
+                outcomes.append(
+                    MediaOutcome(media=media, status="error", path=None, error=str(exc))
+                )
+        return outcomes
+
+    def _flat_dir(self, artist_slug: str) -> Path:
+        # Double platform segment — gallery-dl layout parity (module docstring).
+        return self.images_root / artist_slug / "pixiv" / "pixiv"
+
+    # -- per-item ----------------------------------------------------------
+
+    def _download_one(
+        self,
+        post: dict,
+        media,
+        flat_dir: Path,
+        artist_slug: str,
+        is_seen: Callable[[object], bool],
+        *,
+        recapture: bool = False,
+    ) -> MediaOutcome:
+        seen = is_seen(media)
+        if seen and not recapture:
+            return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)
+
+        # The client's filename already carries the {id}_{title50}_{NN} shape
+        # (raw title, gallery-dl-template order); clean it to the byte-exact
+        # name gallery-dl wrote on disk so tier-2 disk-skip matches (else a
+        # re-download of the whole work). See gdl_clean_filename.
+        media_path = flat_dir / gdl_clean_filename(media.filename)
+
+        if media_path.exists():  # tier-2: already on disk
+            return MediaOutcome(
+                media=media, status="skipped_disk", path=media_path, error=None
+            )
+        # recapture: a seen item not on disk is NOT re-downloaded (recovery's job).
+        if seen:
+            return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)
+
+        flat_dir.mkdir(parents=True, exist_ok=True)
+        if self._rate_limit > 0:
+            time.sleep(self._rate_limit)
+
+        out_path = self._fetch_get(media.url, media_path)
+        reason, quarantine_dest = self._validate_path(out_path, artist_slug, media.url)
+        if reason is not None:
+            return MediaOutcome(
+                media=media, status="quarantined", path=quarantine_dest, error=reason,
+            )
+        self._write_minimal_sidecar(post, out_path, source_url=media.url)
+        return MediaOutcome(media=media, status="downloaded", path=out_path, error=None)
+
+    # -- post record ---------------------------------------------------------
+
+    def write_post_record(self, post: dict, artist_slug: str) -> PostRecordOutcome:
+        """Write the post-first `_post_<id>.json` — the sole writer of the post
+        body/metadata on the native path. Beyond the standard body fields, the
+        record carries pixiv's own structure (tags + EN translations, rating,
+        series, view/bookmark counts, AI flag, dimensions, author, ugoira frame
+        delays) so the archive keeps what the platform knows about the work."""
+        attrs = post.get("attributes") or {}
+        work = post.get("_work") or {}
+        title = attrs.get("title") if isinstance(attrs.get("title"), str) else None
+        post_type = attrs.get("post_type") if isinstance(attrs.get("post_type"), str) else None
+        pid = str(post.get("id") or "")
+        if not pid:
+            return PostRecordOutcome(
+                path=None, post_type=post_type, title=title, body_chars=0,
+            )
+
+        content = attrs.get("content")
+        content = content if isinstance(content, str) else ""
+        data: dict = {
+            "category": "pixiv",
+            "id": pid,
+            "title": title or "",
+            "content": content,
+            "published_at": attrs.get("published_at"),
+            # The post permalink is synthesized by platforms/pixiv.py
+            # derive_post_url from `id` at parse time — no url key here.
+            "rating": rating_label(work.get("x_restrict")),
+        }
+        for key in _WORK_PASSTHROUGH_KEYS:
+            if key in work:
+                data[key] = work[key]
+        tags = work.get("tags")
+        if isinstance(tags, list):
+            data["tags"] = [
+                {
+                    "name": t.get("name"),
+                    "translated_name": t.get("translated_name"),
+                }
+                for t in tags
+                if isinstance(t, dict)
+            ]
+        user = work.get("user")
+        if isinstance(user, dict):
+            data["user"] = {
+                "id": user.get("id"),
+                "account": user.get("account"),
+                "name": user.get("name"),
+            }
+        # Memoized by extract_media's ugoira branch; the zip has no timings.
+        frames = work.get("_ugoira_frames")
+        if frames:
+            data["ugoira_frames"] = frames
+
+        flat_dir = self._flat_dir(artist_slug)
+        flat_dir.mkdir(parents=True, exist_ok=True)
+        path = flat_dir / f"_post_{pid}.json"
+        path.write_text(json.dumps(data, indent=2, ensure_ascii=False))
+        return PostRecordOutcome(
+            path=path, post_type=post_type, title=title, body_chars=len(content),
+        )
@@ -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)
@@ -1,8 +1,14 @@
 """Pixiv — one quirk.

-post_url: gallery-dl's `url` is the image URL on `i.pximg.net`. The
-post permalink follows /artworks/<id>. external_post_id (= `id`) was
-already correct, so no override there.
+post_url: the sidecar's `url` (legacy gallery-dl era) is the image URL
+on `i.pximg.net`, and the native post record (#129) writes no url key
+at all — the permalink is synthesized from `id` here either way:
+/artworks/<id>. external_post_id (= `id`) was already correct, so no
+override there.
+
+Downloads run through the native ingester (pixiv_ingester.py), not
+gallery-dl; this registry entry still owns URL validation, sidecar
+parsing, and the credential surface (the OAuth refresh token).
 """

 from .base import GD_DEFAULTS, PlatformInfo, str_id_value
@@ -298,6 +298,15 @@ class SourceService:

        for key, value in fields.items():
            setattr(source, key, value)
+        # Disabling a source clears its failure state (operator: disable the subs
+        # you're not paying for without them lingering as "failing"). Re-enabling
+        # then starts clean; the next real run re-derives health. Only on the
+        # explicit disable — an unrelated edit to an already-disabled source
+        # leaves its (already-clear) state alone.
+        if fields.get("enabled") is False:
+            source.last_error = None
+            source.error_type = None
+            source.consecutive_failures = 0
        try:
            await self.session.flush()
        except IntegrityError as exc:
@@ -226,13 +226,35 @@ def _attachment_item(
    )


+def _normalize_ss_host(netloc: str) -> str:
+    """Rewrite the `subscribestar.art` host to `subscribestar.adult`.
+
+    The age wall on the `.art` domain does not clear with the
+    `18_plus_agreement_generic` cookie (unlike `.com`/`.adult`): a `.art`
+    creator page keeps 302'ing to `/age_confirmation_warning` even with the
+    cookie set (Elasid, event #54116). The same creator is reachable on
+    `.adult`, where the cookie works — so `.art` behaves as an alias that
+    doesn't honor the age gate. Normalize it to `.adult` at request time (the
+    stored Source.url is left untouched). `.com`/`.adult` pass through.
+    """
+    host = netloc.lower()
+    if host == "subscribestar.art" or host.endswith(".subscribestar.art"):
+        rewritten = netloc[: -len("art")] + "adult"
+        log.info(
+            "SubscribeStar: rewrote age-gated .art host %r → %r", netloc, rewritten
+        )
+        return rewritten
+    return netloc
+
+
 def _split_creator_url(campaign_id: str) -> tuple[str, str]:
    """`campaign_id` is the creator URL → (base, slug).

-    base = scheme://host (preserving .com vs .adult); slug = first path segment.
+    base = scheme://host (preserving .com vs .adult; .art → .adult, see
+    _normalize_ss_host); slug = first path segment.
    """
    parts = urlsplit(campaign_id)
-    base = f"{parts.scheme or 'https'}://{parts.netloc}"
+    base = f"{parts.scheme or 'https'}://{_normalize_ss_host(parts.netloc)}"
    slug = parts.path.strip("/").split("/")[0] if parts.path else ""
    return base, slug

@@ -61,9 +61,14 @@ class SubscribeStarIngester(Ingester):
        validate: bool = True,
        rate_limit: float = 0.0,
        request_sleep: float = 0.0,
+        auth_token: str | None = None,
        client: SubscribeStarClient | None = None,
        downloader: SubscribeStarDownloader | None = None,
    ):
+        # auth_token: accepted for the uniform native-ingester construction
+        # (download_backends passes it to every adapter); SubscribeStar
+        # authenticates by cookies, so it's unused here.
+        del auth_token
        self.images_root = Path(images_root)
        self.cookies_path = str(cookies_path) if cookies_path else None
        resolved_client = (
@@ -76,7 +76,8 @@ const recovering = computed(() => !!props.source.backfill_bypass_seen)
 const recapturing = computed(() => !!props.source.backfill_recapture)
 // Recover / recapture are native-ingester features (ledger-bypass re-walk and
 // post-text re-grab), available to every native platform — not just Patreon.
-const NATIVE_PLATFORMS = ['patreon', 'subscribestar']
+// Mirrors backend download_backends.NATIVE_INGESTER_PLATFORMS.
+const NATIVE_PLATFORMS = ['patreon', 'subscribestar', 'pixiv']
 const isNative = computed(() => NATIVE_PLATFORMS.includes(props.source.platform))
 </script>

@@ -0,0 +1,129 @@
+{
+  "illusts": [
+    {
+      "id": 111,
+      "title": "Multi Page Adventure",
+      "type": "illust",
+      "caption": "Two-page set. <a href=\"https://example.com/wip\">WIP thread</a>",
+      "create_date": "2026-06-20T18:00:00+09:00",
+      "user": {"id": 99, "name": "Example Artist", "account": "exartist"},
+      "tags": [
+        {"name": "オリジナル", "translated_name": "original"},
+        {"name": "女の子", "translated_name": "girl"}
+      ],
+      "page_count": 2,
+      "width": 1200,
+      "height": 1600,
+      "x_restrict": 0,
+      "series": {"id": 4242, "title": "Adventure Series"},
+      "total_view": 1000,
+      "total_bookmarks": 250,
+      "is_bookmarked": false,
+      "illust_ai_type": 1,
+      "meta_single_page": {},
+      "meta_pages": [
+        {
+          "image_urls": {
+            "square_medium": "https://i.pximg.net/c/360x360_70/img-master/img/2026/06/20/18/00/00/111_p0_square1200.jpg",
+            "original": "https://i.pximg.net/img-original/img/2026/06/20/18/00/00/111_p0.png"
+          }
+        },
+        {
+          "image_urls": {
+            "square_medium": "https://i.pximg.net/c/360x360_70/img-master/img/2026/06/20/18/00/00/111_p1_square1200.jpg",
+            "original": "https://i.pximg.net/img-original/img/2026/06/20/18/00/00/111_p1.png"
+          }
+        }
+      ]
+    },
+    {
+      "id": 222,
+      "title": "Single Piece",
+      "type": "illust",
+      "caption": "",
+      "create_date": "2026-06-18T12:30:00+09:00",
+      "user": {"id": 99, "name": "Example Artist", "account": "exartist"},
+      "tags": [{"name": "落書き", "translated_name": "doodle"}],
+      "page_count": 1,
+      "width": 900,
+      "height": 900,
+      "x_restrict": 1,
+      "series": null,
+      "total_view": 500,
+      "total_bookmarks": 60,
+      "is_bookmarked": true,
+      "illust_ai_type": 0,
+      "meta_single_page": {
+        "original_image_url": "https://i.pximg.net/img-original/img/2026/06/18/12/30/00/222_p0.jpg"
+      },
+      "meta_pages": []
+    },
+    {
+      "id": 333,
+      "title": "Wiggle Loop",
+      "type": "ugoira",
+      "caption": "animated",
+      "create_date": "2026-06-15T09:00:00+09:00",
+      "user": {"id": 99, "name": "Example Artist", "account": "exartist"},
+      "tags": [{"name": "うごイラ", "translated_name": "ugoira"}],
+      "page_count": 1,
+      "width": 600,
+      "height": 600,
+      "x_restrict": 0,
+      "series": null,
+      "total_view": 300,
+      "total_bookmarks": 40,
+      "is_bookmarked": false,
+      "illust_ai_type": 0,
+      "meta_single_page": {
+        "original_image_url": "https://i.pximg.net/img-original/img/2026/06/15/09/00/00/333_ugoira0.jpg"
+      },
+      "meta_pages": []
+    },
+    {
+      "id": 444,
+      "title": "Blocked Work",
+      "type": "illust",
+      "caption": "",
+      "create_date": "2026-06-10T00:00:00+09:00",
+      "user": {"id": 99, "name": "Example Artist", "account": "exartist"},
+      "tags": [],
+      "page_count": 1,
+      "width": 0,
+      "height": 0,
+      "x_restrict": 2,
+      "series": null,
+      "total_view": 0,
+      "total_bookmarks": 0,
+      "is_bookmarked": false,
+      "illust_ai_type": 0,
+      "meta_single_page": {
+        "original_image_url": "https://s.pximg.net/common/images/limit_sanity_level_360.png"
+      },
+      "meta_pages": []
+    },
+    {
+      "id": 555,
+      "title": "Ghost Work",
+      "type": "illust",
+      "caption": "",
+      "create_date": "2026-06-01T00:00:00+09:00",
+      "user": {"id": 0, "name": "", "account": ""},
+      "tags": [],
+      "page_count": 1,
+      "width": 0,
+      "height": 0,
+      "x_restrict": 0,
+      "series": null,
+      "total_view": 0,
+      "total_bookmarks": 0,
+      "is_bookmarked": false,
+      "illust_ai_type": 0,
+      "meta_single_page": {
+        "original_image_url": "https://i.pximg.net/img-original/img/2026/06/01/00/00/00/555_p0.png"
+      },
+      "meta_pages": []
+    }
+  ],
+  "next_url": "https://app-api.pixiv.net/v1/user/illusts?user_id=99&offset=30"
+}
@@ -3,12 +3,15 @@ native ingester vs. gallery-dl. Pure, no DB."""

 from backend.app.services.download_backends import (
    NATIVE_INGESTER_PLATFORMS,
+    _campaign_resolution_error,
+    _native_ingester_cls,
    uses_native_ingester,
 )
+from backend.app.services.pixiv_ingester import PixivIngester


 def test_native_platforms():
-    for platform in ("patreon", "subscribestar"):
+    for platform in ("patreon", "subscribestar", "pixiv"):
        assert uses_native_ingester(platform) is True
        assert platform in NATIVE_INGESTER_PLATFORMS

@@ -16,9 +19,19 @@ def test_native_platforms():
 def test_gallery_dl_platforms_are_not_native():
    # The platforms still served by gallery-dl must NOT route to the native
    # ingester — guards an accidental over-broad migration.
-    for platform in ("hentaifoundry", "discord", "pixiv", "deviantart"):
+    for platform in ("hentaifoundry", "discord", "deviantart"):
        assert uses_native_ingester(platform) is False


 def test_unknown_platform_is_not_native():
    assert uses_native_ingester("nonsense") is False
+
+
+def test_pixiv_dispatches_to_its_ingester():
+    assert _native_ingester_cls("pixiv") is PixivIngester
+
+
+def test_pixiv_resolution_error_names_the_expected_url_shape():
+    msg = _campaign_resolution_error("pixiv", "https://www.pixiv.net/artworks/1")
+    assert "pixiv user id" in msg
+    assert "users/<id>" in msg
@@ -322,7 +322,7 @@ async def test_run_download_native_resolves_vanity_and_runs(
    ctx = {
        "platform": "patreon",
        "source_id": source.id, "url": "https://patreon.com/alice",
-        "artist_slug": "alice", "cookies_path": None,
+        "artist_slug": "alice", "cookies_path": None, "auth_token": None,
        "config_overrides": {},
    }
    result, resolved = await db_mod.run_download(
@@ -355,7 +355,7 @@ async def test_run_download_native_unresolvable_fails_loud(
    ctx = {
        "platform": "patreon",
        "source_id": source.id, "url": "https://patreon.com/alice",
-        "artist_slug": "alice", "cookies_path": None,
+        "artist_slug": "alice", "cookies_path": None, "auth_token": None,
        "config_overrides": {},
    }
    result, resolved = await db_mod.run_download(
@@ -111,6 +111,49 @@ async def test_threshold_override_surfaces_below_cut(db):
    assert any(s.canonical_tag_id == tag.id for s in flooded.by_category["general"])


+@pytest.mark.asyncio
+async def test_system_tag_surfaces_at_flat_floor_despite_high_threshold(db):
+    # A system tag's head uses the flat _SYSTEM_TAG_SUGGEST_FLOOR (0.65), not its
+    # precision-tuned suggest_threshold — so a matching image (~0.73) surfaces it
+    # for rejection even though its 0.9 auto threshold would hide it.
+    from backend.app.services.ml.heads import _SYSTEM_TAG_SUGGEST_FLOOR
+    assert _SYSTEM_TAG_SUGGEST_FLOOR == 0.65
+    tag = await TagService(db).find_or_create("wip sketchy", TagKind.general)
+    tag.is_system = True
+    img = await _img(db, "1" * 64, _emb(0))            # matches head → ~0.73
+    await _head(db, tag.id, slot=0, suggest_threshold=0.9)
+    await db.commit()
+    sl = await SuggestionService(db).for_image(img.id)
+    assert any(
+        s.canonical_tag_id == tag.id for s in sl.by_category.get("general", [])
+    )
+
+
+@pytest.mark.asyncio
+async def test_system_tag_below_floor_stays_hidden(db):
+    # Below the 0.65 floor (~0.5 on an orthogonal image) the system tag is not
+    # surfaced — the floor still gates, it isn't "always show".
+    tag = await TagService(db).find_or_create("wip faint", TagKind.general)
+    tag.is_system = True
+    img = await _img(db, "2" * 64, _emb(1))            # orthogonal → 0.5
+    await _head(db, tag.id, slot=0, suggest_threshold=0.9)
+    await db.commit()
+    sl = await SuggestionService(db).for_image(img.id)
+    assert not sl.by_category.get("general")
+
+
+@pytest.mark.asyncio
+async def test_content_tag_ignores_system_floor(db):
+    # A NON-system head with the same high threshold does NOT get the floor: the
+    # ~0.73 match stays hidden below its 0.9 auto threshold (floor is system-only).
+    tag = await TagService(db).find_or_create("glasses hi", TagKind.general)
+    img = await _img(db, "3" * 64, _emb(0))            # matches head → ~0.73
+    await _head(db, tag.id, slot=0, suggest_threshold=0.9)
+    await db.commit()
+    sl = await SuggestionService(db).for_image(img.id)
+    assert not sl.by_category.get("general")
+
+
@pytest.mark.asyncio
 async def test_concept_region_surfaces_via_max_over_bag(db):
    # Max-over-bag: the whole-image vector is orthogonal to the head (scores the
@@ -0,0 +1,369 @@
+"""PixivClient tests — parsing + iteration against canned pages, no network.
+
+The fixture mirrors a real `/v1/user/illusts` page (multi-page work, single
+page, ugoira, sanity-limited placeholder, deleted-author ghost). HTTP is
+stubbed at the requests-session seam (oauth POST + API GET), so the exact
+gallery-dl-parity request profile — headers, oauth form, pagination params —
+is asserted rather than assumed.
+"""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from backend.app.services.pixiv_client import (
+    PIXIV_APP_HEADERS,
+    MediaItem,
+    PixivAPIError,
+    PixivAuthError,
+    PixivClient,
+    PixivDriftError,
+    rating_label,
+    user_id_from_url,
+)
+
+_FIXTURE = Path(__file__).parent / "fixtures" / "pixiv_user_illusts_page1.json"
+
+
+class FakeResponse:
+    def __init__(self, status_code=200, json_data=None, headers=None):
+        self.status_code = status_code
+        self._json = json_data
+        self.headers = headers or {}
+
+    def json(self):
+        if self._json is None:
+            raise ValueError("no JSON")
+        return self._json
+
+
+class FakeSession:
+    """Minimal requests.Session stand-in: canned responses per (method, url
+    fragment), recording every call for profile assertions."""
+
+    def __init__(self):
+        self.headers = dict(PIXIV_APP_HEADERS)
+        self.responses = []
+        self.calls = []
+
+    def queue(self, response):
+        self.responses.append(response)
+        return self
+
+    def _next(self):
+        if not self.responses:
+            raise AssertionError("FakeSession ran out of queued responses")
+        return self.responses.pop(0)
+
+    def post(self, url, data=None, headers=None, timeout=None):
+        self.calls.append(("POST", url, data, headers))
+        return self._next()
+
+    def get(self, url, params=None, timeout=None, headers=None):
+        self.calls.append(("GET", url, params, headers))
+        return self._next()
+
+
+def _oauth_ok():
+    return FakeResponse(200, {
+        "response": {
+            "access_token": "acc-token",
+            "expires_in": 3600,
+            "user": {"id": "77", "account": "operator", "name": "Op"},
+        }
+    })
+
+
+@pytest.fixture
+def page1():
+    return json.loads(_FIXTURE.read_text())
+
+
+@pytest.fixture
+def client():
+    # Never issues a request in pure-parsing tests.
+    return PixivClient("refresh-tok", session=FakeSession())
+
+
+def _post_for(client, page1, work_id):
+    for work in page1["illusts"]:
+        if work["id"] == work_id:
+            return client._normalize(work)
+    raise AssertionError(f"no work {work_id} in fixture")
+
+
+# -- URL → user id ----------------------------------------------------------
+
+def test_user_id_from_url_variants():
+    assert user_id_from_url("https://www.pixiv.net/users/12345678") == "12345678"
+    assert user_id_from_url("https://www.pixiv.net/en/users/42") == "42"
+    assert user_id_from_url("https://pixiv.net/users/7/artworks") == "7"
+    assert user_id_from_url("https://www.pixiv.net/member.php?id=99") == "99"
+
+
+def test_user_id_from_url_rejects_non_matches():
+    assert user_id_from_url("https://www.pixiv.net/artworks/111") is None
+    assert user_id_from_url("https://www.pixiv.net/users/notdigits") is None
+    assert user_id_from_url("https://example.com/users/5") is None
+    assert user_id_from_url("") is None
+
+
+# -- normalization ------------------------------------------------------------
+
+def test_normalize_maps_attributes(client, page1):
+    post = _post_for(client, page1, 111)
+    attrs = post["attributes"]
+    assert post["id"] == 111
+    assert attrs["title"] == "Multi Page Adventure"
+    assert "Two-page set." in attrs["content"]
+    assert attrs["published_at"] == "2026-06-20T18:00:00+09:00"
+    assert attrs["post_type"] == "illust"
+    assert post["_work"]["total_bookmarks"] == 250
+
+
+def test_post_record_key_and_meta(client, page1):
+    post = _post_for(client, page1, 222)
+    assert client.post_record_key(post) == ("post:222", "222")
+    meta = client.post_meta(post)
+    assert meta["title"] == "Single Piece"
+    assert meta["date"] == "2026-06-18T12:30:00+09:00"
+    assert client.post_record_key({"id": None}) is None
+
+
+# -- gating -------------------------------------------------------------------
+
+def test_post_is_gated_limit_placeholder(client, page1):
+    assert client.post_is_gated(_post_for(client, page1, 444)) is True
+
+
+def test_post_is_gated_deleted_author(client, page1):
+    assert client.post_is_gated(_post_for(client, page1, 555)) is True
+
+
+def test_post_is_gated_normal_works(client, page1):
+    assert client.post_is_gated(_post_for(client, page1, 111)) is False
+    assert client.post_is_gated(_post_for(client, page1, 222)) is False
+
+
+# -- extract_media --------------------------------------------------------------
+
+def test_extract_media_multi_page(client, page1):
+    items = client.extract_media(_post_for(client, page1, 111), {})
+    assert len(items) == 2
+    assert all(isinstance(m, MediaItem) for m in items)
+    assert [m.media_id for m in items] == ["p0", "p1"]
+    assert items[0].url.endswith("/111_p0.png")
+    assert items[1].url.endswith("/111_p1.png")
+    # gallery-dl filename parity: {id}_{title[:50]}_{num:>02}.{extension}
+    assert items[0].filename == "111_Multi Page Adventure_00.png"
+    assert items[1].filename == "111_Multi Page Adventure_01.png"
+    assert all(m.post_id == "111" for m in items)
+
+
+def test_extract_media_single_page(client, page1):
+    items = client.extract_media(_post_for(client, page1, 222), {})
+    assert len(items) == 1
+    assert items[0].media_id == "p0"
+    assert items[0].url.endswith("/222_p0.jpg")
+    assert items[0].filename == "222_Single Piece_00.jpg"
+
+
+def test_extract_media_gated_yields_nothing(client, page1):
+    assert client.extract_media(_post_for(client, page1, 444), {}) == []
+    assert client.extract_media(_post_for(client, page1, 555), {}) == []
+
+
+def test_extract_media_ugoira_zip_swap(client, page1, monkeypatch):
+    frames = [{"file": "000000.jpg", "delay": 90}, {"file": "000001.jpg", "delay": 90}]
+
+    def fake_call(endpoint, params):
+        assert endpoint == "/v1/ugoira/metadata"
+        assert params == {"illust_id": "333"}
+        return {"ugoira_metadata": {
+            "zip_urls": {"medium": (
+                "https://i.pximg.net/img-zip-ugoira/img/2026/06/15/09/00/00/"
+                "333_ugoira600x600.zip"
+            )},
+            "frames": frames,
+        }}
+
+    monkeypatch.setattr(client, "_call", fake_call)
+    post = _post_for(client, page1, 333)
+    items = client.extract_media(post, {})
+    assert len(items) == 1
+    assert items[0].media_id == "ugoira"
+    assert items[0].kind == "ugoira"
+    assert items[0].url.endswith("333_ugoira1920x1080.zip")
+    assert items[0].filename == "333_Wiggle Loop_00.zip"
+    # Frame delays memoized for the post record (future video conversion).
+    assert post["_work"]["_ugoira_frames"] == frames
+
+
+def test_extract_media_ugoira_metadata_failure_downgrades(
+    client, page1, monkeypatch
+):
+    def fake_call(endpoint, params):
+        raise PixivAPIError("boom", status_code=500)
+
+    monkeypatch.setattr(client, "_call", fake_call)
+    assert client.extract_media(_post_for(client, page1, 333), {}) == []
+
+
+def test_extract_media_ugoira_auth_failure_stays_loud(
+    client, page1, monkeypatch
+):
+    def fake_call(endpoint, params):
+        raise PixivAuthError("token dead", status_code=400)
+
+    monkeypatch.setattr(client, "_call", fake_call)
+    with pytest.raises(PixivAuthError):
+        client.extract_media(_post_for(client, page1, 333), {})
+
+
+# -- iteration ------------------------------------------------------------------
+
+def test_iter_posts_paginates_and_carries_cursor(page1, monkeypatch):
+    client = PixivClient("refresh-tok", session=FakeSession())
+    page2 = {
+        "illusts": [dict(page1["illusts"][1], id=666, title="Older")],
+        "next_url": None,
+    }
+    seen_params = []
+
+    def fake_call(endpoint, params):
+        assert endpoint == "/v1/user/illusts"
+        seen_params.append(dict(params))
+        return page1 if len(seen_params) == 1 else page2
+
+    monkeypatch.setattr(client, "_call", fake_call)
+    rows = list(client.iter_posts("99"))
+
+    assert seen_params[0] == {"user_id": "99"}
+    # Page 2 params come verbatim from next_url's query string.
+    assert seen_params[1] == {"user_id": "99", "offset": "30"}
+    # Page-1 posts carry cursor None; page-2 posts carry the fetching cursor.
+    assert [c for _, _, c in rows[:5]] == [None] * 5
+    assert rows[5][2] == "user_id=99&offset=30"
+    assert rows[5][0]["id"] == 666
+
+
+def test_iter_posts_resumes_from_cursor(page1, monkeypatch):
+    client = PixivClient("refresh-tok", session=FakeSession())
+    captured = {}
+
+    def fake_call(endpoint, params):
+        captured["params"] = dict(params)
+        return {"illusts": [], "next_url": None}
+
+    monkeypatch.setattr(client, "_call", fake_call)
+    list(client.iter_posts("99", cursor="user_id=99&offset=60"))
+    assert captured["params"] == {"user_id": "99", "offset": "60"}
+
+
+def test_iter_posts_rejects_non_numeric_campaign():
+    client = PixivClient("refresh-tok", session=FakeSession())
+    with pytest.raises(PixivDriftError):
+        next(client.iter_posts("https://www.pixiv.net/users/99"))
+
+
+def test_iter_posts_drift_on_missing_illusts(monkeypatch):
+    client = PixivClient("refresh-tok", session=FakeSession())
+    monkeypatch.setattr(client, "_call", lambda e, p: {"error": None, "body": 1})
+    with pytest.raises(PixivDriftError):
+        next(client.iter_posts("99"))
+
+
+# -- auth ------------------------------------------------------------------------
+
+def test_login_sends_gallery_dl_profile_and_sets_bearer():
+    session = FakeSession().queue(_oauth_ok())
+    client = PixivClient("refresh-tok", session=session)
+    client._login()
+
+    method, url, data, headers = session.calls[0]
+    assert method == "POST"
+    assert url == "https://oauth.secure.pixiv.net/auth/token"
+    assert data["grant_type"] == "refresh_token"
+    assert data["refresh_token"] == "refresh-tok"
+    assert data["client_id"] == "MOBrBDS8blbauoSck0ZfDbtuzpyT"
+    assert data["get_secure_url"] == "1"
+    # X-Client-Time/-Hash pair: ISO + literal +00:00, md5 hex digest.
+    assert headers["X-Client-Time"].endswith("+00:00")
+    assert len(headers["X-Client-Hash"]) == 32
+    assert session.headers["Authorization"] == "Bearer acc-token"
+    # Token is fresh → a second login is a no-op (no extra POST).
+    client._login()
+    assert len(session.calls) == 1
+
+
+def test_login_maps_rejection_to_auth_error():
+    session = FakeSession().queue(FakeResponse(400, {"has_error": True}))
+    client = PixivClient("refresh-tok", session=session)
+    with pytest.raises(PixivAuthError):
+        client._login()
+
+
+def test_login_without_token_is_auth_error():
+    client = PixivClient(None, session=FakeSession())
+    with pytest.raises(PixivAuthError):
+        client._login()
+
+
+def test_call_maps_rate_limit_message():
+    session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(403, {
+        "error": {"message": "Rate Limit", "user_message": ""},
+    }))
+    client = PixivClient("refresh-tok", session=session)
+    with pytest.raises(PixivAPIError) as exc_info:
+        client._call("/v1/user/illusts", {"user_id": "1"})
+    err = exc_info.value
+    assert not isinstance(err, PixivAuthError)
+    assert err.status_code == 429
+    assert err.retry_after == 300.0
+
+
+def test_call_maps_403_to_auth_error():
+    session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(403, {
+        "error": {"message": "invalid access token", "user_message": ""},
+    }))
+    client = PixivClient("refresh-tok", session=session)
+    with pytest.raises(PixivAuthError):
+        client._call("/v1/user/illusts", {"user_id": "1"})
+
+
+def test_call_maps_404_status():
+    session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(404, {
+        "error": {"message": "Not Found", "user_message": ""},
+    }))
+    client = PixivClient("refresh-tok", session=session)
+    with pytest.raises(PixivAPIError) as exc_info:
+        client._call("/v1/user/illusts", {"user_id": "1"})
+    assert exc_info.value.status_code == 404
+
+
+def test_verify_auth_reports_account():
+    session = FakeSession().queue(_oauth_ok())
+    client = PixivClient("refresh-tok", session=session)
+    ok, message = client.verify_auth()
+    assert ok is True
+    assert "operator" in message
+
+
+def test_verify_auth_bad_token():
+    session = FakeSession().queue(FakeResponse(400, {"has_error": True}))
+    client = PixivClient("bad", session=session)
+    ok, message = client.verify_auth()
+    assert ok is False
+    assert "rotate" in message.lower() or "rejected" in message.lower()
+
+
+# -- rating ------------------------------------------------------------------------
+
+def test_rating_label():
+    assert rating_label(0) == "General"
+    assert rating_label(1) == "R-18"
+    assert rating_label(2) == "R-18G"
+    assert rating_label(None) is None
+    assert rating_label(True) is None
+    assert rating_label(9) is None
@@ -0,0 +1,275 @@
+"""Unit tests for PixivDownloader — no network, no DB.
+
+The HTTP layer is stubbed via the `session` seam (a fake session whose `.get`
+returns canned streaming bytes). Media items come from the real
+PixivClient.extract_media over the shared fixture page, so the downloader is
+exercised against the exact shapes the client produces.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import requests
+
+from backend.app.services.pixiv_client import MediaItem, PixivClient
+from backend.app.services.pixiv_downloader import PixivDownloader
+
+_FIXTURE = Path(__file__).parent / "fixtures" / "pixiv_user_illusts_page1.json"
+
+# Minimal valid PNG so the file_validator passes on the happy path.
+_PNG_HEAD = b"\x89PNG\r\n\x1a\n"
+_PNG_TAIL = b"\x00\x00\x00\x00IEND\xaeB`\x82"
+_PNG_BYTES = _PNG_HEAD + (b"\x00" * 16) + _PNG_TAIL
+
+
+class _FakeResponse:
+    def __init__(self, payload: bytes, status_code: int = 200, headers=None):
+        self._payload = payload
+        self.status_code = status_code
+        self.headers = headers or {}
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise requests.HTTPError(f"HTTP {self.status_code}")
+        return None
+
+    def iter_content(self, chunk_size=65536):
+        for i in range(0, len(self._payload), chunk_size):
+            yield self._payload[i : i + chunk_size]
+
+
+class _FakeSession:
+    """Records GETs and returns canned bytes (per-URL, default _PNG_BYTES)."""
+
+    def __init__(self, payloads: dict[str, bytes] | None = None):
+        self.payloads = payloads or {}
+        self.calls: list[str] = []
+
+    def get(self, url, stream=False, timeout=None, headers=None):
+        self.calls.append(url)
+        return _FakeResponse(self.payloads.get(url, _PNG_BYTES))
+
+
+def _client():
+    # Parsing only — extract_media for non-ugoira works never issues a request.
+    return PixivClient("tok", session=_FakeSession())
+
+
+def _post_and_media(work_id):
+    page = json.loads(_FIXTURE.read_text())
+    client = _client()
+    for work in page["illusts"]:
+        if work["id"] == work_id:
+            post = client._normalize(work)
+            return post, client.extract_media(post, {})
+    raise AssertionError(f"no work {work_id} in fixture")
+
+
+def _post_only(work_id):
+    """Normalized post WITHOUT media extraction — the write_post_record tests
+    don't need media, and extracting an ugoira would hit the API seam."""
+    page = json.loads(_FIXTURE.read_text())
+    for work in page["illusts"]:
+        if work["id"] == work_id:
+            return _client()._normalize(work)
+    raise AssertionError(f"no work {work_id} in fixture")
+
+
+def _downloader(tmp_path, session=None):
+    # validate=False: the stub payload is PNG bytes regardless of the target
+    # extension, so real validation would quarantine the .jpg cases. The
+    # validate/quarantine plumbing is BaseNativeDownloader's, covered by the
+    # Patreon downloader tests.
+    return PixivDownloader(
+        tmp_path, validate=False,
+        session=session if session is not None else _FakeSession(),
+    )
+
+
+def test_download_post_writes_gallery_dl_layout(tmp_path):
+    post, media = _post_and_media(111)
+    dl = _downloader(tmp_path)
+    outcomes = dl.download_post(post, media, "artist-a")
+
+    assert [o.status for o in outcomes] == ["downloaded", "downloaded"]
+    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
+    p0 = flat / "111_Multi Page Adventure_00.png"
+    p1 = flat / "111_Multi Page Adventure_01.png"
+    assert p0.is_file()
+    assert p1.is_file()
+    # No per-post directory — pixiv's layout is flat (gallery-dl parity).
+    assert {p.parent for p in (p0, p1)} == {flat}
+
+
+def test_download_post_writes_minimal_sidecar(tmp_path):
+    post, media = _post_and_media(222)
+    dl = _downloader(tmp_path)
+    outcomes = dl.download_post(post, media, "artist-a")
+
+    assert outcomes[0].status == "downloaded"
+    sidecar = outcomes[0].path.with_suffix(".json")
+    data = json.loads(sidecar.read_text())
+    # Post-first: image identity ONLY — the body lives in the post record.
+    assert data == {
+        "category": "pixiv",
+        "id": "222",
+        "source_url": media[0].url,
+    }
+
+
+def test_download_post_skips_seen(tmp_path):
+    post, media = _post_and_media(222)
+    session = _FakeSession()
+    dl = _downloader(tmp_path, session=session)
+    outcomes = dl.download_post(post, media, "artist-a", is_seen=lambda m: True)
+    assert [o.status for o in outcomes] == ["skipped_seen"]
+    assert session.calls == []
+
+
+def test_download_post_skips_on_disk(tmp_path):
+    post, media = _post_and_media(222)
+    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
+    flat.mkdir(parents=True)
+    existing = flat / "222_Single Piece_00.jpg"
+    existing.write_bytes(b"already here")
+
+    session = _FakeSession()
+    dl = _downloader(tmp_path, session=session)
+    outcomes = dl.download_post(post, media, "artist-a")
+    assert [o.status for o in outcomes] == ["skipped_disk"]
+    assert outcomes[0].path == existing
+    assert session.calls == []
+    assert existing.read_bytes() == b"already here"
+
+
+def test_recapture_surfaces_on_disk_path_for_seen_media(tmp_path):
+    post, media = _post_and_media(222)
+    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
+    flat.mkdir(parents=True)
+    (flat / "222_Single Piece_00.jpg").write_bytes(b"kept")
+
+    dl = _downloader(tmp_path)
+    outcomes = dl.download_post(
+        post, media, "artist-a", is_seen=lambda m: True, recapture=True,
+    )
+    # On disk → surfaced with its path (relink channel); seen-but-missing
+    # stays skipped_seen (recovery's job, not recapture's).
+    assert outcomes[0].status == "skipped_disk"
+    assert outcomes[0].path is not None
+
+
+def test_recapture_does_not_redownload_missing_seen_media(tmp_path):
+    post, media = _post_and_media(222)
+    session = _FakeSession()
+    dl = _downloader(tmp_path, session=session)
+    outcomes = dl.download_post(
+        post, media, "artist-a", is_seen=lambda m: True, recapture=True,
+    )
+    assert [o.status for o in outcomes] == ["skipped_seen"]
+    assert session.calls == []
+
+
+def test_download_post_honours_should_stop(tmp_path):
+    post, media = _post_and_media(111)
+    dl = _downloader(tmp_path)
+    outcomes = dl.download_post(post, media, "artist-a", should_stop=lambda: True)
+    assert outcomes == []
+
+
+def test_filename_matches_gallery_dl_linux_restrict(tmp_path):
+    # gallery-dl on Linux replaces ONLY "/" and deletes control chars — the
+    # Windows-forbidden set (<>:"|?*) stays RAW in the on-disk name. Matching
+    # that byte-for-byte is what lets the tier-2 disk-skip recognize
+    # pre-cutover files; a stricter sanitizer would re-download them.
+    post, _ = _post_and_media(222)
+    weird = MediaItem(
+        url="https://i.pximg.net/img-original/img/2026/06/18/12/30/00/222_p0.jpg",
+        filename='222_What? A "Title"/Slash\x0900.jpg',  # ? " stay, / → _, \t dropped
+        kind="image",
+        filehash=None,
+        post_id="222",
+        media_id="p0",
+    )
+    dl = _downloader(tmp_path)
+    outcomes = dl.download_post(post, [weird], "artist-a")
+    assert outcomes[0].status == "downloaded"
+    assert outcomes[0].path.name == '222_What? A "Title"_Slash00.jpg'
+
+
+def test_gdl_clean_filename_parity():
+    from backend.app.services.pixiv_downloader import gdl_clean_filename
+    # Only "/" is replaced; the Windows-forbidden set is untouched.
+    assert gdl_clean_filename('a<>:"|?*b.png') == 'a<>:"|?*b.png'
+    assert gdl_clean_filename("a/b/c.png") == "a_b_c.png"
+    # Control chars (newline, tab, DEL) are deleted, not replaced.
+    assert gdl_clean_filename("a\nb\tc\x7fd.png") == "abcd.png"
+    # Trailing dots/spaces are NOT stripped on Linux.
+    assert gdl_clean_filename("title . .png") == "title . .png"
+
+
+def test_own_session_carries_app_referer(tmp_path):
+    # Built session (no injection) must carry the app profile — i.pximg.net
+    # 403s GETs without the app-api Referer.
+    dl = PixivDownloader(tmp_path)
+    assert dl.session.headers["Referer"] == "https://app-api.pixiv.net/"
+    assert dl.session.headers["App-OS"] == "ios"
+
+
+def test_write_post_record_enriched(tmp_path):
+    post = _post_only(111)
+    dl = _downloader(tmp_path)
+    outcome = dl.write_post_record(post, "artist-a")
+
+    assert outcome.path == (
+        tmp_path / "artist-a" / "pixiv" / "pixiv" / "_post_111.json"
+    )
+    assert outcome.title == "Multi Page Adventure"
+    assert outcome.post_type == "illust"
+    assert outcome.body_chars == len(post["attributes"]["content"])
+
+    data = json.loads(outcome.path.read_text())
+    assert data["category"] == "pixiv"
+    assert data["id"] == "111"
+    assert data["published_at"] == "2026-06-20T18:00:00+09:00"
+    assert data["rating"] == "General"
+    assert data["page_count"] == 2
+    assert data["total_bookmarks"] == 250
+    assert data["total_view"] == 1000
+    assert data["illust_ai_type"] == 1
+    assert data["series"] == {"id": 4242, "title": "Adventure Series"}
+    assert data["tags"] == [
+        {"name": "オリジナル", "translated_name": "original"},
+        {"name": "女の子", "translated_name": "girl"},
+    ]
+    assert data["user"] == {"id": 99, "account": "exartist", "name": "Example Artist"}
+    # JP tags stay human-readable on disk (ensure_ascii=False).
+    assert "オリジナル" in outcome.path.read_text()
+
+
+def test_write_post_record_rating_r18(tmp_path):
+    post = _post_only(222)
+    dl = _downloader(tmp_path)
+    outcome = dl.write_post_record(post, "artist-a")
+    data = json.loads(outcome.path.read_text())
+    assert data["rating"] == "R-18"
+    assert data["is_bookmarked"] is True
+
+
+def test_write_post_record_includes_ugoira_frames(tmp_path):
+    post = _post_only(333)
+    frames = [{"file": "000000.jpg", "delay": 90}]
+    post["_work"]["_ugoira_frames"] = frames
+    dl = _downloader(tmp_path)
+    outcome = dl.write_post_record(post, "artist-a")
+    data = json.loads(outcome.path.read_text())
+    assert data["type"] == "ugoira"
+    assert data["ugoira_frames"] == frames
+
+
+def test_write_post_record_without_id(tmp_path):
+    dl = _downloader(tmp_path)
+    outcome = dl.write_post_record({"id": None, "attributes": {}}, "artist-a")
+    assert outcome.path is None
+    assert outcome.body_chars == 0
@@ -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
@@ -135,6 +135,57 @@ async def test_update_changes_fields(db):
    assert updated.config_overrides == {"videos": False}


+@pytest.mark.asyncio
+async def test_disable_clears_failure_state(db):
+    """Disabling a source wipes its failure state so it stops showing as
+    'failing' (operator: disable subs you're not paying for)."""
+    artist = await _artist(db)
+    svc = SourceService(db)
+    rec = await svc.create(
+        artist_id=artist.id, platform="patreon", url="https://patreon.com/a",
+    )
+    source = (await db.execute(
+        select(Source).where(Source.id == rec.id)
+    )).scalar_one()
+    source.last_error = "auth failed"
+    source.error_type = "auth_error"
+    source.consecutive_failures = 5
+    await db.commit()
+
+    updated = await svc.update(rec.id, enabled=False)
+    assert updated.enabled is False
+    assert updated.last_error is None
+    assert updated.consecutive_failures == 0
+    refetched = (await db.execute(
+        select(Source).where(Source.id == rec.id)
+    )).scalar_one()
+    assert refetched.error_type is None
+
+
+@pytest.mark.asyncio
+async def test_update_while_enabled_keeps_failure_state(db):
+    """A non-disable edit must NOT wipe failure state — only the explicit
+    disable clears it (else a config tweak would hide a real failure)."""
+    artist = await _artist(db)
+    svc = SourceService(db)
+    rec = await svc.create(
+        artist_id=artist.id, platform="patreon", url="https://patreon.com/a",
+    )
+    source = (await db.execute(
+        select(Source).where(Source.id == rec.id)
+    )).scalar_one()
+    source.last_error = "auth failed"
+    source.consecutive_failures = 3
+    await db.commit()
+
+    await svc.update(rec.id, config_overrides={"videos": False})
+    refetched = (await db.execute(
+        select(Source).where(Source.id == rec.id)
+    )).scalar_one()
+    assert refetched.last_error == "auth failed"
+    assert refetched.consecutive_failures == 3
+
+
@pytest.mark.asyncio
 async def test_list_hides_sidecar_synthetic_anchors(db):
    """Filesystem-import synthetic Sources (url='sidecar:<platform>:<slug>',
@@ -22,6 +22,7 @@ from backend.app.services.subscribestar_client import (
    SubscribeStarClient,
    SubscribeStarDriftError,
    _parse_ss_datetime,
+    _split_creator_url,
 )
 from backend.app.services.subscribestar_downloader import SubscribeStarDownloader
 from backend.app.services.subscribestar_ingester import (
@@ -82,6 +83,25 @@ def test_parse_ss_datetime_variants():
    assert _parse_ss_datetime("nonsense") is None


+def test_split_creator_url_rewrites_art_to_adult():
+    # The .art age wall doesn't clear with the 18+ cookie; the same creator is
+    # reachable on .adult (Elasid, event #54116). Requests must route to .adult.
+    base, slug = _split_creator_url("https://subscribestar.art/elasid")
+    assert base == "https://subscribestar.adult"
+    assert slug == "elasid"
+    # www. prefix + trailing path still normalizes the host only.
+    base, slug = _split_creator_url("https://www.subscribestar.art/elasid/posts")
+    assert base == "https://www.subscribestar.adult"
+    assert slug == "elasid"
+
+
+def test_split_creator_url_leaves_com_and_adult_untouched():
+    assert _split_creator_url("https://subscribestar.adult/sabu")[0] == \
+        "https://subscribestar.adult"
+    assert _split_creator_url("https://www.subscribestar.com/sabu")[0] == \
+        "https://www.subscribestar.com"
+
+
 def test_date_parses_when_wrapped_in_permalink_anchor():
    # Image posts wrap the date in an <a> permalink; a plain regex missed them
    # → null dates (cheunart 2026-06-17). gallery-dl's method handles both.