feat(regions): image_region storage + service for the crop pipeline (#114 slice 2)

The storage backbone both crop jobs write to and read from. image_region =
normalized bbox (rx/ry/rw/rh) + kind ('face'/'figure' → CCIP character id;
'concept' → SigLIP head bag) + the crop's embedding (nullable Vector(768) CCIP /
Vector(1152) SigLIP, one per kind) + version stamps for compute-once gating. The
bbox doubles as grounded-tag provenance. Migration 0061.

RegionService.replace_regions (scoped BY KIND so the figure + concept pipelines
don't clobber each other) + get_regions — the GPU agent's results endpoint will
call the writer; the character matcher + bag scorer read. Server-side, no GPU.

Tests: replace/get round-trip, kind-scoped replacement, CCIP vector round-trip.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa

alembic/versions/0061_image_region.py

@@ -0,0 +1,57 @@
+"""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),
+        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")

backend/app/models/__init__.py

@@ -15,6 +15,7 @@ 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
@@ -60,6 +61,7 @@ __all__ = [
     "ImageRecord",
     "ImagePrediction",
     "ImageProvenance",
+    "ImageRegion",
     "Tag",
     "TagKind",
     "image_tag",

backend/app/models/image_region.py

@@ -0,0 +1,57 @@
+"""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
+    )
+    # 'face' | 'figure' (→ CCIP character id) | 'concept' (→ SigLIP head bag).
+    kind: Mapped[str] = mapped_column(String(16), nullable=False)
+    # 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/services/ml/regions.py

@@ -0,0 +1,58 @@
+"""Region read/write for the crop pipeline (#114).
+
+The GPU agent's results endpoint calls replace_regions() to store a freshly
+detected/embedded set; the character matcher + concept-bag scorer read via
+get_regions(). Replacement is scoped BY KIND so the figure pipeline and the
+concept pipeline don't clobber each other.
+"""
+
+from typing import Any
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ...models import ImageRegion
+
+
+class RegionService:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def get_regions(
+        self, image_id: int, kinds: list[str] | None = None
+    ) -> list[ImageRegion]:
+        stmt = select(ImageRegion).where(ImageRegion.image_record_id == image_id)
+        if kinds:
+            stmt = stmt.where(ImageRegion.kind.in_(kinds))
+        return list(
+            (await self.session.execute(stmt.order_by(ImageRegion.id))).scalars()
+        )
+
+    async def replace_regions(
+        self, image_id: int, kinds: list[str], regions: list[dict[str, Any]]
+    ) -> int:
+        """Replace this image's regions OF THE GIVEN KINDS with `regions` (a
+        re-detect/re-propose supersedes the prior set without touching other
+        kinds). Each region dict: {kind, bbox:(x,y,w,h), score?, detector_version?,
+        crop_version?, embedding_version?, ccip_embedding?, siglip_embedding?}.
+        Returns the number inserted."""
+        await self.session.execute(
+            delete(ImageRegion)
+            .where(ImageRegion.image_record_id == image_id)
+            .where(ImageRegion.kind.in_(kinds))
+        )
+        n = 0
+        for r in regions:
+            rx, ry, rw, rh = r["bbox"]
+            self.session.add(ImageRegion(
+                image_record_id=image_id, kind=r["kind"],
+                rx=rx, ry=ry, rw=rw, rh=rh,
+                score=r.get("score"),
+                detector_version=r.get("detector_version"),
+                crop_version=r.get("crop_version"),
+                embedding_version=r.get("embedding_version"),
+                ccip_embedding=r.get("ccip_embedding"),
+                siglip_embedding=r.get("siglip_embedding"),
+            ))
+            n += 1
+        return n

tests/test_regions.py

@@ -0,0 +1,70 @@
+"""Region storage/service for the crop pipeline (#114)."""
+import pytest
+
+from backend.app.models import ImageRecord
+from backend.app.services.ml.regions import RegionService
+
+pytestmark = pytest.mark.integration
+
+
+async def _img(db, sha) -> ImageRecord:
+    img = ImageRecord(
+        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
+        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
+    )
+    db.add(img)
+    await db.flush()
+    return img
+
+
+@pytest.mark.asyncio
+async def test_replace_and_get_regions(db):
+    img = await _img(db, "a" * 64)
+    svc = RegionService(db)
+    n = await svc.replace_regions(img.id, ["figure"], [
+        {"kind": "figure", "bbox": (0.1, 0.1, 0.3, 0.4),
+         "score": 0.9, "detector_version": "det-v1"},
+    ])
+    await db.commit()
+    assert n == 1
+    regs = await svc.get_regions(img.id)
+    assert len(regs) == 1
+    r = regs[0]
+    assert r.kind == "figure"
+    assert r.rw == pytest.approx(0.3) and r.rh == pytest.approx(0.4)
+    assert r.score == pytest.approx(0.9)
+
+
+@pytest.mark.asyncio
+async def test_replace_is_scoped_by_kind(db):
+    img = await _img(db, "b" * 64)
+    svc = RegionService(db)
+    await svc.replace_regions(img.id, ["figure"], [
+        {"kind": "figure", "bbox": (0.0, 0.0, 0.5, 0.5)},
+    ])
+    await svc.replace_regions(img.id, ["concept"], [
+        {"kind": "concept", "bbox": (0.5, 0.5, 0.2, 0.2)},
+    ])
+    await db.commit()
+    # Re-running the figure detector must NOT wipe the concept region.
+    await svc.replace_regions(img.id, ["figure"], [
+        {"kind": "figure", "bbox": (0.1, 0.1, 0.4, 0.4)},
+    ])
+    await db.commit()
+    kinds = sorted(r.kind for r in await svc.get_regions(img.id))
+    assert kinds == ["concept", "figure"]
+
+
+@pytest.mark.asyncio
+async def test_ccip_vector_round_trips(db):
+    img = await _img(db, "c" * 64)
+    svc = RegionService(db)
+    await svc.replace_regions(img.id, ["figure"], [
+        {"kind": "figure", "bbox": (0.0, 0.0, 0.5, 0.5),
+         "ccip_embedding": [0.1] * 768, "embedding_version": "ccip-test"},
+    ])
+    await db.commit()
+    r = (await svc.get_regions(img.id, kinds=["figure"]))[0]
+    assert r.ccip_embedding is not None
+    assert len(list(r.ccip_embedding)) == 768
+    assert r.siglip_embedding is None