diff --git a/alembic/versions/3c2f1a8b9d4e_add_upload_table.py b/alembic/versions/3c2f1a8b9d4e_add_upload_table.py
new file mode 100644
index 0000000..67716cb
--- /dev/null
+++ b/alembic/versions/3c2f1a8b9d4e_add_upload_table.py
@@ -0,0 +1,50 @@
+"""Add upload table
+
+One row per accepted image upload; the object store holds one object per sha256.
+
+Revision ID: 3c2f1a8b9d4e
+Revises: 7a670908f3fa
+Create Date: 2026-08-14 12:00:00.000000
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '3c2f1a8b9d4e'
+down_revision = '7a670908f3fa'
+branch_labels = None
+depends_on = None
+
+# Named, rather than the ``None`` autogenerate emits: an unnamed constraint
+# leaves the downgrade with no name to drop.
+FK_NAME = "upload_uploader_id_fkey"
+
+
+def upgrade():
+    op.create_table(
+        'upload',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('uploader_id', sa.Integer(), nullable=False),
+        sa.Column('sha256', sa.String(), nullable=False),
+        sa.Column('content_type', sa.String(), nullable=False),
+        sa.Column('size_bytes', sa.Integer(), nullable=False),
+        sa.Column('original_filename', sa.String(), nullable=True),
+        sa.Column('purpose', sa.String(), nullable=False),
+        sa.Column('storage_bucket', sa.String(), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(['uploader_id'], ['user.id'], name=FK_NAME),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    op.create_index(op.f('ix_upload_id'), 'upload', ['id'], unique=False)
+    op.create_index(op.f('ix_upload_uploader_id'), 'upload', ['uploader_id'], unique=False)
+    op.create_index(op.f('ix_upload_sha256'), 'upload', ['sha256'], unique=False)
+
+
+def downgrade():
+    op.drop_index(op.f('ix_upload_sha256'), table_name='upload')
+    op.drop_index(op.f('ix_upload_uploader_id'), table_name='upload')
+    op.drop_index(op.f('ix_upload_id'), table_name='upload')
+    op.drop_constraint(FK_NAME, 'upload', type_='foreignkey')
+    op.drop_table('upload')
diff --git a/chafan_core/app/api/api_v1/endpoints/upload.py b/chafan_core/app/api/api_v1/endpoints/upload.py
index 8a89941..11321a2 100644
--- a/chafan_core/app/api/api_v1/endpoints/upload.py
+++ b/chafan_core/app/api/api_v1/endpoints/upload.py
@@ -9,6 +9,7 @@
 from chafan_core.app.limiter import limiter
 from chafan_core.app.services import uploads as uploads_service
 from chafan_core.utils.base import HTTPException_
+from chafan_core.utils.constants import upload_purpose_T
 
 router = APIRouter()
 
@@ -22,7 +23,11 @@ def upload_image(
     ctx: RequestContext = Depends(deps.get_request_context_logged_in),
     file: UploadFile = File(...),
     file_size: int = Depends(valid_content_length),
-    purpose: str = Form("figure"),
+    # A closed set, not a free-form string: anything outside it would skip the
+    # karma gate (which tests for "figure") and be invisible to the read-time
+    # misuse detection (which tests for "avatar"). The default is the gated
+    # value, so an old client cannot slip past the karma check either.
+    purpose: upload_purpose_T = Form("figure"),
 ) -> Any:
     if not object_storage.is_configured():
         raise HTTPException_(
diff --git a/chafan_core/app/crud/crud_upload.py b/chafan_core/app/crud/crud_upload.py
index 49fc2ff..538240f 100644
--- a/chafan_core/app/crud/crud_upload.py
+++ b/chafan_core/app/crud/crud_upload.py
@@ -1,19 +1,10 @@
-"""Stub persistence for image uploads.
-
-TODO: replace with the real ``Upload`` model + migration. Until then the
-endpoint works against the object store but leaves no database record, and
-every upload looks new -- so re-uploading identical bytes is charged again and
-``find_orphans`` / ``misdeclared_avatars`` see nothing. The real crud, model
-and migration land together in the final "SQL" PR; this module is the seam that
-lets the rest of the code merge first.
-"""
-
-from __future__ import annotations
-
+import datetime
 from typing import Any, List, Optional
 
 from sqlalchemy.orm import Session
 
+from chafan_core.app.models.upload import Upload
+
 
 def create(
     db: Session,
@@ -25,21 +16,34 @@ def create(
     purpose: str,
     storage_bucket: str,
     original_filename: Optional[str] = None,
-) -> Any:
-    # TODO: write an Upload row.
-    return None
+) -> Upload:
+    upload = Upload(
+        uploader_id=uploader_id,
+        sha256=sha256,
+        content_type=content_type,
+        size_bytes=size_bytes,
+        original_filename=original_filename,
+        purpose=purpose,
+        storage_bucket=storage_bucket,
+        created_at=datetime.datetime.now(tz=datetime.timezone.utc),
+    )
+    db.add(upload)
+    db.flush()
+    db.refresh(upload)
+    return upload
 
 
 def exists_with_sha(db: Session, *, sha: str) -> bool:
-    # TODO: query the Upload table. Until then every upload looks new.
-    return False
+    return db.query(Upload).filter(Upload.sha256 == sha).first() is not None
 
 
-def get_multi_by_sha(db: Session, *, sha: str) -> List[Any]:
-    # TODO: return the Upload rows for `sha`.
-    return []
+def get_multi_by_sha(db: Session, *, sha: str) -> List[Upload]:
+    return db.query(Upload).filter(Upload.sha256 == sha).all()
 
 
 def all_shas(db: Session) -> List[str]:
-    # TODO: return every distinct sha in the Upload table.
-    return []
+    return [row[0] for row in db.query(Upload.sha256).distinct().all()]
+
+
+def get(db: Session, id: Any) -> Optional[Upload]:
+    return db.query(Upload).filter(Upload.id == id).first()
diff --git a/chafan_core/app/models/__init__.py b/chafan_core/app/models/__init__.py
index 28ebac1..8016c4e 100644
--- a/chafan_core/app/models/__init__.py
+++ b/chafan_core/app/models/__init__.py
@@ -32,6 +32,7 @@
 from .submission_suggestion import SubmissionSuggestion
 from .task import Task
 from .topic import Topic
+from .upload import Upload
 from .user import User
 from .webhook import Webhook
 from .viewcount import ViewCountArticle, ViewCountAnswer, ViewCountQuestion, ViewCountSubmission
diff --git a/chafan_core/app/models/upload.py b/chafan_core/app/models/upload.py
new file mode 100644
index 0000000..b8b401b
--- /dev/null
+++ b/chafan_core/app/models/upload.py
@@ -0,0 +1,34 @@
+from typing import TYPE_CHECKING, Optional
+
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, String
+from sqlalchemy.orm import relationship
+
+from chafan_core.db.base_class import Base
+
+if TYPE_CHECKING:
+    from . import *  # noqa: F401, F403
+
+
+class Upload(Base):
+    """One row per *accepted* upload; the bucket holds one object per sha256.
+
+    The two non-obvious columns:
+
+      * ``sha256`` is deliberately NOT unique. The bucket stores one object per
+        sha; this table records one row per accepted upload, so a free re-upload
+        still leaves an audit trail and the second person to upload the same
+        bytes is not erased.
+      * ``storage_bucket`` records which bucket holds those bytes, so a future
+        vendor migration is a verifiable list rather than a guess.
+    """
+
+    id = Column(Integer, primary_key=True, index=True)
+    uploader_id = Column(Integer, ForeignKey("user.id"), nullable=False, index=True)
+    uploader: Optional["User"] = relationship("User", back_populates="uploads")  # type: ignore
+    sha256 = Column(String, nullable=False, index=True)
+    content_type = Column(String, nullable=False)
+    size_bytes = Column(Integer, nullable=False)  # after sanitizing
+    original_filename = Column(String)
+    purpose = Column(String, nullable=False)  # "figure" | "avatar"
+    storage_bucket = Column(String, nullable=False)
+    created_at = Column(DateTime(timezone=True), nullable=False)
diff --git a/chafan_core/app/models/user.py b/chafan_core/app/models/user.py
index 771aab7..b6b77dc 100755
--- a/chafan_core/app/models/user.py
+++ b/chafan_core/app/models/user.py
@@ -25,6 +25,7 @@
 from chafan_core.app.models.submission import Submission
 from chafan_core.app.models.submission_suggestion import SubmissionSuggestion
 from chafan_core.app.models.task import Task
+from chafan_core.app.models.upload import Upload
 from chafan_core.db.base_class import Base
 from chafan_core.utils.base import UUID_LENGTH
 from chafan_core.utils.validators import StrippedNonEmptyBasicStr
@@ -247,6 +248,11 @@ class User(Base):
         back_populates="user",
         order_by="AuditLog.created_at.desc()",
     )
+    uploads: List["Upload"] = relationship(  # type: ignore
+        "Upload",
+        back_populates="uploader",
+        order_by="Upload.created_at.desc()",
+    )
     initiated_tasks: List["Task"] = relationship(  # type: ignore
         "Task",
         back_populates="initiator",
diff --git a/chafan_core/app/object_storage.py b/chafan_core/app/object_storage.py
index 8a10a77..d1502f1 100644
--- a/chafan_core/app/object_storage.py
+++ b/chafan_core/app/object_storage.py
@@ -35,7 +35,21 @@
 
 
 def is_configured() -> bool:
-    return settings.UPLOADS_S3_ENDPOINT_URL is not None
+    """True when every setting the write path needs is present.
+
+    All three, not just the endpoint: the bucket is a NOT NULL column on the
+    upload row and the public base is what the response URL is built from, so
+    a half-configured deployment used to fail with a 500 *after* the object had
+    already been stored. Checked up front, that becomes a 503 and nothing is
+    written anywhere.
+    """
+    return all(
+        (
+            settings.UPLOADS_S3_ENDPOINT_URL,
+            settings.UPLOADS_S3_BUCKET,
+            settings.UPLOADS_PUBLIC_URL_BASE,
+        )
+    )
 
 
 @lru_cache(maxsize=1)
diff --git a/chafan_core/app/services/uploads.py b/chafan_core/app/services/uploads.py
index 82f8cac..88357f7 100644
--- a/chafan_core/app/services/uploads.py
+++ b/chafan_core/app/services/uploads.py
@@ -21,6 +21,7 @@
 from chafan_core.app.common import MAX_UPLOAD_BYTES, report_msg
 from chafan_core.app.config import settings
 from chafan_core.utils.base import HTTPException_
+from chafan_core.utils.constants import upload_purpose_T
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +45,9 @@ def _read_bytes(file: Any) -> bytes:
     return bytes(buf)
 
 
-def upload_image(ctx, *, file, file_size: int, purpose: str) -> schemas.UploadedImage:
+def upload_image(
+    ctx, *, file, file_size: int, purpose: upload_purpose_T
+) -> schemas.UploadedImage:
     current_user = ctx.get_current_active_user()
     db = ctx.get_db()
 
@@ -98,7 +101,12 @@ def upload_image(ctx, *, file, file_size: int, purpose: str) -> schemas.Uploaded
 # the correct answer with no join table to maintain. Nothing here deletes.
 # ---------------------------------------------------------------------------
 
-_BODY_COLUMNS = [
+# Every column that can hold the URL of an uploaded image. Bodies are the
+# obvious ones. The two avatar columns matter just as much: a purpose="avatar"
+# upload is never embedded in anyone's body, so without them every avatar ever
+# uploaded reads as an orphan -- and orphans are the input to any future
+# garbage collection.
+_REFERENCE_COLUMNS = [
     (models.Article, "body", "article"),
     (models.Article, "body_draft", "article"),
     (models.ArticleArchive, "body", "article_archive"),
@@ -106,31 +114,31 @@ def upload_image(ctx, *, file, file_size: int, purpose: str) -> schemas.Uploaded
     (models.Answer, "body_draft", "answer"),
     (models.Archive, "body", "answer_archive"),
     (models.Comment, "body", "comment"),
+    (models.Question, "description", "question"),
+    (models.QuestionArchive, "description", "question_archive"),
     (models.Submission, "description", "submission"),
     (models.SubmissionArchive, "description", "submission_archive"),
+    (models.User, "avatar_url", "user_avatar"),
+    (models.User, "gif_avatar_url", "user_gif_avatar"),
 ]
 
 
 def find_usages(db: Session, *, sha: str) -> List[str]:
-    """Locations (``table:id``) whose body text embeds ``sha``.
+    """Locations (``table:id``) that reference ``sha``.
 
     The public URL embeds the sha, so a ``LIKE`` hit is exact.
     """
     usages: List[str] = []
-    for model, column, label in _BODY_COLUMNS:
+    for model, column, label in _REFERENCE_COLUMNS:
         col = getattr(model, column)
         for (id_,) in db.query(model.id).filter(col.like(f"%{sha}%")).all():
             usages.append(f"{label}:{id_}")
     return usages
 
 
-def find_orphans(db: Session) -> List[Any]:
-    """Uploads whose sha appears in no body or archive text. Report only.
-
-    Returns nothing until the Upload table lands (``crud.upload`` is stubbed);
-    typed ``Any`` so the real model can arrive later without touching this.
-    """
-    orphans: List[Any] = []
+def find_orphans(db: Session) -> List[models.Upload]:
+    """Uploads whose sha appears in no body or archive text. Report only."""
+    orphans: List[models.Upload] = []
     for sha in crud.upload.all_shas(db):
         if not find_usages(db, sha=sha):
             orphans.extend(crud.upload.get_multi_by_sha(db, sha=sha))
diff --git a/chafan_core/tests/app/api/api_v1/test_upload.py b/chafan_core/tests/app/api/api_v1/test_upload.py
index 45ef8b2..b2a85e3 100644
--- a/chafan_core/tests/app/api/api_v1/test_upload.py
+++ b/chafan_core/tests/app/api/api_v1/test_upload.py
@@ -1,17 +1,22 @@
+import datetime
 import hashlib
 import io
 import random
+from types import SimpleNamespace
 
 import pytest
 from fastapi.testclient import TestClient
 from PIL import Image
 from sqlalchemy.orm import Session
 
-from chafan_core.app import crud, image_sanitize, karma, object_storage
+from chafan_core.app import crud, image_sanitize, karma, models, object_storage, rules
 from chafan_core.app.common import get_redis_cli
 from chafan_core.app.config import settings
+from chafan_core.app.infra.request_context import RequestContext
+from chafan_core.app.services import uploads as uploads_service
 from chafan_core.tests.utils.user import authentication_token_from_email
 from chafan_core.tests.utils.utils import random_email
+from chafan_core.utils.base import get_uuid
 
 UPLOAD_BASE = "https://uploads.cha.fan"
 
@@ -88,8 +93,8 @@ def configure_uploads(monkeypatch):
 @pytest.fixture(autouse=True)
 def clear_upload_rate_limit():
     # The endpoint is rate-limited to 20/hour. Redis outlives a pytest run, so
-    # a developer re-running the suite would otherwise trip the limiter and see
-    # spurious 429s.
+    # a developer re-running the suite (or several modules hitting the same
+    # endpoint) would otherwise trip the limiter and see spurious 429s.
     redis = get_redis_cli()
     for key in redis.scan_iter("*upload/images*"):
         redis.delete(key)
@@ -132,10 +137,31 @@ def test_upload_png_success(client, db, uploader, fake_put_image):
     assert fake_put_image[0]["content_type"] == content_type
 
     db.expire_all()
+    assert db.query(models.Upload).filter_by(uploader_id=uploader["id"]).count() == 1
     user = crud.user.get(db, id=uploader["id"])
     assert user.remaining_coins == 98
 
 
+def test_upload_same_bytes_free(client, db, uploader, fake_put_image):
+    _set_karma(db, uploader["id"], 100)
+    _set_coins(db, uploader["id"], 100)
+    data = _png(_random_rgb())
+
+    r1 = _upload(client, uploader["headers"], data, purpose="figure")
+    assert r1.status_code == 200, r1.json()
+    r2 = _upload(client, uploader["headers"], data, purpose="figure")
+    assert r2.status_code == 200, r2.json()
+
+    assert r1.json()["url"] == r2.json()["url"]
+    assert len(fake_put_image) == 1, "the same bytes must be stored only once"
+
+    db.expire_all()
+    sha = fake_put_image[0]["sha"]
+    assert db.query(models.Upload).filter_by(uploader_id=uploader["id"], sha256=sha).count() == 2
+    user = crud.user.get(db, id=uploader["id"])
+    assert user.remaining_coins == 98, "a re-upload of the same bytes is free"
+
+
 def test_jpeg_exif_stripped(client, db, uploader, fake_put_image):
     _set_karma(db, uploader["id"], 100)
     _set_coins(db, uploader["id"], 100)
@@ -152,6 +178,26 @@ def test_jpeg_exif_stripped(client, db, uploader, fake_put_image):
     )
 
 
+def test_same_photo_different_exif_dedupes(client, db, uploader, fake_put_image):
+    _set_karma(db, uploader["id"], 100)
+    _set_coins(db, uploader["id"], 100)
+    color = _random_rgb()
+    a = _jpeg(rgb=color, description="first copy")
+    b = _jpeg(rgb=color, description="second copy")
+
+    r1 = _upload(client, uploader["headers"], a, filename="a.jpg", content_type="image/jpeg", purpose="figure")
+    assert r1.status_code == 200, r1.json()
+    r2 = _upload(client, uploader["headers"], b, filename="b.jpg", content_type="image/jpeg", purpose="figure")
+    assert r2.status_code == 200, r2.json()
+
+    assert r1.json()["url"] == r2.json()["url"], "same pixels must dedupe to one object"
+    assert len(fake_put_image) == 1
+
+    db.expire_all()
+    user = crud.user.get(db, id=uploader["id"])
+    assert user.remaining_coins == 98, "the second copy is free"
+
+
 def test_animated_gif_stays_animated(client, db, uploader, fake_put_image):
     _set_karma(db, uploader["id"], 100)
     _set_coins(db, uploader["id"], 100)
@@ -176,6 +222,44 @@ def test_figure_requires_karma(client, db, uploader, fake_put_image):
     assert r.status_code == 200, r.json()
 
 
+def test_unknown_purpose_rejected(client, db, uploader, fake_put_image):
+    # `purpose` is a closed set. A third value would be neither gated by karma
+    # (which tests for "figure") nor visible to the read-time misuse detection
+    # (which tests for "avatar"), so it must be refused outright -- karma 0
+    # here, to show the gate is not simply being sidestepped.
+    _set_karma(db, uploader["id"], 0)
+    _set_coins(db, uploader["id"], 100)
+    data = _png(_random_rgb())
+
+    r = _upload(client, uploader["headers"], data, purpose="banana")
+    assert r.status_code == 422, r.json()
+    assert fake_put_image == []
+
+    clean, _ = image_sanitize.sanitize(data)
+    sha = hashlib.sha256(clean).hexdigest()
+    db.expire_all()
+    assert db.query(models.Upload).filter_by(sha256=sha).count() == 0
+
+
+@pytest.mark.parametrize(
+    "unset",
+    ["UPLOADS_S3_ENDPOINT_URL", "UPLOADS_S3_BUCKET", "UPLOADS_PUBLIC_URL_BASE"],
+)
+def test_incomplete_configuration_returns_503(
+    client, db, uploader, fake_put_image, monkeypatch, unset
+):
+    # Any one of the three missing is a 503, checked before anything is stored:
+    # the bucket is a NOT NULL column and the public base builds the response,
+    # so a later failure would leave bytes in the store with no row.
+    _set_karma(db, uploader["id"], 100)
+    _set_coins(db, uploader["id"], 100)
+    monkeypatch.setattr(settings, unset, None)
+
+    r = _upload(client, uploader["headers"], _png(_random_rgb()), purpose="figure")
+    assert r.status_code == 503, r.json()
+    assert fake_put_image == []
+
+
 def test_zero_coins(client, db, uploader, fake_put_image):
     _set_karma(db, uploader["id"], 100)
     _set_coins(db, uploader["id"], 0)
@@ -187,6 +271,14 @@ def test_zero_coins(client, db, uploader, fake_put_image):
     _set_coins(db, uploader["id"], 100)
     assert _upload(client, uploader["headers"], data, purpose="figure").status_code == 200
 
+    _set_coins(db, uploader["id"], 0)
+    r = _upload(client, uploader["headers"], data, purpose="figure")
+    assert r.status_code == 200, "re-uploading already-stored bytes must not charge coins"
+
+    db.expire_all()
+    user = crud.user.get(db, id=uploader["id"])
+    assert user.remaining_coins == 0
+
 
 def test_text_file_rejected(client, db, uploader, fake_put_image):
     _set_karma(db, uploader["id"], 100)
@@ -218,3 +310,236 @@ def test_upload_requires_auth(client, db, uploader):
 def test_vditor_endpoint_removed(client):
     r = client.post(f"{settings.API_V1_STR}/upload/vditor/")
     assert r.status_code == 404
+
+
+def test_find_usages_and_orphans(db, uploader):
+    sha = hashlib.sha256(_png()).hexdigest()
+    crud.upload.create(
+        db,
+        uploader_id=uploader["id"],
+        sha256=sha,
+        content_type="image/png",
+        size_bytes=10,
+        purpose="figure",
+        storage_bucket="test-bucket",
+    )
+    db.commit()
+
+    assert uploads_service.find_usages(db, sha=sha) == []
+    assert any(u.sha256 == sha for u in uploads_service.find_orphans(db))
+
+    now = datetime.datetime.now(tz=datetime.timezone.utc)
+    comment = models.Comment(
+        uuid=get_uuid(),
+        author_id=uploader["id"],
+        body=f"<img src='https://uploads.cha.fan/{sha}.png'>",
+        body_text="an image",
+        created_at=now,
+        updated_at=now,
+    )
+    db.add(comment)
+    db.commit()
+
+    assert uploads_service.find_usages(db, sha=sha) == [f"comment:{comment.id}"]
+    assert not any(u.sha256 == sha for u in uploads_service.find_orphans(db))
+
+
+def test_find_usages_covers_avatars(db, uploader):
+    # An avatar is never embedded in a body: it lives on the user row. Without
+    # those two columns every avatar ever uploaded reads as an orphan.
+    sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    gif_sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    crud.upload.create(
+        db,
+        uploader_id=uploader["id"],
+        sha256=sha,
+        content_type="image/png",
+        size_bytes=10,
+        purpose="avatar",
+        storage_bucket="test-bucket",
+    )
+    db.commit()
+    assert uploads_service.find_usages(db, sha=sha) == []
+
+    user = crud.user.get(db, id=uploader["id"])
+    crud.user.update(
+        db,
+        db_obj=user,
+        obj_in={
+            "avatar_url": f"{UPLOAD_BASE}/{sha}.png",
+            "gif_avatar_url": f"{UPLOAD_BASE}/{gif_sha}.gif",
+        },
+    )
+    db.commit()
+
+    assert uploads_service.find_usages(db, sha=sha) == [f"user_avatar:{user.id}"]
+    assert uploads_service.find_usages(db, sha=gif_sha) == [
+        f"user_gif_avatar:{user.id}"
+    ]
+    assert not any(u.sha256 == sha for u in uploads_service.find_orphans(db))
+
+
+def test_find_usages_covers_question_description(
+    db, uploader, normal_user_authored_question_uuid
+):
+    sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    crud.upload.create(
+        db,
+        uploader_id=uploader["id"],
+        sha256=sha,
+        content_type="image/png",
+        size_bytes=10,
+        purpose="figure",
+        storage_bucket="test-bucket",
+    )
+    db.commit()
+    assert uploads_service.find_usages(db, sha=sha) == []
+
+    question = crud.question.get_by_uuid(db, uuid=normal_user_authored_question_uuid)
+    assert question is not None
+    question.description = f"<img src='{UPLOAD_BASE}/{sha}.png'>"
+    db.add(question)
+    db.commit()
+
+    assert uploads_service.find_usages(db, sha=sha) == [f"question:{question.id}"]
+    assert not any(u.sha256 == sha for u in uploads_service.find_orphans(db))
+
+
+def _create_upload(db, uploader_id, sha, purpose):
+    crud.upload.create(
+        db,
+        uploader_id=uploader_id,
+        sha256=sha,
+        content_type="image/png",
+        size_bytes=10,
+        purpose=purpose,
+        storage_bucket="test-bucket",
+    )
+    db.commit()
+
+
+def test_misdeclared_avatars_logic(db, uploader):
+    avatar_sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    figure_sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    both_sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    never_sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+
+    _create_upload(db, uploader["id"], avatar_sha, "avatar")
+    _create_upload(db, uploader["id"], figure_sha, "figure")
+    _create_upload(db, uploader["id"], both_sha, "avatar")
+    _create_upload(db, uploader["id"], both_sha, "figure")
+
+    body = (
+        f"<img src='https://uploads.cha.fan/{avatar_sha}.png'>"
+        f"<img src='https://uploads.cha.fan/{figure_sha}.png'>"
+        f"<img src='https://uploads.cha.fan/{both_sha}.png'>"
+        f"<img src='https://uploads.cha.fan/{never_sha}.png'>"
+    )
+    ctx = RequestContext(principal_id=uploader["id"])
+    try:
+        result = uploads_service.misdeclared_avatars(
+            ctx, author_id=uploader["id"], body=body
+        )
+    finally:
+        ctx.close()
+    assert result == [avatar_sha]
+
+
+def test_misdeclared_avatar_reported_once(db, uploader, monkeypatch):
+    sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    _create_upload(db, uploader["id"], sha, "avatar")
+
+    reports = []
+    monkeypatch.setattr(uploads_service, "report_msg", reports.append)
+
+    article = SimpleNamespace(
+        id=random.randint(10**6, 10**7),
+        uuid="fake-article-uuid",
+        author_id=uploader["id"],
+        body=f"<img src='https://uploads.cha.fan/{sha}.png'>",
+    )
+    ctx = RequestContext(principal_id=uploader["id"])
+    try:
+        uploads_service.check_article_for_misdeclared_avatars(ctx, article=article)
+        uploads_service.check_article_for_misdeclared_avatars(ctx, article=article)
+    finally:
+        ctx.close()
+
+    assert len(reports) == 1
+    assert sha in reports[0]
+
+
+def test_misdeclared_avatar_swallowed_on_failure(db, uploader, monkeypatch):
+    def boom(ctx, *, author_id, body):
+        raise RuntimeError("upload table down")
+
+    monkeypatch.setattr(uploads_service, "misdeclared_avatars", boom)
+    article = SimpleNamespace(id=1, uuid="x", author_id=uploader["id"], body="x")
+    ctx = RequestContext(principal_id=uploader["id"])
+    try:
+        uploads_service.check_article_for_misdeclared_avatars(ctx, article=article)
+    finally:
+        ctx.close()
+
+
+def test_misdeclared_avatar_swallowed_on_redis_failure(db, uploader, monkeypatch):
+    sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    _create_upload(db, uploader["id"], sha, "avatar")
+
+    class BoomRedis:
+        def set(self, *args, **kwargs):
+            raise RuntimeError("redis down")
+
+    article = SimpleNamespace(
+        id=2,
+        uuid="y",
+        author_id=uploader["id"],
+        body=f"<img src='https://uploads.cha.fan/{sha}.png'>",
+    )
+    ctx = RequestContext(principal_id=uploader["id"])
+    monkeypatch.setattr(ctx, "get_redis", lambda: BoomRedis())
+    try:
+        uploads_service.check_article_for_misdeclared_avatars(ctx, article=article)
+    finally:
+        ctx.close()
+
+
+def test_article_read_reports_misdeclared_avatar_once(client, db, uploader, monkeypatch):
+    sha = hashlib.sha256(_png(_random_rgb())).hexdigest()
+    _create_upload(db, uploader["id"], sha, "avatar")
+    _set_karma(db, uploader["id"], 100)
+    _set_coins(db, uploader["id"], 100)
+
+    r = client.post(
+        f"{settings.API_V1_STR}/article-columns/",
+        headers=uploader["headers"],
+        json={"name": f"col {get_uuid()}", "description": "test"},
+    )
+    assert r.status_code == 200, r.json()
+    column_uuid = r.json()["uuid"]
+
+    body = f"<img src='https://uploads.cha.fan/{sha}.png'>"
+    r = client.post(
+        f"{settings.API_V1_STR}/articles/",
+        headers=uploader["headers"],
+        json={
+            "title": "figure misuse",
+            "content": {"source": body, "editor": "tiptap"},
+            "article_column_uuid": column_uuid,
+            "is_published": True,
+            "writing_session_uuid": get_uuid(),
+            "visibility": "anyone",
+        },
+    )
+    assert r.status_code == 200, r.json()
+    article_uuid = r.json()["uuid"]
+
+    reports = []
+    monkeypatch.setattr(uploads_service, "report_msg", reports.append)
+
+    assert client.get(f"{settings.API_V1_STR}/articles/{article_uuid}").status_code == 200
+    assert client.get(f"{settings.API_V1_STR}/articles/{article_uuid}").status_code == 200
+
+    assert len(reports) == 1, reports
+    assert sha in reports[0]
+    assert article_uuid in reports[0]
diff --git a/chafan_core/utils/constants.py b/chafan_core/utils/constants.py
index 9c17085..38a65b5 100644
--- a/chafan_core/utils/constants.py
+++ b/chafan_core/utils/constants.py
@@ -23,6 +23,12 @@
 ]
 indexed_object_T = Literal["question", "answer", "article", "submission", "site"]
 
+# What an uploaded image was declared to be. The set is closed on purpose: the
+# karma gate applies to "figure" and the read-time misuse detection looks for
+# "avatar", so a third value would be an upload that is neither gated nor
+# detected.
+upload_purpose_T = Literal["figure", "avatar"]
+
 
 feedback_status_T = Union[
     Literal["sent"], Literal["processing"], Literal["closed"], Literal["wontfix"]
diff --git a/smoke/scenarios/s15_upload.py b/smoke/scenarios/s15_upload.py
index 74086e0..c858587 100644
--- a/smoke/scenarios/s15_upload.py
+++ b/smoke/scenarios/s15_upload.py
@@ -1,46 +1,43 @@
-"""s15 — image upload: the spec (KNOWN-FAILING until the endpoint is built).
+"""s15 — image upload: the spec, now enforced.
 
-This scenario is the executable spec for ``POST /api/v1/upload/images/``. It is
-expected to print XFAIL (known failure) against a backend that has not yet
-implemented the endpoint, and to flip to OK once the implementation lands.
-Review the *steps*, not the pass/fail: each step is a statement about what the
-endpoint must do, with "expected … got …" on the XFAIL line.
+This scenario is the executable spec for ``POST /api/v1/upload/images/``. It was
+written before the endpoint existed and reported XFAIL on every point the
+backend did not yet satisfy; the endpoint has since landed, so each point is now
+a hard assertion and a regression fails the suite.
 
 Spec points (see the image-upload proposal):
 
   1. Anonymous uploads are refused: no token → 401/403.
   2. ``purpose="figure"`` is gated on karma ≥ 100 (``rules.MIN_KARMA_UPLOAD_IMAGE``);
      ``purpose="avatar"`` is exempt and succeeds for a low-karma account.
-  3. A new upload costs ``rules.UPLOAD_IMAGE_COST`` = 2 coins, and the response
+  3. ``purpose`` is a closed set: anything else → 422, never a third class of
+     upload that is neither gated nor detected.
+  4. A new upload costs ``rules.UPLOAD_IMAGE_COST`` = 2 coins, and the response
      URL is content-addressed — ``{base}/{sha}.{ext}`` with a 64-hex sha — never
      a placeholder.
-  4. Re-uploading identical bytes is free (no extra coin) and returns the same
+  5. Re-uploading identical bytes is free (no extra coin) and returns the same
      URL.
-  5. Non-image bytes are refused with 415.
-  6. Files over the size cap (``common.MAX_UPLOAD_BYTES`` = 5 MB) are refused.
-  7. The legacy ``POST /upload/vditor/`` endpoint is gone → 404.
-
-The object-store-dependent steps are exercised against whatever store the
-deployment configures via ``UPLOADS_S3_*`` (MinIO in CI); the deduplicate step
-is skipped when the endpoint is not live, since the legacy placeholder endpoint
-returns the same URL for any bytes and charges nothing, which would be a false
-pass.
+  6. Non-image bytes are refused with 415.
+  7. Files over the size cap (``common.MAX_UPLOAD_BYTES`` = 5 MB) are refused.
+  8. The legacy ``POST /upload/vditor/`` endpoint is gone → 404.
+
+Every check that uploads a real image builds its own bytes with
+``_unique_png()``. Sharing one fixture between checks is what made the coin
+assertion wrong before: the server deduplicates on the sha of the *sanitized*
+bytes, so a "new" upload of bytes another check had already sent is correctly
+free, and the "costs 2 coins" point failed against a backend that was right.
 """
 from __future__ import annotations
 
-import base64
+import random
 import re
+import struct
+import zlib
 
 from client import ApiClient, ApiError, note, ok
 
 TAG = "s15_upload"
 
-# A valid 1x1 PNG, so the server-side sanitizer accepts it once the endpoint
-# lands (and the legacy endpoint accepts anything).
-_PNG = base64.b64decode(
-    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADElEQVR4nGP4z8AAAAMBAQDJ/pLvAAAAAElFTkSuQmCC"
-)
-
 # rules.MIN_KARMA_UPLOAD_IMAGE / rules.UPLOAD_IMAGE_COST, duplicated here
 # because the smoke suite does not import app code.
 _KARMA_GATE = 100
@@ -48,52 +45,77 @@
 
 _SHA_RE = re.compile(r"[0-9a-f]{64}")
 
+_UPLOAD_PATH = "/api/v1/upload/images/"
 
-def _content_addressed(url) -> bool:
-    return bool(url) and "picsum.photos" not in url and bool(_SHA_RE.search(url))
 
+def _png(pixels: bytes) -> bytes:
+    """A 4x1 truecolour PNG of ``pixels``, built with the standard library."""
 
-def _describe(resp) -> str:
-    """A short "got X, expected a content-addressed URL" for an XFAIL note."""
-    if resp is None:
-        return "request was refused, expected a content-addressed URL"
-    url = resp.get("url", "") if isinstance(resp, dict) else resp
-    return f"got {url!r}, expected a content-addressed URL"
+    def chunk(kind: bytes, data: bytes) -> bytes:
+        return (
+            struct.pack(">I", len(data))
+            + kind
+            + data
+            + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
+        )
 
+    ihdr = struct.pack(">IIBBBBB", 4, 1, 8, 2, 0, 0, 0)  # 4x1, 8-bit, truecolour
+    idat = zlib.compress(b"\x00" + pixels)  # one scanline, filter type 0
+    return (
+        b"\x89PNG\r\n\x1a\n"
+        + chunk(b"IHDR", ihdr)
+        + chunk(b"IDAT", idat)
+        + chunk(b"IEND", b"")
+    )
+
+
+def _unique_png() -> bytes:
+    """Distinct bytes on every call, so "this upload is new" is actually true.
+
+    The uniqueness has to come from the pixels: the server strips metadata
+    before hashing, so two images differing only in a comment would dedupe to
+    one object. Twelve random bytes also keeps a re-run against an already
+    seeded database honest.
+    """
+    return _png(bytes(random.randrange(256) for _ in range(12)))
+
+
+def _content_addressed(url) -> bool:
+    return bool(url) and "picsum.photos" not in url and bool(_SHA_RE.search(url))
 
-def _upload_ok(client: ApiClient, **kwargs):
-    """Upload and return the response body, or None if the server refused."""
-    try:
-        return client.upload("/api/v1/upload/images/", **kwargs)
-    except ApiError:
-        return None
 
+def _upload_status(client: ApiClient, path: str = _UPLOAD_PATH, **kwargs) -> int:
+    """Upload and return the status: 200 on success, the error's otherwise.
 
-def _upload_refused(client: ApiClient, statuses, **kwargs) -> bool:
-    """Upload and return True if it was refused with one of `statuses`."""
+    Returning the status rather than a bool keeps the "expected X, got Y" that
+    the XFAIL notes used to print, now in the assertion message.
+    """
     try:
-        client.upload("/api/v1/upload/images/", **kwargs)
+        client.upload(path, **kwargs)
     except ApiError as e:
-        return e.status_code in statuses
-    return False
+        return e.status_code
+    return 200
 
 
 def _check_anon_upload_refused(anon: ApiClient) -> None:
-    if _upload_refused(
+    status = _upload_status(
         anon,
-        (401, 403),
-        file_bytes=_PNG,
+        file_bytes=_unique_png(),
         filename="a.png",
         content_type="image/png",
         purpose="figure",
-    ):
-        ok(TAG, "anon upload → 401/403")
-    else:
-        note(TAG, "anon upload → 401/403", "XFAIL", "anonymous upload was accepted")
+    )
+    assert status in (401, 403), f"anon upload: expected 401/403, got {status}"
+    ok(TAG, "anon upload → 401/403")
 
 
 def _accounts_on_opposite_sides(me_a: dict, me_b: dict) -> bool:
-    """A must be ≥ the karma gate and B below it, or this scenario can't run."""
+    """A must be ≥ the karma gate and B below it, or this scenario can't run.
+
+    A precondition of the seeded dataset rather than an assertion about the
+    endpoint, so it skips (like s14's site-creation preconditions) instead of
+    failing.
+    """
     if me_a.get("karma", 0) < _KARMA_GATE or me_b.get("karma", 0) >= _KARMA_GATE:
         note(
             TAG,
@@ -106,134 +128,125 @@ def _accounts_on_opposite_sides(me_a: dict, me_b: dict) -> bool:
 
 
 def _check_low_karma_figure_refused(b: ApiClient) -> None:
-    if _upload_refused(
+    status = _upload_status(
         b,
-        (403,),
-        file_bytes=_PNG,
+        file_bytes=_unique_png(),
         filename="b.png",
         content_type="image/png",
         purpose="figure",
-    ):
-        ok(TAG, "low-karma figure → 403")
-    else:
-        note(TAG, "low-karma figure → 403", "XFAIL", "low-karma figure was accepted")
+    )
+    assert status == 403, f"low-karma figure: expected 403, got {status}"
+    ok(TAG, "low-karma figure → 403")
 
 
 def _check_low_karma_avatar_exempt(b: ApiClient) -> None:
-    resp = _upload_ok(
-        b,
-        file_bytes=_PNG,
+    resp = b.upload(
+        _UPLOAD_PATH,
+        file_bytes=_unique_png(),
         filename="b-avatar.png",
         content_type="image/png",
         purpose="avatar",
     )
-    url = (resp or {}).get("url", "") if isinstance(resp, dict) else ""
-    if _content_addressed(url):
-        ok(TAG, "low-karma avatar → 200 (exempt)", f"url={url}")
-    else:
-        note(TAG, "low-karma avatar → 200 (exempt)", "XFAIL", _describe(resp))
+    url = resp.get("url", "")
+    assert _content_addressed(url), f"got {url!r}, expected a content-addressed URL"
+    ok(TAG, "low-karma avatar → 200 (exempt)", f"url={url}")
+
+
+def _check_unknown_purpose_refused(b: ApiClient) -> None:
+    """An unknown purpose is refused rather than becoming a third class.
+
+    Sent from the low-karma account: if the server accepted it, that account
+    would have uploaded a figure without the karma the gate requires, and the
+    read-time avatar-misuse detection would never see it either.
+    """
+    status = _upload_status(
+        b,
+        file_bytes=_unique_png(),
+        filename="b-other.png",
+        content_type="image/png",
+        purpose="banana",
+    )
+    assert status == 422, f"unknown purpose: expected 422, got {status}"
+    ok(TAG, "unknown purpose → 422")
 
 
 def _check_new_figure(a: ApiClient, me_a: dict):
     """New figure must return a content-addressed URL and cost 2 coins.
 
-    Returns (live, url, coins_after): ``live`` is True when the endpoint is up
-    and storing real objects (a content-addressed URL came back), which is what
-    gates the deduplicate step.
+    Returns (data, url, coins_after) for the deduplicate check to reuse.
     """
     coins_before = me_a.get("remaining_coins", 0)
-    resp = _upload_ok(
-        a,
-        file_bytes=_PNG,
+    data = _unique_png()
+    resp = a.upload(
+        _UPLOAD_PATH,
+        file_bytes=data,
         filename="a.png",
         content_type="image/png",
         purpose="figure",
     )
-    url = (resp or {}).get("url", "") if isinstance(resp, dict) else ""
-    if not _content_addressed(url):
-        note(TAG, "new figure → content-addressed URL, costs 2 coins", "XFAIL", _describe(resp))
-        return False, "", 0
+    url = resp.get("url", "")
+    assert _content_addressed(url), f"got {url!r}, expected a content-addressed URL"
 
     coins_after = a.get("/api/v1/me")["remaining_coins"]
-    if coins_before - coins_after == _COIN_COST:
-        ok(TAG, "new figure → content-addressed URL, costs 2 coins", f"url={url}")
-    else:
-        note(
-            TAG,
-            "new figure → content-addressed URL, costs 2 coins",
-            "XFAIL",
-            f"charged {coins_before - coins_after} coins, expected {_COIN_COST}",
-        )
-    return True, url, coins_after
-
-
-def _check_identical_bytes_free(a: ApiClient, live: bool, url: str, coins_after: int) -> None:
-    if not live:
-        note(TAG, "identical bytes are free and deduplicate", "SKIP", "endpoint not live")
-        return
-    resp = _upload_ok(
-        a,
-        file_bytes=_PNG,
+    charged = coins_before - coins_after
+    assert charged == _COIN_COST, f"charged {charged} coins, expected {_COIN_COST}"
+    ok(TAG, "new figure → content-addressed URL, costs 2 coins", f"url={url}")
+    return data, url, coins_after
+
+
+def _check_identical_bytes_free(
+    a: ApiClient, data: bytes, url: str, coins_after: int
+) -> None:
+    resp = a.upload(
+        _UPLOAD_PATH,
+        file_bytes=data,
         filename="a-again.png",
         content_type="image/png",
         purpose="figure",
     )
-    again = (resp or {}).get("url", "") if isinstance(resp, dict) else ""
+    again = resp.get("url", "")
+    assert again == url, f"got {again!r}, expected the same URL {url!r}"
     coins_now = a.get("/api/v1/me")["remaining_coins"]
-    if again == url and coins_now == coins_after:
-        ok(TAG, "identical bytes are free and deduplicate", f"url={again}")
-    else:
-        note(
-            TAG,
-            "identical bytes are free and deduplicate",
-            "XFAIL",
-            f"url={again!r}, coins {coins_after} → {coins_now}",
-        )
+    assert coins_now == coins_after, (
+        f"re-upload charged {coins_after - coins_now} coins, expected 0"
+    )
+    ok(TAG, "identical bytes are free and deduplicate", f"url={again}")
 
 
 def _check_non_image_refused(a: ApiClient) -> None:
-    if _upload_refused(
+    status = _upload_status(
         a,
-        (415,),
         file_bytes=b"this is definitely not an image",
         filename="fake.png",
         content_type="image/png",
         purpose="figure",
-    ):
-        ok(TAG, "non-image bytes → 415")
-    else:
-        note(TAG, "non-image bytes → 415", "XFAIL", "non-image bytes were accepted")
+    )
+    assert status == 415, f"non-image bytes: expected 415, got {status}"
+    ok(TAG, "non-image bytes → 415")
 
 
 def _check_oversize_refused(a: ApiClient) -> None:
-    if _upload_refused(
+    status = _upload_status(
         a,
-        (413, 422),
         file_bytes=b"\x00" * (5_000_000 + 1),
         filename="big.png",
         content_type="image/png",
         purpose="figure",
-    ):
-        ok(TAG, "oversize file → 413/422")
-    else:
-        note(TAG, "oversize file → 413/422", "XFAIL", "oversize file was accepted")
+    )
+    assert status in (413, 422), f"oversize file: expected 413/422, got {status}"
+    ok(TAG, "oversize file → 413/422")
 
 
 def _check_vditor_removed(a: ApiClient) -> None:
-    try:
-        a.upload(
-            "/api/v1/upload/vditor/",
-            file_bytes=_PNG,
-            filename="a.png",
-            content_type="image/png",
-        )
-    except ApiError as e:
-        if e.status_code == 404:
-            ok(TAG, "vditor endpoint removed → 404")
-        else:
-            note(TAG, "vditor endpoint removed → 404", "XFAIL", f"expected 404, got {e.status_code}")
-    else:
-        note(TAG, "vditor endpoint removed → 404", "XFAIL", "endpoint still exists")
+    status = _upload_status(
+        a,
+        "/api/v1/upload/vditor/",
+        file_bytes=_unique_png(),
+        filename="a.png",
+        content_type="image/png",
+    )
+    assert status == 404, f"vditor endpoint: expected 404, got {status}"
+    ok(TAG, "vditor endpoint removed → 404")
 
 
 def run(state: dict) -> None:
@@ -251,8 +264,9 @@ def run(state: dict) -> None:
 
     _check_low_karma_figure_refused(b)
     _check_low_karma_avatar_exempt(b)
-    live, url, coins_after = _check_new_figure(a, me_a)
-    _check_identical_bytes_free(a, live, url, coins_after)
+    _check_unknown_purpose_refused(b)
+    data, url, coins_after = _check_new_figure(a, me_a)
+    _check_identical_bytes_free(a, data, url, coins_after)
     _check_non_image_refused(a)
     _check_oversize_refused(a)
     _check_vditor_removed(a)