Merge pull request '#69 dev→main: release v26.06.04.0' from dev into main

pg_dump + zstd in runtime image, lift Quart body cap to 1 GiB. See PR #2.

.forgejo/workflows/build.yml

@@ -329,41 +329,3 @@ jobs:
           file: Dockerfile.ml
           push: true
           tags: ${{ steps.tag.outputs.tags }}
-
-  # The desktop GPU agent (#114) — published so the operator pulls + runs it on
-  # the GPU machine instead of building locally. Independent of web/ml (its own
-  # CUDA + onnxruntime-gpu image, context = agent/). Same tag cadence.
-  build-agent:
-    runs-on: python-ci
-    container:
-      image: git.fabledsword.com/bvandeusen/ci-python:3.14
-    steps:
-      - uses: actions/checkout@v4
-
-      - name: Determine tag
-        id: tag
-        run: |
-          SHORT_SHA=$(printf '%s' "$GITHUB_SHA" | cut -c1-7)
-          if [ "${GITHUB_REF#refs/tags/}" != "${GITHUB_REF}" ]; then
-            TAG_NAME="${GITHUB_REF#refs/tags/}"
-            echo "tags=git.fabledsword.com/bvandeusen/fabledcurator-agent:${TAG_NAME}" >> "$GITHUB_OUTPUT"
-          elif [ "${GITHUB_REF##*/}" = "main" ]; then
-            echo "tags=git.fabledsword.com/bvandeusen/fabledcurator-agent:main,git.fabledsword.com/bvandeusen/fabledcurator-agent:latest,git.fabledsword.com/bvandeusen/fabledcurator-agent:c-${SHORT_SHA}" >> "$GITHUB_OUTPUT"
-          else
-            echo "tags=git.fabledsword.com/bvandeusen/fabledcurator-agent:dev" >> "$GITHUB_OUTPUT"
-          fi
-
-      - name: Login to Forgejo registry
-        uses: docker/login-action@v3
-        with:
-          registry: git.fabledsword.com
-          username: ${{ github.actor }}
-          password: ${{ secrets.RELEASE_TOKEN }}
-
-      - name: Build and push agent image
-        uses: docker/build-push-action@v5
-        with:
-          context: agent
-          file: agent/Dockerfile
-          push: true
-          tags: ${{ steps.tag.outputs.tags }}

.gitignore

@@ -69,4 +69,3 @@ Thumbs.db
 alembic/versions/__pycache__/
 *.sqlite
 *.sqlite-journal
-.superpowers/

Dockerfile

@@ -18,16 +18,13 @@ ENV PYTHONUNBUFFERED=1 \
 
 # System deps: ffmpeg (transcode + thumbnails, FC-2), unar (archives, FC-2),
 # libpq for psycopg, postgresql-client + zstd for FC-5 backup/restore
-# (pg_dump + tar --zstd), image libs, megatools (mega.nz public-link downloads
-# for off-platform file-host links, #830 — `megatools dl`; Debian-native, no
-# external MEGA apt repo needed).
+# (pg_dump + tar --zstd), image libs.
 RUN apt-get update && apt-get install -y --no-install-recommends \
         ffmpeg \
         unar \
         libpq5 \
         postgresql-client \
         zstd \
-        megatools \
         libjpeg62-turbo \
         libwebp7 \
         libpng16-16 \

agent/Dockerfile

@@ -1,27 +0,0 @@
-# FabledCurator GPU agent — runs on the desktop with the GPU.
-# CUDA + cuDNN runtime so onnxruntime-gpu can use the card (it needs cuDNN 9 —
-# the plain -runtime image lacks it: "libcudnn.so.9: cannot open shared object
-# file"); ffmpeg for video frames.
-FROM nvidia/cuda:12.4.1-cudnn-runtime-ubuntu22.04
-
-ENV DEBIAN_FRONTEND=noninteractive PYTHONUNBUFFERED=1
-RUN apt-get update \
-    && apt-get install -y --no-install-recommends python3 python3-pip ffmpeg \
-    && rm -rf /var/lib/apt/lists/*
-
-WORKDIR /app
-# torch from the CUDA-12.4 wheel index (matches the base image); its wheels
-# bundle their own CUDA + cuDNN and coexist with onnxruntime-gpu. Installed
-# first + separately so the GPU build of torch is deterministic and layer-cached.
-RUN pip3 install --no-cache-dir torch==2.6.0 --index-url https://download.pytorch.org/whl/cu124
-COPY requirements.txt .
-RUN pip3 install --no-cache-dir -r requirements.txt
-COPY fc_agent ./fc_agent
-
-# imgutils ONNX models + the transformers SigLIP weights both cache here; mount
-# a volume to persist them across restarts (the SigLIP download is ~3.5 GB once).
-ENV HF_HOME=/models
-EXPOSE 8770
-
-# The control UI; the worker is started from it (or POST /start).
-CMD ["uvicorn", "fc_agent.app:app", "--host", "0.0.0.0", "--port", "8770"]

agent/README.md

@@ -1,71 +0,0 @@
-# FabledCurator GPU agent
-
-A desktop-GPU worker that embeds characters (CCIP) + figure crops for
-FabledCurator. It talks to FC **only over HTTP** — it leases jobs, fetches image
-pixels, runs the models on your GPU, and posts results back. Your FC database and
-Redis stay private; the agent never touches them.
-
-You run it when you want a burst and stop it to reclaim the card.
-
-## 0. Host prerequisite — NVIDIA Container Toolkit
-Docker needs the toolkit to hand the GPU to a container (else: *"could not select
-device driver nvidia with capabilities [[gpu]]"*). On Arch/CachyOS:
-```sh
-sudo pacman -S nvidia-container-toolkit
-sudo nvidia-ctk runtime configure --runtime=docker
-sudo systemctl restart docker
-# verify:
-docker run --rm --gpus all nvidia/cuda:12.4.1-base-ubuntu22.04 nvidia-smi
-```
-
-## 1. Get a token
-In FC: **Settings → Tagging → GPU agent → Generate token** (or Rotate). Copy it.
-
-## 2. Pull (CI publishes it alongside the web/ml images)
-```sh
-docker pull git.fabledsword.com/bvandeusen/fabledcurator-agent:latest
-```
-> Local build for development instead: `docker build -t fc-gpu-agent agent/`
-
-## 3. Run (on the machine with the GPU)
-```sh
-docker run --rm --gpus all -p 8770:8770 \
-  -e FC_URL=http://curator.traefik.internal \
-  -e FC_TOKEN=<paste-the-token> \
-  -v fc-agent-models:/models \
-  git.fabledsword.com/bvandeusen/fabledcurator-agent:latest
-```
-Then open <http://localhost:8770> — the control page. Click **Start** to begin
-draining the queue; **Pause**/**Stop** to yield the GPU. The `-v fc-agent-models`
-volume caches the downloaded ONNX models so restarts are fast.
-
-Kick off a backfill from FC (**GPU agent card → Queue character embedding**), then
-watch the queue counts on the control page (or FC's card) drain.
-
-## Config (env)
-| var | default | meaning |
-|---|---|---|
-| `FC_URL` | `http://localhost:8000` | FC base URL |
-| `FC_TOKEN` | — | the bearer token (required) |
-| `AGENT_ID` | `desktop-agent` | identifies this agent's leases |
-| `BATCH_SIZE` | `4` | jobs leased per round (still processed one at a time) |
-| `CCIP_MODEL` | imgutils default | CCIP model name |
-| `DETECTOR_LEVEL` | `m` | person-detector size: `n` < `s` < `m` < `x` |
-| `POLL_IDLE_SECONDS` | `10` | wait between empty leases |
-
-## ⚠️ Verify on first run
-This part can't be CI-tested (no GPU/models in CI), so confirm against your
-installed `dghs-imgutils` (`pip show dghs-imgutils`) — see `fc_agent/models.py`:
-- `imgutils.detect.detect_person(image, level=...)` returns
-  `[((x0,y0,x1,y1), label, score), ...]`.
-- `imgutils.metrics.ccip_extract_feature(image, model=...)` returns a vector
-  (768-d for caformer). If you want the F1-0.94 variant, set
-  `CCIP_MODEL=ccip-caformer_b36-24` (verify the exact string in imgutils).
-
-If FC's matcher under/over-fires, tune the cosine threshold in
-`backend/app/services/ml/ccip.py` (`DEFAULT_SIM_THRESHOLD`) and use
-`GET /api/ccip/overview` + `/api/ccip/images/<id>` to spot-check.
-
-## CPU fallback
-Swap `onnxruntime-gpu` → `onnxruntime` in `requirements.txt` and drop `--gpus all`
-to grind it slowly on the server instead. Same agent, no card.

agent/docker-compose.yml

@@ -1,53 +0,0 @@
-# FabledCurator GPU agent — desktop run via docker compose.
-#
-# Usage:
-#   1. Generate a token: FC → Settings → Tagging → GPU agent → Generate token.
-#   2. Create a .env next to this file:
-#        FC_URL=http://curator.traefik.internal
-#        FC_TOKEN=<paste-the-token>
-#        # optional: CCIP_MODEL=ccip-caformer_b36-24   (the F1-0.94 variant)
-#   3. docker compose up -d     (pulls the published image)
-#   4. Open http://localhost:8770 → Start. Pause/Stop hands the GPU back.
-#      docker compose down       to stop the container entirely.
-#
-# Surviving a curator redeploy (you're away, can't touch the agent):
-#   - A running agent rides out curator being unreachable on its own — it retries
-#     leasing with capped backoff and resumes when the server is back. In-flight
-#     work is handed back (not failed), so a redeploy never poisons good jobs.
-#   - AUTO_START=1 (below) also resumes the worker if the AGENT container itself
-#     restarts (host reboot / crash via `restart: unless-stopped`) — no click.
-#
-# Needs the NVIDIA Container Toolkit installed on the host for --gpus.
-
-services:
-  fc-gpu-agent:
-    image: git.fabledsword.com/bvandeusen/fabledcurator-agent:latest
-    pull_policy: always
-    ports:
-      - "8770:8770"
-    environment:
-      FC_URL: ${FC_URL:-http://curator.traefik.internal}
-      FC_TOKEN: ${FC_TOKEN:?set FC_TOKEN in .env (FC → GPU agent → Generate token)}
-      CCIP_MODEL: ${CCIP_MODEL:-}
-      DETECTOR_LEVEL: ${DETECTOR_LEVEL:-m}
-      BATCH_SIZE: ${BATCH_SIZE:-4}
-      # Resume the worker automatically on container start (survive a reboot /
-      # crash-restart while you're away). Set to 0 to require a manual Start.
-      AUTO_START: ${AUTO_START:-1}
-      # Crop embedder (SigLIP concept bag): float16 keeps VRAM low on a shared
-      # desktop GPU; the model itself is announced by the server.
-      SIGLIP_DTYPE: ${SIGLIP_DTYPE:-float16}
-    volumes:
-      # Persist the downloaded ONNX models so restarts are fast.
-      - fc-agent-models:/models
-    restart: unless-stopped
-    deploy:
-      resources:
-        reservations:
-          devices:
-            - driver: nvidia
-              count: all
-              capabilities: [gpu]
-
-volumes:
-  fc-agent-models:

agent/fc_agent/app.py

@@ -1,134 +0,0 @@
-"""FastAPI control surface for the agent (served on localhost).
-
-Start / stop the worker pool, tune the worker count live (trades desktop
-responsiveness for throughput), and watch GPU load + progress + the server-side
-queue. Config is env-seeded; the worker count is adjustable here on the fly.
-"""
-from fastapi import FastAPI, Request
-from fastapi.responses import HTMLResponse, JSONResponse
-
-from .config import Config
-from .gpu import read_gpu
-from .worker import Worker
-
-cfg = Config.from_env()
-worker = Worker(cfg)
-app = FastAPI(title="FabledCurator GPU agent")
-
-
-@app.on_event("startup")
-def _maybe_autostart() -> None:
-    # With AUTO_START set, a container restart (host reboot, or `restart:
-    # unless-stopped` after a crash) resumes the worker on its own — the slots
-    # then ride out a still-down curator via lease backoff. Lets the agent
-    # survive a redeploy with nobody at the desktop to click Start.
-    if cfg.auto_start and cfg.token:
-        worker.start()
-
-
-@app.get("/", response_class=HTMLResponse)
-def index() -> str:
-    return _PAGE
-
-
-@app.post("/start")
-def start():
-    worker.start()
-    return JSONResponse(worker.status())
-
-
-@app.post("/stop")
-def stop():
-    worker.stop()
-    return JSONResponse(worker.status())
-
-
-@app.post("/concurrency")
-async def concurrency(request: Request):
-    body = await request.json()
-    worker.set_concurrency(int(body.get("value", 1)))
-    return JSONResponse(worker.status())
-
-
-@app.get("/status")
-def status():
-    s = worker.status()
-    s["fc_url"] = cfg.fc_url
-    s["configured"] = bool(cfg.token)
-    s["gpu"] = read_gpu()
-    try:
-        s["queue"] = worker.client.queue_status()
-    except Exception:
-        s["queue"] = None
-    return JSONResponse(s)
-
-
-_PAGE = """<!doctype html><html><head><meta charset=utf-8>
-<title>FabledCurator GPU agent</title>
-<style>
- body{font:14px system-ui;margin:2rem;max-width:680px;background:#14171a;color:#e8e8e8}
- h1{font-size:18px} button{font:14px system-ui;padding:.5rem 1rem;border:0;border-radius:6px;
- margin-right:.5rem;cursor:pointer;color:#fff} .start{background:#2e7d32}.stop{background:#b3261e}
- .step{background:#33373b;padding:.4rem .7rem;font-weight:700}
- .stat{display:inline-block;margin-right:1.5rem;vertical-align:top}
- .n{font-size:22px;font-weight:700} code{background:#222;padding:2px 6px;border-radius:4px}
- .q,.gpu{margin-top:1rem;color:#9aa} .bar{height:8px;border-radius:4px;background:#222;overflow:hidden;
- max-width:320px;margin-top:4px} .bar>i{display:block;height:100%;background:#3f7d3f}
- .row{margin:.8rem 0}
-</style></head><body>
-<h1>FabledCurator GPU agent</h1>
-<p>FC: <code id=fc>—</code> · token <code id=cfg>—</code></p>
-<div class=row>
- <button class=start onclick=act('start')>Start</button>
- <button class=stop onclick=act('stop')>Stop</button>
-</div>
-<div class=row>
- workers
- <button class=step onclick=setc(-1)>−</button>
- <input id=conc type=number min=1 value=1
-  style="width:3.5rem;font:700 16px system-ui;text-align:center;background:#222;color:#e8e8e8;border:1px solid #444;border-radius:6px;padding:.3rem"
-  onchange="setv(this.value)">
- <button class=step onclick=setc(1)>+</button>
- <span class=cap style=color:#9aa>(more = overlap I/O, fill the GPU) max <b id=capn>8</b></span>
-</div>
-<div class=row>
- <span class=stat><span class=n id=state>stopped</span><br>state</span>
- <span class=stat><span class=n id=active>0</span><br>active now</span>
- <span class=stat><span class=n id=done>0</span><br>processed</span>
- <span class=stat><span class=n id=err>0</span><br>errors</span>
- <span class=stat><span class=n id=wait>0</span><br>waited out</span>
-</div>
-<div id=banner style="display:none;margin:.6rem 0;padding:.5rem .8rem;border-radius:6px;background:#5a4a17;color:#ffe28a">
- curator unreachable — holding work + retrying, will resume on its own (no restart needed)
-</div>
-<div class=gpu id=gpu>GPU — …</div>
-<div class=bar><i id=gpubar style=width:0%></i></div>
-<div class=q id=queue></div>
-<script>
- let CAP=8
- async function act(p){await fetch('/'+p,{method:'POST'});refresh()}
- function setc(d){ setv((parseInt(conc.value||'1'))+d) }
- async function setv(v){
-  v=Math.max(1,Math.min(CAP,parseInt(v)||1)); conc.value=v
-  await fetch('/concurrency',{method:'POST',headers:{'Content-Type':'application/json'},
-   body:JSON.stringify({value:v})});refresh()
- }
- async function refresh(){
-  const s=await (await fetch('/status')).json()
-  CAP=s.max_concurrency||8; capn.textContent=CAP
-  state.textContent=s.state; active.textContent=s.active; done.textContent=s.processed
-  err.textContent=s.errors; fc.textContent=s.fc_url; wait.textContent=s.transient||0
-  // Running but the queue read failed → curator is unreachable; show we're
-  // riding it out rather than erroring.
-  banner.style.display=(s.state==='running' && !s.queue)?'block':'none'
-  if(document.activeElement!==conc) conc.value=s.concurrency
-  conc.max=CAP
-  cfg.textContent=s.configured?'set':'MISSING'
-  if(s.gpu){
-   gpu.textContent=`GPU — ${s.gpu.util_pct}% util · VRAM ${s.gpu.mem_used_mb}/${s.gpu.mem_total_mb} MB · ${s.gpu.temp_c}°C`
-   gpubar.style.width=Math.round(100*s.gpu.mem_used_mb/s.gpu.mem_total_mb)+'%'
-  } else { gpu.textContent='GPU — n/a (CPU fallback?)'; gpubar.style.width='0%' }
-  queue.textContent=s.queue?`queue — pending ${s.queue.pending} · in flight ${s.queue.leased} · done ${s.queue.done} · errored ${s.queue.error}`:'queue — unreachable'
- }
- refresh(); setInterval(refresh,3000)
-</script></body></html>"""

agent/fc_agent/client.py

@@ -1,85 +0,0 @@
-"""HTTP client for the FabledCurator GPU-job API.
-
-The agent's ONLY contact with FC — lease/submit/heartbeat/fail + fetch image
-bytes, all over HTTP with the bearer token. No DB/Redis.
-"""
-import requests
-from requests.adapters import HTTPAdapter
-
-
-class FcClient:
-    def __init__(self, base_url: str, token: str, agent_id: str):
-        self.base = base_url.rstrip("/")
-        self.agent_id = agent_id
-        self.s = requests.Session()
-        self.s.headers["Authorization"] = f"Bearer {token}"
-        # Many worker threads share this Session; the default pool (10) would
-        # throttle them + spam "connection pool is full". Size it for the cap.
-        adapter = HTTPAdapter(pool_connections=64, pool_maxsize=64)
-        self.s.mount("http://", adapter)
-        self.s.mount("https://", adapter)
-
-    def lease(self, batch_size: int) -> list[dict]:
-        r = self.s.post(
-            f"{self.base}/api/gpu/jobs/lease",
-            json={"agent_id": self.agent_id, "batch_size": batch_size},
-            timeout=30,
-        )
-        r.raise_for_status()
-        return r.json().get("jobs", [])
-
-    def submit(self, job_id: int, regions: list[dict], replace_kinds: list[str]) -> dict:
-        r = self.s.post(
-            f"{self.base}/api/gpu/jobs/submit",
-            json={
-                "agent_id": self.agent_id, "job_id": job_id,
-                "regions": regions, "replace_kinds": replace_kinds,
-            },
-            timeout=120,
-        )
-        r.raise_for_status()
-        return r.json()
-
-    def heartbeat(self, job_ids: list[int]) -> None:
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/heartbeat",
-                json={"agent_id": self.agent_id, "job_ids": job_ids},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
-
-    def fail(self, job_id: int, error: str) -> None:
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/fail",
-                json={"agent_id": self.agent_id, "job_id": job_id, "error": error},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
-
-    def release(self, job_ids: list[int]) -> None:
-        # Graceful hand-back on stop so orphaned work is re-leased at once.
-        if not job_ids:
-            return
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/release",
-                json={"agent_id": self.agent_id, "job_ids": job_ids},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
-
-    def fetch_image(self, image_url: str) -> bytes:
-        # image_url is a server-relative path ("/images/...").
-        r = self.s.get(f"{self.base}{image_url}", timeout=180)
-        r.raise_for_status()
-        return r.content
-
-    def queue_status(self) -> dict:
-        r = self.s.get(f"{self.base}/api/gpu/status", timeout=15)
-        r.raise_for_status()
-        return r.json()

agent/fc_agent/config.py

@@ -1,36 +0,0 @@
-"""Agent config, all from env (the control container is configured at run)."""
-import os
-from dataclasses import dataclass
-
-
-@dataclass
-class Config:
-    fc_url: str          # base URL of the FabledCurator web service
-    token: str           # the bearer token from Settings → Tagging → GPU agent
-    agent_id: str        # identifies this agent's leases
-    batch_size: int      # jobs a worker leases per round
-    concurrency: int     # INITIAL parallel workers (tunable live from the UI)
-    ccip_model: str      # imgutils CCIP model name ("" → imgutils default)
-    detector_level: str  # imgutils person-detector level: n|s|m|x
-    poll_idle_seconds: float   # wait between empty leases
-    embed_dtype: str     # torch dtype for the crop embedder: float16|float32
-    embed_model_override: str  # force a SigLIP-family model ("" → use the one
-    #                            the server announces in the lease)
-    auto_start: bool     # start the worker pool on boot (so a container restart
-    #                      resumes processing without anyone clicking Start)
-
-    @classmethod
-    def from_env(cls) -> "Config":
-        return cls(
-            fc_url=os.environ.get("FC_URL", "http://localhost:8000").rstrip("/"),
-            token=os.environ.get("FC_TOKEN", ""),
-            agent_id=os.environ.get("AGENT_ID", "desktop-agent"),
-            batch_size=int(os.environ.get("BATCH_SIZE", "4")),
-            concurrency=int(os.environ.get("CONCURRENCY", "1")),
-            ccip_model=os.environ.get("CCIP_MODEL", ""),
-            detector_level=os.environ.get("DETECTOR_LEVEL", "m"),
-            poll_idle_seconds=float(os.environ.get("POLL_IDLE_SECONDS", "10")),
-            embed_dtype=os.environ.get("SIGLIP_DTYPE", "float16"),
-            embed_model_override=os.environ.get("EMBED_MODEL_NAME", ""),
-            auto_start=os.environ.get("AUTO_START", "").lower() in ("1", "true", "yes"),
-        )

agent/fc_agent/crops.py

@@ -1,36 +0,0 @@
-"""Crop primitive — vendored from backend/app/services/ml/crops.py so the agent
-is self-contained. Keep in sync if the floor logic changes."""
-from PIL import Image
-
-MIN_CROP_FRACTION = 0.10
-MIN_CROP_PX = 64
-
-
-def crop_region(
-    img: Image.Image,
-    bbox: tuple[float, float, float, float],
-    *,
-    pad: float = 0.0,
-    min_fraction: float = MIN_CROP_FRACTION,
-    min_px: int = MIN_CROP_PX,
-) -> Image.Image | None:
-    """Crop a NORMALIZED bbox (x, y, w, h in [0,1]); None if below the size
-    floor (max of a fraction-of-short-side and an absolute pixel floor)."""
-    iw, ih = img.size
-    x, y, w, h = bbox
-    px, py, pw, ph = x * iw, y * ih, w * iw, h * ih
-    if pad:
-        px -= pw * pad / 2.0
-        py -= ph * pad / 2.0
-        pw *= (1.0 + pad)
-        ph *= (1.0 + pad)
-    left = max(0, int(round(px)))
-    top = max(0, int(round(py)))
-    right = min(iw, int(round(px + pw)))
-    bottom = min(ih, int(round(py + ph)))
-    if right <= left or bottom <= top:
-        return None
-    floor = max(min_px, int(min_fraction * min(iw, ih)))
-    if min(right - left, bottom - top) < floor:
-        return None
-    return img.crop((left, top, right, bottom)).convert("RGB")

agent/fc_agent/embedder.py

@@ -1,69 +0,0 @@
-"""Crop EMBEDDER for the concept bag — model-agnostic (CLIP/SigLIP-family).
-
-The server trains its per-concept heads in the embedding space of whatever model
-its `embedder_model_version` names; a crop must be embedded with the SAME model
-or its vector lands in a different coordinate system and every head misfires. So
-the model identity (HF name + version) is ANNOUNCED BY THE SERVER in the lease —
-nothing here is hardcoded to SigLIP. Whatever name the server sends is loaded via
-transformers `get_image_features` (the CLIP/SigLIP-family image-tower call); a
-non-CLIP backbone (e.g. a DINO encoder) would need its own pooling adapter.
-
-torch on CUDA, fp16 by default to keep VRAM low on a shared desktop GPU — the
-tiny fp16-vs-fp32 difference is negligible for the linear heads (cosine ~0.999).
-A single inference lock serializes the forward pass: the pipeline is I/O-bound,
-so the GPU isn't the bottleneck, and one model shared across worker threads is
-safest behind a lock.
-"""
-import threading
-
-import numpy as np
-from PIL import Image
-
-
-class CropEmbedder:
-    def __init__(self, model_name: str, dtype: str = "float16"):
-        self._name = model_name
-        self._dtype_name = dtype
-        self._model = None
-        self._processor = None
-        self._torch = None
-        self._device = None
-        self._dt = None
-        self._load_lock = threading.Lock()
-        self._infer_lock = threading.Lock()
-
-    @property
-    def model_name(self) -> str:
-        return self._name
-
-    def load(self) -> None:
-        if self._model is not None:
-            return
-        with self._load_lock:
-            if self._model is not None:
-                return
-            import torch
-            from transformers import AutoImageProcessor, AutoModel
-
-            self._torch = torch
-            self._device = "cuda" if torch.cuda.is_available() else "cpu"
-            dt = getattr(torch, self._dtype_name, torch.float16)
-            if self._device == "cpu":
-                dt = torch.float32   # fp16 matmul is unsupported/slow on CPU
-            self._dt = dt
-            self._processor = AutoImageProcessor.from_pretrained(self._name)
-            model = AutoModel.from_pretrained(self._name, torch_dtype=dt)
-            model.eval().to(self._device)
-            self._model = model
-
-    def embed(self, image: Image.Image) -> list[float]:
-        """A crop → its embedding as a plain float list, ready to POST."""
-        self.load()
-        torch = self._torch
-        enc = self._processor(images=image, return_tensors="pt")
-        pixel_values = enc["pixel_values"].to(self._device, self._dt)
-        with self._infer_lock, torch.no_grad():
-            out = self._model.get_image_features(pixel_values=pixel_values)
-        pooled = out.pooler_output if hasattr(out, "pooler_output") else out
-        vec = pooled[0].float().cpu().numpy().astype(np.float32).reshape(-1)
-        return vec.tolist()

agent/fc_agent/gpu.py

@@ -1,30 +0,0 @@
-"""GPU load readout via nvidia-smi (present in the container thanks to the
-NVIDIA Container Toolkit's `utility` capability). Returns None if unavailable —
-the UI just shows n/a (e.g. CPU-fallback run)."""
-import subprocess
-
-
-def read_gpu() -> dict | None:
-    try:
-        out = subprocess.run(
-            [
-                "nvidia-smi",
-                "--query-gpu=utilization.gpu,memory.used,memory.total,temperature.gpu",
-                "--format=csv,noheader,nounits",
-            ],
-            capture_output=True, text=True, timeout=5, check=True,
-        ).stdout.strip().splitlines()
-    except (OSError, subprocess.SubprocessError):
-        return None
-    if not out:
-        return None
-    parts = [p.strip() for p in out[0].split(",")]
-    try:
-        return {
-            "util_pct": int(float(parts[0])),
-            "mem_used_mb": int(float(parts[1])),
-            "mem_total_mb": int(float(parts[2])),
-            "temp_c": int(float(parts[3])),
-        }
-    except (ValueError, IndexError):
-        return None

agent/fc_agent/media.py

@@ -1,63 +0,0 @@
-"""Image + video handling. Stills load directly; videos are sampled into frames
-(ffmpeg) at the cadence FC sends — so a video becomes a bag of per-frame
-instances, each with a timestamp."""
-import io
-import os
-import subprocess
-import tempfile
-
-from PIL import Image
-
-
-def is_video(mime: str) -> bool:
-    return bool(mime) and (mime.startswith("video/") or mime in {"image/gif"})
-
-
-def to_rgb(img: Image.Image) -> Image.Image:
-    """RGB, flattening any transparency onto white first. A naive convert('RGB')
-    on a palette-with-transparency image (common for character PNGs on a clear
-    background) lets PIL guess the transparent pixels — usually black artifacts
-    that bleed into the crop + the embedding (and the "should be converted to
-    RGBA" warning). Compositing over white gives a clean, consistent background."""
-    if img.mode in ("RGBA", "LA", "PA") or (
-        img.mode == "P" and "transparency" in img.info
-    ):
-        img = img.convert("RGBA")
-        bg = Image.new("RGBA", img.size, (255, 255, 255, 255))
-        return Image.alpha_composite(bg, img).convert("RGB")
-    return img.convert("RGB")
-
-
-def load_image(data: bytes) -> Image.Image:
-    return to_rgb(Image.open(io.BytesIO(data)))
-
-
-def sample_frames(
-    data: bytes, interval_seconds: float, max_frames: int
-) -> list[tuple[float, Image.Image]]:
-    """Extract up to max_frames frames at one-every-interval_seconds via ffmpeg.
-    Returns [(timestamp_seconds, frame)]. Empty on failure (caller falls back)."""
-    interval = max(0.5, float(interval_seconds or 4.0))
-    cap = max(1, int(max_frames or 64))
-    with tempfile.TemporaryDirectory() as tmp:
-        src = os.path.join(tmp, "in")
-        with open(src, "wb") as fh:
-            fh.write(data)
-        pattern = os.path.join(tmp, "f_%05d.jpg")
-        try:
-            subprocess.run(
-                [
-                    "ffmpeg", "-nostdin", "-loglevel", "error", "-i", src,
-                    "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
-                    "-q:v", "3", pattern,
-                ],
-                check=True, timeout=600,
-            )
-        except (subprocess.SubprocessError, FileNotFoundError):
-            return []
-        out: list[tuple[float, Image.Image]] = []
-        names = sorted(n for n in os.listdir(tmp) if n.startswith("f_"))
-        for i, name in enumerate(names[:cap]):
-            with Image.open(os.path.join(tmp, name)) as im:
-                out.append((round(i * interval, 2), to_rgb(im)))
-        return out

agent/fc_agent/models.py

@@ -1,39 +0,0 @@
-"""imgutils model wrappers — the figure DETECTOR + the CCIP EMBEDDER.
-
-⚠️ VERIFY ON FIRST RUN: the exact imgutils function names/signatures + the CCIP
-model string can drift between dghs-imgutils releases. These are the two seams to
-check against your installed version (`pip show dghs-imgutils`):
-  - detect_person(image, level=...) -> [((x0,y0,x1,y1), label, score), ...]
-  - ccip_extract_feature(image, model=...) -> a vector (768-d for caformer)
-imgutils auto-downloads the ONNX models from HuggingFace on first use; GPU is
-used when onnxruntime-gpu is installed.
-"""
-import numpy as np
-from PIL import Image
-
-
-def detect_figures(image: Image.Image, level: str = "m") -> list[tuple[tuple, float | None]]:
-    """Person/figure bounding boxes, NORMALIZED (x, y, w, h in [0,1]) + score.
-    Returns [] if detection finds nothing (caller falls back to whole-image)."""
-    from imgutils.detect import detect_person
-
-    iw, ih = image.size
-    out = []
-    for (x0, y0, x1, y1), _label, score in detect_person(image, level=level):
-        out.append((
-            (x0 / iw, y0 / ih, (x1 - x0) / iw, (y1 - y0) / ih),
-            float(score),
-        ))
-    return out
-
-
-def ccip_vector(image: Image.Image, model: str | None = None) -> list[float]:
-    """The CCIP identity embedding of a (cropped) character image, as a plain
-    float list ready to POST."""
-    from imgutils.metrics import ccip_extract_feature
-
-    feat = (
-        ccip_extract_feature(image, model=model)
-        if model else ccip_extract_feature(image)
-    )
-    return np.asarray(feat, dtype=np.float32).reshape(-1).tolist()

agent/fc_agent/worker.py

@@ -1,274 +0,0 @@
-"""The lease → fetch → detect+embed → submit loop, run by a pool of worker
-slots whose count is tunable live from the UI.
-
-Each slot is an independent loop (its own leases; the server's SKIP-LOCKED lease
-keeps them from colliding). More slots = more GPU load + throughput; the model is
-loaded once and shared, so slots add concurrent inference, not N× model VRAM.
-That's the dial the operator turns to trade desktop responsiveness for speed.
-
-Stop (or shrinking the pool) RELEASES a slot's still-leased jobs immediately so
-orphaned work is re-picked at once rather than waiting out the lease.
-"""
-import threading
-
-import requests
-
-from . import media, models
-from .client import FcClient
-from .config import Config
-from .crops import crop_region
-
-# Cap on the lease-retry backoff: when curator is unreachable (e.g. you redeploy
-# it while away), each slot retries leasing with exponential backoff up to this
-# many seconds, then resumes within this window once the server is back — no
-# restart needed.
-MAX_BACKOFF_SECONDS = 60.0
-
-
-def _is_transient(exc: "requests.RequestException") -> bool:
-    """A server/transport problem (wait it out) vs a job-specific fault (fail it).
-    No response → connection refused/timeout → curator is down → transient. With
-    a response: 5xx, auth (401/403, e.g. a token blip on redeploy), 408/409/429
-    (timeout / our lease reclaimed / rate-limited) are all 'not this job's fault'.
-    A specific 4xx like 404 (image gone) / 400 IS the job's fault → fail it."""
-    resp = getattr(exc, "response", None)
-    if resp is None:
-        return True
-    return resp.status_code >= 500 or resp.status_code in (401, 403, 408, 409, 429)
-
-# Generous cap: the pipeline is usually I/O-bound (downloading + decoding images
-# over HTTP), so the GPU stays underused until many workers overlap that I/O.
-# Push it up while watching the GPU util + VRAM in the UI.
-MAX_CONCURRENCY = 32
-
-# Fallbacks only — the server ANNOUNCES the embedding model (name + version) in
-# the lease so the agent stays model-agnostic and in lock-step with the space
-# the heads were trained in. These cover an older server that doesn't send them.
-DEFAULT_EMBED_MODEL = "google/siglip-so400m-patch14-384"
-DEFAULT_EMBED_VERSION = "siglip-so400m-patch14-384"
-
-
-class _Slot:
-    """One worker loop. `inflight` = jobs leased but not yet processed, so a
-    graceful stop can hand them back."""
-    __slots__ = ("stop", "inflight")
-
-    def __init__(self):
-        self.stop = threading.Event()
-        self.inflight: list[int] = []
-
-
-class Worker:
-    def __init__(self, cfg: Config):
-        self.cfg = cfg
-        self.client = FcClient(cfg.fc_url, cfg.token, cfg.agent_id)
-        self._lock = threading.Lock()
-        self._running = False
-        self._target = max(1, min(MAX_CONCURRENCY, cfg.concurrency))
-        self._slots: list[_Slot] = []
-        self.processed = 0
-        self.errors = 0
-        self.transient = 0        # jobs handed back due to a server outage (NOT
-        #                           failed) — the "waiting out curator" counter
-        self._active = 0          # slots currently mid-image
-        # The crop embedder (SigLIP-family) is built lazily on the first job that
-        # needs it, from the model the server announces — one shared instance.
-        self._embedder = None
-        self._embedder_lock = threading.Lock()
-
-    # --- control -----------------------------------------------------------
-    def start(self):
-        with self._lock:
-            self._running = True
-            self._reconcile_locked()
-
-    def stop(self):
-        with self._lock:
-            self._running = False
-            slots, self._slots = self._slots, []
-        for s in slots:
-            s.stop.set()          # each slot releases its inflight on exit
-
-    def set_concurrency(self, n: int):
-        with self._lock:
-            self._target = max(1, min(MAX_CONCURRENCY, int(n)))
-            if self._running:
-                self._reconcile_locked()
-
-    def _reconcile_locked(self):
-        while len(self._slots) < self._target:
-            slot = _Slot()
-            self._slots.append(slot)
-            threading.Thread(target=self._loop, args=(slot,), daemon=True).start()
-        while len(self._slots) > self._target:
-            self._slots.pop().stop.set()
-
-    def status(self) -> dict:
-        with self._lock:
-            return {
-                "state": "running" if self._running else "stopped",
-                "concurrency": self._target,
-                "max_concurrency": MAX_CONCURRENCY,
-                "workers": len(self._slots),
-                "active": self._active,
-                "processed": self.processed,
-                "errors": self.errors,
-                "transient": self.transient,
-            }
-
-    def _bump(self, *, processed=0, errors=0, active=0, transient=0):
-        with self._lock:
-            self.processed += processed
-            self.errors += errors
-            self.transient += transient
-            self._active += active
-
-    # --- per-slot loop -----------------------------------------------------
-    def _loop(self, slot: _Slot):
-        backoff = self.cfg.poll_idle_seconds
-        while not slot.stop.is_set() and self._running:
-            try:
-                jobs = self.client.lease(self.cfg.batch_size)
-                backoff = self.cfg.poll_idle_seconds      # server answered → reset
-            except Exception:
-                # curator unreachable (redeploy, network drop): wait it out with
-                # exponential backoff, capped — resume on our own when it returns.
-                self._interruptible_sleep(slot, backoff)
-                backoff = min(backoff * 2, MAX_BACKOFF_SECONDS)
-                continue
-            if not jobs:
-                self._interruptible_sleep(slot, self.cfg.poll_idle_seconds)
-                continue
-            slot.inflight = [j["job_id"] for j in jobs]
-            for job in jobs:
-                if slot.stop.is_set() or not self._running:
-                    break
-                ok = self._process(job)
-                slot.inflight = [i for i in slot.inflight if i != job["job_id"]]
-                if not ok:
-                    # Server went away mid-batch: hand the rest back (best effort)
-                    # and back off instead of hammering a recovering server or
-                    # burning the jobs' attempt budgets on fail().
-                    if slot.inflight:
-                        self.client.release(slot.inflight)
-                        slot.inflight = []
-                    self._interruptible_sleep(slot, backoff)
-                    backoff = min(backoff * 2, MAX_BACKOFF_SECONDS)
-                    break
-                if slot.inflight:
-                    self.client.heartbeat(slot.inflight)
-        # Graceful hand-back of anything leased but not processed.
-        if slot.inflight:
-            self.client.release(slot.inflight)
-            slot.inflight = []
-
-    def _interruptible_sleep(self, slot: _Slot, seconds: float):
-        """Sleep, but wake immediately if the slot is told to stop — so a Stop or
-        a pool-shrink doesn't hang for a full backoff window."""
-        slot.stop.wait(timeout=seconds)
-
-    def _ensure_embedder(self, model_name: str):
-        if self._embedder is not None:
-            return self._embedder
-        with self._embedder_lock:
-            if self._embedder is None:
-                from .embedder import CropEmbedder
-                self._embedder = CropEmbedder(model_name, self.cfg.embed_dtype)
-        return self._embedder
-
-    def _process(self, job: dict) -> bool:
-        """Process one job. Returns True when handled (completed, or hard-failed
-        because the job itself is bad) and False on a TRANSPORT error (curator
-        unreachable / 5xx / our lease was reclaimed mid-flight) — which is not
-        the job's fault, so the caller backs off and the job is left to be
-        re-leased rather than fail()ed into its attempt budget."""
-        self._bump(active=1)
-        try:
-            data = self.client.fetch_image(job["image_url"])
-            if media.is_video(job.get("mime", "")):
-                frames = media.sample_frames(
-                    data, job.get("frame_interval_seconds", 4.0),
-                    job.get("max_frames", 64),
-                ) or [(None, media.load_image(data))]
-            else:
-                frames = [(None, media.load_image(data))]
-
-            # task picks what to produce per crop:
-            #   'siglip' (backfill existing images) → concept (SigLIP) regions
-            #     ONLY, so it never churns their figure/CCIP regions or the
-            #     character-reference cache.
-            #   'ccip' / 'both' (a new image's first pass) → figure (CCIP) AND
-            #     concept (SigLIP) in one go, off the same crop.
-            task = job.get("task") or "ccip"
-            want_ccip = task in ("ccip", "both")
-            want_siglip = task in ("ccip", "siglip", "both")
-            replace_kinds = (
-                ["concept"] if task == "siglip" else ["figure", "face", "concept"]
-            )
-
-            embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION
-            embedder = None
-            if want_siglip:
-                model_name = (
-                    self.cfg.embed_model_override
-                    or job.get("embed_model_name")
-                    or DEFAULT_EMBED_MODEL
-                )
-                embedder = self._ensure_embedder(model_name)
-
-            regions = []
-            ccip_ev = self.cfg.ccip_model or "ccip-default"
-            dv = f"person-{self.cfg.detector_level}"
-            for t, frame in frames:
-                figs = models.detect_figures(frame, self.cfg.detector_level)
-                if not figs:
-                    figs = [((0.0, 0.0, 1.0, 1.0), None)]   # whole-frame fallback
-                for bbox, score in figs:
-                    crop = crop_region(frame, bbox)
-                    if crop is None:
-                        continue
-                    if want_ccip:
-                        regions.append({
-                            "kind": "figure",
-                            "bbox": list(bbox),
-                            "frame_time": t,
-                            "score": score,
-                            "ccip_embedding": models.ccip_vector(
-                                crop, self.cfg.ccip_model or None
-                            ),
-                            "embedding_version": ccip_ev,
-                            "detector_version": dv,
-                        })
-                    if want_siglip:
-                        regions.append({
-                            "kind": "concept",
-                            "bbox": list(bbox),
-                            "frame_time": t,
-                            "score": score,
-                            "siglip_embedding": embedder.embed(crop),
-                            "embedding_version": embed_version,
-                            "detector_version": dv,
-                        })
-            self.client.submit(job["job_id"], regions, replace_kinds)
-            self._bump(processed=1)
-            return True
-        except requests.RequestException as exc:
-            if _is_transient(exc):
-                # curator down/redeploying, a 5xx, or our lease was reclaimed
-                # while we worked. NOT the job's fault — hand it back (best
-                # effort; no-ops if the server is still down, then the server's
-                # orphan-recovery reclaims it) and signal the loop to wait.
-                self._bump(transient=1)
-                self.client.release([job["job_id"]])
-                return False
-            # A job-specific HTTP fault (404 image gone, 400) → fail it so it
-            # doesn't re-lease forever.
-            self._bump(errors=1)
-            self.client.fail(job["job_id"], str(exc)[:500])
-            return True
-        except Exception as exc:   # noqa: BLE001 — a genuine job fault: report it
-            self._bump(errors=1)
-            self.client.fail(job["job_id"], str(exc)[:500])
-            return True
-        finally:
-            self._bump(active=-1)

agent/requirements.txt

@@ -1,15 +0,0 @@
-# CCIP + figure detection (ONNX models, auto-downloaded from HuggingFace).
-dghs-imgutils>=0.4
-# GPU inference for the ONNX models. Swap to onnxruntime (CPU) for a slow
-# server-side fallback run.
-onnxruntime-gpu
-# The crop EMBEDDER (concept bag). torch is installed separately in the
-# Dockerfile from the CUDA-12.4 wheel index so the GPU build is deterministic;
-# transformers loads whatever SigLIP-family model the server announces.
-transformers>=4.45
-# Control surface + HTTP.
-fastapi
-uvicorn[standard]
-requests
-pillow
-numpy

alembic/env.py

@@ -1,28 +1,13 @@
 """Alembic environment — reads DATABASE_URL from app config."""
 
-import os
-import re
 from logging.config import fileConfig
 
-from sqlalchemy import engine_from_config, pool, text
+from sqlalchemy import engine_from_config, pool
 
 from alembic import context
 from backend.app.config import get_config
 from backend.app.models import Base
 
-# Fail a blocked migration FAST instead of hanging forever. Migrations run
-# against the live DB while workers hold locks; 0040's `ALTER series_page` queued
-# behind a tag-merge that held a series_page lock for minutes (the merge runs an
-# unindexed full scan over image_record while repointing series_page) and hung
-# with no timeout — silent, indefinite (operator-flagged 2026-06-07). With a
-# lock_timeout a blocked DDL errors ("canceling statement due to lock timeout")
-# and the entrypoint's `alembic upgrade head` exits non-zero, so the deploy
-# retries / surfaces loudly rather than wedging. Override via env when a known
-# slow-lock window is expected.
-_MIGRATION_LOCK_TIMEOUT = os.environ.get("MIGRATION_LOCK_TIMEOUT", "30s")
-if not re.fullmatch(r"\d+\s*(ms|s|min)?", _MIGRATION_LOCK_TIMEOUT.strip()):
-    _MIGRATION_LOCK_TIMEOUT = "30s"  # ignore a malformed override
-
 config = context.config
 
 if config.config_file_name is not None:
@@ -53,15 +38,6 @@ def run_migrations_online() -> None:
         poolclass=pool.NullPool,
     )
     with connectable.connect() as connection:
-        # Session-level lock_timeout for every DDL statement in this run. Set
-        # (and commit) before alembic opens its own transaction so the GUC
-        # persists on this connection regardless of how alembic structures its
-        # transactions. Value is from our own env, so f-string interpolation is
-        # safe (and it's been pattern-validated above); SET takes no bind params.
-        connection.execute(
-            text(f"SET lock_timeout = '{_MIGRATION_LOCK_TIMEOUT}'")
-        )
-        connection.commit()
         context.configure(
             connection=connection,
             target_metadata=target_metadata,

alembic/versions/0037_patreon_seen_media.py

@@ -1,53 +0,0 @@
-"""patreon_seen_media: per-source ledger of already-ingested Patreon media
-
-Revision ID: 0037
-Revises: 0036
-Create Date: 2026-06-05
-
-Native Patreon ingester (build step 2a). Replaces gallery-dl's
-archive.sqlite3 with our own queryable table. The downloader upserts one
-row per (source, media) so routine walks skip media we've already
-processed; a future "recovery" mode bypasses the ledger to re-walk.
-
-`filehash` is a 32-hex Patreon CDN MD5, OR a video sentinel of the form
-``video:<post_id>:<media_id>`` — hence String(128). The unique
-constraint on (source_id, filehash) is the dedup upsert key.
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0037"
-down_revision: Union[str, None] = "0036"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "patreon_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_patreon_seen_media_source_id"
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_table("patreon_seen_media")

alembic/versions/0038_patreon_failed_media.py

@@ -1,58 +0,0 @@
-"""patreon_failed_media: per-source dead-letter ledger for failing Patreon media
-
-Revision ID: 0038
-Revises: 0037
-Create Date: 2026-06-06
-
-Plan #705 (#7). Media that keeps failing to download/validate (404'd CDN,
-deleted post, geo-blocked Mux, persistently-corrupt bytes) gets recorded here
-with an attempt counter; once it crosses the dead-letter threshold the ingester
-skips it on routine walks (recovery still re-attempts). A clean download clears
-the row. UNIQUE (source_id, filehash) is the upsert key (same media key the
-seen-ledger uses).
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0038"
-down_revision: Union[str, None] = "0037"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "patreon_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_patreon_failed_media_source_id"
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_table("patreon_failed_media")

alembic/versions/0039_library_audit_resume.py

@@ -1,40 +0,0 @@
-"""library_audit_run: resume cursor + progress timestamp for chunked scans
-
-Revision ID: 0039
-Revises: 0038
-Create Date: 2026-06-07
-
-scan_library_for_rule used to run one 2h pass that timed out on large libraries
-and monopolized the concurrency-1 maintenance queue (operator-flagged). It now
-runs short time-boxed chunks that re-enqueue: `resume_after_id` persists the
-keyset cursor so the next chunk continues where it left off, and
-`last_progress_at` lets the recovery sweep tell a progressing multi-chunk audit
-from a genuinely stuck one.
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0039"
-down_revision: Union[str, None] = "0038"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "library_audit_run",
-        sa.Column(
-            "resume_after_id", sa.Integer, nullable=False, server_default="0"
-        ),
-    )
-    op.add_column(
-        "library_audit_run",
-        sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("library_audit_run", "last_progress_at")
-    op.drop_column("library_audit_run", "resume_after_id")

alembic/versions/0040_series_chapters.py

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

alembic/versions/0041_series_suggestions.py

@@ -1,98 +0,0 @@
-"""series suggestions: assisted-continuation matcher (FC-6.3)
-
-Revision ID: 0041
-Revises: 0040
-Create Date: 2026-06-07
-
-A confirm-only queue of "this post may continue this series" hints, plus two
-import_settings knobs (enable + score threshold) for the matcher.
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0041"
-down_revision: Union[str, None] = "0040"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "series_suggestion",
-        sa.Column("id", sa.Integer, primary_key=True),
-        sa.Column(
-            "post_id",
-            sa.Integer,
-            sa.ForeignKey("post.id", ondelete="CASCADE"),
-            nullable=False,
-        ),
-        sa.Column(
-            "series_tag_id",
-            sa.Integer,
-            sa.ForeignKey("tag.id", ondelete="CASCADE"),
-            nullable=False,
-        ),
-        sa.Column("score", sa.Float, nullable=False),
-        sa.Column("signals", sa.JSON, nullable=True),
-        sa.Column(
-            "status", sa.String(16), nullable=False, server_default="pending"
-        ),
-        sa.Column(
-            "created_at",
-            sa.DateTime(timezone=True),
-            nullable=False,
-            server_default=sa.text("now()"),
-        ),
-        sa.Column(
-            "updated_at",
-            sa.DateTime(timezone=True),
-            nullable=False,
-            server_default=sa.text("now()"),
-        ),
-        sa.UniqueConstraint(
-            "post_id", "series_tag_id", name="uq_series_suggestion_post_series"
-        ),
-    )
-    op.create_index(
-        "ix_series_suggestion_post_id", "series_suggestion", ["post_id"]
-    )
-    op.create_index(
-        "ix_series_suggestion_series_tag_id",
-        "series_suggestion",
-        ["series_tag_id"],
-    )
-    op.create_index(
-        "ix_series_suggestion_status", "series_suggestion", ["status"]
-    )
-
-    op.add_column(
-        "import_settings",
-        sa.Column(
-            "series_suggest_enabled",
-            sa.Boolean,
-            nullable=False,
-            server_default=sa.true(),
-        ),
-    )
-    op.add_column(
-        "import_settings",
-        sa.Column(
-            "series_suggest_threshold",
-            sa.Float,
-            nullable=False,
-            server_default="0.5",
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("import_settings", "series_suggest_threshold")
-    op.drop_column("import_settings", "series_suggest_enabled")
-    op.drop_index("ix_series_suggestion_status", table_name="series_suggestion")
-    op.drop_index(
-        "ix_series_suggestion_series_tag_id", table_name="series_suggestion"
-    )
-    op.drop_index("ix_series_suggestion_post_id", table_name="series_suggestion")
-    op.drop_table("series_suggestion")

alembic/versions/0042_series_chapter_stated_part.py

@@ -1,32 +0,0 @@
-"""series chapter stated_part: operator-facing Part N label (FC-6.4)
-
-Revision ID: 0042
-Revises: 0041
-Create Date: 2026-06-07
-
-A chapter's positional chapter_number is auto-managed (rewritten 1..N on
-reorder/delete), so it can't double as the installment number the operator wants
-to type (e.g. a series authored from a post that is Part 2). Add a nullable
-stated_part alongside it — the same split as series_page.page_number (order) vs
-series_page.stated_page (printed number). Nullable; the UI falls back to
-chapter_number when unset.
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0042"
-down_revision: Union[str, None] = "0041"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "series_chapter", sa.Column("stated_part", sa.Integer, nullable=True)
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("series_chapter", "stated_part")

alembic/versions/0043_post_attachment_per_post_unique.py

@@ -1,62 +0,0 @@
-"""post_attachment: per-post sha uniqueness (empty-post flood fix)
-
-Revision ID: 0043
-Revises: 0042
-Create Date: 2026-06-08
-
-PostAttachment.sha256 was GLOBALLY unique, so a non-art file the creator attaches
-to many posts (a standard pdf/zip/link-card) only ever got ONE row — on the first
-post — leaving every later post a bare shell (no image, no attachment). The native
-Patreon backfill of Anduo surfaced 1589 such shells (operator-flagged 2026-06-08).
-
-Switch to PER-POST uniqueness: the on-disk blob stays sha-deduped, but each post
-gets its own row. Replace the unique sha256 index with a plain lookup index plus
-two partial uniques — (post_id, sha256) for real posts and (sha256) for the
-NULL-post filesystem case (still one row per file there).
-
-Existing data has ≤1 row per sha (the old global unique), so the new partial
-uniques can't be violated on upgrade — no data backfill needed here. The bare-post
-shells themselves are removed by the separate prune-empty-posts cleanup tool.
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0043"
-down_revision: Union[str, None] = "0042"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    # Drop the global unique index; recreate it as a plain (non-unique) lookup
-    # index so sha-based reads keep their index (matches the model's index=True).
-    op.drop_index("ix_post_attachment_sha256", table_name="post_attachment")
-    op.create_index(
-        "ix_post_attachment_sha256", "post_attachment", ["sha256"],
-    )
-    op.create_index(
-        "uq_post_attachment_post_sha", "post_attachment",
-        ["post_id", "sha256"], unique=True,
-        postgresql_where=sa.text("post_id IS NOT NULL"),
-    )
-    op.create_index(
-        "uq_post_attachment_null_post_sha", "post_attachment",
-        ["sha256"], unique=True,
-        postgresql_where=sa.text("post_id IS NULL"),
-    )
-
-
-def downgrade() -> None:
-    op.drop_index(
-        "uq_post_attachment_null_post_sha", table_name="post_attachment"
-    )
-    op.drop_index(
-        "uq_post_attachment_post_sha", table_name="post_attachment"
-    )
-    op.drop_index("ix_post_attachment_sha256", table_name="post_attachment")
-    op.create_index(
-        "ix_post_attachment_sha256", "post_attachment", ["sha256"],
-        unique=True,
-    )

alembic/versions/0044_ml_settings_tagger_store_floor.py

@@ -1,37 +0,0 @@
-"""ml_settings.tagger_store_floor
-
-The ingest confidence floor below which tagger predictions are not stored,
-promoted from the TAGGER_STORE_FLOOR env var to a DB-backed, UI-tunable
-setting. Default 0.70 (was an env default of 0.05): the suggestion path
-already filters at 0.70 and the centroid/learned path covers low-confidence
-preferred tags, so the sub-0.70 tail was redundant weight — it had grown
-image_record's TOAST to ~100 GB. See plan-task #764.
-
-Revision ID: 0044
-Revises: 0043
-Create Date: 2026-06-10
-
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0044"
-down_revision: Union[str, None] = "0043"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "tagger_store_floor", sa.Float(),
-            nullable=False, server_default="0.7",
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("ml_settings", "tagger_store_floor")

alembic/versions/0045_image_prediction_table.py

@@ -1,69 +0,0 @@
-"""image_prediction table (DDL only — backfill runs as a background task)
-
-Normalizes the per-image tagger predictions out of the JSON blob into a
-queryable table (#768). This migration creates ONLY the table + indexes — it
-is pure DDL and commits instantly, so web boots immediately.
-
-The data backfill from the existing image_record.tagger_predictions JSON is
-deliberately NOT done here. Doing it inline made the whole migration one
-transaction over the ~100 GB TOAST: nothing committed until the very end, it
-was invisible/unmonitorable mid-run, and an early MATERIALIZED-CTE form spilled
-the full 100 GB to temp. Instead the backfill is the
-backend.app.tasks.admin.backfill_image_predictions_task — batched by id window,
-committed per chunk (visible progress + resumable), idempotent
-(ON CONFLICT DO NOTHING). Trigger it from Settings → Maintenance once web is up.
-
-The old image_record.tagger_predictions column is left in place (vestigial) and
-dropped in a follow-up once the backfill + code cutover are verified — dropping
-it needs an ACCESS EXCLUSIVE lock on the hot image_record table (the 0044 lock
-class), so it's deferred to a quiesced-worker window.
-
-Revision ID: 0045
-Revises: 0044
-Create Date: 2026-06-10
-
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0045"
-down_revision: Union[str, None] = "0044"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "image_prediction",
-        sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column(
-            "image_record_id", sa.Integer(),
-            sa.ForeignKey("image_record.id", ondelete="CASCADE"),
-            nullable=False,
-        ),
-        sa.Column("raw_name", sa.String(length=255), nullable=False),
-        sa.Column("category", sa.String(length=64), nullable=False),
-        sa.Column("score", sa.Float(), nullable=False),
-        sa.UniqueConstraint(
-            "image_record_id", "raw_name", name="image_raw_name",
-        ),
-    )
-    op.create_index(
-        "ix_image_prediction_image", "image_prediction", ["image_record_id"],
-    )
-    op.create_index(
-        "ix_image_prediction_name_score", "image_prediction",
-        ["raw_name", "score"],
-    )
-    # No data backfill here — see the module docstring. The one-time copy from
-    # image_record.tagger_predictions runs as backfill_image_predictions_task
-    # (batched, resumable, idempotent), kept out of this transaction so web boots
-    # without waiting on a ~100 GB pass.
-
-
-def downgrade() -> None:
-    op.drop_index("ix_image_prediction_name_score", "image_prediction")
-    op.drop_index("ix_image_prediction_image", "image_prediction")
-    op.drop_table("image_prediction")

alembic/versions/0046_drop_tagger_predictions.py

@@ -1,43 +0,0 @@
-"""drop image_record.tagger_predictions (predictions normalized to image_prediction)
-
-Final step of #768. The per-tag predictions now live in the image_prediction
-table (backfilled from the JSON, read by suggestions + allowlist, written by
-tag_and_embed). The old JSON column is dead weight — and it's the ~100 GB of
-sub-0.70 score tail that bloated image_record's TOAST and broke DB backups
-(#739). Dropping it is a fast catalog change; it does NOT reclaim the disk on
-its own — run `VACUUM FULL image_record` (or pg_repack) afterward, off-hours,
-to return the space to the OS so backups go small.
-
-DROP COLUMN needs a brief ACCESS EXCLUSIVE lock on image_record; env.py's
-lock_timeout guards it, so quiesce the ml-worker if a tagging run is in flight
-(see the migration-lock reference). tagger_model_version is kept — it's the
-"has this been tagged / is it current?" signal the backfill sweep reads.
-
-Revision ID: 0046
-Revises: 0045
-Create Date: 2026-06-11
-
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0046"
-down_revision: Union[str, None] = "0045"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.drop_column("image_record", "tagger_predictions")
-
-
-def downgrade() -> None:
-    # Re-add the column empty. The JSON data is not restored (it lived only in
-    # this column); a downgrade would re-tag or backfill from image_prediction
-    # separately if ever needed.
-    op.add_column(
-        "image_record",
-        sa.Column("tagger_predictions", sa.JSON(), nullable=True),
-    )

alembic/versions/0047_series_chapter_dividers.py

@@ -1,175 +0,0 @@
-"""series chapters become cosmetic dividers; pages become one series-global run
-
-FC-6.x reframe (#789). A series is now ONE flat, series-global ordered run of
-pages; chapters stop owning pages and become labeled dividers anchored to the
-page that begins them.
-
-Migration (order matters — series_page.chapter_id cascades, so it must be
-dropped BEFORE any chapter row is deleted, or pages would cascade away):
-  a. Renumber series_page.page_number to a series-global 1..N (ordered by the
-     OLD (chapter_number, page_number)).
-  b. Add series_chapter.anchor_page_id and populate it with each chapter's first
-     page (lowest new page_number).
-  c. Drop series_page.chapter_id (severs the cascade link).
-  d. Prune chapters that shouldn't become dividers: empty/placeholder ones (no
-     anchor) and the redundant unlabeled chapter that would sit at page 1.
-  e. Reshape series_chapter into the divider: drop chapter_number,
-     is_placeholder, stated_page_start/end; make anchor_page_id NOT NULL +
-     UNIQUE + FK→series_page ON DELETE CASCADE.
-
-Revision ID: 0047
-Revises: 0046
-Create Date: 2026-06-11
-
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0047"
-down_revision: Union[str, None] = "0046"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    # a. series-global page numbering, preserving the old reading order.
-    op.execute(
-        """
-        WITH ordered AS (
-            SELECT sp.id,
-                   ROW_NUMBER() OVER (
-                       PARTITION BY sp.series_tag_id
-                       ORDER BY sc.chapter_number, sp.page_number, sp.id
-                   ) AS rn
-            FROM series_page sp
-            JOIN series_chapter sc ON sc.id = sp.chapter_id
-        )
-        UPDATE series_page sp
-        SET page_number = ordered.rn
-        FROM ordered
-        WHERE sp.id = ordered.id
-        """
-    )
-
-    # b. anchor each existing chapter at its first page (lowest new page_number).
-    op.add_column(
-        "series_chapter",
-        sa.Column("anchor_page_id", sa.Integer(), nullable=True),
-    )
-    op.execute(
-        """
-        WITH firsts AS (
-            SELECT DISTINCT ON (sp.chapter_id)
-                   sp.chapter_id, sp.id AS page_id
-            FROM series_page sp
-            ORDER BY sp.chapter_id, sp.page_number, sp.id
-        )
-        UPDATE series_chapter sc
-        SET anchor_page_id = firsts.page_id
-        FROM firsts
-        WHERE firsts.chapter_id = sc.id
-        """
-    )
-
-    # c. sever the ownership link (drops the FK + index with the column) BEFORE
-    #    pruning chapters, so deleting a chapter can't cascade-delete its pages.
-    op.drop_column("series_page", "chapter_id")
-
-    # d. prune chapters that don't become dividers: placeholders / empty ones
-    #    (no anchor), and the unlabeled chapter that would land redundantly at
-    #    page 1 (the series just starts — no divider needed there).
-    op.execute(
-        """
-        DELETE FROM series_chapter sc
-        USING (
-            SELECT sc2.id
-            FROM series_chapter sc2
-            LEFT JOIN series_page sp ON sp.id = sc2.anchor_page_id
-            WHERE sc2.anchor_page_id IS NULL
-               OR (sp.page_number = 1
-                   AND sc2.title IS NULL
-                   AND sc2.stated_part IS NULL)
-        ) gone
-        WHERE sc.id = gone.id
-        """
-    )
-
-    # e. reshape into the divider model.
-    op.drop_column("series_chapter", "chapter_number")
-    op.drop_column("series_chapter", "is_placeholder")
-    op.drop_column("series_chapter", "stated_page_start")
-    op.drop_column("series_chapter", "stated_page_end")
-    op.alter_column("series_chapter", "anchor_page_id", nullable=False)
-    op.create_unique_constraint(
-        "uq_series_chapter_anchor_page", "series_chapter", ["anchor_page_id"]
-    )
-    op.create_foreign_key(
-        "fk_series_chapter_anchor_page",
-        "series_chapter",
-        "series_page",
-        ["anchor_page_id"],
-        ["id"],
-        ondelete="CASCADE",
-    )
-
-
-def downgrade() -> None:
-    # Lossy: dividers can't be reconstructed as owning chapters. Collapse back to
-    # exactly one chapter per series that owns all its pages in order.
-    op.add_column(
-        "series_page", sa.Column("chapter_id", sa.Integer(), nullable=True)
-    )
-    op.drop_constraint(
-        "fk_series_chapter_anchor_page", "series_chapter", type_="foreignkey"
-    )
-    op.drop_constraint(
-        "uq_series_chapter_anchor_page", "series_chapter", type_="unique"
-    )
-    op.drop_column("series_chapter", "anchor_page_id")
-    op.add_column(
-        "series_chapter",
-        sa.Column(
-            "chapter_number", sa.Integer(), nullable=False, server_default="1"
-        ),
-    )
-    op.add_column(
-        "series_chapter",
-        sa.Column(
-            "is_placeholder", sa.Boolean(), nullable=False,
-            server_default="false",
-        ),
-    )
-    op.add_column(
-        "series_chapter",
-        sa.Column("stated_page_start", sa.Integer(), nullable=True),
-    )
-    op.add_column(
-        "series_chapter",
-        sa.Column("stated_page_end", sa.Integer(), nullable=True),
-    )
-    op.execute("DELETE FROM series_chapter")
-    op.execute(
-        """
-        INSERT INTO series_chapter (series_tag_id, chapter_number)
-        SELECT DISTINCT series_tag_id, 1 FROM series_page
-        """
-    )
-    op.execute(
-        """
-        UPDATE series_page sp
-        SET chapter_id = sc.id
-        FROM series_chapter sc
-        WHERE sc.series_tag_id = sp.series_tag_id
-        """
-    )
-    op.alter_column("series_page", "chapter_id", nullable=False)
-    op.create_foreign_key(
-        "fk_series_page_chapter",
-        "series_page",
-        "series_chapter",
-        ["chapter_id"],
-        ["id"],
-        ondelete="CASCADE",
-    )

alembic/versions/0048_series_page_pending_status.py

@@ -1,45 +0,0 @@
-"""series_page pending staging: status + nullable page_number (#789 Phase 2)
-
-Pages added from a post no longer append straight into the run — they land
-'pending' with a NULL page_number, staged grouped by their source post so the
-operator can drop junk (text-free alts, bumpers) and place the keepers into the
-sequence. A page only gets a series-global page_number once it's 'placed'.
-
-Revision ID: 0048
-Revises: 0047
-Create Date: 2026-06-11
-
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0048"
-down_revision: Union[str, None] = "0047"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "series_page",
-        sa.Column(
-            "status", sa.String(length=16), nullable=False,
-            server_default="placed",
-        ),
-    )
-    op.alter_column(
-        "series_page", "page_number",
-        existing_type=sa.Integer(), nullable=True,
-    )
-
-
-def downgrade() -> None:
-    # Lossy: pending pages are unsorted staging rows with no order — drop them.
-    op.execute("DELETE FROM series_page WHERE status = 'pending'")
-    op.alter_column(
-        "series_page", "page_number",
-        existing_type=sa.Integer(), nullable=False,
-    )
-    op.drop_column("series_page", "status")

alembic/versions/0049_external_link_table.py

@@ -1,90 +0,0 @@
-"""external_link table — off-platform file-host links found in post bodies
-
-Creators host the real files on mega.nz / Google Drive / MediaFire / Dropbox /
-Pixeldrain and link them in the post text. This table records each such link
-(so nothing is silently dropped), and doubles as the dedup + dead-letter ledger
-the download worker (a later slice) walks. `url` keeps the FULL link including
-the `#fragment` — mega.nz's decryption key lives there; truncating it makes the
-file undownloadable.
-
-CHECK whitelists for host + status include the full enum up front (incl. the
-download-worker statuses) so the worker slice needs no constraint migration.
-
-Revision ID: 0049
-Revises: 0048
-Create Date: 2026-06-14
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0049"
-down_revision: Union[str, None] = "0048"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "external_link",
-        sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column(
-            "post_id", sa.Integer(),
-            sa.ForeignKey("post.id", ondelete="CASCADE"), nullable=False,
-        ),
-        sa.Column(
-            "artist_id", sa.Integer(),
-            sa.ForeignKey("artist.id", ondelete="SET NULL"), nullable=True,
-        ),
-        sa.Column("host", sa.String(length=16), nullable=False),
-        sa.Column("url", sa.Text(), nullable=False),
-        sa.Column("label", sa.Text(), nullable=True),
-        sa.Column(
-            "status", sa.String(length=16), nullable=False,
-            server_default="pending",
-        ),
-        sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("last_error", sa.Text(), nullable=True),
-        sa.Column(
-            "attachment_id", sa.Integer(),
-            sa.ForeignKey("post_attachment.id", ondelete="SET NULL"),
-            nullable=True,
-        ),
-        sa.Column(
-            "created_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("duration_seconds", sa.Float(), nullable=True),
-        sa.CheckConstraint(
-            "host IN ('mega','gdrive','mediafire','dropbox','pixeldrain')",
-            name="ck_external_link_host",
-        ),
-        sa.CheckConstraint(
-            "status IN ('pending','downloading','downloaded','failed',"
-            "'skipped','dead')",
-            name="ck_external_link_status",
-        ),
-    )
-    op.create_index(
-        "ix_external_link_post_id", "external_link", ["post_id"],
-    )
-    op.create_index(
-        "ix_external_link_artist_id", "external_link", ["artist_id"],
-    )
-    op.create_index(
-        "ix_external_link_status", "external_link", ["status"],
-    )
-    op.create_index(
-        "uq_external_link_post_url", "external_link", ["post_id", "url"],
-        unique=True,
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("uq_external_link_post_url", table_name="external_link")
-    op.drop_index("ix_external_link_status", table_name="external_link")
-    op.drop_index("ix_external_link_artist_id", table_name="external_link")
-    op.drop_index("ix_external_link_post_id", table_name="external_link")
-    op.drop_table("external_link")

alembic/versions/0050_external_link_host_toggles.py

@@ -1,38 +0,0 @@
-"""import_settings: per-host enable toggles for external file-host downloads
-
-Operator levers (#830): disable a single host (e.g. mega.nz when it's
-rate-limiting/banning) without touching the others. The worker reads these via
-getattr and defaults to enabled, so the toggles default TRUE (works out of the
-box, rule #26).
-
-Revision ID: 0050
-Revises: 0049
-Create Date: 2026-06-14
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0050"
-down_revision: Union[str, None] = "0049"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-_HOSTS = ("mega", "gdrive", "mediafire", "dropbox", "pixeldrain")
-
-
-def upgrade() -> None:
-    for host in _HOSTS:
-        op.add_column(
-            "import_settings",
-            sa.Column(
-                f"extdl_{host}_enabled", sa.Boolean(), nullable=False,
-                server_default=sa.true(),
-            ),
-        )
-
-
-def downgrade() -> None:
-    for host in _HOSTS:
-        op.drop_column("import_settings", f"extdl_{host}_enabled")

alembic/versions/0051_image_source_provenance.py

@@ -1,38 +0,0 @@
-"""image_record: source_url + source_filehash (inline-image localization)
-
-#830 Phase 2. To render a post body faithfully we serve LOCAL copies of inline
-images instead of hotlinking the public CDN. The join key between a body
-`<img src=CDN>` and the local file is the CDN's 32-hex filehash (the same
-identity extract_media dedups by). Persist it (indexed) plus the full source
-URL for provenance/debugging. Both NULL for filesystem-imported / pre-existing
-rows — those fall back to hotlinking until re-downloaded.
-
-Revision ID: 0051
-Revises: 0050
-Create Date: 2026-06-14
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0051"
-down_revision: Union[str, None] = "0050"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column("image_record", sa.Column("source_url", sa.Text(), nullable=True))
-    op.add_column(
-        "image_record", sa.Column("source_filehash", sa.String(length=32), nullable=True)
-    )
-    op.create_index(
-        "ix_image_record_source_filehash", "image_record", ["source_filehash"]
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("ix_image_record_source_filehash", table_name="image_record")
-    op.drop_column("image_record", "source_filehash")
-    op.drop_column("image_record", "source_url")

alembic/versions/0052_image_duration_seconds.py

@@ -1,32 +0,0 @@
-"""image_record: duration_seconds (Tier-1 video near-dup key)
-
-#871. Videos previously deduped on sha256 only (pHash is images-only), so a
-different encode/remux of the same video imported as a distinct record. Persist
-the container duration so the importer can treat same-artist videos with matching
-duration (+ aspect ratio) as the same content and dedup/supersede like images.
-NULL for images and for video rows imported before this column existed (a
-backfill re-probes those so they participate in dedup).
-
-Revision ID: 0052
-Revises: 0051
-Create Date: 2026-06-16
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0052"
-down_revision: Union[str, None] = "0051"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "image_record", sa.Column("duration_seconds", sa.Float(), nullable=True)
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("image_record", "duration_seconds")

alembic/versions/0053_ml_settings_video_tagging.py

@@ -1,49 +0,0 @@
-"""ml_settings: video tagging knobs (cadence sampling + noise floor)
-
-#747. Video tag quality/perf: sample frames at a fixed cadence (interval) so a
-tag's frame-presence reflects real screen time, cap total frames so long videos
-stay bounded, and keep a tag only if it appears in >= min_tag_frames sampled
-frames. Operator-tunable via Settings → ML (replaces the VIDEO_ML_FRAMES env var).
-
-Revision ID: 0053
-Revises: 0052
-Create Date: 2026-06-16
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0053"
-down_revision: Union[str, None] = "0052"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "video_frame_interval_seconds", sa.Float(), nullable=False,
-            server_default="4.0",
-        ),
-    )
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "video_max_frames", sa.Integer(), nullable=False, server_default="64",
-        ),
-    )
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "video_min_tag_frames", sa.Integer(), nullable=False,
-            server_default="3",
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("ml_settings", "video_min_tag_frames")
-    op.drop_column("ml_settings", "video_max_frames")
-    op.drop_column("ml_settings", "video_frame_interval_seconds")

alembic/versions/0054_subscribestar_ledgers.py

@@ -1,82 +0,0 @@
-"""subscribestar_seen_media + subscribestar_failed_media: per-source ledgers
-
-Revision ID: 0054
-Revises: 0053
-Create Date: 2026-06-17
-
-SubscribeStar native ingester (phase 1 of the gallery-dl → native-core
-migration). Mirrors the Patreon ledger tables (0037/0038): 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. `filehash` is a CDN content hash when present, else a synthesized
-``<post_id>:<filename>`` key — hence String(128). 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 = "0054"
-down_revision: Union[str, None] = "0053"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "subscribestar_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_subscribestar_seen_media_source_id"
-        ),
-    )
-    op.create_table(
-        "subscribestar_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_subscribestar_failed_media_source_id"
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_table("subscribestar_failed_media")
-    op.drop_table("subscribestar_seen_media")

alembic/versions/0055_image_provenance_from_attachment.py

@@ -1,55 +0,0 @@
-"""image_provenance: from_attachment_id (which archive an image was extracted from)
-
-Milestone #87. When an image is pulled out of a .zip/.rar, record WHICH archive
-PostAttachment it came from, so the provenance UI can show the single archive a
-file lives inside instead of every attachment on the post. Nullable FK with
-ON DELETE SET NULL — a loose (non-archive) download leaves it NULL, and deleting
-the archive attachment forgets the linkage without destroying the (image, post)
-provenance edge. Existing rows are NULL until the reextract backfill stamps them.
-
-Revision ID: 0055
-Revises: 0054
-Create Date: 2026-06-22
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0055"
-down_revision: Union[str, None] = "0054"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "image_provenance",
-        sa.Column("from_attachment_id", sa.Integer(), nullable=True),
-    )
-    op.create_index(
-        "ix_image_provenance_from_attachment_id",
-        "image_provenance",
-        ["from_attachment_id"],
-    )
-    op.create_foreign_key(
-        "fk_image_provenance_from_attachment",
-        "image_provenance",
-        "post_attachment",
-        ["from_attachment_id"],
-        ["id"],
-        ondelete="SET NULL",
-    )
-
-
-def downgrade() -> None:
-    op.drop_constraint(
-        "fk_image_provenance_from_attachment",
-        "image_provenance",
-        type_="foreignkey",
-    )
-    op.drop_index(
-        "ix_image_provenance_from_attachment_id",
-        table_name="image_provenance",
-    )
-    op.drop_column("image_provenance", "from_attachment_id")

alembic/versions/0056_tag_eval_run.py

@@ -1,43 +0,0 @@
-"""tag_eval_run: persisted head-vs-centroid tagging eval runs (#1130)
-
-Milestone #114 slice 1. A long ml-queue eval whose full report must SURVIVE
-navigation, so the run + report live in a row the admin card rehydrates from
-(mirrors library_audit_run). running -> ready / error.
-
-Revision ID: 0056
-Revises: 0055
-Create Date: 2026-06-28
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-from sqlalchemy.dialects.postgresql import JSONB
-
-revision: str = "0056"
-down_revision: Union[str, None] = "0055"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "tag_eval_run",
-        sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column("params", JSONB(), nullable=False),
-        sa.Column("status", sa.String(length=16), nullable=False, server_default="running"),
-        sa.Column(
-            "started_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("report", JSONB(), nullable=True),
-        sa.Column("error", sa.Text(), nullable=True),
-        sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True),
-    )
-    op.create_index("ix_tag_eval_run_status", "tag_eval_run", ["status"])
-
-
-def downgrade() -> None:
-    op.drop_index("ix_tag_eval_run_status", table_name="tag_eval_run")
-    op.drop_table("tag_eval_run")

alembic/versions/0057_tag_positive_confirmation.py

@@ -1,40 +0,0 @@
-"""tag_positive_confirmation: operator-affirmed correct positives (#1130)
-
-Mirror of tag_suggestion_rejection. "Keep" on a doubted positive records here so
-the eval's doubts list stops resurfacing confirmed-correct images every run.
-
-Revision ID: 0057
-Revises: 0056
-Create Date: 2026-06-28
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0057"
-down_revision: Union[str, None] = "0056"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "tag_positive_confirmation",
-        sa.Column(
-            "image_record_id", sa.Integer(),
-            sa.ForeignKey("image_record.id", ondelete="CASCADE"), primary_key=True,
-        ),
-        sa.Column(
-            "tag_id", sa.Integer(),
-            sa.ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True, index=True,
-        ),
-        sa.Column(
-            "confirmed_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_table("tag_positive_confirmation")

alembic/versions/0058_tag_head.py

@@ -1,95 +0,0 @@
-"""tag_head + head_training_run: production heads that learn from tags (#114)
-
-The eval (#1130) proved the frozen-embedding + trained-head spine; this lands its
-production form. tag_head stores one logistic-regression head per concept (the
-new suggestion source, replacing Camie + centroid); head_training_run tracks the
-batch that (re)trains them. Adds two head-training tunables to ml_settings.
-
-Revision ID: 0058
-Revises: 0057
-Create Date: 2026-06-28
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-from pgvector.sqlalchemy import Vector
-from sqlalchemy.dialects.postgresql import JSONB
-
-revision: str = "0058"
-down_revision: Union[str, None] = "0057"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-_HEAD_DIM = 1152
-
-
-def upgrade() -> None:
-    op.create_table(
-        "tag_head",
-        sa.Column(
-            "tag_id", sa.Integer(),
-            sa.ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True,
-        ),
-        sa.Column("embedding_version", sa.String(length=128), nullable=False),
-        sa.Column("weights", Vector(_HEAD_DIM), nullable=False),
-        sa.Column("bias", sa.Float(), nullable=False),
-        sa.Column("suggest_threshold", sa.Float(), nullable=False),
-        sa.Column("auto_apply_threshold", sa.Float(), nullable=True),
-        sa.Column("n_pos", sa.Integer(), nullable=False),
-        sa.Column("n_neg", sa.Integer(), nullable=False),
-        sa.Column("ap", sa.Float(), nullable=False),
-        sa.Column("precision_cv", sa.Float(), nullable=False),
-        sa.Column("recall", sa.Float(), nullable=False),
-        sa.Column(
-            "trained_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-        sa.Column("metrics", JSONB(), nullable=True),
-    )
-
-    op.create_table(
-        "head_training_run",
-        sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column("params", JSONB(), nullable=False),
-        sa.Column(
-            "status", sa.String(length=16), nullable=False,
-            server_default="running",
-        ),
-        sa.Column(
-            "started_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("n_trained", sa.Integer(), nullable=True),
-        sa.Column("n_skipped", sa.Integer(), nullable=True),
-        sa.Column("error", sa.Text(), nullable=True),
-        sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True),
-    )
-    op.create_index(
-        "ix_head_training_run_status", "head_training_run", ["status"],
-    )
-
-    # Head-training tunables on the ml_settings singleton.
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "head_min_positives", sa.Integer(), nullable=False,
-            server_default="8",
-        ),
-    )
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "head_auto_apply_precision", sa.Float(), nullable=False,
-            server_default="0.97",
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("ml_settings", "head_auto_apply_precision")
-    op.drop_column("ml_settings", "head_min_positives")
-    op.drop_index("ix_head_training_run_status", table_name="head_training_run")
-    op.drop_table("head_training_run")
-    op.drop_table("tag_head")

alembic/versions/0059_head_auto_apply.py

@@ -1,70 +0,0 @@
-"""head_auto_apply_run + earned-auto-apply settings (#114)
-
-A graduated head can apply its tag without a human, gated by a master switch +
-a support floor. head_auto_apply_run tracks each sweep / dry-run preview.
-
-Revision ID: 0059
-Revises: 0058
-Create Date: 2026-06-29
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-from sqlalchemy.dialects.postgresql import JSONB
-
-revision: str = "0059"
-down_revision: Union[str, None] = "0058"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "head_auto_apply_run",
-        sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column(
-            "dry_run", sa.Boolean(), nullable=False, server_default=sa.false()
-        ),
-        sa.Column("params", JSONB(), nullable=False),
-        sa.Column(
-            "status", sa.String(length=16), nullable=False,
-            server_default="running",
-        ),
-        sa.Column(
-            "started_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("n_applied", sa.Integer(), nullable=True),
-        sa.Column("report", JSONB(), nullable=True),
-        sa.Column("error", sa.Text(), nullable=True),
-        sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True),
-    )
-    op.create_index(
-        "ix_head_auto_apply_run_status", "head_auto_apply_run", ["status"],
-    )
-
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "head_auto_apply_enabled", sa.Boolean(), nullable=False,
-            server_default=sa.true(),  # opt-out: on by default (operator-asked)
-        ),
-    )
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "head_auto_apply_min_positives", sa.Integer(), nullable=False,
-            server_default="30",
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("ml_settings", "head_auto_apply_min_positives")
-    op.drop_column("ml_settings", "head_auto_apply_enabled")
-    op.drop_index(
-        "ix_head_auto_apply_run_status", table_name="head_auto_apply_run"
-    )
-    op.drop_table("head_auto_apply_run")

alembic/versions/0060_head_metrics.py

@@ -1,74 +0,0 @@
-"""head_metric + head_metrics_snapshot: auto-apply observability (#114)
-
-Running misfire/under-fire counters per concept (captured at correction time,
-since image_tag.source is lost on delete) + a daily per-concept time-series so
-the operator can tune the precision target + support floor from real data.
-
-Revision ID: 0060
-Revises: 0059
-Create Date: 2026-06-29
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0060"
-down_revision: Union[str, None] = "0059"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "head_metric",
-        sa.Column(
-            "tag_id", sa.Integer(),
-            sa.ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True,
-        ),
-        sa.Column("n_misfires", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("n_underfires", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column(
-            "updated_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-    )
-
-    op.create_table(
-        "head_metrics_snapshot",
-        sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column(
-            "tag_id", sa.Integer(),
-            sa.ForeignKey("tag.id", ondelete="CASCADE"),
-        ),
-        sa.Column("name", sa.String(length=255), nullable=False),
-        sa.Column(
-            "snapshot_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-        sa.Column("n_auto_applied", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("n_misfires", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("n_underfires", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("ap", sa.Float(), nullable=True),
-        sa.Column("precision_cv", sa.Float(), nullable=True),
-        sa.Column("recall", sa.Float(), nullable=True),
-        sa.Column("n_pos", sa.Integer(), nullable=True),
-    )
-    op.create_index(
-        "ix_head_metrics_snapshot_tag_id", "head_metrics_snapshot", ["tag_id"],
-    )
-    op.create_index(
-        "ix_head_metrics_snapshot_snapshot_at", "head_metrics_snapshot",
-        ["snapshot_at"],
-    )
-
-
-def downgrade() -> None:
-    op.drop_index(
-        "ix_head_metrics_snapshot_snapshot_at", table_name="head_metrics_snapshot"
-    )
-    op.drop_index(
-        "ix_head_metrics_snapshot_tag_id", table_name="head_metrics_snapshot"
-    )
-    op.drop_table("head_metrics_snapshot")
-    op.drop_table("head_metric")

alembic/versions/0061_image_region.py

@@ -1,59 +0,0 @@
-"""image_region: detected/proposed regions + their crop embeddings (#114)
-
-Storage backbone of the crop pipeline. A region = normalized bbox + the crop's
-embedding (CCIP for face/figure → character id; SigLIP for concept regions →
-head bag-of-embeddings). Also serves as grounded-tag bbox provenance.
-
-Revision ID: 0061
-Revises: 0060
-Create Date: 2026-06-29
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-from pgvector.sqlalchemy import Vector
-
-revision: str = "0061"
-down_revision: Union[str, None] = "0060"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-_CCIP_DIM = 768
-_SIGLIP_DIM = 1152
-
-
-def upgrade() -> None:
-    op.create_table(
-        "image_region",
-        sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column(
-            "image_record_id", sa.Integer(),
-            sa.ForeignKey("image_record.id", ondelete="CASCADE"), nullable=False,
-        ),
-        sa.Column("kind", sa.String(length=16), nullable=False),
-        # Video/animated: source frame timestamp (seconds); NULL for stills.
-        sa.Column("frame_time", sa.Float(), nullable=True),
-        sa.Column("rx", sa.Float(), nullable=False),
-        sa.Column("ry", sa.Float(), nullable=False),
-        sa.Column("rw", sa.Float(), nullable=False),
-        sa.Column("rh", sa.Float(), nullable=False),
-        sa.Column("score", sa.Float(), nullable=True),
-        sa.Column("detector_version", sa.String(length=64), nullable=True),
-        sa.Column("crop_version", sa.String(length=64), nullable=True),
-        sa.Column("embedding_version", sa.String(length=128), nullable=True),
-        sa.Column("ccip_embedding", Vector(_CCIP_DIM), nullable=True),
-        sa.Column("siglip_embedding", Vector(_SIGLIP_DIM), nullable=True),
-        sa.Column(
-            "created_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-    )
-    op.create_index(
-        "ix_image_region_image_record_id", "image_region", ["image_record_id"],
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("ix_image_region_image_record_id", table_name="image_region")
-    op.drop_table("image_region")

alembic/versions/0062_gpu_job.py

@@ -1,55 +0,0 @@
-"""gpu_job: the HTTP-leased GPU work queue for the desktop agent (#114)
-
-The agent stays HTTP-only — the server enqueues per-(image, task) jobs here and
-the agent leases/submits over the web API; Redis/Postgres stay private.
-
-Revision ID: 0062
-Revises: 0061
-Create Date: 2026-06-29
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0062"
-down_revision: Union[str, None] = "0061"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "gpu_job",
-        sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column(
-            "image_record_id", sa.Integer(),
-            sa.ForeignKey("image_record.id", ondelete="CASCADE"), nullable=False,
-        ),
-        sa.Column("task", sa.String(length=32), nullable=False),
-        sa.Column(
-            "status", sa.String(length=16), nullable=False,
-            server_default="pending",
-        ),
-        sa.Column("lease_token", sa.String(length=64), nullable=True),
-        sa.Column("leased_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("error", sa.Text(), nullable=True),
-        sa.Column(
-            "created_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-        sa.Column(
-            "updated_at", sa.DateTime(timezone=True), nullable=False,
-            server_default=sa.func.now(),
-        ),
-    )
-    op.create_index("ix_gpu_job_image_record_id", "gpu_job", ["image_record_id"])
-    op.create_index("ix_gpu_job_status", "gpu_job", ["status"])
-
-
-def downgrade() -> None:
-    op.drop_index("ix_gpu_job_status", table_name="gpu_job")
-    op.drop_index("ix_gpu_job_image_record_id", table_name="gpu_job")
-    op.drop_table("gpu_job")

alembic/versions/0063_ccip_match_threshold.py

@@ -1,33 +0,0 @@
-"""ml_settings.ccip_match_threshold — tunable CCIP character-match cut (#114)
-
-The v1 matcher used a flat 0.75 cosine; live data showed that over-fires (a
-high-reference character matched a scatter of images). 0.85 keeps the confident
-single-character matches and drops the noise. Tunable from the GPU agent card.
-
-Revision ID: 0063
-Revises: 0062
-Create Date: 2026-06-29
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0063"
-down_revision: Union[str, None] = "0062"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "ccip_match_threshold", sa.Float(), nullable=False,
-            server_default="0.85",
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("ml_settings", "ccip_match_threshold")

alembic/versions/0064_ccip_auto_apply.py

@@ -1,42 +0,0 @@
-"""ml_settings: CCIP auto-apply switch + threshold (#114)
-
-Confident CCIP character matches auto-tag (source='ccip_auto') on a daily sweep,
-so identity tags keep flowing without pressing a button. ON by default (opt-out,
-like head auto-apply); the high threshold (0.92, above the 0.85 suggest cut) +
-single-character references keep it safe, and every auto-tag is reversible.
-
-Revision ID: 0064
-Revises: 0063
-Create Date: 2026-06-30
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "0064"
-down_revision: Union[str, None] = "0063"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "ccip_auto_apply_enabled", sa.Boolean(), nullable=False,
-            server_default=sa.true(),
-        ),
-    )
-    op.add_column(
-        "ml_settings",
-        sa.Column(
-            "ccip_auto_apply_threshold", sa.Float(), nullable=False,
-            server_default="0.92",
-        ),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("ml_settings", "ccip_auto_apply_threshold")
-    op.drop_column("ml_settings", "ccip_auto_apply_enabled")

backend/app/api/__init__.py

@@ -20,14 +20,11 @@ def all_blueprints() -> list[Blueprint]:
     from .artist import artist_bp
     from .artists import artists_bp
     from .attachments import attachments_bp
-    from .ccip import ccip_bp
     from .cleanup import cleanup_bp
     from .credentials import credentials_bp
     from .downloads import downloads_bp
     from .extension import extension_bp
     from .gallery import gallery_bp
-    from .gpu import gpu_bp
-    from .heads import heads_bp
     from .import_admin import import_admin_bp
     from .ml_admin import ml_admin_bp
     from .platforms import platforms_bp
@@ -39,7 +36,6 @@ def all_blueprints() -> list[Blueprint]:
     from .suggestions import suggestions_bp
     from .system_activity import system_activity_bp
     from .system_backup import system_backup_bp
-    from .tag_eval import tag_eval_bp
     from .tags import tags_bp
     from .thumbnails import thumbnails_bp
     return [
@@ -60,10 +56,6 @@ def all_blueprints() -> list[Blueprint]:
         suggestions_bp,
         allowlist_bp,
         aliases_bp,
-        tag_eval_bp,
-        heads_bp,
-        gpu_bp,
-        ccip_bp,
         ml_admin_bp,
         thumbnails_bp,
         sources_bp,

backend/app/api/admin.py

@@ -6,7 +6,6 @@ Five action surfaces:
   DELETE /api/admin/tags/<int:tag_id>              (Tier B)
   POST   /api/admin/tags/<int:dest_id>/merge       (Tier B)
   POST   /api/admin/tags/prune-unused              (Tier A)
-  POST   /api/admin/posts/prune-bare               (Tier A)
   POST   /api/admin/tags/purge-legacy              (Tier A)
   GET    /api/admin/tags/<int:tag_id>/usage-count  (helper)
 
@@ -39,31 +38,6 @@ def _bulk_image_confirm_token(image_ids: list[int]) -> str:
     return digest[:8]
 
 
-async def _run_dry_run_op(service_fn, **service_kwargs):
-    """Shared body for the Tier-A dry-run/apply endpoints: read the `dry_run`
-    flag, run the cleanup_service predicate under `run_sync`, and return its
-    result dict. The SAME `service_fn` drives both preview and apply (the flag
-    just toggles), so a handler physically can't let its preview diverge from
-    its delete (rule 93). Default False preserves the existing contract — the UI
-    always passes `dry_run` explicitly (true to preview, false to apply). Extra
-    service kwargs (e.g. `source_id`) pass straight through."""
-    body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", False))
-    async with get_session() as session:
-        result = await session.run_sync(
-            lambda sync_sess: service_fn(sync_sess, dry_run=dry_run, **service_kwargs)
-        )
-    return jsonify(result)
-
-
-def _queued(async_result):
-    """Standard 202 for an operator-triggered maintenance task: hand the UI the
-    Celery task id so it can tail /maintenance/task-result (or the activity
-    dashboard) for the summary. (trigger_vacuum stays bespoke — the UI doesn't
-    poll it, so it returns no task id.)"""
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
-
-
 @admin_bp.route("/artists/<slug>/cascade-delete", methods=["POST"])
 async def artist_cascade_delete(slug: str):
     body = await request.get_json(silent=True) or {}
@@ -171,30 +145,6 @@ async def tag_merge(dest_id: int):
     if not isinstance(source_id, int) or source_id == dest_id:
         return _bad("invalid_source_id", detail="source_id must be int and differ from dest")
 
-    # dry_run: non-mutating preview (counts + sample) so the operator can
-    # confirm the target before the irreversible merge (#8, rule 93 parity).
-    if body.get("dry_run"):
-        async with get_session() as session:
-            try:
-                p = await TagService(session).merge_preview(
-                    source_id=source_id, target_id=dest_id,
-                )
-            except TagValidationError as exc:
-                return _bad("tag_not_found", status=404, detail=str(exc))
-        return jsonify({
-            "preview": {
-                "source_id": p.source_id, "source_name": p.source_name,
-                "target_id": p.target_id, "target_name": p.target_name,
-                "compatible": p.compatible,
-                "images_moving": p.images_moving,
-                "images_already_on_target": p.images_already_on_target,
-                "source_total": p.source_total,
-                "series_pages": p.series_pages,
-                "will_alias": p.will_alias,
-                "sample_thumbnails": p.sample_thumbnails,
-            },
-        })
-
     async with get_session() as session:
         try:
             result = await TagService(session).merge(
@@ -242,39 +192,16 @@ async def tags_prune_unused():
     re-call with dry_run=false."""
     from ..services.cleanup_service import prune_unused_tags
 
-    return await _run_dry_run_op(prune_unused_tags)
-
-
-@admin_bp.route("/posts/prune-bare", methods=["POST"])
-async def posts_prune_bare():
-    """Tier-A: delete bare posts — Post rows with no linked images (primary OR
-    provenance) and no attachments. Dry-run preview list IS the prompt: UI calls
-    with dry_run=true first, shows the count + sample, operator confirms by
-    re-calling with dry_run=false. Same preview/apply-parity predicate as the
-    prune itself, so the preview can't diverge from the delete."""
-    from ..services.cleanup_service import prune_bare_posts
-
-    return await _run_dry_run_op(prune_bare_posts)
-
-
-@admin_bp.route("/posts/reconcile-duplicates", methods=["POST"])
-async def posts_reconcile_duplicates():
-    """Tier-A: unify duplicate post rows for the same real post — the gallery-dl
-    (attachment-id) + native (post-id) duplicates — onto ONE post-id-keyed keeper,
-    moving image/provenance/attachment/link rows over. Images are untouched.
-    dry_run=true returns {groups, posts_to_merge, sample}; dry_run=false applies
-    and returns {groups, merged, sample}. Optional source_id scopes to one source.
-    Same find_duplicate_post_groups predicate drives preview + apply (rule 93)."""
-    from ..services.cleanup_service import reconcile_duplicate_posts
-
     body = await request.get_json(silent=True) or {}
-    raw_source = body.get("source_id")
-    try:
-        source_id = int(raw_source) if raw_source is not None else None
-    except (TypeError, ValueError):
-        return _bad("invalid_source_id", detail="source_id must be an integer")
+    dry_run = bool(body.get("dry_run", False))
 
-    return await _run_dry_run_op(reconcile_duplicate_posts, source_id=source_id)
+    async with get_session() as session:
+        result = await session.run_sync(
+            lambda sync_sess: prune_unused_tags(
+                sync_sess, dry_run=dry_run,
+            )
+        )
+    return jsonify(result)
 
 
 @admin_bp.route("/tags/purge-legacy", methods=["POST"])
@@ -287,7 +214,14 @@ async def tags_purge_legacy():
     operator confirms with dry_run=false."""
     from ..services.cleanup_service import purge_legacy_tags
 
-    return await _run_dry_run_op(purge_legacy_tags)
+    body = await request.get_json(silent=True) or {}
+    dry_run = bool(body.get("dry_run", False))
+
+    async with get_session() as session:
+        result = await session.run_sync(
+            lambda sync_sess: purge_legacy_tags(sync_sess, dry_run=dry_run)
+        )
+    return jsonify(result)
 
 
 @admin_bp.route("/tags/reset-content", methods=["POST"])
@@ -295,39 +229,20 @@ async def tags_reset_content():
     """Tier-A: delete ALL general + character tags (the Camie-suggestable
     content vocabulary) so the operator can re-tag from scratch via
     auto-suggest. fandom + series tags + series_page ordering are preserved,
-    and image_prediction rows are untouched so suggestions repopulate.
+    and image tagger_predictions are untouched so suggestions repopulate.
     dry-run preview returns per-kind counts + applications + a sample so the
     UI shows exactly what'll go before the operator confirms (dry_run=false).
     Irreversible except via DB backup restore."""
     from ..services.cleanup_service import reset_content_tagging
 
-    return await _run_dry_run_op(reset_content_tagging)
-
-
-@admin_bp.route("/tags/normalize", methods=["POST"])
-async def tags_normalize():
-    """#714: retro-normalize existing tags to the #701 canonical form (Title
-    Case + collapsed whitespace) and merge case/whitespace-variant duplicates.
-
-    dry_run=true (default) returns a projection inline — group/collision/rename
-    counts + a sample of the changes — so the UI shows exactly what'll happen.
-    dry_run=false dispatches the long-running maintenance task (the merge FK
-    repoints can touch many tags); the UI tails the activity dashboard for the
-    summary. Idempotent; back up first (the merges are irreversible)."""
-    from ..services.tag_service import normalize_existing_tags
-
     body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", True))
+    dry_run = bool(body.get("dry_run", False))
 
-    if dry_run:
-        async with get_session() as session:
-            result = await normalize_existing_tags(session, dry_run=True)
-        return jsonify(result)
-
-    from ..tasks.admin import normalize_tags_task
-
-    async_result = normalize_tags_task.delay()
-    return _queued(async_result)
+    async with get_session() as session:
+        result = await session.run_sync(
+            lambda sync_sess: reset_content_tagging(sync_sess, dry_run=dry_run)
+        )
+    return jsonify(result)
 
 
 @admin_bp.route("/maintenance/db-stats", methods=["GET"])
@@ -374,74 +289,3 @@ async def trigger_vacuum():
 
     vacuum_analyze.delay()
     return jsonify({"status": "queued"}), 202
-
-
-@admin_bp.route("/maintenance/reextract-archives", methods=["POST"])
-async def trigger_reextract_archives():
-    """Operator-triggered re-extract (#713): PostAttachments that are actually
-    archives but were filed opaquely (pre magic-byte gate) get extracted and
-    their members linked to the post. Idempotent; runs on the maintenance queue."""
-    from ..tasks.admin import reextract_archive_attachments_task
-
-    async_result = reextract_archive_attachments_task.delay()
-    return _queued(async_result)
-
-
-@admin_bp.route("/maintenance/prune-missing-files", methods=["POST"])
-async def trigger_prune_missing_files():
-    """Operator-triggered orphan repair (#859): delete ImageRecords whose backing
-    file is gone from disk (e.g. left by the external-attach unlink bug), so they
-    stop 404-ing on playback. The task aborts WITHOUT deleting if a large fraction
-    of files look missing (a filesystem/NFS stall). Maintenance queue;
-    operator-triggered only — never an unattended sweep."""
-    from ..tasks.admin import prune_missing_file_records_task
-
-    async_result = prune_missing_file_records_task.delay()
-    return _queued(async_result)
-
-
-@admin_bp.route("/maintenance/dedup-videos", methods=["POST"])
-async def trigger_dedup_videos():
-    """Tier-1 video dedup (#871). Body {"dry_run": bool}: dry_run=true previews
-    what would be removed (groups / redundant count / reclaimable bytes) WITHOUT
-    deleting; dry_run=false applies it (re-link posts to the keeper, then delete
-    the redundant copies). Either way it first re-probes NULL-duration videos so
-    the existing library participates. Returns the Celery task id — poll
-    /maintenance/task-result/<id> for the summary."""
-    from ..tasks.admin import dedup_videos_task
-
-    body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", True))  # default to the SAFE preview
-    async_result = dedup_videos_task.delay(dry_run=dry_run)
-    return _queued(async_result)
-
-
-@admin_bp.route("/maintenance/purge-gated-previews", methods=["POST"])
-async def trigger_purge_gated_previews():
-    """Cleanup (#874 follow-up). Body {"dry_run": bool}: dry_run=true previews how
-    many blurred locked-preview images (grabbed from tier-gated Patreon posts
-    before the fix) would be removed WITHOUT deleting; dry_run=false applies it.
-    Re-walks every enabled Patreon source read-only and matches by content hash, so
-    real content downloaded when access existed is provably spared. Returns the
-    Celery task id — poll /maintenance/task-result/<id> for the summary."""
-    from ..tasks.admin import purge_gated_previews_task
-
-    body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", True))  # default to the SAFE preview
-    async_result = purge_gated_previews_task.delay(dry_run=dry_run)
-    return _queued(async_result)
-
-
-@admin_bp.route("/maintenance/task-result/<task_id>", methods=["GET"])
-async def maintenance_task_result(task_id: str):
-    """Poll a maintenance Celery task's result (the summary dict it returns).
-    Used by the video-dedup card to show the dry-run projection before apply."""
-    from ..celery_app import celery
-
-    res = celery.AsyncResult(task_id)
-    ready = res.ready()
-    return jsonify({
-        "ready": ready,
-        "successful": res.successful() if ready else None,
-        "result": res.result if (ready and res.successful()) else None,
-    })

backend/app/api/allowlist.py

@@ -20,37 +20,12 @@ async def list_allowlist():
                 "tag_name": r.tag_name,
                 "tag_kind": r.tag_kind,
                 "min_confidence": r.min_confidence,
-                "applied_count": r.applied_count,
-                "coverage_count": r.coverage_count,
             }
             for r in rows
         ]
     )
 
 
-@allowlist_bp.route("/tags/<int:tag_id>/allowlist/coverage", methods=["GET"])
-async def coverage(tag_id: int):
-    """Live "at threshold T, a sweep would cover ~N images" projection for the
-    allowlist tuning dashboard. Defaults to the tag's stored threshold."""
-    raw = request.args.get("threshold")
-    async with get_session() as session:
-        svc = AllowlistService(session)
-        if raw is not None:
-            try:
-                threshold = float(raw)
-            except ValueError:
-                return jsonify({"error": "threshold must be a float"}), 400
-            if not (0 < threshold <= 1):
-                return jsonify({"error": "threshold must be in (0, 1]"}), 400
-        else:
-            row = await session.get(TagAllowlist, tag_id)
-            if row is None:
-                return jsonify({"error": "not on allowlist"}), 404
-            threshold = row.min_confidence
-        count = await svc.coverage(tag_id, threshold)
-    return jsonify({"count": count, "threshold": threshold})
-
-
 @allowlist_bp.route("/tags/<int:tag_id>/allowlist", methods=["GET"])
 async def get_one(tag_id: int):
     async with get_session() as session:

backend/app/api/ccip.py

@@ -1,124 +0,0 @@
-"""CCIP / region observability API (#114) — read-only, analysis-shaped.
-
-So the work can be checked through an API as the agent fills in vectors: overall
-coverage (regions by kind, how many images have figure CCIP vectors, which
-characters have enough reference examples to match on) + a per-image drill-down
-(its regions + the CCIP character matches it would get). Mirrors the heads
-metrics endpoint; no GPU, just reads what's stored.
-"""
-
-from quart import Blueprint, jsonify
-from sqlalchemy import distinct, func, select
-
-from ..extensions import get_session
-from ..models import ImageRegion, Tag, TagKind
-from ..models.tag import image_tag
-from ..services.ml.ccip import match_image
-
-ccip_bp = Blueprint("ccip", __name__, url_prefix="/api/ccip")
-
-_FIGURE_KINDS = ("face", "figure")
-
-
-@ccip_bp.route("/overview", methods=["GET"])
-async def overview():
-    async with get_session() as session:
-        by_kind = dict(
-            (
-                await session.execute(
-                    select(ImageRegion.kind, func.count()).group_by(ImageRegion.kind)
-                )
-            ).all()
-        )
-        images_with_figure_ccip = (
-            await session.execute(
-                select(func.count(distinct(ImageRegion.image_record_id)))
-                .where(ImageRegion.kind.in_(_FIGURE_KINDS))
-                .where(ImageRegion.ccip_embedding.is_not(None))
-            )
-        ).scalar_one()
-        # Concept-crop (SigLIP bag) coverage — how far the back-catalogue embed
-        # has progressed, so the max-over-bag scorer's reach is checkable.
-        images_with_concept_siglip = (
-            await session.execute(
-                select(func.count(distinct(ImageRegion.image_record_id)))
-                .where(ImageRegion.kind == "concept")
-                .where(ImageRegion.siglip_embedding.is_not(None))
-            )
-        ).scalar_one()
-        # Per-character reference counts (no vectors loaded) — which characters
-        # have enough examples to match on.
-        ref_rows = (
-            await session.execute(
-                select(image_tag.c.tag_id, Tag.name, func.count())
-                .select_from(ImageRegion)
-                .join(
-                    image_tag,
-                    image_tag.c.image_record_id == ImageRegion.image_record_id,
-                )
-                .join(Tag, Tag.id == image_tag.c.tag_id)
-                .where(Tag.kind == TagKind.character)
-                .where(ImageRegion.kind.in_(_FIGURE_KINDS))
-                .where(ImageRegion.ccip_embedding.is_not(None))
-                .group_by(image_tag.c.tag_id, Tag.name)
-                .order_by(func.count().desc())
-            )
-        ).all()
-        versions = [
-            v for (v,) in (
-                await session.execute(
-                    select(distinct(ImageRegion.embedding_version))
-                )
-            ).all() if v
-        ]
-        auto_applied = (
-            await session.execute(
-                select(func.count()).select_from(image_tag).where(
-                    image_tag.c.source == "ccip_auto"
-                )
-            )
-        ).scalar_one()
-    return jsonify({
-        "regions_by_kind": by_kind,
-        "images_with_figure_ccip": images_with_figure_ccip,
-        "images_with_concept_siglip": images_with_concept_siglip,
-        "characters_with_references": len(ref_rows),
-        "character_references": [
-            {"tag_id": t, "name": n, "n_refs": c} for (t, n, c) in ref_rows
-        ],
-        "embedding_versions": versions,
-        "auto_applied": auto_applied,
-    })
-
-
-@ccip_bp.route("/images/<int:image_id>", methods=["GET"])
-async def image_detail(image_id: int):
-    """An image's stored regions + the CCIP character matches it would get —
-    for spot-checking the agent's output + the matcher."""
-    async with get_session() as session:
-        regions = (
-            await session.execute(
-                select(ImageRegion)
-                .where(ImageRegion.image_record_id == image_id)
-                .order_by(ImageRegion.id)
-            )
-        ).scalars().all()
-        matches = await match_image(session, image_id)
-    return jsonify({
-        "image_id": image_id,
-        "regions": [
-            {
-                "id": r.id,
-                "kind": r.kind,
-                "bbox": [r.rx, r.ry, r.rw, r.rh],
-                "frame_time": r.frame_time,
-                "score": r.score,
-                "detector_version": r.detector_version,
-                "embedding_version": r.embedding_version,
-                "has_ccip": r.ccip_embedding is not None,
-                "has_siglip": r.siglip_embedding is not None,
-            }
-            for r in regions
-        ],
-        "ccip_matches": matches,
-    })

backend/app/api/credentials.py

@@ -121,15 +121,13 @@ async def delete_credential(platform: str):
 
 @credentials_bp.route("/<platform>/verify", methods=["POST"])
 async def verify_credential(platform: str):
-    """Test the stored credential against one of the platform's enabled sources,
-    WITHOUT downloading. Routes through the platform's backend
-    (download_backends.verify_credential) — native ingester for Patreon, an
-    authenticated API page; gallery-dl --simulate for the rest. On success
-    stamps last_verified. Returns {valid: bool|null, reason, last_verified?};
-    valid=null means "couldn't test" (no credential, no enabled source, or an
-    inconclusive network/drift result)."""
+    """Test the stored credential by running gallery-dl --simulate
+    against one of the platform's enabled sources. On success stamps
+    last_verified. Returns {valid: bool|null, reason, last_verified?}.
+    valid=null means "couldn't test" (no credential, or no enabled
+    source to point at)."""
     from ..models import Artist, Source
-    from ..services.download_backends import verify_source_credential
+    from ..services.gallery_dl import GalleryDLService, SourceConfig
 
     async with get_session() as session:
         if not await _ext_key_ok(session):
@@ -156,14 +154,14 @@ async def verify_credential(platform: str):
         cookies_path = await svc.get_cookies_path(platform)
         auth_token = await svc.get_token(platform)
 
-    ok, message = await verify_source_credential(
-        platform=platform,
+    gdl = GalleryDLService(images_root=Path("/images"))
+    ok, message = await gdl.verify(
         url=source.url,
         artist_slug=artist.slug,
-        config_overrides=source.config_overrides or {},
+        platform=platform,
+        source_config=SourceConfig.from_dict(source.config_overrides or {}),
         cookies_path=str(cookies_path) if cookies_path else None,
         auth_token=auth_token,
-        images_root=Path("/images"),
     )
 
     last_verified = None

backend/app/api/downloads.py

@@ -44,9 +44,6 @@ def _list_record(event: DownloadEvent, source: Source | None, artist: Artist | N
         "bytes_downloaded": event.bytes_downloaded,
         "error": event.error,
         "summary": _summary_from_metadata(event.metadata_),
-        # plan #709: mid-walk live counts for a RUNNING native-ingester event
-        # (None otherwise; phase 3 overwrites metadata with run_stats on finish).
-        "live": (event.metadata_ or {}).get("live"),
     }
 
 

backend/app/api/gallery.py

@@ -37,30 +37,16 @@ def _parse_filters():
     """Parse the composable gallery filters from query args, returning
     ``(filters_dict, sort)``. Raises ValueError (→ 400) on malformed ids/dates.
 
-    The structured tag filter (#6) is AND-of-OR plus exclusions:
-    - `tag_id` accepts a single id or a comma-separated list — all ANDed
-      (the include common case; back-compat).
-    - `tag_or` is REPEATABLE; each instance is a comma-separated OR-group, and
-      the image must match at least one tag from EACH group (groups ANDed).
-    - `tag_not` is a comma-separated exclude list (image must carry none).
-
-    `media` is image|video; `sort` is newest|oldest; `platform` selects one
-    platform (or the UNSOURCED_PLATFORM sentinel); `untagged`/`no_artist` are
-    boolean flags; `date_from`/`date_to` are inclusive calendar-day bounds
-    (date_to is widened by a day so the whole day is covered by the service's
-    half-open `< date_to`)."""
+    `tag_id` accepts a single id or a comma-separated list (AND); `media` is
+    image|video; `sort` is newest|oldest; `platform` selects one platform
+    (or the UNSOURCED_PLATFORM sentinel); `untagged`/`no_artist` are boolean
+    flags; `date_from`/`date_to` are inclusive calendar-day bounds (date_to is
+    widened by a day so the whole day is covered by the service's half-open
+    `< date_to`)."""
     tag_raw = request.args.get("tag_id")
     tag_ids = (
         [int(x) for x in tag_raw.split(",") if x.strip()] if tag_raw else None
     ) or None
-    tag_or_groups = [
-        grp for raw in request.args.getlist("tag_or")
-        if (grp := [int(x) for x in raw.split(",") if x.strip()])
-    ] or None
-    not_raw = request.args.get("tag_not")
-    tag_exclude = (
-        [int(x) for x in not_raw.split(",") if x.strip()] if not_raw else None
-    ) or None
     post_id_raw = request.args.get("post_id")
     post_id = int(post_id_raw) if post_id_raw else None
     artist_id_raw = request.args.get("artist_id")
@@ -78,9 +64,7 @@ def _parse_filters():
         date_to += timedelta(days=1)  # inclusive of the date_to calendar day
     filters = {
         "tag_ids": tag_ids, "post_id": post_id, "artist_id": artist_id,
-        "media_type": media_type,
-        "tag_or_groups": tag_or_groups, "tag_exclude": tag_exclude,
-        "platform": platform,
+        "media_type": media_type, "platform": platform,
         "untagged": untagged, "no_artist": no_artist,
         "date_from": date_from, "date_to": date_to,
     }

backend/app/api/gpu.py

@@ -1,220 +0,0 @@
-"""GPU-job API (#114): the HTTP surface the desktop agent pulls work from.
-
-The agent stays HTTP-only — it leases jobs, fetches image pixels via the normal
-FC image URLs, and submits embeddings/regions back, all over this API. Redis and
-Postgres are never exposed. The agent endpoints are gated by a bearer token
-(Authorization: Bearer <token>) stored in AppSetting; the admin endpoints
-(token / backfill / status) ride the browser session like the rest of FC's
-homelab admin.
-"""
-
-import secrets
-
-from quart import Blueprint, jsonify, request
-from sqlalchemy import func, select
-from sqlalchemy.dialects.postgresql import insert as pg_insert
-
-from ..extensions import get_session
-from ..models import AppSetting, GpuJob, ImageRecord, MLSettings
-from ..services.gallery_service import image_url
-from ..services.ml.embedder import MODEL_NAME as EMBED_MODEL_NAME
-from ..services.ml.gpu_jobs import GpuJobService
-from ..services.ml.regions import RegionService
-
-gpu_bp = Blueprint("gpu", __name__, url_prefix="/api/gpu")
-
-_TOKEN_KEY = "gpu_agent_token"
-
-
-def _bearer() -> str | None:
-    h = request.headers.get("Authorization", "")
-    return h[7:].strip() if h.startswith("Bearer ") else None
-
-
-async def _agent_authed(session) -> bool:
-    supplied = _bearer()
-    if not supplied:
-        return False
-    stored = (
-        await session.execute(
-            select(AppSetting.value).where(AppSetting.key == _TOKEN_KEY)
-        )
-    ).scalar_one_or_none()
-    return stored is not None and secrets.compare_digest(supplied, stored)
-
-
-# --- Admin (browser): token + backfill + status -------------------------
-
-@gpu_bp.route("/token", methods=["GET"])
-async def get_token():
-    async with get_session() as session:
-        tok = (
-            await session.execute(
-                select(AppSetting.value).where(AppSetting.key == _TOKEN_KEY)
-            )
-        ).scalar_one_or_none()
-    return jsonify({"token": tok, "configured": tok is not None})
-
-
-@gpu_bp.route("/token/rotate", methods=["POST"])
-async def rotate_token():
-    token = secrets.token_urlsafe(32)
-    async with get_session() as session:
-        await session.execute(
-            pg_insert(AppSetting)
-            .values(key=_TOKEN_KEY, value=token)
-            .on_conflict_do_update(index_elements=["key"], set_={"value": token})
-        )
-        await session.commit()
-    return jsonify({"token": token})
-
-
-@gpu_bp.route("/status", methods=["GET"])
-async def status():
-    async with get_session() as session:
-        rows = (
-            await session.execute(
-                select(GpuJob.status, func.count()).group_by(GpuJob.status)
-            )
-        ).all()
-    counts = dict(rows)
-    return jsonify({
-        "pending": counts.get("pending", 0),
-        "leased": counts.get("leased", 0),
-        "done": counts.get("done", 0),
-        "error": counts.get("error", 0),
-    })
-
-
-@gpu_bp.route("/backfill", methods=["POST"])
-async def backfill():
-    """Enqueue a job for every image that doesn't already have one for `task`."""
-    body = await request.get_json(silent=True) or {}
-    task = str(body.get("task") or "ccip")
-    from ..tasks.ml import enqueue_gpu_backfill
-
-    r = enqueue_gpu_backfill.delay(task)
-    return jsonify({"celery_task_id": r.id, "task": task}), 202
-
-
-# --- Agent (bearer token): lease / submit / heartbeat / fail ------------
-
-@gpu_bp.route("/jobs/lease", methods=["POST"])
-async def lease():
-    body = await request.get_json(silent=True) or {}
-    agent_id = str(body.get("agent_id") or "agent")
-    try:
-        batch = min(max(int(body.get("batch_size", 8)), 1), 64)
-    except (TypeError, ValueError):
-        batch = 8
-    async with get_session() as session:
-        if not await _agent_authed(session):
-            return jsonify({"error": "unauthorized"}), 401
-        jobs = await GpuJobService(session).lease(agent_id, batch_size=batch)
-        ml = (
-            await session.execute(select(MLSettings).where(MLSettings.id == 1))
-        ).scalar_one()
-        # image rows for url/mime in one shot
-        ids = [j.image_record_id for j in jobs]
-        imgs = {
-            i.id: i for i in (
-                await session.execute(
-                    select(ImageRecord).where(ImageRecord.id.in_(ids))
-                )
-            ).scalars()
-        } if ids else {}
-        await session.commit()
-        out = []
-        for j in jobs:
-            img = imgs.get(j.image_record_id)
-            if img is None:
-                continue
-            out.append({
-                "job_id": j.id,
-                "image_id": j.image_record_id,
-                "task": j.task,
-                "mime": img.mime,
-                "image_url": image_url(img.path),
-                # For video/animated: the agent samples at this cadence.
-                "frame_interval_seconds": ml.video_frame_interval_seconds,
-                "max_frames": ml.video_max_frames,
-                # The embedding model the agent must use for concept crops, so
-                # its region vectors land in the SAME space the heads trained in.
-                # Server-announced → the agent stays model-agnostic; a swap is a
-                # server setting + a re-embed migration, never an agent change.
-                "embed_model_name": EMBED_MODEL_NAME,
-                "embed_version": ml.embedder_model_version,
-            })
-    return jsonify({"jobs": out})
-
-
-@gpu_bp.route("/jobs/heartbeat", methods=["POST"])
-async def heartbeat():
-    body = await request.get_json(silent=True) or {}
-    agent_id = str(body.get("agent_id") or "agent")
-    job_ids = [int(x) for x in (body.get("job_ids") or [])]
-    async with get_session() as session:
-        if not await _agent_authed(session):
-            return jsonify({"error": "unauthorized"}), 401
-        n = await GpuJobService(session).heartbeat(agent_id, job_ids)
-        await session.commit()
-    return jsonify({"extended": n})
-
-
-@gpu_bp.route("/jobs/submit", methods=["POST"])
-async def submit():
-    """Store a job's regions + close it. regions: [{kind, bbox:[x,y,w,h],
-    frame_time?, score?, *_version?, ccip_embedding?, siglip_embedding?}].
-    replace_kinds defaults to the kinds present in the submitted regions."""
-    body = await request.get_json(silent=True) or {}
-    agent_id = str(body.get("agent_id") or "agent")
-    job_id = body.get("job_id")
-    regions = body.get("regions") or []
-    if job_id is None:
-        return jsonify({"error": "job_id required"}), 400
-    kinds = body.get("replace_kinds") or sorted({r["kind"] for r in regions})
-    async with get_session() as session:
-        if not await _agent_authed(session):
-            return jsonify({"error": "unauthorized"}), 401
-        job = await session.get(GpuJob, int(job_id))
-        if job is None or job.status != "leased" or job.lease_token != agent_id:
-            return jsonify({"error": "lease_invalid"}), 409
-        if kinds:
-            await RegionService(session).replace_regions(
-                job.image_record_id, kinds, regions
-            )
-        await GpuJobService(session).complete(agent_id, int(job_id))
-        await session.commit()
-    return jsonify({"ok": True, "stored": len(regions)})
-
-
-@gpu_bp.route("/jobs/fail", methods=["POST"])
-async def fail():
-    body = await request.get_json(silent=True) or {}
-    agent_id = str(body.get("agent_id") or "agent")
-    job_id = body.get("job_id")
-    if job_id is None:
-        return jsonify({"error": "job_id required"}), 400
-    async with get_session() as session:
-        if not await _agent_authed(session):
-            return jsonify({"error": "unauthorized"}), 401
-        ok = await GpuJobService(session).fail(
-            agent_id, int(job_id), str(body.get("error") or "")
-        )
-        await session.commit()
-    return jsonify({"ok": ok})
-
-
-@gpu_bp.route("/jobs/release", methods=["POST"])
-async def release():
-    """Graceful stop: the agent hands its still-leased jobs back to pending so
-    they're picked up immediately instead of waiting out the lease."""
-    body = await request.get_json(silent=True) or {}
-    agent_id = str(body.get("agent_id") or "agent")
-    job_ids = [int(x) for x in (body.get("job_ids") or [])]
-    async with get_session() as session:
-        if not await _agent_authed(session):
-            return jsonify({"error": "unauthorized"}), 401
-        n = await GpuJobService(session).release(agent_id, job_ids)
-        await session.commit()
-    return jsonify({"released": n})

backend/app/api/heads.py

@@ -1,285 +0,0 @@
-"""Heads API (#114): train + inspect the per-concept heads that power
-suggestions (replacing Camie + centroid).
-
-POST /api/heads/train   — (re)train all eligible heads (one run at a time).
-GET  /api/heads         — status: head count, last-trained, running run, the
-                          per-concept head table (strength + auto-apply ready),
-                          and recent training runs. The card rehydrates from
-                          here so status survives navigation.
-"""
-
-from quart import Blueprint, jsonify, request
-from sqlalchemy import desc, func, select
-
-from ..extensions import get_session
-from ..models import (
-    HeadAutoApplyRun,
-    HeadMetric,
-    HeadMetricsSnapshot,
-    HeadTrainingRun,
-    Tag,
-    TagHead,
-)
-from ..models.tag import image_tag
-from ..services.ml.heads import (
-    HeadAutoApplyAlreadyRunning,
-    HeadAutoApplyDisabled,
-    HeadTrainingAlreadyRunning,
-    start_head_auto_apply_run,
-    start_head_training_run,
-)
-
-heads_bp = Blueprint("heads", __name__, url_prefix="/api/heads")
-
-
-def _serialize_run(run: HeadTrainingRun) -> dict:
-    return {
-        "id": run.id,
-        "params": run.params,
-        "status": run.status,
-        "started_at": run.started_at.isoformat() if run.started_at else None,
-        "finished_at": run.finished_at.isoformat() if run.finished_at else None,
-        "n_trained": run.n_trained,
-        "n_skipped": run.n_skipped,
-        "error": run.error,
-    }
-
-
-@heads_bp.route("/train", methods=["POST"])
-async def train():
-    body = await request.get_json(silent=True) or {}
-    params = body.get("params") or body or {}
-    async with get_session() as session:
-        try:
-            run_id = await session.run_sync(
-                lambda s: start_head_training_run(s, params)
-            )
-        except HeadTrainingAlreadyRunning as running:
-            return jsonify({
-                "error": "training_already_running",
-                "running_id": int(running.args[0]),
-            }), 409
-        await session.commit()
-    return jsonify({"run_id": run_id, "status": "running"}), 202
-
-
-@heads_bp.route("", methods=["GET"])
-async def status():
-    async with get_session() as session:
-        count, last_trained = (
-            await session.execute(
-                select(func.count(), func.max(TagHead.trained_at))
-            )
-        ).one()
-        graduated = (
-            await session.execute(
-                select(func.count()).where(
-                    TagHead.auto_apply_threshold.is_not(None)
-                )
-            )
-        ).scalar_one()
-        running = (
-            await session.execute(
-                select(HeadTrainingRun.id)
-                .where(HeadTrainingRun.status == "running")
-                .order_by(HeadTrainingRun.id.desc())
-                .limit(1)
-            )
-        ).scalar_one_or_none()
-        runs = (
-            await session.execute(
-                select(HeadTrainingRun)
-                .order_by(HeadTrainingRun.id.desc())
-                .limit(10)
-            )
-        ).scalars().all()
-        # The per-concept table: strongest first, capped for the admin card.
-        head_rows = (
-            await session.execute(
-                select(
-                    TagHead.tag_id, Tag.name, Tag.kind,
-                    TagHead.n_pos, TagHead.n_neg, TagHead.ap,
-                    TagHead.precision_cv, TagHead.recall,
-                    TagHead.auto_apply_threshold, TagHead.trained_at,
-                )
-                .join(Tag, Tag.id == TagHead.tag_id)
-                .order_by(desc(TagHead.ap))
-                .limit(500)
-            )
-        ).all()
-        heads = [
-            {
-                "tag_id": r.tag_id,
-                "name": r.name,
-                "category": r.kind.value if hasattr(r.kind, "value") else str(r.kind),
-                "n_pos": r.n_pos,
-                "n_neg": r.n_neg,
-                "ap": r.ap,
-                "precision": r.precision_cv,
-                "recall": r.recall,
-                "auto_apply": r.auto_apply_threshold is not None,
-                "trained_at": r.trained_at.isoformat() if r.trained_at else None,
-            }
-            for r in head_rows
-        ]
-    return jsonify({
-        "head_count": count,
-        "graduated_count": graduated,
-        "last_trained_at": last_trained.isoformat() if last_trained else None,
-        "running_id": running,
-        "runs": [_serialize_run(r) for r in runs],
-        "heads": heads,
-    })
-
-
-def _serialize_apply_run(run: HeadAutoApplyRun) -> dict:
-    return {
-        "id": run.id,
-        "dry_run": run.dry_run,
-        "status": run.status,
-        "started_at": run.started_at.isoformat() if run.started_at else None,
-        "finished_at": run.finished_at.isoformat() if run.finished_at else None,
-        "n_applied": run.n_applied,
-        "report": run.report,
-        "error": run.error,
-    }
-
-
-@heads_bp.route("/auto-apply", methods=["POST"])
-async def auto_apply():
-    """Trigger an earned-auto-apply sweep. {dry_run:true} previews (writes
-    nothing); a real sweep needs head_auto_apply_enabled on."""
-    body = await request.get_json(silent=True) or {}
-    params = {"dry_run": bool(body.get("dry_run", False))}
-    async with get_session() as session:
-        try:
-            run_id = await session.run_sync(
-                lambda s: start_head_auto_apply_run(s, params)
-            )
-        except HeadAutoApplyAlreadyRunning as running:
-            return jsonify({
-                "error": "auto_apply_already_running",
-                "running_id": int(running.args[0]),
-            }), 409
-        except HeadAutoApplyDisabled:
-            return jsonify({"error": "auto_apply_disabled"}), 400
-        await session.commit()
-    return jsonify({"run_id": run_id, "status": "running"}), 202
-
-
-@heads_bp.route("/auto-apply", methods=["GET"])
-async def auto_apply_status():
-    async with get_session() as session:
-        running = (
-            await session.execute(
-                select(HeadAutoApplyRun.id)
-                .where(HeadAutoApplyRun.status == "running")
-                .order_by(HeadAutoApplyRun.id.desc())
-                .limit(1)
-            )
-        ).scalar_one_or_none()
-        runs = (
-            await session.execute(
-                select(HeadAutoApplyRun)
-                .order_by(HeadAutoApplyRun.id.desc())
-                .limit(10)
-            )
-        ).scalars().all()
-    return jsonify({
-        "running_id": running,
-        "runs": [_serialize_apply_run(r) for r in runs],
-    })
-
-
-@heads_bp.route("/metrics", methods=["GET"])
-async def metrics():
-    """Auto-apply observability: per-concept current counts (volume, misfires,
-    under-fires, realized misfire rate, head quality) + the daily time-series so
-    the operator can tune the precision target + support floor from real data."""
-    async with get_session() as session:
-        head_rows = (
-            await session.execute(
-                select(
-                    TagHead.tag_id, Tag.name, TagHead.ap, TagHead.precision_cv,
-                    TagHead.recall, TagHead.auto_apply_threshold, TagHead.n_pos,
-                ).join(Tag, Tag.id == TagHead.tag_id)
-            )
-        ).all()
-        heads = {r.tag_id: r for r in head_rows}
-        metric_rows = (
-            await session.execute(
-                select(
-                    HeadMetric.tag_id, HeadMetric.n_misfires, HeadMetric.n_underfires
-                )
-            )
-        ).all()
-        mets = {r.tag_id: r for r in metric_rows}
-        applied = dict(
-            (
-                await session.execute(
-                    select(image_tag.c.tag_id, func.count())
-                    .where(image_tag.c.source == "head_auto")
-                    .group_by(image_tag.c.tag_id)
-                )
-            ).all()
-        )
-        names = {r.tag_id: r.name for r in head_rows}
-        # Names for metric-only tags (head pruned but corrections recorded).
-        missing = [t for t in mets if t not in names]
-        if missing:
-            for tid, nm in (
-                await session.execute(
-                    select(Tag.id, Tag.name).where(Tag.id.in_(missing))
-                )
-            ).all():
-                names[tid] = nm
-
-        concepts = []
-        for tid in set(heads) | set(mets):
-            h = heads.get(tid)
-            m = mets.get(tid)
-            n_applied = applied.get(tid, 0)
-            n_mis = m.n_misfires if m else 0
-            denom = n_applied + n_mis
-            concepts.append({
-                "tag_id": tid,
-                "name": names.get(tid, str(tid)),
-                "n_auto_applied": n_applied,
-                "n_misfires": n_mis,
-                "n_underfires": m.n_underfires if m else 0,
-                # Of everything this head ever auto-applied, the fraction you
-                # removed — the misfire rate (null until something fired).
-                "misfire_rate": round(n_mis / denom, 4) if denom else None,
-                "ap": h.ap if h else None,
-                "precision_cv": h.precision_cv if h else None,
-                "recall": h.recall if h else None,
-                "auto_apply": bool(h and h.auto_apply_threshold is not None),
-                "n_pos": h.n_pos if h else None,
-            })
-        concepts.sort(key=lambda c: (c["n_misfires"], c["n_auto_applied"]), reverse=True)
-
-        snaps = (
-            await session.execute(
-                select(HeadMetricsSnapshot)
-                .order_by(HeadMetricsSnapshot.snapshot_at.desc())
-                .limit(1000)
-            )
-        ).scalars().all()
-    return jsonify({
-        "concepts": concepts,
-        "snapshots": [
-            {
-                "tag_id": s.tag_id,
-                "name": s.name,
-                "snapshot_at": s.snapshot_at.isoformat() if s.snapshot_at else None,
-                "n_auto_applied": s.n_auto_applied,
-                "n_misfires": s.n_misfires,
-                "n_underfires": s.n_underfires,
-                "ap": s.ap,
-                "precision_cv": s.precision_cv,
-                "recall": s.recall,
-                "n_pos": s.n_pos,
-            }
-            for s in snaps
-        ],
-    })

backend/app/api/ml_admin.py

@@ -13,17 +13,6 @@ _EDITABLE = (
     "suggestion_threshold_general",
     "centroid_similarity_threshold",
     "min_reference_images",
-    "tagger_store_floor",
-    "video_frame_interval_seconds",
-    "video_max_frames",
-    "video_min_tag_frames",
-    "head_min_positives",
-    "head_auto_apply_precision",
-    "head_auto_apply_enabled",
-    "head_auto_apply_min_positives",
-    "ccip_match_threshold",
-    "ccip_auto_apply_enabled",
-    "ccip_auto_apply_threshold",
 )
 
 
@@ -41,19 +30,8 @@ async def get_settings():
             "suggestion_threshold_general": s.suggestion_threshold_general,
             "centroid_similarity_threshold": s.centroid_similarity_threshold,
             "min_reference_images": s.min_reference_images,
-            "tagger_store_floor": s.tagger_store_floor,
-            "video_frame_interval_seconds": s.video_frame_interval_seconds,
-            "video_max_frames": s.video_max_frames,
-            "video_min_tag_frames": s.video_min_tag_frames,
             "tagger_model_version": s.tagger_model_version,
             "embedder_model_version": s.embedder_model_version,
-            "head_min_positives": s.head_min_positives,
-            "head_auto_apply_precision": s.head_auto_apply_precision,
-            "head_auto_apply_enabled": s.head_auto_apply_enabled,
-            "head_auto_apply_min_positives": s.head_auto_apply_min_positives,
-            "ccip_match_threshold": s.ccip_match_threshold,
-            "ccip_auto_apply_enabled": s.ccip_auto_apply_enabled,
-            "ccip_auto_apply_threshold": s.ccip_auto_apply_threshold,
         }
     )
 
@@ -69,65 +47,13 @@ async def patch_settings():
         s = (
             await session.execute(select(MLSettings).where(MLSettings.id == 1))
         ).scalar_one()
-
-        # Merge the patch over current values, then validate the result as a
-        # whole — the store-floor invariant couples three fields, so they
-        # can't be checked one at a time.
-        proposed = {f: getattr(s, f) for f in _EDITABLE}
         for field in _EDITABLE:
             if field in body:
-                proposed[field] = body[field]
-
-        err = _validate(proposed)
-        if err is not None:
-            return jsonify({"error": err}), 400
-
-        for field in _EDITABLE:
-            setattr(s, field, proposed[field])
+                setattr(s, field, body[field])
         await session.commit()
     return await get_settings()
 
 
-def _validate(p: dict) -> str | None:
-    """Returns an error string if the proposed settings are invalid, else None.
-
-    Invariant (plan-task #764): the per-category suggestion thresholds can't
-    drop below tagger_store_floor — nothing below the floor is stored, so a
-    lower threshold would silently surface nothing in that gap. The UI clamps
-    the sliders to the floor; this is the server-side backstop.
-    """
-    floor = p["tagger_store_floor"]
-    if not (0.0 <= floor <= 1.0):
-        return "tagger_store_floor must be between 0 and 1"
-    for cat in ("character", "general"):
-        if p[f"suggestion_threshold_{cat}"] < floor:
-            return (
-                f"suggestion_threshold_{cat} cannot be below tagger_store_floor "
-                f"({floor}) — predictions below the floor are not stored"
-            )
-    # Video tagging (#747).
-    if p["video_frame_interval_seconds"] <= 0:
-        return "video_frame_interval_seconds must be > 0"
-    if p["video_max_frames"] < 1:
-        return "video_max_frames must be >= 1"
-    if p["video_min_tag_frames"] < 1:
-        return "video_min_tag_frames must be >= 1"
-    if p["video_min_tag_frames"] > p["video_max_frames"]:
-        return "video_min_tag_frames cannot exceed video_max_frames"
-    # Head training (#114).
-    if int(p["head_min_positives"]) < 1:
-        return "head_min_positives must be >= 1"
-    if not (0.5 <= float(p["head_auto_apply_precision"]) <= 0.999):
-        return "head_auto_apply_precision must be between 0.5 and 0.999"
-    if int(p["head_auto_apply_min_positives"]) < 1:
-        return "head_auto_apply_min_positives must be >= 1"
-    if not (0.5 <= float(p["ccip_match_threshold"]) <= 0.999):
-        return "ccip_match_threshold must be between 0.5 and 0.999"
-    if not (0.5 <= float(p["ccip_auto_apply_threshold"]) <= 0.999):
-        return "ccip_auto_apply_threshold must be between 0.5 and 0.999"
-    return None
-
-
 @ml_admin_bp.route("/backfill", methods=["POST"])
 async def trigger_backfill():
     from ..tasks.ml import backfill

backend/app/api/posts.py

@@ -17,7 +17,6 @@ async def list_posts():
     cursor = args.get("cursor") or None
     artist_id_raw = args.get("artist_id")
     platform = args.get("platform") or None
-    q = (args.get("q") or "").strip() or None
     limit_raw = args.get("limit", "24")
     direction = args.get("direction", "older")
     around_raw = args.get("around")
@@ -57,7 +56,7 @@ async def list_posts():
         if around_id is not None:
             result = await svc.around(
                 post_id=around_id, artist_id=artist_id,
-                platform=platform, q=q, limit=limit,
+                platform=platform, limit=limit,
             )
             if result is None:
                 return _bad("not_found", status=404, detail=f"post id={around_id}")
@@ -65,7 +64,7 @@ async def list_posts():
         try:
             page = await svc.scroll(
                 cursor=cursor, artist_id=artist_id,
-                platform=platform, q=q, limit=limit, direction=direction,
+                platform=platform, limit=limit, direction=direction,
             )
         except ValueError as exc:
             # Service raises ValueError for malformed cursors only;

backend/app/api/settings.py

@@ -25,22 +25,6 @@ _EDITABLE_FIELDS = (
     "download_schedule_default_seconds",
     "download_event_retention_days",
     "download_failure_warning_threshold",
-    "series_suggest_enabled",
-    "series_suggest_threshold",
-    "extdl_mega_enabled",
-    "extdl_gdrive_enabled",
-    "extdl_mediafire_enabled",
-    "extdl_dropbox_enabled",
-    "extdl_pixeldrain_enabled",
-)
-
-# Per-host external-download toggles — all plain booleans, validated uniformly.
-_EXTDL_TOGGLE_FIELDS = (
-    "extdl_mega_enabled",
-    "extdl_gdrive_enabled",
-    "extdl_mediafire_enabled",
-    "extdl_dropbox_enabled",
-    "extdl_pixeldrain_enabled",
 )
 
 
@@ -62,13 +46,6 @@ async def get_import_settings():
         "download_schedule_default_seconds": row.download_schedule_default_seconds,
         "download_event_retention_days": row.download_event_retention_days,
         "download_failure_warning_threshold": row.download_failure_warning_threshold,
-        "series_suggest_enabled": row.series_suggest_enabled,
-        "series_suggest_threshold": row.series_suggest_threshold,
-        "extdl_mega_enabled": row.extdl_mega_enabled,
-        "extdl_gdrive_enabled": row.extdl_gdrive_enabled,
-        "extdl_mediafire_enabled": row.extdl_mediafire_enabled,
-        "extdl_dropbox_enabled": row.extdl_dropbox_enabled,
-        "extdl_pixeldrain_enabled": row.extdl_pixeldrain_enabled,
     })
 
 
@@ -119,22 +96,6 @@ async def update_import_settings():
         if not isinstance(v, int) or isinstance(v, bool) or v < 1 or v > 100:
             return _bad_int("download_failure_warning_threshold", 1, 100)
 
-    if "series_suggest_enabled" in body and not isinstance(
-        body["series_suggest_enabled"], bool
-    ):
-        return jsonify(
-            {"error": "series_suggest_enabled must be a boolean"}
-        ), 400
-    for tog in _EXTDL_TOGGLE_FIELDS:
-        if tog in body and not isinstance(body[tog], bool):
-            return jsonify({"error": f"{tog} must be a boolean"}), 400
-    if "series_suggest_threshold" in body:
-        v = body["series_suggest_threshold"]
-        if not isinstance(v, (int, float)) or isinstance(v, bool) or v < 0 or v > 1:
-            return jsonify(
-                {"error": "series_suggest_threshold must be a number in [0, 1]"}
-            ), 400
-
     async with get_session() as session:
         row = await ImportSettings.load(session)
         for field in _EDITABLE_FIELDS:

backend/app/api/sources.py

@@ -85,22 +85,6 @@ async def create_source():
             return _bad("empty_url", detail=str(exc))
         except DuplicateSourceError as exc:
             return _bad("duplicate", status=409, existing_id=exc.existing_id)
-
-        # Immediate kickoff: a new enabled source is armed for backfill (#693)
-        # but would otherwise sit idle until the next scheduler tick (~60s).
-        # Enqueue the first walk now, skipping only if the platform is in a
-        # rate-limit cooldown (the scheduler picks it up when that clears).
-        dispatch_id = None
-        if record.enabled:
-            cooldowns = await active_platform_cooldowns(session)
-            if record.platform not in cooldowns:
-                session.add(DownloadEvent(source_id=record.id, status="pending"))
-                await session.commit()
-                dispatch_id = record.id
-
-    if dispatch_id is not None:
-        from ..tasks.download import download_source
-        download_source.delay(dispatch_id)
     return jsonify(record.to_dict()), 201
 
 
@@ -138,123 +122,29 @@ async def delete_source(source_id: int):
 
 @sources_bp.route("/<int:source_id>/backfill", methods=["POST"])
 async def set_backfill(source_id: int):
-    """Plan #693/#697 + #830: start/stop a backfill, or start a recovery /
-    recapture. Body: `{"action": "start" | "stop" | "recover" | "recapture"}`
-    (default "start"). 'start' walks the full post history in time-boxed chunks
-    until it reaches the bottom (then the source shows 'complete'); 'recover' is
-    the same walk but bypasses the Patreon seen-ledger to re-fetch
-    dropped-and-deleted near-dups under the current pHash threshold; 'recapture'
-    re-grabs EVERY post's body + external links and localizes on-disk inline
-    images WITHOUT re-downloading media; 'stop' cancels any back to tick mode.
-    Returns the updated source dict (incl. backfill_state / backfill_chunks /
-    backfill_bypass_seen / backfill_recapture)."""
-    from pathlib import Path
-
-    from ..services.credential_service import CredentialService
-    from ..services.download_backends import (
-        uses_native_ingester,
-        verify_source_credential,
-    )
-    from .credentials import _get_crypto
-
+    """Plan #544: arm a source for backfill mode for the next N download
+    runs. Body: `{"runs": int}` (1..10, default 3). Returns the updated
+    source dict. While backfill_runs_remaining > 0, downloads use
+    gallery-dl's full-walk config (skip: True + 30-min timeout) instead
+    of the catch-up default (skip: "exit:20" + 14.5-min timeout)."""
     payload = await request.get_json(silent=True) or {}
-    action = payload.get("action", "start")
-    if action not in ("start", "stop", "recover", "recapture"):
-        return _bad(
-            "invalid_action",
-            detail="action must be 'start', 'stop', 'recover', or 'recapture'",
-        )
-
-    # Pre-flight (plan #703 #2): before arming a deep walk on a native-ingester
-    # platform (where verify is one cheap API page), refuse if the credential is
-    # DEFINITIVELY rejected — don't burn chunks against expired cookies. Proceed
-    # on valid OR inconclusive (a network blip shouldn't block). Gated to native
-    # platforms: gallery-dl verify is a slow --simulate subprocess, too heavy for
-    # an arm action. The credential read happens in a session that's CLOSED
-    # before the verify network call (don't hold a DB conn across the request).
-    if action in ("start", "recover", "recapture"):
-        async with get_session() as session:
-            rec = await SourceService(session).get(source_id)
-            if rec is None:
-                return _bad("not_found", status=404)
-            native = uses_native_ingester(rec.platform)
-            if native:
-                cred = CredentialService(session, _get_crypto())
-                cookies_path = await cred.get_cookies_path(rec.platform)
-                auth_token = await cred.get_token(rec.platform)
-        if native:
-            ok, message = await verify_source_credential(
-                platform=rec.platform,
-                url=rec.url,
-                artist_slug=rec.artist_slug,
-                config_overrides=rec.config_overrides or {},
-                cookies_path=str(cookies_path) if cookies_path else None,
-                auth_token=auth_token,
-                images_root=Path("/images"),
-            )
-            if ok is False:
-                return _bad("credential_rejected", detail=message, status=409)
-
+    runs = payload.get("runs", 3)
+    try:
+        runs = int(runs)
+    except (TypeError, ValueError):
+        return _bad("invalid_runs", detail="runs must be an integer")
     async with get_session() as session:
         try:
-            svc = SourceService(session)
-            if action == "start":
-                record = await svc.start_backfill(source_id)
-            elif action == "recover":
-                record = await svc.start_recovery(source_id)
-            elif action == "recapture":
-                record = await svc.start_recapture(source_id)
-            else:
-                record = await svc.stop_backfill(source_id)
+            record = await SourceService(session).set_backfill_runs(
+                source_id, runs,
+            )
         except LookupError:
             return _bad("not_found", status=404)
+        except ValueError as exc:
+            return _bad("invalid_runs", detail=str(exc))
     return jsonify(record.to_dict())
 
 
-@sources_bp.route("/<int:source_id>/preview", methods=["POST"])
-async def preview_source_endpoint(source_id: int):
-    """Plan #708 B4: dry-run — count what a backfill WOULD download for a native
-    platform (Patreon today), without downloading. Walks the first few feed pages
-    and counts media not already in the seen/dead ledgers. Returns
-    {total_new, posts_scanned, pages_scanned, has_more, sample[]} or 409 + reason
-    (unresolvable campaign id / auth / drift). 400 for gallery-dl platforms (no
-    cheap dry-run — their verify is a slow --simulate)."""
-    from pathlib import Path
-
-    from ..services.credential_service import CredentialService
-    from ..services.download_backends import preview_source, uses_native_ingester
-    from ..tasks._sync_engine import sync_session_factory
-    from .credentials import _get_crypto
-
-    async with get_session() as session:
-        rec = await SourceService(session).get(source_id)
-        if rec is None:
-            return _bad("not_found", status=404)
-        if not uses_native_ingester(rec.platform):
-            return _bad(
-                "unsupported",
-                detail="Preview is only available for native-ingester platforms.",
-                status=400,
-            )
-        cred = CredentialService(session, _get_crypto())
-        cookies_path = await cred.get_cookies_path(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.
-    result = await preview_source(
-        platform=rec.platform,
-        url=rec.url,
-        source_id=source_id,
-        config_overrides=rec.config_overrides or {},
-        cookies_path=str(cookies_path) if cookies_path else None,
-        images_root=Path("/images"),
-        sync_session_factory=sync_session_factory(),
-    )
-    if "error" in result:
-        return _bad("preview_failed", detail=result["error"], status=409)
-    return jsonify(result)
-
-
 @sources_bp.route("/<int:source_id>/check", methods=["POST"])
 async def check_source(source_id: int):
     """FC-3c: enqueue a download for this source.

backend/app/api/suggestions.py

@@ -3,48 +3,16 @@
 from quart import Blueprint, jsonify, request
 
 from ..extensions import get_session
-from ..models import Tag, TagAllowlist
 from ..services.ml.allowlist import AllowlistService
 from ..services.ml.suggestions import SuggestionService
 
 suggestions_bp = Blueprint("suggestions", __name__, url_prefix="/api")
 
 
-async def _accept_payload(session, svc, newly_added: bool, tag_id: int) -> dict:
-    """Shape the accept/alias response. When accepting newly allowlists a tag,
-    include the coverage PROJECTION (at the tag's threshold) so the UI can show
-    a non-blocking "auto-applying to ~N images" toast — the actual apply runs
-    async via apply_allowlist_tags, so this is an estimate, not a post-hoc
-    count (#7)."""
-    payload = {"allowlisted": newly_added}
-    if newly_added:
-        tag = await session.get(Tag, tag_id)
-        row = await session.get(TagAllowlist, tag_id)
-        payload["tag_id"] = tag_id
-        payload["tag_name"] = tag.name if tag is not None else None
-        payload["projected_count"] = await svc.coverage(
-            tag_id, row.min_confidence if row is not None else 0.90,
-        )
-    return payload
-
-
 @suggestions_bp.route("/images/<int:image_id>/suggestions", methods=["GET"])
 async def get_suggestions(image_id: int):
-    # ?min=<float> overrides the configured per-category thresholds so the typed
-    # tag-input dropdown can surface EVERY stored prediction (min=0), including
-    # low-confidence actions/features, in canonical formatting. Omitted → the
-    # curated above-threshold list the Suggestions panel uses.
-    override = None
-    raw_min = request.args.get("min")
-    if raw_min is not None:
-        try:
-            override = min(1.0, max(0.0, float(raw_min)))
-        except ValueError:
-            return jsonify({"error": "min must be a float in [0,1]"}), 400
     async with get_session() as session:
-        sl = await SuggestionService(session).for_image(
-            image_id, threshold_override=override
-        )
+        sl = await SuggestionService(session).for_image(image_id)
     return jsonify(
         {
             "by_category": {
@@ -56,15 +24,6 @@ async def get_suggestions(image_id: int):
                         "score": round(s.score, 4),
                         "source": s.source,
                         "creates_new_tag": s.creates_new_tag,
-                        # raw model key (alias is stored under this) + whether an
-                        # operator alias produced this suggestion — drive the
-                        # modal's "Treat as alias"/"Remove alias" affordances.
-                        "raw_name": s.raw_name,
-                        "via_alias": s.via_alias,
-                        # operator dismissed this tag for this image — surfaced
-                        # (not dropped) so the rail can show it rejected + offer
-                        # one-click un-reject.
-                        "rejected": s.rejected,
                     }
                     for s in items
                 ]
@@ -83,15 +42,13 @@ async def accept_suggestion(image_id: int):
         return jsonify({"error": "tag_id required"}), 400
     tag_id = body["tag_id"]
     async with get_session() as session:
-        svc = AllowlistService(session)
-        newly_added = await svc.accept(image_id, tag_id)
-        payload = await _accept_payload(session, svc, newly_added, tag_id)
+        newly_added = await AllowlistService(session).accept(image_id, tag_id)
         await session.commit()
     if newly_added:
         from ..tasks.ml import apply_allowlist_tags
 
         apply_allowlist_tags.delay(tag_id=tag_id)
-    return jsonify(payload)
+    return "", 204
 
 
 @suggestions_bp.route(
@@ -102,24 +59,19 @@ async def alias_suggestion(image_id: int):
     required = {"alias_string", "alias_category", "canonical_tag_id"}
     if not body or not required.issubset(body):
         return jsonify({"error": f"required: {sorted(required)}"}), 400
-    canonical_tag_id = body["canonical_tag_id"]
     async with get_session() as session:
-        svc = AllowlistService(session)
-        newly_added = await svc.add_alias_and_accept(
+        newly_added = await AllowlistService(session).add_alias_and_accept(
             image_id,
             body["alias_string"],
             body["alias_category"],
-            canonical_tag_id,
-        )
-        payload = await _accept_payload(
-            session, svc, newly_added, canonical_tag_id,
+            body["canonical_tag_id"],
         )
         await session.commit()
     if newly_added:
         from ..tasks.ml import apply_allowlist_tags
 
-        apply_allowlist_tags.delay(tag_id=canonical_tag_id)
-    return jsonify(payload)
+        apply_allowlist_tags.delay(tag_id=body["canonical_tag_id"])
+    return "", 204
 
 
 @suggestions_bp.route(
@@ -135,21 +87,6 @@ async def dismiss_suggestion(image_id: int):
     return "", 204
 
 
-@suggestions_bp.route(
-    "/images/<int:image_id>/suggestions/undismiss", methods=["POST"]
-)
-async def undismiss_suggestion(image_id: int):
-    """Reverse a per-image dismissal (reject-recovery). Idempotent — undoing a
-    tag that isn't rejected is a no-op delete."""
-    body = await request.get_json()
-    if not body or "tag_id" not in body:
-        return jsonify({"error": "tag_id required"}), 400
-    async with get_session() as session:
-        await AllowlistService(session).undismiss(image_id, body["tag_id"])
-        await session.commit()
-    return "", 204
-
-
 @suggestions_bp.route("/suggestions/bulk", methods=["POST"])
 async def bulk_suggestions():
     body = await request.get_json()

backend/app/api/system_activity.py

@@ -31,7 +31,7 @@ system_activity_bp = Blueprint(
 # absent.
 _QUEUE_NAMES = (
     "default", "import", "thumbnail", "ml",
-    "download", "scan", "maintenance", "maintenance_long",
+    "download", "scan", "maintenance",
 )
 
 # Cache module-level so all requests share the cache between polls.
@@ -147,7 +147,6 @@ async def list_runs():
     """Paginated task_run history. Query params:
       queue=<name>     filter to one queue
       status=<status>  filter to one status (running/ok/error/timeout/retry)
-      task=<substr>    case-insensitive substring match on task_name
       limit=<int>      default 50, max 200
       before_id=<int>  cursor for keyset pagination
 
@@ -162,7 +161,6 @@ async def list_runs():
 
     queue = request.args.get("queue")
     status = request.args.get("status")
-    task = request.args.get("task")
     before_id_raw = request.args.get("before_id")
     before_id = int(before_id_raw) if before_id_raw else None
 
@@ -172,11 +170,6 @@ async def list_runs():
             stmt = stmt.where(TaskRun.queue == queue)
         if status:
             stmt = stmt.where(TaskRun.status == status)
-        if task:
-            # Task names contain literal underscores (download_source,
-            # vacuum_analyze) — escape LIKE wildcards so a search for
-            # "vacuum_analyze" doesn't treat "_" as a single-char match.
-            stmt = stmt.where(TaskRun.task_name.ilike(f"%{_escape_like(task)}%", escape="\\"))
         if before_id is not None:
             stmt = stmt.where(TaskRun.id < before_id)
         stmt = stmt.limit(limit + 1)
@@ -232,12 +225,6 @@ async def list_failures():
     })
 
 
-def _escape_like(value: str) -> str:
-    """Escape SQL LIKE/ILIKE metacharacters so user search text is matched
-    literally. Pairs with `escape="\\"` on the .ilike() call."""
-    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
-
-
 def _row_to_dict(r: TaskRun) -> dict:
     return {
         "id": r.id,

backend/app/api/tag_eval.py

@@ -1,70 +0,0 @@
-"""Tag-eval API (#1130): trigger + revisit the head-vs-centroid eval.
-
-The run + full report live in the tag_eval_run row, so the admin card rehydrates
-from GET (history / detail) on mount — the report survives navigation rather than
-living in transient frontend state.
-"""
-
-from quart import Blueprint, jsonify, request
-from sqlalchemy import select
-
-from ..extensions import get_session
-from ..models import TagEvalRun
-from ..services.ml.tag_eval import EvalAlreadyRunning, start_tag_eval_run
-
-tag_eval_bp = Blueprint("tag_eval", __name__, url_prefix="/api/tag-eval")
-
-
-def _serialize(run: TagEvalRun, *, include_report: bool) -> dict:
-    out = {
-        "id": run.id,
-        "params": run.params,
-        "status": run.status,
-        "started_at": run.started_at.isoformat() if run.started_at else None,
-        "finished_at": run.finished_at.isoformat() if run.finished_at else None,
-        "error": run.error,
-    }
-    if include_report:
-        out["report"] = run.report
-    return out
-
-
-@tag_eval_bp.route("", methods=["POST"])
-async def create():
-    body = await request.get_json(silent=True) or {}
-    params = body.get("params") or body or {}
-    async with get_session() as session:
-        try:
-            run_id = await session.run_sync(
-                lambda s: start_tag_eval_run(s, params)
-            )
-        except EvalAlreadyRunning as running:
-            return jsonify({
-                "error": "eval_already_running",
-                "running_id": int(running.args[0]),
-            }), 409
-        await session.commit()
-    return jsonify({"run_id": run_id, "status": "running"}), 202
-
-
-@tag_eval_bp.route("", methods=["GET"])
-async def history():
-    try:
-        limit = min(int(request.args.get("limit", "20")), 100)
-    except ValueError:
-        return jsonify({"error": "invalid_limit"}), 400
-    async with get_session() as session:
-        rows = (await session.execute(
-            select(TagEvalRun).order_by(TagEvalRun.id.desc()).limit(limit)
-        )).scalars().all()
-        # List is light — no full report (the detail endpoint carries it).
-        return jsonify({"runs": [_serialize(r, include_report=False) for r in rows]})
-
-
-@tag_eval_bp.route("/<int:run_id>", methods=["GET"])
-async def detail(run_id: int):
-    async with get_session() as session:
-        run = await session.get(TagEvalRun, run_id)
-        if run is None:
-            return jsonify({"error": "not_found"}), 404
-        return jsonify(_serialize(run, include_report=True))

backend/app/api/tags.py

@@ -2,23 +2,18 @@
 
 from quart import Blueprint, jsonify, request
 from sqlalchemy import exists, select
-from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.exc import IntegrityError
 
 from ..extensions import get_session
-from ..models import Tag, TagKind, TagPositiveConfirmation
+from ..models import Tag, TagKind
 from ..models.tag_allowlist import TagAllowlist
 from ..services.bulk_tag_service import BulkTagService
-from ..services.ml.aliases import AliasService
-from ..services.series_match_service import SeriesMatchService
 from ..services.series_service import SeriesError, SeriesService
 from ..services.tag_directory_service import TagDirectoryService
-from ..services.tag_query import serialize_tag
 from ..services.tag_service import (
     TagMergeConflict,
     TagService,
     TagValidationError,
-    normalize_tag_name,
 )
 from ..utils.tag_prefix import parse_kind_prefix
 
@@ -75,7 +70,17 @@ async def autocomplete():
         hits = await svc.autocomplete(q, kind=kind, limit=limit)
 
     return jsonify(
-        [{**serialize_tag(h), "image_count": h.image_count} for h in hits]
+        [
+            {
+                "id": h.id,
+                "name": h.name,
+                "kind": h.kind,
+                "fandom_id": h.fandom_id,
+                "fandom_name": h.fandom_name,
+                "image_count": h.image_count,
+            }
+            for h in hits
+        ]
     )
 
 
@@ -136,11 +141,6 @@ async def create_tag():
 
     fandom_id = body.get("fandom_id")
 
-    # #701: Title-Case operator-entered tags. Only here (the explicit create
-    # endpoint), NOT in the shared find_or_create — the ML tagger uses that path
-    # and must keep the booru vocabulary's casing for allowlist matching.
-    name = normalize_tag_name(name)
-
     async with get_session() as session:
         svc = TagService(session)
         try:
@@ -158,7 +158,17 @@ async def list_tags_for_image(image_id: int):
     async with get_session() as session:
         svc = TagService(session)
         tags = await svc.list_for_image(image_id)
-    return jsonify([serialize_tag(t) for t in tags])
+    return jsonify(
+        [
+            {
+                "id": t.id,
+                "name": t.name,
+                "kind": t.kind.value,
+                "fandom_id": t.fandom_id,
+            }
+            for t in tags
+        ]
+    )
 
 
 @tags_bp.route("/images/<int:image_id>/tags", methods=["POST"])
@@ -184,20 +194,6 @@ async def remove_tag_from_image(image_id: int, tag_id: int):
     return "", 204
 
 
-@tags_bp.route("/images/<int:image_id>/tags/<int:tag_id>/confirm", methods=["POST"])
-async def confirm_tag_on_image(image_id: int, tag_id: int):
-    """Operator affirmed an applied tag is correct ("keep" on a doubted positive).
-    Idempotent; recorded so the eval's doubts list stops resurfacing it (#1130)."""
-    async with get_session() as session:
-        await session.execute(
-            pg_insert(TagPositiveConfirmation)
-            .values(image_record_id=image_id, tag_id=tag_id)
-            .on_conflict_do_nothing(index_elements=["image_record_id", "tag_id"])
-        )
-        await session.commit()
-    return "", 204
-
-
 @tags_bp.route("/tags/<int:tag_id>", methods=["GET"])
 async def get_tag(tag_id: int):
     """Resolve a single tag (used by the gallery to label its active
@@ -216,25 +212,6 @@ async def get_tag(tag_id: int):
         )
 
 
-@tags_bp.route("/tags/<int:tag_id>/aliases", methods=["GET"])
-async def list_tag_aliases(tag_id: int):
-    """Model keys that fold into this tag (tag-side alias view). Remove via the
-    shared DELETE /api/aliases/<string>/<category>."""
-    async with get_session() as session:
-        if await session.get(Tag, tag_id) is None:
-            return jsonify({"error": "tag not found"}), 404
-        rows = await AliasService(session).list_for_tag(tag_id)
-    return jsonify(
-        [
-            {
-                "alias_string": r.alias_string,
-                "alias_category": r.alias_category,
-            }
-            for r in rows
-        ]
-    )
-
-
 @tags_bp.route("/tags/<int:tag_id>", methods=["PATCH"])
 async def update_tag(tag_id: int):
     """Rename and/or re-fandom a tag. Body may carry `name` and/or
@@ -385,31 +362,6 @@ def _series_err(exc: SeriesError):
     return jsonify({"error": msg}), status
 
 
-def _opt_int(body, key: str):
-    """(value, error) — value is None when absent, error is (json, status)."""
-    if not body or body.get(key) is None:
-        return None, None
-    try:
-        return int(body[key]), None
-    except (TypeError, ValueError):
-        return None, (jsonify({"error": f"{key} must be an integer"}), 400)
-
-
-def _parse_int_list(body, key: str, *, max_ids: int = 500):
-    """(list, error) for a required list of ints under `key`."""
-    if not body or key not in body:
-        return None, (jsonify({"error": f"{key} required"}), 400)
-    raw = body[key]
-    if not isinstance(raw, list) or not raw:
-        return None, (jsonify({"error": f"{key} must be a non-empty list"}), 400)
-    if len(raw) > max_ids:
-        return None, (jsonify({"error": f"too many ids (max {max_ids})"}), 400)
-    try:
-        return [int(x) for x in raw], None
-    except (TypeError, ValueError):
-        return None, (jsonify({"error": f"{key} must be integers"}), 400)
-
-
 @tags_bp.route("/series/<int:tag_id>/pages", methods=["GET"])
 async def series_pages(tag_id: int):
     async with get_session() as session:
@@ -450,26 +402,15 @@ async def series_remove(tag_id: int):
     return jsonify({"removed_count": n})
 
 
-@tags_bp.route("/series/<int:tag_id>/pages/number", methods=["POST"])
-async def series_set_page_number(tag_id: int):
-    """Set one placed page's number — the operator's value (sparse, gaps
-    allowed); pass page_number: null to leave it unnumbered."""
-    body = await request.get_json() or {}
-    image_id, ierr = _opt_int(body, "image_id")
-    if ierr:
-        return ierr
-    if image_id is None:
-        return jsonify({"error": "image_id required"}), 400
-    if "page_number" not in body:
-        return jsonify({"error": "page_number required (may be null)"}), 400
-    page_number, perr = _opt_int(body, "page_number")
-    if perr:
-        return perr
+@tags_bp.route("/series/<int:tag_id>/reorder", methods=["POST"])
+async def series_reorder(tag_id: int):
+    body = await request.get_json()
+    ids, err = _parse_bulk_ids(body, max_ids=500)
+    if err:
+        return err
     async with get_session() as session:
         try:
-            await SeriesService(session).set_page_number(
-                tag_id, image_id, page_number
-            )
+            await SeriesService(session).reorder(tag_id, ids)
         except SeriesError as exc:
             return _series_err(exc)
         await session.commit()
@@ -492,201 +433,3 @@ async def series_cover(tag_id: int):
             return _series_err(exc)
         await session.commit()
     return jsonify({"ok": True})
-
-
-# ---- chapter dividers (FC-6.x) -------------------------------------------
-# A chapter is a cosmetic divider anchored to the page that begins it; it owns
-# no pages. Page ordering follows each page's operator-set number (the
-# /pages/number endpoint), so there is no per-chapter reorder/merge — those are
-# gone.
-
-
-@tags_bp.route("/series/<int:tag_id>/chapters", methods=["POST"])
-async def series_chapter_create(tag_id: int):
-    body = await request.get_json() or {}
-    anchor, aerr = _opt_int(body, "anchor_image_id")
-    if aerr:
-        return aerr
-    if anchor is None:
-        return jsonify({"error": "anchor_image_id required"}), 400
-    title = body.get("title")
-    if title is not None and not isinstance(title, str):
-        return jsonify({"error": "title must be a string"}), 400
-    part, perr = _opt_int(body, "stated_part")
-    if perr:
-        return perr
-    async with get_session() as session:
-        try:
-            ch = await SeriesService(session).create_divider(
-                tag_id, anchor, title=title, stated_part=part,
-            )
-        except SeriesError as exc:
-            return _series_err(exc)
-        await session.commit()
-    return jsonify(ch)
-
-
-@tags_bp.route(
-    "/series/<int:tag_id>/chapters/<int:chapter_id>", methods=["PATCH"]
-)
-async def series_chapter_update(tag_id: int, chapter_id: int):
-    body = await request.get_json() or {}
-    kwargs: dict = {}
-    if "title" in body:
-        if body["title"] is not None and not isinstance(body["title"], str):
-            return jsonify({"error": "title must be a string"}), 400
-        kwargs.update(set_title=True, title=body["title"])
-    if "stated_part" in body:
-        part, perr = _opt_int(body, "stated_part")
-        if perr:
-            return perr
-        kwargs.update(set_part=True, stated_part=part)
-    if "anchor_image_id" in body:
-        anchor, aerr = _opt_int(body, "anchor_image_id")
-        if aerr:
-            return aerr
-        if anchor is None:
-            return jsonify(
-                {"error": "anchor_image_id must be an integer"}
-            ), 400
-        kwargs.update(set_anchor=True, anchor_image_id=anchor)
-    async with get_session() as session:
-        try:
-            await SeriesService(session).update_divider(
-                tag_id, chapter_id, **kwargs
-            )
-        except SeriesError as exc:
-            return _series_err(exc)
-        await session.commit()
-    return jsonify({"ok": True})
-
-
-@tags_bp.route(
-    "/series/<int:tag_id>/chapters/<int:chapter_id>", methods=["DELETE"]
-)
-async def series_chapter_delete(tag_id: int, chapter_id: int):
-    async with get_session() as session:
-        try:
-            await SeriesService(session).delete_divider(tag_id, chapter_id)
-        except SeriesError as exc:
-            return _series_err(exc)
-        await session.commit()
-    return jsonify({"ok": True})
-
-
-# ---- browse list + post→series flows (FC-6.2) -----------------------------
-
-
-@tags_bp.route("/series", methods=["GET"])
-async def series_list():
-    args = request.args
-    sort = args.get("sort", "recent")
-    if sort not in ("recent", "name", "size"):
-        return jsonify({"error": "sort must be recent|name|size"}), 400
-    artist_id = None
-    if args.get("artist_id") is not None:
-        try:
-            artist_id = int(args["artist_id"])
-        except ValueError:
-            return jsonify({"error": "artist_id must be an integer"}), 400
-    async with get_session() as session:
-        rows = await SeriesService(session).list_series(
-            sort=sort, artist_id=artist_id
-        )
-    return jsonify({"series": rows})
-
-
-@tags_bp.route("/series/from-post", methods=["POST"])
-async def series_from_post():
-    body = await request.get_json()
-    post_id, err = _opt_int(body, "post_id")
-    if err:
-        return err
-    if post_id is None:
-        return jsonify({"error": "post_id required"}), 400
-    async with get_session() as session:
-        try:
-            out = await SeriesService(session).promote_post_to_series(post_id)
-        except SeriesError as exc:
-            return _series_err(exc)
-        await session.commit()
-    return jsonify(out)
-
-
-@tags_bp.route("/series/<int:tag_id>/add-post", methods=["POST"])
-async def series_add_post(tag_id: int):
-    body = await request.get_json()
-    post_id, err = _opt_int(body, "post_id")
-    if err:
-        return err
-    if post_id is None:
-        return jsonify({"error": "post_id required"}), 400
-    async with get_session() as session:
-        try:
-            out = await SeriesService(session).add_post(tag_id, post_id)
-        except SeriesError as exc:
-            return _series_err(exc)
-        await session.commit()
-    return jsonify(out)
-
-
-@tags_bp.route("/series/<int:tag_id>/pending/place", methods=["POST"])
-async def series_place_pending(tag_id: int):
-    """Place staged (pending) pages into the run, numbered sequentially from
-    `start_page` in the given order (#789). start_page null → unnumbered."""
-    body = await request.get_json()
-    ids, err = _parse_bulk_ids(body, max_ids=500)
-    if err:
-        return err
-    start, serr = _opt_int(body, "start_page")
-    if serr:
-        return serr
-    async with get_session() as session:
-        try:
-            n = await SeriesService(session).place_pending(
-                tag_id, ids, start_page=start
-            )
-        except SeriesError as exc:
-            return _series_err(exc)
-        await session.commit()
-    return jsonify({"placed_count": n})
-
-
-# ---- suggestion queue (FC-6.3) --------------------------------------------
-
-
-@tags_bp.route("/series/suggestions", methods=["GET"])
-async def series_suggestions_list():
-    async with get_session() as session:
-        rows = await SeriesMatchService(session).list_pending()
-    return jsonify({"suggestions": rows})
-
-
-@tags_bp.route("/series/suggestions/<int:sid>/accept", methods=["POST"])
-async def series_suggestion_accept(sid: int):
-    async with get_session() as session:
-        try:
-            out = await SeriesMatchService(session).accept(sid)
-        except SeriesError as exc:
-            return _series_err(exc)
-        await session.commit()
-    return jsonify(out)
-
-
-@tags_bp.route("/series/suggestions/<int:sid>/dismiss", methods=["POST"])
-async def series_suggestion_dismiss(sid: int):
-    async with get_session() as session:
-        try:
-            await SeriesMatchService(session).dismiss(sid)
-        except SeriesError as exc:
-            return _series_err(exc)
-        await session.commit()
-    return jsonify({"ok": True})
-
-
-@tags_bp.route("/series/suggestions/rescan", methods=["POST"])
-async def series_suggestions_rescan():
-    from ..tasks.admin import rescan_series_suggestions_task
-
-    res = rescan_series_suggestions_task.delay()
-    return jsonify({"task_id": res.id})

backend/app/celery_app.py

@@ -30,7 +30,6 @@ def make_celery() -> Celery:
             "backend.app.tasks.maintenance",
             "backend.app.tasks.ml",
             "backend.app.tasks.download",
-            "backend.app.tasks.external",
             "backend.app.tasks.backup",
             "backend.app.tasks.admin",
             "backend.app.tasks.library_audit",
@@ -43,51 +42,16 @@ def make_celery() -> Celery:
             "backend.app.tasks.ml.*": {"queue": "ml"},
             "backend.app.tasks.thumbnail.*": {"queue": "thumbnail"},
             "backend.app.tasks.download.*": {"queue": "download"},
-            # External file-host fetches are downloads — same lane (they can run
-            # long, but the download worker already tolerates long backfills).
-            "backend.app.tasks.external.*": {"queue": "download"},
             "backend.app.tasks.scan.*": {"queue": "scan"},
-            # `maintenance` is the QUICK lane — recovery sweeps, vacuum, cleanup
-            # (concurrency-1 on the scheduler). The long one-shots (DB backups,
-            # library audits, admin maintenance: normalize/re-extract/cascade-
-            # delete) run on a SEPARATE `maintenance_long` lane + worker so they
-            # can never starve the quick self-healing sweeps (operator-flagged
-            # 2026-06-07: a 2h audit blocked vacuum/backup/normalize for hours).
             "backend.app.tasks.maintenance.*": {"queue": "maintenance"},
-            "backend.app.tasks.backup.*": {"queue": "maintenance_long"},
-            "backend.app.tasks.admin.*": {"queue": "maintenance_long"},
-            "backend.app.tasks.library_audit.*": {"queue": "maintenance_long"},
+            "backend.app.tasks.backup.*": {"queue": "maintenance"},
+            "backend.app.tasks.admin.*": {"queue": "maintenance"},
+            "backend.app.tasks.library_audit.*": {"queue": "maintenance"},
         },
         # Heavy ML tasks need fair dispatch — see ImageRepo's precedent.
         task_acks_late=True,
         worker_prefetch_multiplier=1,
-        # Broker resilience (2026-06-24): a swarm overlay-network blip after a
-        # redeploy left Redis healthy but transiently unreachable, and a worker
-        # starting in that window crash-looped on the initial broker connect
-        # (kombu OperationalError) instead of waiting it out — needing a manual
-        # Redis reset to recover. Retry the broker FOREVER (None) on startup and
-        # at runtime so a transient outage self-heals when routing returns,
-        # rather than the worker exiting.
         broker_connection_retry_on_startup=True,
-        broker_connection_retry=True,
-        broker_connection_max_retries=None,
-        # Redis-transport socket options (apply to the BROKER connection): a
-        # short connect timeout + TCP keepalive so a dead/blocked socket is
-        # noticed and retried, and a periodic health check that proactively
-        # reconnects a live worker through a network hiccup.
-        broker_transport_options={
-            "socket_connect_timeout": 5,
-            "socket_timeout": 30,
-            "socket_keepalive": True,
-            "retry_on_timeout": True,
-            "health_check_interval": 30,
-        },
-        # Same hardening for the Redis RESULT backend (separate connection pool).
-        redis_socket_connect_timeout=5,
-        redis_socket_timeout=30,
-        redis_socket_keepalive=True,
-        redis_retry_on_timeout=True,
-        redis_backend_health_check_interval=30,
         beat_schedule={
             "recover-interrupted-tasks": {
                 "task": "backend.app.tasks.maintenance.recover_interrupted_tasks",
@@ -109,36 +73,6 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.ml.apply_allowlist_tags",
                 "schedule": 86400.0,
             },
-            "train-heads-nightly": {
-                "task": "backend.app.tasks.ml.scheduled_train_heads",
-                "schedule": 86400.0,  # passive cadence; manual retrain stays available
-            },
-            "apply-head-tags-daily": {
-                "task": "backend.app.tasks.ml.scheduled_apply_head_tags",
-                "schedule": 86400.0,  # no-op unless head_auto_apply_enabled
-            },
-            "recover-orphaned-gpu-jobs": {
-                "task": "backend.app.tasks.ml.recover_orphaned_gpu_jobs",
-                "schedule": 60.0,  # quick pickup of work a dead agent orphaned
-            },
-            "enqueue-ccip-backfill-hourly": {
-                "task": "backend.app.tasks.ml.enqueue_gpu_backfill",
-                "schedule": 3600.0,  # auto-feed new images (+ retry errored) so
-                "args": ("ccip",),   # the queue keeps moving without the button
-            },
-            "enqueue-siglip-backfill-daily": {
-                "task": "backend.app.tasks.ml.enqueue_gpu_backfill",
-                "schedule": 86400.0,  # drain the concept-crop back-catalogue +
-                "args": ("siglip",),  # retry failed embeds, no button needed
-            },
-            "ccip-auto-apply-daily": {
-                "task": "backend.app.tasks.ml.scheduled_ccip_auto_apply",
-                "schedule": 86400.0,  # no-op unless ccip_auto_apply_enabled
-            },
-            "snapshot-head-metrics-daily": {
-                "task": "backend.app.tasks.maintenance.snapshot_head_metrics",
-                "schedule": 86400.0,
-            },
             "integrity-verify-weekly": {
                 "task": "backend.app.tasks.maintenance.verify_integrity",
                 "schedule": 604800.0,  # weekly
@@ -186,18 +120,6 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.maintenance.recover_stalled_library_audit_runs",
                 "schedule": 300.0,
             },
-            "recover-stalled-tag-eval-runs": {
-                "task": "backend.app.tasks.maintenance.recover_stalled_tag_eval_runs",
-                "schedule": 300.0,
-            },
-            "recover-stalled-head-training-runs": {
-                "task": "backend.app.tasks.maintenance.recover_stalled_head_training_runs",
-                "schedule": 300.0,
-            },
-            "recover-stalled-head-auto-apply-runs": {
-                "task": "backend.app.tasks.maintenance.recover_stalled_head_auto_apply_runs",
-                "schedule": 300.0,
-            },
             "recover-stalled-import-batches": {
                 "task": "backend.app.tasks.maintenance.recover_stalled_import_batches",
                 "schedule": 300.0,
@@ -222,21 +144,6 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.thumbnail.backfill_thumbnails",
                 "schedule": 86400.0,
             },
-            # External file-host downloads (#830): a steady sweep catches links
-            # the post-download hook missed (worker down, etc.); recovery re-tries
-            # dead links daily; retention prunes long-dead rows.
-            "extdl-sweep": {
-                "task": "backend.app.tasks.external.sweep_external_links",
-                "schedule": 600.0,  # every 10 min
-            },
-            "extdl-recover-daily": {
-                "task": "backend.app.tasks.external.recover_external_links",
-                "schedule": 86400.0,
-            },
-            "extdl-prune-daily": {
-                "task": "backend.app.tasks.external.prune_external_links",
-                "schedule": 86400.0,
-            },
         },
         timezone="UTC",
     )

backend/app/celery_signals.py

@@ -69,15 +69,7 @@ def _queue_for(task) -> str:
         return "ml"
     if name.startswith("backend.app.tasks.thumbnail."):
         return "thumbnail"
-    if name.startswith((
-        "backend.app.tasks.download.",
-        # External file-host fetches share the download lane (celery_app
-        # routes external.* → download). Mirror it here or TaskRun.queue
-        # lies 'default' for them, so per-queue dashboard filters and the
-        # per-queue threshold override miss them — the same gap the
-        # 2026-06-02 audit fixed for backup/admin/library_audit.
-        "backend.app.tasks.external.",
-    )):
+    if name.startswith("backend.app.tasks.download."):
         return "download"
     if name.startswith("backend.app.tasks.scan."):
         return "scan"

backend/app/models/__init__.py

@@ -7,37 +7,20 @@ from .backup_run import BackupRun
 from .base import Base
 from .credential import Credential
 from .download_event import DownloadEvent
-from .external_link import ExternalLink
-from .gpu_job import GpuJob
-from .head_auto_apply_run import HeadAutoApplyRun
-from .head_metric import HeadMetric
-from .head_metrics_snapshot import HeadMetricsSnapshot
-from .head_training_run import HeadTrainingRun
-from .image_prediction import ImagePrediction
 from .image_provenance import ImageProvenance
 from .image_record import ImageRecord
-from .image_region import ImageRegion
 from .import_batch import ImportBatch
 from .import_settings import ImportSettings
 from .import_task import ImportTask
 from .library_audit_run import LibraryAuditRun
 from .ml_settings import MLSettings
-from .patreon_failed_media import PatreonFailedMedia
-from .patreon_seen_media import PatreonSeenMedia
 from .post import Post
 from .post_attachment import PostAttachment
-from .series_chapter import SeriesChapter
 from .series_page import SeriesPage
-from .series_suggestion import SeriesSuggestion
 from .source import Source
-from .subscribestar_failed_media import SubscribeStarFailedMedia
-from .subscribestar_seen_media import SubscribeStarSeenMedia
 from .tag import Tag, TagKind, image_tag
 from .tag_alias import TagAlias
 from .tag_allowlist import TagAllowlist
-from .tag_eval_run import TagEvalRun
-from .tag_head import TagHead
-from .tag_positive_confirmation import TagPositiveConfirmation
 from .tag_reference_embedding import TagReferenceEmbedding
 from .tag_suggestion_rejection import TagSuggestionRejection
 from .task_run import TaskRun
@@ -50,39 +33,22 @@ __all__ = [
     "BackupRun",
     "Source",
     "Credential",
-    "PatreonFailedMedia",
-    "PatreonSeenMedia",
-    "SubscribeStarFailedMedia",
-    "SubscribeStarSeenMedia",
     "Post",
     "PostAttachment",
-    "SeriesChapter",
     "SeriesPage",
-    "SeriesSuggestion",
     "ImageRecord",
-    "ImagePrediction",
     "ImageProvenance",
-    "ImageRegion",
     "Tag",
     "TagKind",
     "image_tag",
     "DownloadEvent",
-    "ExternalLink",
-    "GpuJob",
     "ImportBatch",
     "ImportTask",
     "ImportSettings",
     "LibraryAuditRun",
     "MLSettings",
-    "HeadAutoApplyRun",
-    "HeadMetric",
-    "HeadMetricsSnapshot",
-    "HeadTrainingRun",
     "TagAlias",
     "TagAllowlist",
-    "TagEvalRun",
-    "TagHead",
-    "TagPositiveConfirmation",
     "TagReferenceEmbedding",
     "TagSuggestionRejection",
     "TaskRun",

backend/app/models/external_link.py

@@ -1,73 +0,0 @@
-"""ExternalLink — an off-platform file-host link found in a post body.
-
-Creators host the actual files (films, packs) on mega.nz / Google Drive /
-MediaFire / Dropbox / Pixeldrain and drop the link in the post text. This row
-is the record that the link existed (so nothing is silently dropped), the
-dedup + dead-letter ledger for fetching it, and the driver the download worker
-walks. `url` keeps the FULL link including the `#fragment` (mega's decryption
-key) — truncating it makes the file undownloadable.
-
-status lifecycle: pending → downloading → downloaded | failed | dead
-(too many attempts) | skipped (host disabled). `attachment_id` links the
-captured file once a download lands (SET NULL so deleting the attachment
-doesn't delete the link record).
-"""
-
-from datetime import datetime
-
-from sqlalchemy import (
-    DateTime,
-    Float,
-    ForeignKey,
-    Index,
-    Integer,
-    String,
-    Text,
-    func,
-    text,
-)
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-# Kept in sync with link_extract.SUPPORTED_HOSTS and the CHECK in migration 0049.
-HOSTS = ("mega", "gdrive", "mediafire", "dropbox", "pixeldrain")
-STATUSES = ("pending", "downloading", "downloaded", "failed", "skipped", "dead")
-
-
-class ExternalLink(Base):
-    __tablename__ = "external_link"
-    __table_args__ = (
-        # One row per (post, url). The full url (incl. #fragment) is the identity
-        # — the same file linked twice in a post collapses to one row.
-        Index("uq_external_link_post_url", "post_id", "url", unique=True),
-        Index("ix_external_link_status", "status"),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    post_id: Mapped[int] = mapped_column(
-        ForeignKey("post.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    artist_id: Mapped[int | None] = mapped_column(
-        ForeignKey("artist.id", ondelete="SET NULL"), nullable=True, index=True
-    )
-    host: Mapped[str] = mapped_column(String(16), nullable=False)
-    url: Mapped[str] = mapped_column(Text, nullable=False)
-    label: Mapped[str | None] = mapped_column(Text, nullable=True)
-    status: Mapped[str] = mapped_column(
-        String(16), nullable=False, server_default="pending"
-    )
-    attempts: Mapped[int] = mapped_column(
-        Integer, nullable=False, server_default=text("0")
-    )
-    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    attachment_id: Mapped[int | None] = mapped_column(
-        ForeignKey("post_attachment.id", ondelete="SET NULL"), nullable=True
-    )
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    completed_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
-    duration_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)

backend/app/models/gpu_job.py

@@ -1,50 +0,0 @@
-"""GpuJob — a unit of GPU work the desktop agent pulls over HTTP (#114).
-
-The durable work list that lets the agent stay HTTP-only: the server enqueues a
-job per (image, task) — e.g. detect figures + CCIP-embed — and the agent LEASES a
-batch, computes on its GPU, then SUBMITS results, all over the already-exposed web
-API. Redis/Postgres stay private. A lease has an expiry; the lease query itself
-re-claims expired leases (agent died / stopped mid-batch), so the queue is
-self-healing without a separate sweep. One job is per ITEM; the agent fans a
-VIDEO out into per-frame instances internally (see image_region.frame_time).
-
-State: pending → leased → done | error (a failure under the attempt cap returns to
-pending for another agent).
-"""
-
-from datetime import datetime
-
-from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class GpuJob(Base):
-    __tablename__ = "gpu_job"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    image_record_id: Mapped[int] = mapped_column(
-        ForeignKey("image_record.id", ondelete="CASCADE"), index=True
-    )
-    # What to compute, e.g. 'ccip' (detect figures + CCIP-embed) or 'siglip_region'.
-    task: Mapped[str] = mapped_column(String(32), nullable=False)
-    status: Mapped[str] = mapped_column(
-        String(16), nullable=False, default="pending", index=True
-    )
-    # pending | leased | done | error
-    lease_token: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    leased_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
-    lease_expires_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
-    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/models/head_auto_apply_run.py

@@ -1,46 +0,0 @@
-"""HeadAutoApplyRun — persisted lifecycle of an earned-auto-apply sweep (#114).
-
-A graduated head can apply its tag to images it scores above the head's
-auto-apply threshold, without a human. This row tracks one such sweep (or a
-dry-run PREVIEW of it) so the result survives navigation and the admin card can
-show what fired / what would fire. Mirrors HeadTrainingRun. State machine:
-running → ready / error. The `report` JSONB holds per-concept counts
-(applied / projected / scanned).
-"""
-
-from datetime import datetime
-from typing import Any
-
-from sqlalchemy import Boolean, DateTime, Integer, String, Text, func
-from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class HeadAutoApplyRun(Base):
-    __tablename__ = "head_auto_apply_run"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    # dry_run=True is a PREVIEW: scores + counts what WOULD apply, writes nothing
-    # (preview/apply parity, rule 93).
-    dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
-    params: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
-    status: Mapped[str] = mapped_column(
-        String(16), nullable=False, default="running", index=True
-    )
-    # running | ready | error
-    started_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    finished_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
-    # Total tags applied across all heads this sweep (0 for a clean dry-run).
-    n_applied: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    # Per-concept breakdown: [{tag_id, name, applied, scanned, threshold}, ...].
-    report: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
-    error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    last_progress_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )

backend/app/models/head_metric.py

@@ -1,32 +0,0 @@
-"""HeadMetric — running correction counters per concept (#114 observability).
-
-Earned auto-apply fires graduated heads; to TUNE it we need to know how often a
-head's auto-applied tag was wrong (the operator removed it = a MISFIRE) and how
-often the operator had to add a tag a head exists for by hand (an UNDER-FIRE,
-the head missed it). image_tag.source is lost when a row is deleted, so these
-are captured as durable cumulative counters at correction time — they survive
-head retrain/prune (keyed by tag, not by the head row). The daily snapshot reads
-them into the time-series.
-"""
-
-from datetime import datetime
-
-from sqlalchemy import DateTime, ForeignKey, Integer, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class HeadMetric(Base):
-    __tablename__ = "head_metric"
-
-    tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True
-    )
-    # An auto-applied (source='head_auto') tag the operator later REMOVED.
-    n_misfires: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    # A tag with a head that the operator added by HAND (the head missed it).
-    n_underfires: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/models/head_metrics_snapshot.py

@@ -1,38 +0,0 @@
-"""HeadMetricsSnapshot — a daily per-concept time-series point (#114).
-
-The "amount of change over time" reporting the operator asked for: once a day,
-record each concept's auto-applied VOLUME (current head_auto tags), cumulative
-misfires/under-fires, and the head's measured quality. Plotting these rows over
-time shows whether auto-apply is landing better/worse and whether tagging more is
-sharpening a concept — the signal for tuning the precision target + support floor.
-"""
-
-from datetime import datetime
-
-from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class HeadMetricsSnapshot(Base):
-    __tablename__ = "head_metrics_snapshot"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), index=True
-    )
-    # Denormalized so a snapshot stays readable even if the tag is later renamed.
-    name: Mapped[str] = mapped_column(String(255), nullable=False)
-    snapshot_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now(), index=True
-    )
-    # Current count of source='head_auto' applications still standing.
-    n_auto_applied: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    n_misfires: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    n_underfires: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    # The head's measured quality at snapshot time (null if no head exists).
-    ap: Mapped[float | None] = mapped_column(Float, nullable=True)
-    precision_cv: Mapped[float | None] = mapped_column(Float, nullable=True)
-    recall: Mapped[float | None] = mapped_column(Float, nullable=True)
-    n_pos: Mapped[int | None] = mapped_column(Integer, nullable=True)

backend/app/models/head_training_run.py

@@ -1,44 +0,0 @@
-"""HeadTrainingRun — persisted lifecycle of a head-training batch (#114).
-
-Mirrors TagEvalRun so the run SURVIVES navigation and the admin card can show
-live + historical status instead of holding it in transient frontend state.
-Training is idempotent (it upserts tag_head rows), so a SIGKILL'd run is harmless
-— a maintenance recovery sweep flips a stalled `running` row to `error`, and the
-next run re-trains. State machine: running → ready / error.
-"""
-
-from datetime import datetime
-from typing import Any
-
-from sqlalchemy import DateTime, Integer, String, Text, func
-from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class HeadTrainingRun(Base):
-    __tablename__ = "head_training_run"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    # Training parameters: {min_positives, neg_ratio, precision_target, ...}.
-    params: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
-    status: Mapped[str] = mapped_column(
-        String(16), nullable=False, default="running", index=True
-    )
-    # running | ready | error
-    started_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    finished_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
-    # How many concepts got a (re)trained head vs were skipped (too few labels).
-    n_trained: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    n_skipped: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    # Last time the task made progress — the recovery sweep tells a live run from
-    # a SIGKILL'd one by this (mirrors TagEvalRun).
-    last_progress_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )

backend/app/models/image_prediction.py

@@ -1,37 +0,0 @@
-"""ImagePrediction — one row per (image, tagger vocab prediction).
-
-Replaces the image_record.tagger_predictions JSON blob (#768). Storing the
-raw Camie/booru vocab name (not a tag_id) preserves the suggestion read
-path's semantics: raw_name → canonical Tag resolution happens at read time
-via the alias map, and accepting a prediction can CREATE the Tag. The store
-floor (ml_settings.tagger_store_floor) is applied at WRITE time, so only
-predictions >= the floor land here.
-"""
-
-from sqlalchemy import Float, ForeignKey, Index, String, UniqueConstraint
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class ImagePrediction(Base):
-    __tablename__ = "image_prediction"
-    __table_args__ = (
-        UniqueConstraint(
-            "image_record_id", "raw_name", name="image_raw_name",
-        ),
-        # Per-image read (suggestion build) and the "images with tag X above
-        # Y" query the JSON blob never allowed.
-        Index("ix_image_prediction_image", "image_record_id"),
-        Index("ix_image_prediction_name_score", "raw_name", "score"),
-    )
-
-    id: Mapped[int] = mapped_column(primary_key=True)
-    image_record_id: Mapped[int] = mapped_column(
-        ForeignKey("image_record.id", ondelete="CASCADE"), nullable=False,
-    )
-    # The raw tagger vocab key (booru form) — NOT a tag_id. Resolved to a
-    # canonical Tag at read time, exactly as the old JSON keys were.
-    raw_name: Mapped[str] = mapped_column(String(255), nullable=False)
-    category: Mapped[str] = mapped_column(String(64), nullable=False)
-    score: Mapped[float] = mapped_column(Float, nullable=False)

backend/app/models/image_provenance.py

@@ -41,16 +41,6 @@ class ImageProvenance(Base):
     source_id: Mapped[int | None] = mapped_column(
         ForeignKey("source.id", ondelete="SET NULL"), nullable=True, index=True
     )
-    # The archive PostAttachment this image was extracted FROM, when it came
-    # out of a .zip/.rar rather than as a loose file (milestone #87). Lets the
-    # provenance UI show the exact archive a file lives inside instead of every
-    # attachment on the post. NULL for loose downloads and pre-backfill rows.
-    # SET NULL so deleting the archive attachment never destroys the (image,
-    # post) edge — it just forgets which archive it came from.
-    from_attachment_id: Mapped[int | None] = mapped_column(
-        ForeignKey("post_attachment.id", ondelete="SET NULL"),
-        nullable=True, index=True,
-    )
     captured_metadata: Mapped[dict | None] = mapped_column(JSON, nullable=True)
     captured_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()

backend/app/models/image_record.py

@@ -13,7 +13,6 @@ from sqlalchemy import (
     BigInteger,
     DateTime,
     Enum,
-    Float,
     ForeignKey,
     Integer,
     String,
@@ -40,10 +39,6 @@ class ImageRecord(Base):
     mime: Mapped[str] = mapped_column(String(64), nullable=False)
     width: Mapped[int | None] = mapped_column(Integer, nullable=True)
     height: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    # Video container duration (seconds); NULL for images. The Tier-1 video
-    # near-dup key (#871): two videos of the same artist with matching duration
-    # (+ aspect) are the same content across re-encodes — dedup like image pHash.
-    duration_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)
 
     # Integrity verification status. FC-2e populates this; FC-2a leaves rows at 'unknown'.
     # Values: 'unknown' (default), 'ok', 'corrupt', 'failed_verification'.
@@ -54,18 +49,6 @@ class ImageRecord(Base):
     # Thumbnail (populated by FC-2)
     thumbnail_path: Mapped[str | None] = mapped_column(Text, nullable=True)
 
-    # Source provenance for downloaded media (#830 Phase 2). `source_url` is the
-    # CDN/origin URL the file was fetched from (debugging + future re-fetch).
-    # `source_filehash` is the URL's 32-hex CDN identity segment
-    # (utils.paths.filehash_from_url) — the JOIN KEY that maps a post body's
-    # inline `<img src=CDN>` back to this local copy so the rendered body serves
-    # our stored image instead of hotlinking the public source. Indexed for the
-    # render-time lookup. NULL for filesystem-imported / pre-Phase-2 rows.
-    source_url: Mapped[str | None] = mapped_column(Text, nullable=True)
-    source_filehash: Mapped[str | None] = mapped_column(
-        String(32), nullable=True, index=True
-    )
-
     # Origin / provenance pointers
     origin: Mapped[str] = mapped_column(Enum(*ORIGIN_CHOICES, name="origin_enum"), nullable=False)
     primary_post_id: Mapped[int | None] = mapped_column(
@@ -77,10 +60,8 @@ class ImageRecord(Base):
         ForeignKey("artist.id", ondelete="SET NULL"), nullable=True, index=True
     )
 
-    # ML fields (populated by FC-2's ml-worker). Per-tag predictions live in the
-    # normalized image_prediction table (#768) — the tagger_predictions JSON
-    # column was dropped in migration 0046. tagger_model_version stays as the
-    # "has this been tagged / is it current?" signal the backfill sweep reads.
+    # ML fields (populated by FC-2's ml-worker)
+    tagger_predictions: Mapped[dict | None] = mapped_column(JSON, nullable=True)
     tagger_model_version: Mapped[str | None] = mapped_column(String(128), nullable=True)
     # 1152 = SigLIP-so400m embedding dim. Swapping models in FC-2 may require
     # a column-width migration.

backend/app/models/image_region.py

@@ -1,62 +0,0 @@
-"""ImageRegion — a detected/proposed sub-region of an image + its crop embedding.
-
-The storage backbone of the crop pipeline (#114). A region is a normalized bbox
-plus the embedding of its crop:
-- kind='face' / 'figure' → embedded by CCIP for cross-artist character identity.
-- kind='concept' → embedded by SigLIP, a localized instance for a concept head's
-  bag-of-embeddings (a concept is "present if ANY instance matches").
-One row carries the embedding appropriate to its kind (the other is null). The
-bbox doubles as grounded-tag provenance (hover a tag → highlight its region; a
-wrong box is a precise negative). The GPU agent writes these via the job API;
-the few-shot character matcher + bag scorer read them — both server-side, no GPU.
-"""
-
-from datetime import datetime
-
-from pgvector.sqlalchemy import Vector
-from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-CCIP_DIM = 768       # deepghs/imgutils CCIP character embedding
-SIGLIP_DIM = 1152    # matches image_record.siglip_embedding
-
-
-class ImageRegion(Base):
-    __tablename__ = "image_region"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    image_record_id: Mapped[int] = mapped_column(
-        ForeignKey("image_record.id", ondelete="CASCADE"), index=True
-    )
-    # 'frame' (a whole video frame → SigLIP bag) | 'face' | 'figure' (→ CCIP
-    # character id) | 'concept' (→ SigLIP head bag).
-    kind: Mapped[str] = mapped_column(String(16), nullable=False)
-    # For video/animated media: the source frame's timestamp in SECONDS. NULL for
-    # static images. Lets a video be a BAG of per-frame instances (fixes the
-    # mean-embedding muddle) + grounds a tag to "appears at 0:42".
-    frame_time: Mapped[float | None] = mapped_column(Float, nullable=True)
-    # Normalized bbox in [0,1]: top-left (rx, ry) + size (rw, rh). Named rx/ry/…
-    # rather than x/y/by to dodge SQL keyword ambiguity ('by').
-    rx: Mapped[float] = mapped_column(Float, nullable=False)
-    ry: Mapped[float] = mapped_column(Float, nullable=False)
-    rw: Mapped[float] = mapped_column(Float, nullable=False)
-    rh: Mapped[float] = mapped_column(Float, nullable=False)
-    # Proposer/detector confidence (null for deterministic proposers).
-    score: Mapped[float | None] = mapped_column(Float, nullable=True)
-    # Version stamps so a re-detect / re-crop / re-embed can be gated (compute
-    # once; only redo when the producing model version changes).
-    detector_version: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    crop_version: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    embedding_version: Mapped[str | None] = mapped_column(String(128), nullable=True)
-    # Exactly one is set, per kind.
-    ccip_embedding: Mapped[list[float] | None] = mapped_column(
-        Vector(CCIP_DIM), nullable=True
-    )
-    siglip_embedding: Mapped[list[float] | None] = mapped_column(
-        Vector(SIGLIP_DIM), nullable=True
-    )
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/models/import_settings.py

@@ -64,34 +64,6 @@ class ImportSettings(Base):
         Integer, nullable=False, default=3,
     )
 
-    # FC-6.3 series continuation matcher. enabled gates the rescan; threshold is
-    # the weighted-score cut-off (0..1) above which a pending suggestion is made.
-    series_suggest_enabled: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=True,
-    )
-    series_suggest_threshold: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.5,
-    )
-
-    # #830 off-platform file-host downloads — per-host enable lever (default on,
-    # rule #26). Column names are extdl_<host>_enabled so the worker reads them
-    # via getattr(settings, f"extdl_{host}_enabled", True).
-    extdl_mega_enabled: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=True, server_default="true",
-    )
-    extdl_gdrive_enabled: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=True, server_default="true",
-    )
-    extdl_mediafire_enabled: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=True, server_default="true",
-    )
-    extdl_dropbox_enabled: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=True, server_default="true",
-    )
-    extdl_pixeldrain_enabled: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=True, server_default="true",
-    )
-
     @classmethod
     async def load(cls, session) -> ImportSettings:
         """The singleton settings row (id=1), via an async session."""

backend/app/models/library_audit_run.py

@@ -35,10 +35,3 @@ class LibraryAuditRun(Base):
     matched_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     matched_ids: Mapped[list[int]] = mapped_column(JSONB, nullable=False, default=list)
     error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    # Chunked-scan state (alembic 0039): keyset cursor the next chunk resumes
-    # from, and the last time a chunk made progress (so the recovery sweep can
-    # tell a progressing multi-chunk audit from a stuck one).
-    resume_after_id: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    last_progress_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True,
-    )

backend/app/models/ml_settings.py

@@ -2,15 +2,7 @@
 
 from datetime import datetime
 
-from sqlalchemy import (
-    Boolean,
-    CheckConstraint,
-    DateTime,
-    Float,
-    Integer,
-    String,
-    func,
-)
+from sqlalchemy import CheckConstraint, DateTime, Float, Integer, String, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from .base import Base
@@ -36,71 +28,9 @@ class MLSettings(Base):
     centroid_similarity_threshold: Mapped[float] = mapped_column(
         Float, nullable=False, default=0.55
     )
-    # Ingest floor: tagger predictions below this confidence are not stored
-    # (tagger.Tagger.infer). Default 0.70 — the suggestion path already
-    # filters at 0.70 and the centroid/learned path covers low-confidence
-    # preferred tags, so the sub-0.70 tail is redundant weight (it had
-    # bloated image_record's TOAST to ~100 GB; plan-task #764). Operator-
-    # tunable via Settings → ML; must stay ≤ the suggestion thresholds.
-    tagger_store_floor: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.70
-    )
     min_reference_images: Mapped[int] = mapped_column(
         Integer, nullable=False, default=5
     )
-    # Video tagging (#747). Sample one frame every N seconds (fixed CADENCE, not a
-    # fixed count) so a tag's frame-presence reflects real screen time regardless
-    # of video length; cap the total so a long video can't explode into hundreds
-    # of inferences (the cadence stretches past the cap). A tag is kept only if it
-    # appears in >= video_min_tag_frames sampled frames (≈ that many × interval
-    # seconds on screen) — duration-independent noise rejection. Operator-tunable.
-    video_frame_interval_seconds: Mapped[float] = mapped_column(
-        Float, nullable=False, default=4.0
-    )
-    video_max_frames: Mapped[int] = mapped_column(
-        Integer, nullable=False, default=64
-    )
-    video_min_tag_frames: Mapped[int] = mapped_column(
-        Integer, nullable=False, default=3
-    )
-    # Tagging-v2 head training (#114). The head is the suggestion source that
-    # LEARNS from the operator's tags (replacing Camie + centroid). A concept
-    # needs >= head_min_positives labelled images before a head is trained;
-    # head_auto_apply_precision is the precision bar a head must clear (at some
-    # operating point) to "graduate" into earned auto-apply. Operator-tunable.
-    head_min_positives: Mapped[int] = mapped_column(
-        Integer, nullable=False, default=8
-    )
-    head_auto_apply_precision: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.97
-    )
-    # Earned auto-apply (#114). A graduated head fires (tags images without a
-    # human) when this master switch is on AND the head has at least
-    # head_auto_apply_min_positives clean labels — so a precise-looking but
-    # under-supported low-N head can't spray tags across the library. ON by
-    # default (operator-asked 2026-06-29: opt-OUT, not opt-in); the support +
-    # measured-precision gates keep it safe, and every auto-tag is reversible.
-    head_auto_apply_enabled: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=True
-    )
-    head_auto_apply_min_positives: Mapped[int] = mapped_column(
-        Integer, nullable=False, default=30
-    )
-    # CCIP character-match cosine cut (#114). 0.85 default — the v1 flat 0.75
-    # over-fired (high-reference characters matched a scatter of images); 0.85
-    # keeps the confident single-character matches. Tunable from the agent card.
-    ccip_match_threshold: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.85
-    )
-    # CCIP auto-apply (#114). Confident matches (>= ccip_auto_apply_threshold,
-    # above the suggest cut) auto-tag on a daily sweep. ON by default (opt-out);
-    # single-character references + the high bar keep it safe, every tag reversible.
-    ccip_auto_apply_enabled: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=True
-    )
-    ccip_auto_apply_threshold: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.92
-    )
     tagger_model_version: Mapped[str] = mapped_column(
         String(128), nullable=False, default="camie-tagger-v2"
     )

backend/app/models/patreon_failed_media.py

@@ -1,45 +0,0 @@
-"""PatreonFailedMedia — per-source dead-letter ledger of Patreon media that
-keeps failing to download/validate.
-
-Plan #705 (#7). A media that fails every walk (404'd CDN URL, deleted post,
-geo-blocked Mux stream, 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 it — the operator's "try everything again"). A later clean download
-clears the row (the media recovered).
-
-`filehash` is the same per-media key the seen-ledger uses (32-hex CDN MD5, or a
-``video:`` / ``post:filename`` synthesized key) — hence String(128). 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 PatreonFailedMedia(Base):
-    __tablename__ = "patreon_failed_media"
-    __table_args__ = (
-        UniqueConstraint(
-            "source_id", "filehash", name="uq_patreon_failed_media_source_id"
-        ),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    source_id: Mapped[int] = mapped_column(
-        ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    filehash: Mapped[str] = mapped_column(String(128), nullable=False)
-    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
-    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    first_failed_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    last_failed_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/models/patreon_seen_media.py

@@ -1,38 +0,0 @@
-"""PatreonSeenMedia — per-source ledger of Patreon media already
-downloaded+processed.
-
-Replaces gallery-dl's archive.sqlite3 with our own queryable table so
-routine walks can skip media we've already ingested (and a future
-"recovery" mode can deliberately bypass the ledger to re-walk).
-
-`filehash` is normally a Patreon CDN MD5 (32 hex chars), but videos —
-which have no stable content hash at discovery time — use a sentinel of
-the form ``video:<post_id>:<media_id>``, hence String(128) rather than 32.
-"""
-
-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 PatreonSeenMedia(Base):
-    __tablename__ = "patreon_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_patreon_seen_media_source_id"),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    source_id: Mapped[int] = mapped_column(
-        ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    filehash: Mapped[str] = mapped_column(String(128), nullable=False)
-    post_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    seen_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/models/post_attachment.py

@@ -14,12 +14,10 @@ from sqlalchemy import (
     BigInteger,
     DateTime,
     ForeignKey,
-    Index,
     Integer,
     String,
     Text,
     func,
-    text,
 )
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -28,24 +26,6 @@ from .base import Base
 
 class PostAttachment(Base):
     __tablename__ = "post_attachment"
-    # Dedup is PER-POST, not global (2026-06-08): the same non-art file attached
-    # to many posts gets one row per post over a single sha-addressed blob, so no
-    # post is left a bare shell. Partial uniques: (post_id, sha256) for real posts;
-    # (sha256) alone for the NULL-post filesystem case (one row per file there).
-    __table_args__ = (
-        Index(
-            "uq_post_attachment_post_sha",
-            "post_id", "sha256",
-            unique=True,
-            postgresql_where=text("post_id IS NOT NULL"),
-        ),
-        Index(
-            "uq_post_attachment_null_post_sha",
-            "sha256",
-            unique=True,
-            postgresql_where=text("post_id IS NULL"),
-        ),
-    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     post_id: Mapped[int | None] = mapped_column(
@@ -55,7 +35,7 @@ class PostAttachment(Base):
         ForeignKey("artist.id", ondelete="SET NULL"), nullable=True, index=True
     )
     sha256: Mapped[str] = mapped_column(
-        String(64), nullable=False, index=True
+        String(64), nullable=False, unique=True, index=True
     )
     path: Mapped[str] = mapped_column(Text, nullable=False)
     original_filename: Mapped[str] = mapped_column(Text, nullable=False)

backend/app/models/series_chapter.py

@@ -1,47 +0,0 @@
-"""SeriesChapter — a cosmetic chapter DIVIDER within a series (FC-6.x reframe).
-
-A series is ONE flat, series-global ordered run of SeriesPages. A chapter is NOT
-a container — it owns no pages. It is a labeled divider anchored to the page that
-BEGINS the chapter (anchor_page_id → series_page): "a new chapter starts here."
-A page's chapter is derived at read time as the nearest preceding divider.
-
-Dividers never affect page ordering or the series-global page numbers; they stay
-pinned to their anchor page across reorders. anchor_page_id is UNIQUE — at most
-one chapter begins at a given page — and FK-cascades, so removing the anchor page
-from the series drops the divider (the chapter merges into the preceding run).
-
-title is the optional chapter name; stated_part is the optional operator-facing
-"Part N" label (shown instead of a derived ordinal when set).
-"""
-
-from datetime import datetime
-
-from sqlalchemy import DateTime, ForeignKey, Integer, Text, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class SeriesChapter(Base):
-    __tablename__ = "series_chapter"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    series_tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    anchor_page_id: Mapped[int] = mapped_column(
-        ForeignKey("series_page.id", ondelete="CASCADE"),
-        nullable=False,
-        unique=True,
-    )
-    title: Mapped[str | None] = mapped_column(Text, nullable=True)
-    stated_part: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True),
-        nullable=False,
-        server_default=func.now(),
-        onupdate=func.now(),
-    )

backend/app/models/series_page.py

@@ -1,20 +1,14 @@
 """SeriesPage — ordered image membership for a series-kind Tag.
 
-A series IS a Tag with kind='series'; series_page gives it a SINGLE flat,
-series-global ordered run of pages (FC-6.x divider reframe). An image belongs to
-at most one series (UNIQUE image_id). Reading order is `page_number` alone — a
-series-wide ordering key (not unique), rewritten 1..N wholesale on reorder so a
-reorder can't transiently collide on an index.
-
-Chapters are cosmetic DIVIDERS anchored to a page (see SeriesChapter); they do
-NOT own pages, so there is no chapter_id here — a page's chapter is derived at
-read time as the nearest preceding divider. stated_page carries the printed page
-number parsed from the source post, nullable when unknown.
+A series IS a Tag with kind='series'; series_page gives it ordered pages.
+An image belongs to at most one series (UNIQUE image_id). Cover = the
+lowest page_number. page_number is an ordering key only (not unique) —
+reorder rewrites 1..N wholesale.
 """
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy import DateTime, ForeignKey, Integer, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from .base import Base
@@ -32,13 +26,7 @@ class SeriesPage(Base):
         nullable=False,
         unique=True,
     )
-    # 'placed' = in the series-global run (page_number set); 'pending' = staged
-    # from a post awaiting the operator's sort (page_number NULL). (#789 P2)
-    status: Mapped[str] = mapped_column(
-        String(16), nullable=False, server_default="placed"
-    )
-    page_number: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    stated_page: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    page_number: Mapped[int] = mapped_column(Integer, nullable=False)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )

backend/app/models/series_suggestion.py

@@ -1,55 +0,0 @@
-"""SeriesSuggestion — a confirm-only "this post may continue this series" hint.
-
-The matcher (FC-6.3) scores a (post, candidate series) pair from several weighted
-signals and, above the configured threshold, records a pending suggestion. The
-operator confirms (→ the post is added as a chapter) or dismisses it; FC never
-files a post into a series on its own. status is a plain string (no Postgres
-ENUM — see the check-existing-enums lesson): pending | added | dismissed.
-"""
-
-from datetime import datetime
-
-from sqlalchemy import (
-    JSON,
-    DateTime,
-    Float,
-    ForeignKey,
-    Integer,
-    String,
-    UniqueConstraint,
-    func,
-)
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class SeriesSuggestion(Base):
-    __tablename__ = "series_suggestion"
-    __table_args__ = (
-        UniqueConstraint(
-            "post_id", "series_tag_id", name="uq_series_suggestion_post_series"
-        ),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    post_id: Mapped[int] = mapped_column(
-        ForeignKey("post.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    series_tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    score: Mapped[float] = mapped_column(Float, nullable=False)
-    signals: Mapped[dict | None] = mapped_column(JSON, nullable=True)
-    status: Mapped[str] = mapped_column(
-        String(16), nullable=False, server_default="pending", index=True
-    )
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True),
-        nullable=False,
-        server_default=func.now(),
-        onupdate=func.now(),
-    )

backend/app/models/subscribestar_failed_media.py

@@ -1,44 +0,0 @@
-"""SubscribeStarFailedMedia — per-source dead-letter ledger of SubscribeStar
-media that keeps failing to download/validate.
-
-Mirror of PatreonFailedMedia. Media that fails every walk (404'd CDN URL,
-deleted post, 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 per-media key the seen-ledger uses (CDN content hash or a
-synthesized ``<post_id>:<filename>`` key) — hence String(128). 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 SubscribeStarFailedMedia(Base):
-    __tablename__ = "subscribestar_failed_media"
-    __table_args__ = (
-        UniqueConstraint(
-            "source_id", "filehash", name="uq_subscribestar_failed_media_source_id"
-        ),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    source_id: Mapped[int] = mapped_column(
-        ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    filehash: Mapped[str] = mapped_column(String(128), nullable=False)
-    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
-    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    first_failed_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    last_failed_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/models/subscribestar_seen_media.py

@@ -1,40 +0,0 @@
-"""SubscribeStarSeenMedia — per-source ledger of SubscribeStar media already
-downloaded+processed.
-
-Mirror of PatreonSeenMedia for the SubscribeStar 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.
-
-`filehash` is a CDN content hash when the media URL carries one, else a
-synthesized ``<post_id>:<filename>`` key (SubscribeStar URLs aren't always
-content-addressed) — hence String(128) rather than 32.
-"""
-
-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 SubscribeStarSeenMedia(Base):
-    __tablename__ = "subscribestar_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_subscribestar_seen_media_source_id"
-        ),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    source_id: Mapped[int] = mapped_column(
-        ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    filehash: Mapped[str] = mapped_column(String(128), nullable=False)
-    post_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    seen_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/models/tag_alias.py

@@ -1,6 +1,6 @@
 """TagAlias — maps a model's (name, category) prediction to the operator's
-canonical tag. Resolved at suggestion-read time so the raw predictions stored
-in image_prediction stay unmolested.
+canonical tag. Resolved at suggestion-read time so raw predictions stay
+unmolested in image_record.tagger_predictions.
 """
 
 from datetime import datetime

backend/app/models/tag_allowlist.py

@@ -22,11 +22,7 @@ class TagAllowlist(Base):
     tag_id: Mapped[int] = mapped_column(
         ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True
     )
-    # Default auto-apply threshold for a newly-accepted tag. 0.90 (lowered from
-    # 0.95 on operator evidence 2026-06-07: 0.95 was too strict and skipped
-    # confident-enough applications). Per-tag value is still tunable in the
-    # allowlist table; existing rows keep whatever they were stored with.
-    min_confidence: Mapped[float] = mapped_column(Float, nullable=False, default=0.90)
+    min_confidence: Mapped[float] = mapped_column(Float, nullable=False, default=0.95)
     added_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )

backend/app/models/tag_eval_run.py

@@ -1,45 +0,0 @@
-"""TagEvalRun — persisted lifecycle of a head-vs-centroid tagging eval (#1130).
-
-Mirrors LibraryAuditRun so the result SURVIVES navigation: the run + its full
-report live in this row, and the admin card rehydrates from it on mount instead
-of holding the report in transient frontend state. State machine:
-running → ready / error. The async ml-queue task writes `report` (JSONB) when
-done; a maintenance recovery sweep flips a stalled `running` row to `error`.
-"""
-
-from datetime import datetime
-from typing import Any
-
-from sqlalchemy import DateTime, Integer, String, Text, func
-from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class TagEvalRun(Base):
-    __tablename__ = "tag_eval_run"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    # The eval parameters: {concepts: [...], curve_points: [...], neg_ratio,
-    # cv_folds, ...} — echoed back so the report is self-describing.
-    params: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
-    status: Mapped[str] = mapped_column(
-        String(16), nullable=False, default="running", index=True,
-    )
-    # running | ready | error
-    started_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now(),
-    )
-    finished_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True,
-    )
-    # The full result: per-concept metrics (head vs centroid), learning-curve
-    # points, and example image ids. Null until the task finishes.
-    report: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
-    error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    # Last time the task made progress — the recovery sweep tells a live run
-    # from a SIGKILL'd one by this (mirrors LibraryAuditRun).
-    last_progress_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True,
-    )

backend/app/models/tag_head.py

@@ -1,77 +0,0 @@
-"""TagHead — a small per-concept classifier trained on the operator's tags.
-
-Milestone #114, tagging-v2: the production form of the head the eval (#1130)
-proved. One row per concept (general or character) that has enough labelled
-positives. The head is a logistic-regression boundary over the FROZEN SigLIP
-embedding (L2-normalized), trained on the operator's positives + negatives
-(rejections + sampled unlabeled). It REPLACES the Camie prediction + per-tag
-centroid as the suggestion source — and unlike them it LEARNS: every accept /
-reject re-trains it sharper.
-
-Scoring (suggestion path, API worker, NO numpy): p = sigmoid(weights · x̂ + bias)
-where x̂ is the L2-normalized image embedding. Surface as a suggestion when
-p >= suggest_threshold; auto-apply only once auto_apply_threshold is set (the
-head "graduated" — a precision-targeted operating point was achievable). The
-thresholds come from CROSS-VALIDATED out-of-fold scores so they're honest, not
-in-sample-optimistic; the deployable weights are fit on all data.
-"""
-
-from datetime import datetime
-from typing import Any
-
-from pgvector.sqlalchemy import Vector
-from sqlalchemy import (
-    DateTime,
-    Float,
-    ForeignKey,
-    Integer,
-    String,
-    func,
-)
-from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-# Matches image_record.siglip_embedding's dimensionality — the head operates in
-# the same space. A model-version change re-embeds AND retrains (embedding_version
-# guards staleness).
-HEAD_DIM = 1152
-
-
-class TagHead(Base):
-    __tablename__ = "tag_head"
-
-    # One head per concept tag; cascade so deleting a tag retires its head.
-    tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True
-    )
-    # The embedding the head was trained against (image_record's
-    # embedder_model_version). A mismatch with the current embedder means the
-    # head is stale and must be retrained, not scored.
-    embedding_version: Mapped[str] = mapped_column(String(128), nullable=False)
-    # Logistic-regression coefficients over the L2-normalized embedding, stored
-    # as a pgvector for compactness + a future in-DB dot-product path. NOT a
-    # similarity target, just a serialized weight vector.
-    weights: Mapped[list[float]] = mapped_column(Vector(HEAD_DIM), nullable=False)
-    bias: Mapped[float] = mapped_column(Float, nullable=False)
-    # Probability cutoff for SURFACING as a suggestion (F1-best on CV scores).
-    suggest_threshold: Mapped[float] = mapped_column(Float, nullable=False)
-    # Probability cutoff for EARNED auto-apply: the operating point that holds
-    # precision >= the configured target while maximizing recall. NULL = the head
-    # hasn't graduated (can't auto-apply without a human yet).
-    auto_apply_threshold: Mapped[float | None] = mapped_column(Float, nullable=True)
-    # Training-set sizes + cross-validated quality, surfaced in the admin card so
-    # the operator can see which concepts are strong / need more tags.
-    n_pos: Mapped[int] = mapped_column(Integer, nullable=False)
-    n_neg: Mapped[int] = mapped_column(Integer, nullable=False)
-    ap: Mapped[float] = mapped_column(Float, nullable=False)
-    # 'precision' is a SQL reserved word → store as precision_cv (the
-    # cross-validated precision at the suggest operating point).
-    precision_cv: Mapped[float] = mapped_column(Float, nullable=False)
-    recall: Mapped[float] = mapped_column(Float, nullable=False)
-    trained_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )
-    # Extra detail (auto-apply operating point, F1, etc.) — non-load-bearing.
-    metrics: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)

backend/app/models/tag_positive_confirmation.py

@@ -1,28 +0,0 @@
-"""TagPositiveConfirmation — operator affirmed an applied tag is correct.
-
-The mirror of TagSuggestionRejection (#1130). When the operator "keeps" a
-positive the head doubts (low-scoring), record it so the eval's doubts list
-stops resurfacing the same confirmed-correct images every run. Does not change
-training (it's already a positive) — purely a "I've reviewed this" marker.
-"""
-
-from datetime import datetime
-
-from sqlalchemy import DateTime, ForeignKey, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class TagPositiveConfirmation(Base):
-    __tablename__ = "tag_positive_confirmation"
-
-    image_record_id: Mapped[int] = mapped_column(
-        ForeignKey("image_record.id", ondelete="CASCADE"), primary_key=True
-    )
-    tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True, index=True
-    )
-    confirmed_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/services/archive_extractor.py

@@ -16,45 +16,9 @@ log = logging.getLogger(__name__)
 
 ARCHIVE_EXTS = {".zip", ".cbz", ".rar", ".7z"}
 
-# Magic-byte signatures, so an archive with a mangled / extension-less filename
-# is still recognised. Patreon attachment download URLs sanitize to names like
-# `01_https___www.patreon.com_media-u_v3_131083093`, whose `Path.suffix` is junk
-# (`.com_media-u_v3_131083093`), never `.zip` — an extension-only gate filed
-# those as opaque PostAttachments and NEVER extracted them (operator-flagged
-# 2026-06-06). Detection is by extension first (cheap), then header sniff.
-_RAR_MAGIC = b"Rar!\x1a\x07"
-_7Z_MAGIC = b"7z\xbc\xaf\x27\x1c"
-
-
-def detect_archive_format(path: Path) -> str | None:
-    """Return "zip" | "rar" | "7z" for an archive, else None.
-
-    Trusts a known extension first, then falls back to magic-byte sniffing so a
-    mis-named or extension-less archive is still handled. (zip covers .cbz too.)
-    """
-    ext = Path(path).suffix.lower()
-    if ext in (".zip", ".cbz"):
-        return "zip"
-    if ext == ".rar":
-        return "rar"
-    if ext == ".7z":
-        return "7z"
-    try:
-        if zipfile.is_zipfile(path):
-            return "zip"
-        with open(path, "rb") as fh:
-            head = fh.read(8)
-    except OSError:
-        return None
-    if head.startswith(_RAR_MAGIC):
-        return "rar"
-    if head.startswith(_7Z_MAGIC):
-        return "7z"
-    return None
-
 
 def is_archive(path: Path) -> bool:
-    return detect_archive_format(path) is not None
+    return Path(path).suffix.lower() in ARCHIVE_EXTS
 
 
 @contextmanager
@@ -68,16 +32,16 @@ def extract_archive(path: Path):
     members: list[tuple[str, Path]] = []
     try:
         try:
-            fmt = detect_archive_format(path)
-            if fmt == "zip":
+            ext = Path(path).suffix.lower()
+            if ext in (".zip", ".cbz"):
                 with zipfile.ZipFile(path) as zf:
                     zf.extractall(base)
-            elif fmt == "rar":
+            elif ext == ".rar":
                 import rarfile
 
                 with rarfile.RarFile(path) as rf:
                     rf.extractall(base)
-            elif fmt == "7z":
+            elif ext == ".7z":
                 import py7zr
 
                 with py7zr.SevenZipFile(path, "r") as zf:

backend/app/services/artist_service.py

@@ -10,6 +10,7 @@ from dataclasses import dataclass
 
 from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import (
@@ -23,9 +24,7 @@ from ..models import (
 )
 from ..models.tag import image_tag
 from ..utils.slug import slugify
-from .db_helpers import get_or_create
-from .gallery_service import thumbnail_url
-from .pagination import decode_cursor, encode_cursor
+from .gallery_service import decode_cursor, encode_cursor, thumbnail_url
 
 
 @dataclass(frozen=True)
@@ -251,10 +250,12 @@ class ArtistService:
         )
 
     async def find_or_create(self, name: str) -> tuple[Artist, bool]:
-        """Return (artist, created). Slug-keyed; idempotent under races via the
-        shared race-safe db_helpers.get_or_create (savepoint + IntegrityError
-        recovery). A new artist also seeds an ArtistVisit so the directory's
-        `+N new` badge starts at 0.
+        """Return (artist, created). Slug-keyed; idempotent under races.
+
+        Audit 2026-06-02: switched from session.rollback() to a
+        begin_nested savepoint + IntegrityError recovery so a lost
+        race doesn't unwind the calling request's surrounding work.
+        Mirrors importer._get_or_create.
         """
         cleaned = (name or "").strip()
         if not cleaned:
@@ -262,8 +263,12 @@ class ArtistService:
         slug = slugify(cleaned)
 
         select_existing = select(Artist).where(Artist.slug == slug)
+        existing = (await self.session.execute(select_existing)).scalar_one_or_none()
+        if existing is not None:
+            return existing, False
 
-        async def _create() -> Artist:
+        sp = await self.session.begin_nested()
+        try:
             artist = Artist(name=cleaned, slug=slug)
             self.session.add(artist)
             await self.session.flush()
@@ -274,14 +279,13 @@ class ArtistService:
             # count every image imported in the same session.
             self.session.add(ArtistVisit(artist_id=artist.id))
             await self.session.flush()
-            return artist
-
-        artist, created = await get_or_create(
-            self.session, select_existing, _create
-        )
-        if created:
-            await self.session.commit()
-        return artist, created
+            await sp.commit()
+        except IntegrityError:
+            await sp.rollback()
+            existing = (await self.session.execute(select_existing)).scalar_one()
+            return existing, False
+        await self.session.commit()
+        return artist, True
 
     async def autocomplete(self, prefix: str, limit: int = 20) -> list[Artist]:
         cleaned = (prefix or "").strip()

backend/app/services/backup_service.py

@@ -15,55 +15,17 @@ lifecycle + soft/hard time limits + retention bookkeeping.
 from __future__ import annotations
 
 import json
-import os
-import shutil
 import subprocess
-import tempfile
 from datetime import UTC, datetime
 from pathlib import Path
 
 _BACKUPS_DIRNAME = "_backups"
 
-# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The Celery
-# soft limit signals the Python process; subprocess.Popen in a blocking syscall
-# ignores that signal, so these bound the worst case directly. Each sits just
-# UNDER its task's Celery soft_time_limit so the bounded-kill (_run_bounded) is
-# the primary guard and fires cleanly before Celery's soft/hard limits — which
-# matters because an NFS D-state hang defeats even Celery's SIGKILL (the failure
-# that wedged the maintenance lane for hours, #739).
-#   backup_db_task:     soft=1800s / hard=2100s  → 1700s
-#   backup_images_task: soft=21600s / hard=23400s → 21000s
-_DB_SUBPROCESS_TIMEOUT_S = 1700            # ~28 min, under the 30-min DB soft limit
-_IMAGES_SUBPROCESS_TIMEOUT_S = 21000       # ~5.8 hr, under the 6-hr images soft limit
-# Grace after SIGKILL to reap the child. If it can't be reaped in this window
-# (an uninterruptible NFS D-state — the failure mode that wedged the
-# concurrency-1 maintenance lane for hours, operator-flagged 2026-06-07), we
-# STOP waiting and fail fast, freeing the worker slot. The orphan is reaped by
-# the OS once its blocking syscall clears.
-_KILL_REAP_GRACE_S = 10
-
-
-def _run_bounded(cmd: list[str], timeout: int) -> None:
-    """subprocess.run(check=True, timeout) whose reaper can't itself hang.
-
-    subprocess.run's timeout path SIGKILLs the child then blocks in wait() to
-    reap it — but a process stuck in uninterruptible I/O (NFS) can't be reaped,
-    so wait() blocks for hours. Here we bound the post-kill reap and re-raise
-    TimeoutExpired regardless, so the caller fails fast instead of wedging."""
-    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    try:
-        out, err = proc.communicate(timeout=timeout)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        try:
-            proc.communicate(timeout=_KILL_REAP_GRACE_S)
-        except subprocess.TimeoutExpired:
-            pass  # unkillable (D-state) — abandon the reap, fail fast
-        raise
-    if proc.returncode != 0:
-        raise subprocess.CalledProcessError(
-            proc.returncode, cmd, output=out, stderr=err
-        )
+# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The
+# Celery soft limit signals the Python process; subprocess.Popen in a
+# blocking syscall ignores that signal. These bound the worst case.
+_DB_SUBPROCESS_TIMEOUT_S = 12 * 60        # 12 min (Celery soft is 10 min)
+_IMAGES_SUBPROCESS_TIMEOUT_S = 7 * 60 * 60  # 7 hr (Celery soft is 6 hr)
 
 
 def _libpq_url(sa_url: str) -> str:
@@ -121,29 +83,15 @@ def backup_db(
     to persist into BackupRun. Raises on subprocess failure."""
     ts = _now_ts()
     out_dir = _backups_dir(images_root)
-    # Custom format (-Fc): compressed (much smaller on NFS) and restored with
-    # pg_restore. The .dump extension marks it as non-SQL. The BackupRun field
-    # is still named sql_path — it's just "the db artifact path".
-    sql_path = out_dir / f"fc_db_{ts}.dump"
-    # Dump to LOCAL disk first, then move the finished file to the (NFS) backups
-    # dir. pg_dump's long phase is then a DB-socket wait + local writes — both
-    # killable — instead of an NFS write that can hang uninterruptibly. Only the
-    # final move touches NFS, and it's a bounded single-file step.
-    fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".dump")
-    os.close(fd)
-    tmp_path = Path(tmp_name)
-    try:
-        _run_bounded(
-            [
-                "pg_dump", "--no-owner", "--no-acl", "-Fc",
-                "-f", str(tmp_path), _libpq_url(db_url),
-            ],
-            _DB_SUBPROCESS_TIMEOUT_S,
-        )
-        shutil.move(str(tmp_path), str(sql_path))
-    finally:
-        if tmp_path.exists():
-            tmp_path.unlink(missing_ok=True)
+    sql_path = out_dir / f"fc_db_{ts}.sql"
+    subprocess.run(
+        [
+            "pg_dump", "--no-owner", "--no-acl",
+            "-f", str(sql_path), _libpq_url(db_url),
+        ],
+        capture_output=True, check=True,
+        timeout=_DB_SUBPROCESS_TIMEOUT_S,
+    )
     manifest_path = _write_manifest(
         out_dir, kind="db", ts=ts, tag=tag, triggered_by=triggered_by,
         artifact_path=sql_path,
@@ -166,17 +114,15 @@ def backup_images(
     ts = _now_ts()
     out_dir = _backups_dir(images_root)
     tar_path = out_dir / f"fc_images_{ts}.tar.zst"
-    # No local-temp here (the archive is hundreds of GB — it can't stage in
-    # /tmp), but bounded-kill still applies so a tar wedged on NFS fails fast
-    # rather than holding the lane for hours.
-    _run_bounded(
+    subprocess.run(
         [
             "tar", "--zstd", "-cf", str(tar_path),
             "-C", str(images_root.parent), images_root.name,
             f"--exclude={images_root.name}/_backups",
             f"--exclude={images_root.name}/_quarantine",
         ],
-        _IMAGES_SUBPROCESS_TIMEOUT_S,
+        capture_output=True, check=True,
+        timeout=_IMAGES_SUBPROCESS_TIMEOUT_S,
     )
     manifest_path = _write_manifest(
         out_dir, kind="images", ts=ts, tag=tag, triggered_by=triggered_by,
@@ -193,8 +139,8 @@ def backup_images(
 
 
 def restore_db(*, db_url: str, sql_path: Path) -> None:
-    """Wipe public schema, then load from the custom-format dump. Raises on
-    subprocess failure; partial-restore state is the caller's concern."""
+    """Wipe public schema, then load from .sql. Raises on subprocess
+    failure; partial-restore state is the caller's concern."""
     libpq = _libpq_url(db_url)
     subprocess.run(
         [
@@ -203,9 +149,8 @@ def restore_db(*, db_url: str, sql_path: Path) -> None:
         ],
         capture_output=True, check=True, timeout=120,
     )
-    # Custom-format (-Fc) dumps are restored with pg_restore, not psql.
     subprocess.run(
-        ["pg_restore", "--no-owner", "--no-acl", "-d", libpq, str(sql_path)],
+        ["psql", libpq, "-f", str(sql_path)],
         capture_output=True, check=True,
         timeout=_DB_SUBPROCESS_TIMEOUT_S,
     )

backend/app/services/db_helpers.py

@@ -1,52 +0,0 @@
-"""Shared DB-access helpers for the async services.
-
-`get_or_create` centralizes the race-safe find-or-create dance — SELECT, then on
-a miss a savepoint INSERT that recovers (NOT a full rollback) when a concurrent
-worker inserted the same row first. It was hand-rolled identically in
-ArtistService, TagService and ExtensionService; divergent copies are exactly how
-the duplicate-row / race bugs in [[reference_scalar_one_or_none_duplicates]] crept
-in, so it lives in one place now (DRY pattern sweep 2026-06-09).
-
-Note: this is the ASYNC sibling of `Importer._get_or_create` (sync, used by the
-filesystem-import path). The two can't share an implementation across the
-sync/async boundary; the importer one stays as the lone sync consumer.
-"""
-
-from __future__ import annotations
-
-from collections.abc import Awaitable, Callable
-
-from sqlalchemy import Select
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.ext.asyncio import AsyncSession
-
-
-async def get_or_create[T](
-    session: AsyncSession,
-    select_stmt: Select,
-    factory: Callable[[], Awaitable[T]],
-) -> tuple[T, bool]:
-    """Race-safe find-or-create. Returns ``(row, created)``.
-
-    Run ``select_stmt`` (scalar_one_or_none); if a row exists, return it with
-    ``created=False``. Otherwise open a SAVEPOINT and ``await factory()`` — which
-    must add its row(s), flush, and return the primary row. On ``IntegrityError``
-    (a concurrent worker inserted the same row first) roll back the SAVEPOINT —
-    NOT the outer transaction, which would lose the caller's surrounding work —
-    and re-run ``select_stmt`` (scalar_one) to return the row the other worker
-    created. The caller owns the outer commit.
-
-    A UNIQUE/partial-unique constraint matching ``select_stmt``'s predicate is
-    required for the recovery to trip; without it a duplicate slips through.
-    """
-    existing = (await session.execute(select_stmt)).scalar_one_or_none()
-    if existing is not None:
-        return existing, False
-    sp = await session.begin_nested()
-    try:
-        row = await factory()
-        await sp.commit()
-        return row, True
-    except IntegrityError:
-        await sp.rollback()
-        return (await session.execute(select_stmt)).scalar_one(), False

backend/app/services/download_backends.py

@@ -1,271 +0,0 @@
-"""Platform → download-backend dispatch (one place that knows which platforms
-are served by the native FC ingester vs. the gallery-dl subprocess).
-
-gallery-dl wasn't built to be driven by an automated scheduler — no native
-checkpoint/resume, no structured logs, per-file HEADs that dominate wall-clock.
-The native ingester (services/patreon_ingester.py, plan #697) replaces it for
-Patreon and is the path we grow as more platforms migrate. To keep that
-migration DRY, every caller that has to behave differently per backend —
-download routing, the credential-verify probe, cursor handling — asks THIS
-module instead of testing ``platform == "patreon"`` inline. When a platform gets
-a native ingester, it moves into ``NATIVE_INGESTER_PLATFORMS`` here and both the
-download path and verify switch over together.
-
-The backend surfaces share a UNIFORM signature so a caller invokes the same
-function regardless of platform:
-  - verify_credential(...)  → (ok: bool|None, message: str)
-  - (download stays in download_service for now; uses_native_ingester() is the
-    shared predicate it routes on, so the decision lives here too.)
-"""
-
-from __future__ import annotations
-
-import asyncio
-from pathlib import Path
-
-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 .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"})
-
-# Mirrors patreon_resolver._CAMPAIGNS_URL — surfaced in resolution-failure
-# messages so the operator sees the exact lookup endpoint that was hit.
-_CAMPAIGNS_API = "https://www.patreon.com/api/campaigns"
-
-
-def _native_ingester_cls(platform: str):
-    """The native ingester class for `platform` (uniform constructor signature).
-    A call-time lookup (not a module-level dict captured at import) so tests can
-    monkeypatch db_mod.PatreonIngester / SubscribeStarIngester and have the
-    dispatch pick up the replacement."""
-    return {
-        "patreon": PatreonIngester,
-        "subscribestar": SubscribeStarIngester,
-    }[platform]
-
-
-def uses_native_ingester(platform: str) -> bool:
-    """True when `platform` is served by the native ingester (not gallery-dl).
-    The single predicate the download path and verify both route on."""
-    return platform in NATIVE_INGESTER_PLATFORMS
-
-
-async def run_download(
-    *,
-    ctx: dict,
-    source_config,
-    skip_value: bool | str,
-    mode: str | None,
-    gdl,
-    sync_session_factory,
-) -> tuple[DownloadResult, str | None]:
-    """Uniform download across backends — the download counterpart to
-    `verify_source_credential`, so this module is the ONE place that knows how
-    each platform both downloads AND verifies (the seam that makes adding a
-    platform a bounded job).
-
-    Returns `(DownloadResult, resolved_campaign_id)`; `resolved_campaign_id` is
-    non-None only when a native vanity lookup ran this call (so phase 3 caches
-    it). Native platforms route through their ingester in `mode`
-    (tick/backfill/recovery); gallery-dl platforms run the subprocess. The caller
-    (download_service) prepares `source_config`/`skip_value`/`mode` from the
-    backfill state machine and owns phase 3.
-    """
-    platform = ctx["platform"]
-    if uses_native_ingester(platform):
-        return await _run_native_ingester(
-            ctx, source_config, mode, gdl, sync_session_factory
-        )
-    result = await gdl.download(
-        url=ctx["url"],
-        artist_slug=ctx["artist_slug"],
-        platform=platform,
-        source_config=source_config,
-        cookies_path=ctx["cookies_path"],
-        auth_token=ctx["auth_token"],
-        skip_value=skip_value,
-    )
-    return result, None
-
-
-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
-    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
-    return await resolve_campaign_id_for_source(url, cookies_path, overrides)
-
-
-async def _run_native_ingester(
-    ctx: dict, source_config, mode: str | None, gdl, sync_session_factory,
-) -> tuple[DownloadResult, str | None]:
-    """Run the native ingester for a native platform in a worker thread (sync
-    requests/subprocess). Patreon resolves a campaign id from the vanity URL;
-    SubscribeStar's feed id is the creator URL itself. A campaign id we cannot
-    resolve is a loud NOT_FOUND — never a silent empty success.
-
-    `resolved_campaign_id` is non-None only when a lookup ran this call, so phase
-    3 caches it the way the old gallery-dl retry did.
-    """
-    platform = ctx["platform"]
-    overrides = ctx["config_overrides"] or {}
-    campaign_id, resolved_campaign_id = await _resolve_native_campaign_id(
-        platform, ctx["url"], ctx["cookies_path"], overrides
-    )
-    if not campaign_id:
-        # Only reachable for Patreon (SubscribeStar's campaign id is the URL).
-        url = ctx["url"]
-        vanity = extract_vanity(url)
-        return (
-            DownloadResult(
-                success=False,
-                url=url,
-                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?)"
-                ),
-            ),
-            None,
-        )
-
-    # Honor the operator's existing rate-limit knobs on the native path (plan
-    # #703): the global download_rate_limit_seconds (gallery-dl's `rate_limit`,
-    # here on the gdl service) paces media downloads; page fetches use the
-    # per-source sleep_request override, else `max(0.5, rate_limit/4)` — the same
-    # API-pacing default gallery-dl applied as its `sleep-request`.
-    rate_limit = gdl._rate_limit
-    request_sleep = (
-        source_config.sleep_request
-        if source_config.sleep_request is not None
-        else max(0.5, rate_limit / 4)
-    )
-    ingester = _native_ingester_cls(platform)(
-        images_root=gdl.images_root,
-        cookies_path=ctx["cookies_path"],
-        session_factory=sync_session_factory,
-        validate=gdl._validate_files,
-        rate_limit=rate_limit,
-        request_sleep=request_sleep,
-    )
-    loop = asyncio.get_running_loop()
-    dl_result = await loop.run_in_executor(
-        None,
-        lambda: ingester.run(
-            source_id=ctx["source_id"],
-            campaign_id=campaign_id,
-            artist_slug=ctx["artist_slug"],
-            url=ctx["url"],
-            mode=mode,
-            resume_cursor=source_config.resume_cursor,
-            time_budget_seconds=source_config.timeout,
-            posts_base=int(overrides.get("_backfill_posts", 0)),
-            # plan #709: live progress writes to this running event mid-walk.
-            event_id=ctx.get("event_id"),
-        ),
-    )
-    return dl_result, resolved_campaign_id
-
-
-async def preview_source(
-    *,
-    platform: str,
-    url: str,
-    source_id: int,
-    config_overrides: dict | None,
-    cookies_path: str | None,
-    images_root: Path,
-    sync_session_factory,
-    page_limit: int = 3,
-) -> dict:
-    """Dry-run preview for a native platform (plan #708 B4): resolve the campaign
-    id, then walk a few pages counting media not already seen/dead — no download.
-
-    Returns the preview dict (total_new / posts_scanned / pages_scanned /
-    has_more / sample), or `{"error": msg}` on a resolve / auth / drift failure.
-    Native-only — the caller gates on `uses_native_ingester`.
-    """
-    import asyncio
-
-    campaign_id, _ = await _resolve_native_campaign_id(
-        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?)."
-            )
-        }
-    ingester = _native_ingester_cls(platform)(
-        images_root=images_root,
-        cookies_path=cookies_path,
-        session_factory=sync_session_factory,
-    )
-    loop = asyncio.get_running_loop()
-    try:
-        result = await loop.run_in_executor(
-            None,
-            lambda: ingester.preview(source_id, campaign_id, page_limit=page_limit),
-        )
-    except NativeIngestError as exc:
-        return {"error": f"Couldn't preview: {exc}"}
-    return result
-
-
-async def verify_source_credential(
-    *,
-    platform: str,
-    url: str,
-    artist_slug: str,
-    config_overrides: dict | None,
-    cookies_path: str | None,
-    auth_token: str | None,
-    images_root: Path,
-) -> tuple[bool | None, str]:
-    """Uniform credential probe across backends. Returns `(ok, message)`:
-    True = authenticated, False = rejected, None = inconclusive (drift /
-    network / nothing to test). Callers don't branch on platform — they call
-    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.
-        if platform == "subscribestar":
-            from .subscribestar_ingester import verify_subscribestar_credential
-
-            return await verify_subscribestar_credential(url, cookies_path, config_overrides)
-        from .patreon_ingester import verify_patreon_credential
-
-        return await verify_patreon_credential(url, cookies_path, config_overrides)
-
-    # gallery-dl platforms: --simulate one item; the extractor errors before it
-    # can list if auth is bad.
-    from .gallery_dl import GalleryDLService, SourceConfig
-
-    gdl = GalleryDLService(images_root=images_root)
-    return await gdl.verify(
-        url=url,
-        artist_slug=artist_slug,
-        platform=platform,
-        source_config=SourceConfig.from_dict(config_overrides or {}),
-        cookies_path=cookies_path,
-        auth_token=auth_token,
-    )

backend/app/services/download_service.py

@@ -13,6 +13,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import re
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -24,25 +25,44 @@ from sqlalchemy.orm import joinedload
 
 from ..models import Artist, DownloadEvent, Source
 from .credential_service import CredentialService
-from .download_backends import run_download, uses_native_ingester
 from .gallery_dl import (
-    BACKFILL_CHUNK_SECONDS,
     BACKFILL_SKIP_VALUE,
+    BACKFILL_TIMEOUT_SECONDS,
     TICK_SKIP_VALUE,
-    DownloadResult,
     ErrorType,
     GalleryDLService,
     SourceConfig,
-    extract_errors_warnings,
-    truncate_log,
 )
 from .importer import Importer
+from .patreon_resolver import resolve_campaign_id
 from .platforms import auth_type_for
 from .scheduler_service import set_platform_cooldown
 
 log = logging.getLogger(__name__)
 
 
+_PATREON_VANITY_RE = re.compile(
+    r"^https?://(?:www\.)?patreon\.com/(?:c/)?(?!id:)([^/?#]+)",
+    re.IGNORECASE,
+)
+_CAMPAIGN_ID_FAILURE_PATTERN = "failed to extract campaign id"
+
+
+def _extract_patreon_vanity(url: str) -> str | None:
+    m = _PATREON_VANITY_RE.match(url)
+    return m.group(1) if m else None
+
+
+def _looks_like_campaign_id_failure(stdout: str, stderr: str) -> bool:
+    return _CAMPAIGN_ID_FAILURE_PATTERN in f"{stdout}\n{stderr}".lower()
+
+
+def _effective_url(platform: str, source_url: str, overrides: dict) -> str:
+    if platform == "patreon" and overrides.get("patreon_campaign_id"):
+        return f"https://www.patreon.com/id:{overrides['patreon_campaign_id']}"
+    return source_url
+
+
 class DownloadService:
     """Async orchestrator. The Celery task runs `asyncio.run(svc.download_source(N))`.
 
@@ -57,18 +77,12 @@ class DownloadService:
         gdl: GalleryDLService,
         importer: Importer,
         cred_service: CredentialService,
-        sync_session_factory=None,
     ):
         self.async_session = async_session
         self.sync_session = sync_session
         self.gdl = gdl
         self.importer = importer
         self.cred_service = cred_service
-        # Sync sessionmaker the native Patreon ingester opens SHORT-LIVED
-        # sessions from for its seen-ledger reads/writes (never one held across
-        # the multi-minute walk — see PatreonIngester). Only the patreon branch
-        # of phase 2 uses it; gallery-dl sources leave it None.
-        self.sync_session_factory = sync_session_factory
 
     async def download_source(self, source_id: int) -> int:
         """Returns DownloadEvent.id. Idempotent: in-flight events are returned as-is."""
@@ -93,91 +107,63 @@ class DownloadService:
         self.sync_session.close()
 
         source_config = SourceConfig.from_dict(ctx["config_overrides"] or {})
-        # Backfill mode (plan #693): the source's `_backfill_state == "running"`
-        # selects a time-boxed deep-walk chunk — skip: True (walk full history)
-        # + the BACKFILL_CHUNK_SECONDS budget, resuming from the cursor
-        # checkpoint (plan #689). State is the single source of truth; it stays
-        # "running" across ticks until the walk reaches the bottom (phase 3
-        # flips it to "complete"). Otherwise tick mode: exit gallery-dl after
-        # 20 contiguous archived items (skip: "exit:20" + the default 870s).
-        # Operator drives this via POST /api/sources/{id}/backfill {action}.
-        overrides = ctx["config_overrides"] or {}
-        in_backfill = overrides.get("_backfill_state") == "running"
-        # Recovery (plan #697) reuses the entire #693 backfill state machine —
-        # cursor checkpoint, time-boxed chunks, complete/stall lifecycle — and
-        # differs only in bypassing the tier-1 seen-ledger so dropped-and-deleted
-        # near-dups get re-fetched and re-evaluated under the CURRENT pHash
-        # threshold (tier-2 disk still spares files we kept). The
-        # `_backfill_bypass_seen` flag rides alongside the running backfill state;
-        # download mode is "recovery" when both are set.
-        bypass_seen = bool(overrides.get("_backfill_bypass_seen"))
-        # #830 recapture: re-grab post bodies/links + localize on-disk inline
-        # images, WITHOUT re-downloading media. Rides alongside the backfill
-        # state like _backfill_bypass_seen; mutually exclusive with it.
-        recapture = bool(overrides.get("_backfill_recapture"))
-        if in_backfill:
+        # alembic 0031 / plan #544: derive skip_value + timeout from the
+        # source's backfill_runs_remaining counter. When > 0, walk the full
+        # post history (skip: True + 1170s); when 0, exit gallery-dl after
+        # 20 contiguous archived items (skip: "exit:20" + the default
+        # 870s). Operator sets backfill via POST /api/sources/{id}/backfill.
+        backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0
+        if backfill_remaining > 0:
             skip_value: bool | str = BACKFILL_SKIP_VALUE
-            source_config.timeout = BACKFILL_CHUNK_SECONDS
-            pending_cursor = overrides.get("_backfill_cursor")
-            if uses_native_ingester(ctx["platform"]) and pending_cursor:
-                source_config.resume_cursor = pending_cursor
+            source_config.timeout = BACKFILL_TIMEOUT_SECONDS
         else:
             skip_value = TICK_SKIP_VALUE
 
-        # Phase 2 dispatch is uniform across backends (download_backends.
-        # run_download — the download counterpart to verify_source_credential):
-        # native platforms run their ingester in `mode` (zero per-file HEADs,
-        # native cursor/resume, loud drift detection); gallery-dl platforms run
-        # the subprocess. Either returns a DownloadResult-shaped object so phase 3
-        # is untouched. `mode` is None for gallery-dl (run_download ignores it).
-        mode: str | None = None
-        if uses_native_ingester(ctx["platform"]):
-            if in_backfill and bypass_seen:
-                mode = "recovery"
-            elif in_backfill and recapture:
-                mode = "recapture"
-            elif in_backfill:
-                mode = "backfill"
-            else:
-                mode = "tick"
-        dl_result, resolved_campaign_id = await self._run_download(
-            ctx=ctx, source_config=source_config, skip_value=skip_value, mode=mode,
+        effective_url = _effective_url(
+            ctx["platform"], ctx["url"], ctx["config_overrides"] or {}
         )
 
-        # A backfill chunk that hit its time-box but made forward progress is
-        # NORMAL, not a failure — reclassify TIMEOUT → PARTIAL so it reads as
-        # "ok/progress" (PARTIAL maps to status "ok"), not a red error, and the
-        # next chunk just resumes from the new cursor (plan #693). A chunk that
-        # timed out with NO progress stays TIMEOUT and feeds phase 3's
-        # stall-guard. Leave RATE_LIMITED alone so the platform-cooldown fires.
-        if in_backfill and dl_result.error_type == ErrorType.TIMEOUT:
-            new_cursor = dl_result.cursor  # plan #704: structured, not scraped
-            advanced = bool(
-                (new_cursor and new_cursor != overrides.get("_backfill_cursor"))
-                or dl_result.files_downloaded > 0
-            )
-            if advanced:
-                dl_result.error_type = ErrorType.PARTIAL
-                dl_result.error_message = (
-                    f"Backfill chunk: {dl_result.files_downloaded} file(s) — continuing"
+        dl_result = await self.gdl.download(
+            url=effective_url,
+            artist_slug=ctx["artist_slug"],
+            platform=ctx["platform"],
+            source_config=source_config,
+            cookies_path=ctx["cookies_path"],
+            auth_token=ctx["auth_token"],
+            skip_value=skip_value,
+        )
+
+        resolved_campaign_id: str | None = None
+        if (
+            ctx["platform"] == "patreon"
+            and not dl_result.success
+            and "patreon_campaign_id" not in (ctx["config_overrides"] or {})
+            and _looks_like_campaign_id_failure(dl_result.stdout, dl_result.stderr)
+        ):
+            vanity = _extract_patreon_vanity(ctx["url"])
+            if vanity:
+                log.info(
+                    "Attempting campaign-ID resolution for %s (%s)",
+                    ctx["artist_slug"], vanity,
                 )
+                resolved_campaign_id = await resolve_campaign_id(
+                    vanity, ctx["cookies_path"]
+                )
+                if resolved_campaign_id:
+                    dl_result = await self.gdl.download(
+                        url=f"https://www.patreon.com/id:{resolved_campaign_id}",
+                        artist_slug=ctx["artist_slug"],
+                        platform=ctx["platform"],
+                        source_config=source_config,
+                        cookies_path=ctx["cookies_path"],
+                        auth_token=ctx["auth_token"],
+                        skip_value=skip_value,
+                    )
 
         return await self._phase3_persist(
             ctx["event_id"], ctx, dl_result, resolved_campaign_id,
         )
 
-    async def _run_download(
-        self, *, ctx: dict, source_config, skip_value, mode: str | None,
-    ) -> tuple[DownloadResult, str | None]:
-        """Phase-2 dispatch → `download_backends.run_download`, passing this
-        service's gdl + sync sessionmaker. Kept as a thin instance method so tests
-        can stub the whole phase-2 dispatch on the service, and so the per-backend
-        construction lives in download_backends (the backend registry)."""
-        return await run_download(
-            ctx=ctx, source_config=source_config, skip_value=skip_value, mode=mode,
-            gdl=self.gdl, sync_session_factory=self.sync_session_factory,
-        )
-
     async def _phase1_setup(self, source_id: int) -> dict[str, Any]:
         source = (await self.async_session.execute(
             select(Source).options(joinedload(Source.artist)).where(Source.id == source_id)
@@ -274,20 +260,7 @@ class DownloadService:
             artist = self.sync_session.get(Artist, ctx["artist_id"])
         source_row = self.sync_session.get(Source, ctx["source_id"])
 
-        # Post-first (#856): on the native ingester the post-record is the sole
-        # writer of the post body/links, so the per-media import must NOT apply post
-        # fields. gallery-dl platforms keep writing them via the sidecar. The
-        # Importer is per-task, so this per-instance flag is safe. uses_native_ingester
-        # is the future-proof seam — a platform migrating onto the native core
-        # flips to post-first automatically (milestone #67, step 3).
-        self.importer.post_first = uses_native_ingester(ctx["platform"])
-
         import_summary = {"attached": 0, "skipped": 0, "errors": 0}
-        # Archives detected but captured WITHOUT extracting any image (probe
-        # rejected / corrupt / missing extractor backend). Surfaced on the event
-        # so a post showing "no images" beside a zip is diagnosable (plan
-        # follow-up 2026-06-06 — the recurring archive-association report).
-        unextracted_archives: list[dict] = []
         bytes_downloaded = 0
 
         loop = asyncio.get_running_loop()
@@ -341,17 +314,6 @@ class DownloadService:
                 # mirror duplicate_hash cleanup so we don't keep two copies.
                 # Operator-flagged 2026-06-02 (Lustria OST zip).
                 import_summary["attached"] += 1
-                # An archive captured WITHOUT extracting any image carries a
-                # reason on result.error — record it so the event explains the
-                # "no images beside a zip" symptom instead of staying silent.
-                if result.error:
-                    unextracted_archives.append(
-                        {"file": path.name, "reason": result.error}
-                    )
-                    log.warning(
-                        "archive captured unextracted (%s): %s",
-                        path.name, result.error,
-                    )
                 try:
                     bytes_downloaded += path.stat().st_size  # noqa: ASYNC240
                 except OSError:
@@ -394,68 +356,15 @@ class DownloadService:
             else:
                 import_summary["errors"] += 1
 
-        # Post-only records: media-less posts (pure-text) the native ingester
-        # captured so the artist archive is complete. Upsert each (keyed on
-        # external_post_id → updates the same Post a media import would create,
-        # never doubles). No file to clean up; the sidecar stays on disk.
-        for rec_str in getattr(dl_result, "post_record_paths", None) or []:
-            rec_path = Path(rec_str)
-            if not rec_path.exists():  # noqa: ASYNC240
-                continue
-
-            def _upsert(p=rec_path):
-                return self.importer.upsert_post_record(
-                    p, artist=artist, source=source_row,
-                )
-
-            await loop.run_in_executor(None, _upsert)
-
-        # #830 recapture: backfill source_filehash on EXISTING on-disk images so
-        # their post-body inline <img src=CDN> remaps to the local copy. A
-        # SEPARATE non-deleting channel (NOT the import list — that would unlink
-        # the duplicate file). Empty outside recapture mode.
-        relink_pairs = getattr(dl_result, "relink_source_paths", None) or []
-        relinked = 0
-        for rel_str, rel_url in relink_pairs:
-            rel_path = Path(rel_str)
-            if not rel_path.exists():  # noqa: ASYNC240
-                continue
-
-            def _relink(p=rel_path, u=rel_url):
-                return self.importer.relink_source_filehash(p, u, artist=artist)
-
-            if await loop.run_in_executor(None, _relink):
-                relinked += 1
-        if relink_pairs:
-            # recapture diagnostic: how many on-disk images got their
-            # source_filehash backfilled (inline-image localization). < total is
-            # normal — files already carrying a filehash are skipped (NULL-only).
-            log.info(
-                "recap: relinked source_filehash on %d/%d on-disk image(s)",
-                relinked, len(relink_pairs),
-            )
-
-        # Kick the off-platform file-host downloader for any links this run
-        # recorded (mega/gdrive/…). Global + idempotent (only claims pending/
-        # retryable rows); the beat sweep is the backstop. Lazy import dodges a
-        # task-module import cycle.
-        from ..tasks.external import sweep_external_links
-        sweep_external_links.delay()
-
         ev = (await self.async_session.execute(
             select(DownloadEvent).where(DownloadEvent.id == event_id)
         )).scalar_one()
 
-        # plan #704: the native ingester returns structured run_stats; only the
-        # gallery-dl path needs the regex-over-stdout reconstruction.
-        if dl_result.run_stats is not None:
-            run_stats = dict(dl_result.run_stats)
-        else:
-            run_stats = self.gdl._compute_run_stats(
-                dl_result.return_code, dl_result.stdout, dl_result.stderr
-            )
+        run_stats = self.gdl._compute_run_stats(
+            dl_result.return_code, dl_result.stdout, dl_result.stderr
+        )
         run_stats["quarantined_count"] = dl_result.files_quarantined
-        stderr_summary = extract_errors_warnings(dl_result.stderr)
+        stderr_summary = self.gdl._extract_errors_warnings(dl_result.stderr)
 
         # Plan #544: PARTIAL means the run downloaded ≥1 file but the
         # subprocess didn't finish in budget (typically wall-clock timeout
@@ -475,118 +384,50 @@ class DownloadService:
         ev.metadata_ = {
             "run_stats": run_stats,
             "error_type": dl_result.error_type.value if dl_result.error_type else None,
-            "stdout": truncate_log(dl_result.stdout) or None,
-            "stderr": truncate_log(dl_result.stderr) or None,
+            "stdout": self.gdl._truncate_log(dl_result.stdout) or None,
+            "stderr": self.gdl._truncate_log(dl_result.stderr) or None,
             "stderr_errors_warnings": stderr_summary or None,
             "duration_seconds": dl_result.duration_seconds,
             "quarantined_paths": dl_result.quarantined_paths or None,
             "import_summary": import_summary,
-            # Archives detected but captured without extracting an image — the
-            # recurring "post shows a zip but no images" report. Each entry is
-            # {file, reason}; None when every archive extracted cleanly.
-            "unextracted_archives": unextracted_archives or None,
         }
         await self._update_source_health(
             source_id=ctx["source_id"], status=status, error_message=ev.error,
             error_type=dl_result.error_type.value if dl_result.error_type else None,
-            retry_after_seconds=getattr(dl_result, "retry_after_seconds", None),
         )
-        await self._apply_backfill_lifecycle(ctx, dl_result)
+        # Plan #544: backfill lifecycle — auto-complete when a clean
+        # backfill run drained the queue (gallery-dl exited 0 + zero files
+        # downloaded means there was nothing to fetch); otherwise decrement
+        # the counter. Next tick falls back to tick mode once it hits 0.
+        #
+        # Audit 2026-06-02 gating: VALIDATION_FAILED also exits the
+        # subprocess with return_code=0 and files_downloaded=0 (every
+        # file was quarantined), which used to match the auto-complete
+        # predicate exactly — zeroing the operator's armed budget on
+        # the FIRST quarantine run instead of decrementing. Require
+        # dl_result.success + no error_type so only genuinely-empty
+        # successful runs drain the counter.
+        backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0
+        if backfill_remaining > 0:
+            src = (await self.async_session.execute(
+                select(Source).where(Source.id == ctx["source_id"])
+            )).scalar_one()
+            queue_drained = (
+                dl_result.success
+                and dl_result.error_type is None
+                and dl_result.return_code == 0
+                and dl_result.files_downloaded == 0
+            )
+            if queue_drained:
+                src.backfill_runs_remaining = 0
+            else:
+                src.backfill_runs_remaining = max(0, backfill_remaining - 1)
         await self.async_session.commit()
         return event_id
 
-    async def _apply_backfill_lifecycle(self, ctx: dict, dl_result) -> None:
-        """Backfill state machine (plan #693, building on the cursor of #689).
-
-        A backfill runs in time-boxed chunks while
-        `config_overrides["_backfill_state"] == "running"`. Each chunk:
-          - COMPLETES the walk → clean rc=0 (gallery-dl with skip:True exits 0
-            only after exhausting the newest→oldest walk; a chunk cut short by
-            its time-box returns success=False / rc<0 via TimeoutExpired). On
-            completion: state="complete", clear the cursor, return to tick mode.
-          - made PROGRESS (cursor advanced and/or files written) → stay
-            "running", checkpoint the new cursor, bump the chunk counter, and
-            spend one of the safety-cap chunks (backfill_runs_remaining). If the
-            cap is exhausted without finishing → state="stalled".
-          - made NO progress → increment the stall counter; two strikes →
-            state="stalled", clear the cursor (a wedged walk can't loop).
-
-        State is the single source of truth; the cursor lives only inside a
-        running backfill. config_overrides is reassigned (not mutated in place)
-        for SQLAlchemy JSON change detection.
-        """
-        overrides = ctx["config_overrides"] or {}
-        if overrides.get("_backfill_state") != "running":
-            return  # not backfilling — tick mode, nothing to do
-
-        old_cursor = overrides.get("_backfill_cursor")
-        cap_remaining = ctx.get("backfill_runs_remaining", 0) or 0
-
-        src = (await self.async_session.execute(
-            select(Source).where(Source.id == ctx["source_id"])
-        )).scalar_one()
-        new_overrides = dict(src.config_overrides or {})
-        chunks = int(new_overrides.get("_backfill_chunks", 0)) + 1
-        new_overrides["_backfill_chunks"] = chunks
-        # plan #704 (#5): _backfill_posts (the live progress badge) is OWNED by the
-        # ingester now — it writes a monotonic absolute mid-walk at each page
-        # boundary (ingest_core._checkpoint_posts), so the badge climbs DURING a
-        # chunk instead of jumping once per chunk here, and the re-walked resume
-        # page is no longer double-counted. new_overrides (read fresh above)
-        # carries the ingester's committed value forward untouched.
-
-        completed = (
-            dl_result.success
-            and dl_result.error_type is None
-            and dl_result.return_code == 0
-        )
-        if completed:
-            new_overrides["_backfill_state"] = "complete"
-            new_overrides.pop("_backfill_cursor", None)
-            new_overrides.pop("_backfill_cursor_stalls", None)
-            # plan #697: a recovery walk shares this lifecycle; clear its bypass
-            # flag on completion so the next routine tick honors the seen-ledger.
-            new_overrides.pop("_backfill_bypass_seen", None)
-            # #830: same for the recapture flag.
-            new_overrides.pop("_backfill_recapture", None)
-            src.config_overrides = new_overrides
-            src.backfill_runs_remaining = 0
-            return
-
-        # Did not finish. The native ingester checkpoints + resumes via cursor
-        # (carried structurally on the result, plan #704); gallery-dl platforms
-        # have no resumable cursor (every chunk re-walks from the top), so they
-        # advance only by the download archive growing.
-        new_cursor = (
-            dl_result.cursor if uses_native_ingester(ctx["platform"]) else None
-        )
-        advanced = bool(
-            (new_cursor and new_cursor != old_cursor)
-            or dl_result.files_downloaded > 0
-        )
-        if advanced:
-            if new_cursor:
-                new_overrides["_backfill_cursor"] = new_cursor
-            new_overrides.pop("_backfill_cursor_stalls", None)
-            cap_remaining = max(0, cap_remaining - 1)
-            src.backfill_runs_remaining = cap_remaining
-            if cap_remaining == 0:
-                # Safety cap hit before reaching the bottom — pause, don't loop.
-                new_overrides["_backfill_state"] = "stalled"
-        else:
-            stalls = int(new_overrides.get("_backfill_cursor_stalls", 0)) + 1
-            if stalls >= 2:
-                new_overrides["_backfill_state"] = "stalled"
-                new_overrides.pop("_backfill_cursor", None)
-                new_overrides.pop("_backfill_cursor_stalls", None)
-            else:
-                new_overrides["_backfill_cursor_stalls"] = stalls
-
-        src.config_overrides = new_overrides
-
     async def _update_source_health(
         self, *, source_id: int, status: str, error_message: str | None,
-        error_type: str | None = None, retry_after_seconds: float | None = None,
+        error_type: str | None = None,
     ) -> None:
         """FC-3d: update Source.{consecutive_failures, last_error, last_checked_at}.
 
@@ -618,17 +459,7 @@ class DownloadService:
             # by error class without opening Logs per row.
             source.error_type = error_type
             if error_type == "rate_limited":
-                # plan #708 B1: honor the server's Retry-After when the native
-                # client surfaced one (clamped to a sane [60, 3600] window so a
-                # tiny hint can't leave the platform effectively un-cooled and a
-                # huge one can't strand it for hours); else the flat default.
-                if retry_after_seconds is not None:
-                    seconds = int(min(max(retry_after_seconds, 60), 3600))
-                    await set_platform_cooldown(
-                        self.async_session, source.platform, seconds=seconds,
-                    )
-                else:
-                    await set_platform_cooldown(self.async_session, source.platform)
+                await set_platform_cooldown(self.async_session, source.platform)
         elif status == "skipped":
             source.last_error = None
         source.last_checked_at = now

backend/app/services/extension_service.py

@@ -11,12 +11,12 @@ from __future__ import annotations
 import re
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import Artist, Source
 from ..utils.slug import slugify
-from .db_helpers import get_or_create
-from .source_service import BACKFILL_MAX_CHUNKS
+from .source_service import NEW_SOURCE_BACKFILL_RUNS
 
 
 class UnknownPlatformError(Exception):
@@ -169,16 +169,24 @@ class ExtensionService:
         500 against uq_artist_slug.
         """
         slug = slugify(raw_name)
-
-        async def _create() -> Artist:
+        existing = (await self.session.execute(
+            select(Artist).where(Artist.slug == slug)
+        )).scalar_one_or_none()
+        if existing is not None:
+            return existing, False
+        sp = await self.session.begin_nested()
+        try:
             artist = Artist(name=raw_name, slug=slug, is_subscription=True)
             self.session.add(artist)
             await self.session.flush()
-            return artist
-
-        return await get_or_create(
-            self.session, select(Artist).where(Artist.slug == slug), _create
-        )
+            await sp.commit()
+            return artist, True
+        except IntegrityError:
+            await sp.rollback()
+            recovered = (await self.session.execute(
+                select(Artist).where(Artist.slug == slug)
+            )).scalar_one()
+            return recovered, False
 
     async def _find_or_create_source(
         self, *, artist_id: int, platform: str, url: str,
@@ -186,32 +194,40 @@ class ExtensionService:
         """Race-safe — same pattern as _find_or_create_artist above. The
         uq_source_artist_platform_url constraint catches the duplicate
         insert; we roll the savepoint back and re-select."""
-        select_existing = select(Source).where(
-            Source.artist_id == artist_id,
-            Source.platform == platform,
-            Source.url == url,
-        )
-
-        async def _create() -> Source:
-            # New subscription sources arm run-until-done backfill (plan #693)
-            # so the first ticks walk the full history (otherwise gallery-dl's
-            # exit:20 short-circuits before the archive is built). Mirrors
-            # SourceService.create — without it, Firefox quick-add on a creator
-            # with >20 unsynced posts would surface as "check failed" with no
-            # diagnosis. Audit 2026-06-02.
+        existing = (await self.session.execute(
+            select(Source).where(
+                Source.artist_id == artist_id,
+                Source.platform == platform,
+                Source.url == url,
+            )
+        )).scalar_one_or_none()
+        if existing is not None:
+            return existing, False
+        sp = await self.session.begin_nested()
+        try:
+            # New subscription sources arm a few backfill runs so the
+            # first ticks walk the full history (otherwise gallery-dl's
+            # exit:20 short-circuits before the archive is built).
+            # Mirrors SourceService.create — without it, Firefox quick-
+            # add on a creator with >20 unsynced posts would surface
+            # as "check failed" with no diagnosis. Audit 2026-06-02.
             src = Source(
                 artist_id=artist_id, platform=platform,
                 url=url, enabled=True,
-                config_overrides={"_backfill_state": "running"},
-                backfill_runs_remaining=BACKFILL_MAX_CHUNKS,
+                backfill_runs_remaining=NEW_SOURCE_BACKFILL_RUNS,
             )
             self.session.add(src)
             await self.session.flush()
-            return src
-
-        src, created = await get_or_create(
-            self.session, select_existing, _create
-        )
-        if created:
-            await self.session.commit()
-        return src, created
+            await sp.commit()
+        except IntegrityError:
+            await sp.rollback()
+            recovered = (await self.session.execute(
+                select(Source).where(
+                    Source.artist_id == artist_id,
+                    Source.platform == platform,
+                    Source.url == url,
+                )
+            )).scalar_one()
+            return recovered, False
+        await self.session.commit()
+        return src, True