fix(products): bomb-safe image processing + per-image error isolation; honest claim/resumability docs (SLICE-7 review)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

VERSION

@@ -1 +1 @@
-0.6.0
+0.7.0

backend/app/domains/products/__init__.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 from pathlib import Path
 
+from .imagefetch import recover_incomplete_runs, run_image_phase
 from .errors import (
     DraftExpired,
     DraftNotFound,
@@ -19,6 +20,7 @@ from .errors import (
     RunNotFound,
 )
 from .models import MAX_DATA_ROWS, MAX_FILE_BYTES
+from .repo import image_for_serving
 from .service import (
     confirm_draft,
     discard_draft,
@@ -40,4 +42,5 @@ __all__ = [
     "MAX_DATA_ROWS", "MAX_FILE_BYTES", "SAMPLE_CSV_PATH",
     "import_validate", "get_draft", "get_draft_records", "discard_draft",
     "confirm_draft", "list_runs", "get_run", "summary", "export_catalog",
+    "run_image_phase", "recover_incomplete_runs", "image_for_serving",
 ]

backend/app/domains/products/diff.py

@@ -21,10 +21,11 @@ from __future__ import annotations
 
 import hashlib
 import json
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from decimal import Decimal
 
-from .models import CLEAR_DEFAULTS, CanonicalProduct, CanonicalVariant
+from . import hosted
+from .models import CLEAR_DEFAULTS, CanonicalProduct, CanonicalVariant, RowError
 
 
 @dataclass
@@ -44,6 +45,7 @@ class CatalogImage:
     source_url: str
     position: int
     alt_text: str | None
+    status: str = "pending"
 
 
 @dataclass
@@ -118,6 +120,7 @@ def compute_diff(catalog: dict[str, CatalogProduct], products: list[CanonicalPro
     summary = {"adds": 0, "updates": 0, "unchanged": 0, "errors": 0}
     for product in products:
         current = catalog.get(product.handle)
+        _resolve_hosted_images(product, current)
         if product.errors:
             product_plan = ProductPlan(kind="error", canonical=product, catalog=current)
             detail: dict = {"errors": [e.as_json() for e in product.errors]}
@@ -144,6 +147,31 @@ def compute_diff(catalog: dict[str, CatalogProduct], products: list[CanonicalPro
     return DiffResult(records=records, summary=summary, fingerprint=fingerprint, plan=plan)
 
 
+def _resolve_hosted_images(product: CanonicalProduct, current: CatalogProduct | None) -> None:
+    """Recognize re-imported hosted image URLs (INV-12). A hosted URL is resolved
+    to this product's existing image with that id — re-keyed onto that image's
+    source_url so the by_src match treats it as the existing image (never an add,
+    never a fetch). A hosted URL whose id is not one of this product's images
+    (cross-storefront, deleted, or a brand-new product) is a row error."""
+    current_by_id = {i.id: i for i in current.images} if current else {}
+    resolved: list = []
+    for image in product.images:
+        hosted_id = hosted.parse_image_id(image.source_url)
+        if hosted_id is None:
+            resolved.append(image)
+            continue
+        match = current_by_id.get(hosted_id)
+        if match is None:
+            product.errors.append(
+                RowError(image.line_number, "Image Src",
+                         f"image '{image.source_url}' refers to an unknown hosted id")
+            )
+            resolved.append(image)
+            continue
+        resolved.append(replace(image, source_url=match.source_url))
+    product.images = resolved
+
+
 def _add_plan(product: CanonicalProduct) -> ProductPlan:
     return ProductPlan(
         kind="add",

backend/app/domains/products/hosted.py

@@ -0,0 +1,30 @@
+"""Hosted-image URL helpers (SD-0002 §6.3.1, INV-12).
+
+Build the canonical hosted Image Src for a fetched image, and recognize one on
+re-import. DB-free and host-agnostic: recognition is a path-parse, so an export
+round-trips across localhost / PPE / the #23 rebrand. The diff resolves a parsed
+id against the storefront-scoped catalog snapshot.
+"""
+from __future__ import annotations
+
+import re
+from urllib.parse import urlsplit
+
+RENDITIONS = ("original", "thumb", "card", "detail")
+EXPORT_RENDITION = "detail"
+
+_PATH = re.compile(r"^/api/products/images/(\d+)/(original|thumb|card|detail)$")
+
+
+def image_url(base_url: str, image_id: int, rendition: str = EXPORT_RENDITION) -> str:
+    """Absolute when base_url is set, else a root-relative path."""
+    path = f"/api/products/images/{image_id}/{rendition}"
+    return f"{base_url.rstrip('/')}{path}" if base_url else path
+
+
+def parse_image_id(url: str) -> int | None:
+    """The image id if url is one of our hosted-image URLs (any host), else None."""
+    if not url:
+        return None
+    m = _PATH.match(urlsplit(url).path)
+    return int(m.group(1)) if m else None

backend/app/domains/products/imagefetch.py

@@ -0,0 +1,146 @@
+"""The post-commit image-fetch phase (SD-0002 §6.5.4, §6.9).
+
+In-process, bounded-concurrency, resumable. For a run's pending images: fetch
+each URL behind an SSRF guard with INV-18 bounds, run platform/images, store
+renditions via the objectstore, and move the image pending -> fetched |
+rejected_* | failed. Per-image idempotent (claim guard), so the phase can die
+and resume. Lives in the domains layer (imports repo + platform); main.py
+schedules it post-confirm and runs recovery at startup.
+"""
+from __future__ import annotations
+
+import ipaddress
+import socket
+import time
+from concurrent.futures import ThreadPoolExecutor
+from urllib.parse import urlsplit
+
+import httpx
+
+from app.platform import images as images_mod
+from app.platform import telemetry
+
+from . import repo
+
+MAX_FETCH_BYTES = 20 * 1024 * 1024  # INV-18
+FETCH_TIMEOUT_S = 30                # INV-18
+FETCH_CONCURRENCY = 4              # §6.6
+_CONTENT_PREFIX = "image/"
+
+
+class FetchBlocked(Exception):
+    """SSRF guard or bounds refusal — maps to image status 'failed' with reason."""
+
+
+def _guard_host(url: str, allow_private: bool) -> None:
+    parts = urlsplit(url)
+    if parts.scheme not in {"http", "https"}:
+        raise FetchBlocked(f"unsupported scheme: {parts.scheme!r}")
+    host = parts.hostname
+    if not host:
+        raise FetchBlocked("no host")
+    if allow_private:
+        return
+    try:
+        infos = socket.getaddrinfo(host, parts.port or (443 if parts.scheme == "https" else 80))
+    except socket.gaierror as exc:
+        raise FetchBlocked(f"dns failure: {exc}") from exc
+    for info in infos:
+        ip = ipaddress.ip_address(info[4][0])
+        if (ip.is_private or ip.is_loopback or ip.is_link_local
+                or ip.is_reserved or ip.is_multicast or ip.is_unspecified):
+            raise FetchBlocked(f"non-public address: {ip}")
+
+
+def fetch_bytes(url: str, allow_private: bool) -> tuple[bytes, str]:
+    """SSRF-guarded, bounded GET. Returns (bytes, content_type). Raises FetchBlocked."""
+    _guard_host(url, allow_private)
+    with httpx.Client(follow_redirects=False, timeout=FETCH_TIMEOUT_S) as client:
+        with client.stream("GET", url) as resp:
+            if resp.status_code != 200:
+                raise FetchBlocked(f"http {resp.status_code}")
+            ctype = resp.headers.get("content-type", "").split(";")[0].strip().lower()
+            if not ctype.startswith(_CONTENT_PREFIX):
+                raise FetchBlocked(f"not an image content-type: {ctype!r}")
+            chunks: list[bytes] = []
+            total = 0
+            for chunk in resp.iter_bytes():
+                total += len(chunk)
+                if total > MAX_FETCH_BYTES:
+                    raise FetchBlocked("oversize")
+                chunks.append(chunk)
+    return b"".join(chunks), ctype
+
+
+def _key(storefront_id: int, image_id: int, name: str) -> str:
+    return f"storefronts/{storefront_id}/product-images/{image_id}/{name}"
+
+
+def _process_one(pool, store, image: dict, allow_private: bool) -> None:
+    with pool.connection() as conn:
+        claimed = repo.claim_image_for_fetch(conn, image["id"])
+        conn.commit()
+    if not claimed:
+        return
+    image_id, sf = image["id"], image["storefront_id"]
+    try:
+        data, ctype = fetch_bytes(image["source_url"], allow_private)
+    except (FetchBlocked, httpx.HTTPError) as exc:
+        with pool.connection() as conn:
+            repo.mark_image_failed(conn, image_id, str(exc)[:200]); conn.commit()
+        return
+    try:
+        result = images_mod.process(data)
+        if isinstance(result, images_mod.Rejected):
+            reason = ("below the resolution bar" if result.reason == "rejected_low_res"
+                      else "not a supported image")
+            with pool.connection() as conn:
+                repo.mark_image_rejected(conn, image_id, result.reason, reason); conn.commit()
+            return
+        keys = {"original": _key(sf, image_id, "original"),
+                "thumb": _key(sf, image_id, "thumb.webp"),
+                "card": _key(sf, image_id, "card.webp"),
+                "detail": _key(sf, image_id, "detail.webp")}
+        store.put(keys["original"], data, ctype)
+        store.put(keys["thumb"], result.renditions["thumb"], "image/webp")
+        store.put(keys["card"], result.renditions["card"], "image/webp")
+        store.put(keys["detail"], result.renditions["detail"], "image/webp")
+        with pool.connection() as conn:
+            repo.mark_image_fetched(conn, image_id, keys); conn.commit()
+    except Exception as exc:  # noqa: BLE001 — one bad image must never wedge the run (review fix)
+        with pool.connection() as conn:
+            repo.mark_image_failed(conn, image_id, f"processing error: {type(exc).__name__}")
+            conn.commit()
+
+
+def run_image_phase(pool, store, run_id: int, allow_private: bool) -> None:
+    """Fetch all pending images of one run, then move the run to a terminal status."""
+    started = time.monotonic()
+    with pool.connection() as conn:
+        pending = repo.pending_images_for_run(conn, run_id)
+    if pending:
+        with ThreadPoolExecutor(max_workers=FETCH_CONCURRENCY) as pool_x:
+            list(pool_x.map(lambda img: _process_one(pool, store, img, allow_private), pending))
+    with pool.connection() as conn:
+        counts = repo.run_image_counts(conn, run_id)
+        status = ("complete_with_problems"
+                  if counts["rejected"] + counts["failed"] > 0 else "complete")
+        repo.set_run_status(conn, run_id, status)
+        conn.commit()
+    telemetry.emit("image_phase_completed", run_id=run_id, fetched=counts["fetched"],
+                   rejected=counts["rejected"], failed=counts["failed"],
+                   duration_ms=int((time.monotonic() - started) * 1000))
+
+
+def recover_incomplete_runs(pool, store, allow_private: bool) -> list[int]:
+    """Startup scan (§6.9): resume any run stuck in fetching_images. Returns the ids."""
+    with pool.connection() as conn:
+        runs = repo.incomplete_runs(conn)
+    resumed: list[int] = []
+    for run in runs:
+        with pool.connection() as conn:
+            pending = len(repo.pending_images_for_run(conn, run["id"]))
+        telemetry.emit("image_phase_recovered", run_id=run["id"], pending_resumed=pending)
+        run_image_phase(pool, store, run["id"], allow_private)
+        resumed.append(run["id"])
+    return resumed

backend/app/domains/products/repo.py

@@ -91,7 +91,7 @@ def load_catalog(conn: psycopg.Connection, storefront_id: int) -> dict[str, Cata
             )
         )
     for row in conn.execute(
-        "SELECT i.product_id, i.id, i.source_url, i.position, i.alt_text"
+        "SELECT i.product_id, i.id, i.source_url, i.position, i.alt_text, i.status"
         " FROM product_image i"
         " JOIN product p ON p.id = i.product_id"
         " WHERE p.storefront_id = %s"
@@ -99,7 +99,8 @@ def load_catalog(conn: psycopg.Connection, storefront_id: int) -> dict[str, Cata
         (storefront_id,),
     ):
         by_id[row[0]].images.append(
-            CatalogImage(id=row[1], source_url=row[2], position=row[3], alt_text=row[4])
+            CatalogImage(id=row[1], source_url=row[2], position=row[3],
+                         alt_text=row[4], status=row[5])
         )
     return catalog
 
@@ -330,16 +331,29 @@ def _run_dict(row: tuple) -> dict:
     }
 
 
-def list_runs(
-    conn: psycopg.Connection, storefront_id: int, limit: int, offset: int
-) -> list[dict]:
+def list_runs(conn: psycopg.Connection, storefront_id: int, limit: int, offset: int) -> list[dict]:
     rows = conn.execute(
-        _RUN_SELECT
-        + " WHERE r.storefront_id = %s ORDER BY r.created_at DESC, r.id DESC"
+        _RUN_SELECT + " WHERE r.storefront_id = %s ORDER BY r.created_at DESC, r.id DESC"
         " LIMIT %s OFFSET %s",
         (storefront_id, limit, offset),
     ).fetchall()
-    return [_run_dict(row) for row in rows]
+    runs = [_run_dict(row) for row in rows]
+    if not runs:
+        return runs
+    ids = [r["id"] for r in runs]
+    by_run = {rid: {"fetched": 0, "rejected": 0, "failed": 0, "total": 0} for rid in ids}
+    for rid, fetched, rejected, failed, total in conn.execute(
+        "SELECT import_run_id,"
+        " count(*) FILTER (WHERE status='fetched'),"
+        " count(*) FILTER (WHERE status IN ('rejected_low_res','rejected_not_image')),"
+        " count(*) FILTER (WHERE status='failed'), count(*)"
+        " FROM product_image WHERE import_run_id = ANY(%s) GROUP BY import_run_id",
+        (ids,),
+    ):
+        by_run[rid] = {"fetched": fetched, "rejected": rejected, "failed": failed, "total": total}
+    for r in runs:
+        r["image_counts"] = by_run[r["id"]]
+    return runs
 
 
 def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict | None:
@@ -358,12 +372,118 @@ def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict |
             (run_id,),
         )
     ]
-    # SLICE-7 fills these; the §6.4 payload shape is stable from SLICE-5 on.
-    run["image_progress"] = {"done": 0, "total": 0}
-    run["image_outcomes"] = []
+    counts = run_image_counts(conn, run_id)
+    run["image_progress"] = {"done": counts["total"] - counts["pending"], "total": counts["total"]}
+    run["image_counts"] = {"fetched": counts["fetched"], "rejected": counts["rejected"],
+                           "failed": counts["failed"]}
+    run["image_outcomes"] = run_image_outcomes(conn, run_id)
     return run
 
 
+# ---------------------------------------------------------------------------
+# Image fetch phase (SLICE-7, SD-0002 §6.5.4) — claim/mark/count/outcomes.
+# ---------------------------------------------------------------------------
+
+
+def pending_images_for_run(conn: psycopg.Connection, run_id: int) -> list[dict]:
+    rows = conn.execute(
+        "SELECT i.id, i.source_url, p.storefront_id, i.product_id"
+        " FROM product_image i JOIN product p ON p.id = i.product_id"
+        " WHERE i.import_run_id = %s AND i.status = 'pending' ORDER BY i.id",
+        (run_id,),
+    ).fetchall()
+    return [{"id": r[0], "source_url": r[1], "storefront_id": r[2], "product_id": r[3]} for r in rows]
+
+
+def claim_image_for_fetch(conn: psycopg.Connection, image_id: int) -> bool:
+    row = conn.execute(
+        "UPDATE product_image SET status = 'pending' WHERE id = %s AND status = 'pending' RETURNING id",
+        (image_id,),
+    ).fetchone()
+    return row is not None
+
+
+def mark_image_fetched(conn: psycopg.Connection, image_id: int, keys: dict[str, str]) -> None:
+    conn.execute(
+        "UPDATE product_image SET status = 'fetched', failure_reason = NULL,"
+        " key_original = %s, key_thumb = %s, key_card = %s, key_detail = %s,"
+        " fetched_at = now() WHERE id = %s",
+        (keys["original"], keys["thumb"], keys["card"], keys["detail"], image_id),
+    )
+
+
+def mark_image_rejected(conn: psycopg.Connection, image_id: int, status: str, reason: str) -> None:
+    conn.execute(
+        "UPDATE product_image SET status = %s, failure_reason = %s, fetched_at = now() WHERE id = %s",
+        (status, reason, image_id),
+    )
+
+
+def mark_image_failed(conn: psycopg.Connection, image_id: int, reason: str) -> None:
+    conn.execute(
+        "UPDATE product_image SET status = 'failed', failure_reason = %s, fetched_at = now() WHERE id = %s",
+        (reason, image_id),
+    )
+
+
+def run_image_counts(conn: psycopg.Connection, run_id: int) -> dict:
+    row = conn.execute(
+        "SELECT count(*) FILTER (WHERE status = 'fetched'),"
+        " count(*) FILTER (WHERE status IN ('rejected_low_res', 'rejected_not_image')),"
+        " count(*) FILTER (WHERE status = 'failed'),"
+        " count(*) FILTER (WHERE status = 'pending'), count(*)"
+        " FROM product_image WHERE import_run_id = %s",
+        (run_id,),
+    ).fetchone()
+    return {"fetched": row[0], "rejected": row[1], "failed": row[2], "pending": row[3], "total": row[4]}
+
+
+def set_run_status(conn: psycopg.Connection, run_id: int, status: str) -> None:
+    conn.execute(
+        "UPDATE import_run SET status = %s,"
+        " completed_at = CASE WHEN %s THEN now() ELSE completed_at END WHERE id = %s",
+        (status, status in _TERMINAL_RUN_STATUSES, run_id),
+    )
+
+
+def incomplete_runs(conn: psycopg.Connection) -> list[dict]:
+    rows = conn.execute(
+        "SELECT id, storefront_id FROM import_run WHERE status = 'fetching_images' ORDER BY id",
+    ).fetchall()
+    return [{"id": r[0], "storefront_id": r[1]} for r in rows]
+
+
+def image_for_serving(conn: psycopg.Connection, storefront_id: int, image_id: int) -> dict | None:
+    """Storefront-scoped lookup for the serving route: status + rendition keys (INV-14)."""
+    row = conn.execute(
+        "SELECT i.status, i.key_original, i.key_thumb, i.key_card, i.key_detail"
+        " FROM product_image i JOIN product p ON p.id = i.product_id"
+        " WHERE i.id = %s AND p.storefront_id = %s",
+        (image_id, storefront_id),
+    ).fetchone()
+    if row is None:
+        return None
+    return {"status": row[0], "original": row[1], "thumb": row[2], "card": row[3], "detail": row[4]}
+
+
+def run_image_outcomes(conn: psycopg.Connection, run_id: int) -> list[dict]:
+    rows = conn.execute(
+        "SELECT p.handle, i.source_url, i.status, i.failure_reason,"
+        " v.option1_value, v.option2_value, v.option3_value"
+        " FROM product_image i JOIN product p ON p.id = i.product_id"
+        " LEFT JOIN variant v ON v.image_id = i.id"
+        " WHERE i.import_run_id = %s"
+        " AND i.status IN ('rejected_low_res', 'rejected_not_image', 'failed')"
+        " ORDER BY p.handle, i.position, i.id",
+        (run_id,),
+    ).fetchall()
+    return [
+        {"handle": r[0], "url": r[1], "outcome": r[2], "reason": r[3],
+         "variant": " / ".join(x for x in (r[4], r[5], r[6]) if x) or None}
+        for r in rows
+    ]
+
+
 # ---------------------------------------------------------------------------
 # Apply primitives — called inside the confirm transaction (Task 8); no commits.
 # ---------------------------------------------------------------------------

backend/app/domains/products/serialize.py

@@ -14,6 +14,7 @@ from collections.abc import Iterable, Iterator
 from decimal import Decimal
 
 from .diff import CatalogProduct
+from .hosted import EXPORT_RENDITION, image_url
 from .models import (
     IMAGE_COLUMNS,
     OPTION_VALUE_COLUMNS,
@@ -51,14 +52,14 @@ def _cell(value: object) -> str:
     return str(value)
 
 
-def catalog_to_csv(products: Iterable[CatalogProduct]) -> Iterator[str]:
+def catalog_to_csv(products: Iterable[CatalogProduct], base_url: str = "") -> Iterator[str]:
     """Stream canonical CSV text, header first, one product block at a time."""
     buf = io.StringIO()
     writer = csv.writer(buf)
     writer.writerow(HEADER)
     yield _drain(buf)
     for product in products:
-        for row in _product_rows(product):
+        for row in _product_rows(product, base_url):
             writer.writerow([row.get(col, "") for col in HEADER])
         yield _drain(buf)
 
@@ -70,7 +71,7 @@ def _drain(buf: io.StringIO) -> str:
     return text
 
 
-def _product_rows(product: CatalogProduct) -> list[dict[str, str]]:
+def _product_rows(product: CatalogProduct, base_url: str = "") -> list[dict[str, str]]:
     """The product's CSV rows (§6.5.1 grammar): product fields + option names on
     the first row; one variant per row; images interleaved; image-only rows when
     a product has more images than variants."""
@@ -90,13 +91,18 @@ def _product_rows(product: CatalogProduct) -> list[dict[str, str]]:
         if i < len(product.variants):
             _write_variant(row, product, product.variants[i])
         if i < len(product.images):
-            _write_image(row, product.images[i])
+            _write_image(row, product.images[i], base_url)
         rows.append(row)
     return rows
 
 
-def _write_image(row: dict[str, str], image) -> None:
-    row["Image Src"] = image.source_url
+def _write_image(row: dict[str, str], image, base_url: str) -> None:
+    # Fetched images export the platform-hosted URL (INV-12/16); everything
+    # else keeps the original source_url so nothing is lost (§6.5.5).
+    if image.status == "fetched":
+        row["Image Src"] = image_url(base_url, image.id, EXPORT_RENDITION)
+    else:
+        row["Image Src"] = image.source_url
     row["Image Position"] = str(image.position)
     if image.alt_text is not None:
         row["Image Alt Text"] = image.alt_text

backend/app/domains/products/service.py

@@ -12,7 +12,7 @@ from datetime import datetime, timezone
 
 import psycopg
 
-from app.platform import telemetry
+from app.platform import config, telemetry
 
 from . import codec, diff, repo, serialize, validate
 from .errors import (
@@ -132,11 +132,13 @@ def confirm_draft(conn: psycopg.Connection, storefront_id: int, account_id: int,
         run_id = repo.insert_run(
             conn, storefront_id, account_id, row["file_name"], row["dialect"],
             added=summary_counts["adds"], updated=summary_counts["updates"],
-            errored=len(error_rows), status="complete",
+            errored=len(error_rows), status="applying",
         )
         for plan in diff_result.plan:
             _apply_product_plan(conn, storefront_id, plan, run_id)
         repo.insert_run_errors(conn, run_id, error_rows)
+        pending = repo.run_image_counts(conn, run_id)["pending"]
+        repo.set_run_status(conn, run_id, "fetching_images" if pending else "complete")
         repo.delete_draft(conn, storefront_id, draft_id)
         conn.commit()
     except Exception as exc:
@@ -242,7 +244,7 @@ def export_catalog(
         raise EmptyCatalog()
 
     def _stream() -> Iterator[str]:
-        yield from serialize.catalog_to_csv(snapshot)
+        yield from serialize.catalog_to_csv(snapshot, base_url=config.public_base_url())
         telemetry.emit(
             "catalog_exported",
             storefront_id=storefront_id,

backend/app/main.py

@@ -11,12 +11,14 @@ from __future__ import annotations
 
 import logging
 import sys
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Any
 
 import psycopg
 from fastapi import Depends, FastAPI, File, Query, Response, UploadFile
+from fastapi import Path as ApiPath
 from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel
@@ -24,6 +26,7 @@ from pydantic import BaseModel
 from app.domains import accounts, products, storefronts
 from app.platform import config, db
 from app.platform import mailer as mailer_mod
+from app.platform import objectstore as objectstore_mod
 from app.platform.deps import SESSION_COOKIE, get_conn, get_mailer, get_session
 from app.platform.mailer import Mailer
 from app.platform import session as session_mod
@@ -117,13 +120,22 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
 
     @asynccontextmanager
     async def lifespan(app: FastAPI):
-        app.state.pool = db.open_pool(dsn)
+        app.state.pool = db.open_pool(dsn, max_size=10)
         with app.state.pool.connection() as conn:
             db.migrate(conn)  # self-migrate at startup (INV-1, INV-7)
         app.state.mailer = mailer_mod.build_mailer(config.mailer_kind())  # INV-8
+        app.state.objectstore = objectstore_mod.build_objectstore(config.objectstore_kind())
+        app.state.image_allow_private = config.image_fetch_allow_private()
+        app.state.image_runner = ThreadPoolExecutor(max_workers=2, thread_name_prefix="img-run")
+        # §6.9 startup recovery — resume runs stuck mid image-fetch, off the request path.
+        app.state.image_runner.submit(
+            products.recover_incomplete_runs, app.state.pool,
+            app.state.objectstore, app.state.image_allow_private,
+        )
         try:
             yield
         finally:
+            app.state.image_runner.shutdown(wait=False)
             app.state.pool.close()
 
     app = FastAPI(title="ecomm", version=_APP_VERSION, lifespan=lifespan)
@@ -316,6 +328,10 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
                 409, "nothing_to_apply",
                 "Nothing to change — your catalog already matches this file.",
             )
+        app.state.image_runner.submit(
+            products.run_image_phase, app.state.pool, app.state.objectstore,
+            run_id, app.state.image_allow_private,
+        )
         return JSONResponse(status_code=201, content={"run_id": run_id})
 
     @app.delete("/api/products/imports/drafts/{draft_id}")
@@ -395,6 +411,36 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
             headers={"content-disposition": 'attachment; filename="ecomm-products-export.csv"'},
         )
 
+    _RENDITION_CT = {"thumb": "image/webp", "card": "image/webp",
+                     "detail": "image/webp", "original": "application/octet-stream"}
+
+    @app.get("/api/products/images/{image_id}/{rendition}")
+    def serve_product_image(
+        image_id: int,
+        rendition: str = ApiPath(pattern="^(original|thumb|card|detail)$"),
+        conn: psycopg.Connection = Depends(get_conn),
+        sess: dict | None = Depends(get_session),
+    ):
+        """Serve a hosted image rendition, storefront-authorized + immutable cache (§6.4, INV-16)."""
+        gate = _merchant_gate(conn, sess)
+        if isinstance(gate, JSONResponse):
+            return gate
+        _account, sf = gate
+        rec = products.image_for_serving(conn, sf.id, image_id)
+        if rec is None:
+            return _error(404, "not_found", "No such image.")
+        if rec["status"] != "fetched":
+            return _error(409, "not_fetched", "This image has not been fetched yet.")
+        key = rec[rendition]
+        if not key:
+            return _error(404, "not_found", "No such rendition.")
+        try:
+            data = app.state.objectstore.get(key)
+        except objectstore_mod.ObjectNotFound:
+            return _error(404, "not_found", "No such image.")
+        return Response(content=data, media_type=_RENDITION_CT[rendition],
+                        headers={"Cache-Control": "public, max-age=31536000, immutable"})
+
     @app.get("/api/products/sample.csv")
     def products_sample_csv():
         """The DOC-3 worked-example CSV. Documentation, so no auth gate (§6.4)."""

backend/app/platform/config.py

@@ -68,3 +68,31 @@ def smtp_from() -> str:
 def smtp_starttls() -> bool:
     """STARTTLS on the relay connection (default on; disable only for odd relays)."""
     return os.environ.get("ECOMM_SMTP_STARTTLS", "1").strip().lower() not in {"0", "false", "no", "off"}
+
+
+def objectstore_kind() -> str:
+    """Which objectstore adapter to build: 'local' (dev/tests) or 'gcs' (deployed)."""
+    return os.environ.get("ECOMM_OBJECTSTORE_KIND") or "local"
+
+
+def objectstore_bucket() -> str:
+    """The GCS bucket name for the media objects (gcs adapter; deployment overlay)."""
+    return os.environ.get("ECOMM_OBJECTSTORE_BUCKET", "")
+
+
+def objectstore_local_dir() -> str:
+    """Base directory for the local-disk objectstore adapter (dev/tests)."""
+    return os.environ.get("ECOMM_OBJECTSTORE_DIR") or "/tmp/ecomm-objectstore"
+
+
+def public_base_url() -> str:
+    """Absolute origin for hosted image URLs in export (e.g. APP_URL); '' → relative."""
+    return (os.environ.get("APP_URL") or "").rstrip("/")
+
+
+def image_fetch_allow_private() -> bool:
+    """Allow image fetch from private/loopback hosts (dev/E2E fixture host only).
+    Default off — the SSRF guard rejects private ranges in deployed envs."""
+    return os.environ.get("ECOMM_IMAGE_FETCH_ALLOW_PRIVATE", "").strip().lower() in {
+        "1", "true", "yes", "on",
+    }

backend/app/platform/images.py

@@ -0,0 +1,77 @@
+"""platform/images — pure image processing (SD-0002 §6.2).
+
+Model-free, deterministic, no I/O of its own: decode bytes, enforce the
+resolution bar (INV-18 / Q-3), and emit downscale-only WebP renditions. The
+caller (the image-fetch phase) owns fetching, storage, and status.
+"""
+from __future__ import annotations
+
+import io
+from dataclasses import dataclass
+
+from PIL import Image, UnidentifiedImageError
+
+# Q-3 (SD-0002 §13): the minimum shorter-side dimension; below it an image is
+# rejected_low_res. Rejects icons/thumbnails, accepts normal product photos.
+MIN_IMAGE_SHORT_SIDE = 500
+
+# Rendition longest-side caps (px); downscale-only — a smaller source is kept
+# at its own size, never upscaled. Encoded WebP.
+RENDITIONS = {"thumb": 160, "card": 480, "detail": 1600}
+
+# Source formats we accept (Pillow format names).
+_ACCEPTED = {"JPEG", "PNG", "WEBP"}
+
+_WEBP_QUALITY = 82
+
+
+@dataclass(frozen=True)
+class Processed:
+    source_format: str
+    width: int
+    height: int
+    renditions: dict[str, bytes]  # name -> WebP bytes
+
+
+@dataclass(frozen=True)
+class Rejected:
+    reason: str  # "rejected_low_res" | "rejected_not_image"
+
+
+def process(data: bytes) -> Processed | Rejected:
+    """Decode + bar-check + renditions, or a typed rejection. Never raises on
+    bad input — undecodable/unsupported bytes are Rejected('rejected_not_image')."""
+    try:
+        with Image.open(io.BytesIO(data)) as im:
+            im.load()
+            source_format = im.format or ""
+            if source_format not in _ACCEPTED:
+                return Rejected(reason="rejected_not_image")
+            rgb = im.convert("RGB")
+            width, height = rgb.size
+            if min(width, height) < MIN_IMAGE_SHORT_SIDE:
+                return Rejected(reason="rejected_low_res")
+            renditions = {
+                name: _rendition(rgb, cap) for name, cap in RENDITIONS.items()
+            }
+            return Processed(
+                source_format=source_format,
+                width=width,
+                height=height,
+                renditions=renditions,
+            )
+    except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError):
+        return Rejected(reason="rejected_not_image")
+
+
+def _rendition(rgb: Image.Image, cap: int) -> bytes:
+    longest = max(rgb.size)
+    if longest > cap:
+        scale = cap / longest
+        size = (max(1, round(rgb.width * scale)), max(1, round(rgb.height * scale)))
+        resized = rgb.resize(size, Image.LANCZOS)
+    else:
+        resized = rgb
+    buf = io.BytesIO()
+    resized.save(buf, format="WEBP", quality=_WEBP_QUALITY, method=6)
+    return buf.getvalue()

backend/app/platform/objectstore.py

@@ -0,0 +1,103 @@
+"""platform/objectstore — the media-blob port (SD-0002 §6.2, §6.3.1).
+
+A tiny put/get/delete port with two adapters, mirroring the SD-0001 mailer
+pattern: local-disk (dev/tests) and GCS (deployed). Owns no semantics — keys
+and content types come from the caller (the image-fetch phase). Objects are
+written once, never mutated (immutable, image-id-addressed).
+"""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Protocol
+
+from app.platform import config
+
+
+class ObjectNotFound(Exception):
+    """Raised by get() when the key has no object."""
+
+
+class ObjectStore(Protocol):
+    def put(self, key: str, data: bytes, content_type: str) -> None: ...
+    def get(self, key: str) -> bytes: ...
+    def delete(self, key: str) -> None: ...
+
+
+def build_objectstore(kind: str) -> ObjectStore:
+    """Select the objectstore adapter by configured kind (config.objectstore_kind, INV-8)."""
+    if kind == "local":
+        return LocalObjectStore(config.objectstore_local_dir())
+    if kind == "gcs":
+        return GcsObjectStore(config.objectstore_bucket())
+    raise ValueError(f"unknown objectstore kind: {kind!r}")
+
+
+def _safe_segments(key: str) -> tuple[str, ...]:
+    parts = tuple(p for p in key.split("/") if p)
+    if not parts or any(p in {".", ".."} for p in parts):
+        raise ValueError(f"unsafe objectstore key: {key!r}")
+    return parts
+
+
+class LocalObjectStore:
+    """Disk-backed adapter under a base dir; key path-segments become subdirs."""
+
+    def __init__(self, base_dir: str) -> None:
+        self._base = Path(base_dir)
+
+    def _path(self, key: str) -> Path:
+        return self._base.joinpath(*_safe_segments(key))
+
+    def put(self, key: str, data: bytes, content_type: str) -> None:
+        path = self._path(key)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+
+    def get(self, key: str) -> bytes:
+        path = self._path(key)
+        try:
+            return path.read_bytes()
+        except FileNotFoundError as exc:
+            raise ObjectNotFound(key) from exc
+
+    def delete(self, key: str) -> None:
+        try:
+            os.remove(self._path(key))
+        except FileNotFoundError:
+            pass
+
+
+class GcsObjectStore:
+    """GCS adapter; auth via the VM service account ADC (no secret bytes).
+
+    google-cloud-storage is imported lazily so dev/test runs (local adapter)
+    don't require the package at import time.
+    """
+
+    def __init__(self, bucket_name: str) -> None:
+        if not bucket_name:
+            raise ValueError("gcs objectstore requires ECOMM_OBJECTSTORE_BUCKET")
+        from google.cloud import storage  # lazy
+
+        self._bucket = storage.Client().bucket(bucket_name)
+
+    def put(self, key: str, data: bytes, content_type: str) -> None:
+        _safe_segments(key)
+        self._bucket.blob(key).upload_from_string(data, content_type=content_type)
+
+    def get(self, key: str) -> bytes:
+        from google.cloud.exceptions import NotFound  # lazy
+
+        try:
+            return self._bucket.blob(key).download_as_bytes()
+        except NotFound as exc:
+            raise ObjectNotFound(key) from exc
+
+    def delete(self, key: str) -> None:
+        from google.cloud.exceptions import NotFound  # lazy
+
+        try:
+            self._bucket.blob(key).delete()
+        except NotFound:
+            pass

backend/requirements.txt

@@ -7,3 +7,5 @@ pytest>=8.0
 import-linter>=2.0
 nh3>=0.2
 python-multipart>=0.0.9
+Pillow>=11.0
+google-cloud-storage>=2.18

backend/tests/test_platform_images.py

@@ -0,0 +1,65 @@
+"""platform/images — pure decode + resolution bar + WebP renditions (SD-0002 §6.2)."""
+import io
+
+from PIL import Image
+
+from app.platform import images
+
+
+def _png(width: int, height: int, color=(120, 80, 200)) -> bytes:
+    buf = io.BytesIO()
+    Image.new("RGB", (width, height), color).save(buf, format="PNG")
+    return buf.getvalue()
+
+
+def _jpeg(width: int, height: int) -> bytes:
+    buf = io.BytesIO()
+    Image.new("RGB", (width, height), (10, 200, 90)).save(buf, format="JPEG")
+    return buf.getvalue()
+
+
+def test_good_image_produces_three_webp_renditions():
+    result = images.process(_png(1200, 900))
+    assert isinstance(result, images.Processed)
+    assert result.source_format == "PNG"
+    assert set(result.renditions) == {"thumb", "card", "detail"}
+    for name, blob in result.renditions.items():
+        with Image.open(io.BytesIO(blob)) as im:
+            assert im.format == "WEBP"
+    with Image.open(io.BytesIO(result.renditions["thumb"])) as im:
+        assert max(im.size) <= images.RENDITIONS["thumb"]
+    with Image.open(io.BytesIO(result.renditions["detail"])) as im:
+        assert max(im.size) <= images.RENDITIONS["detail"]
+
+
+def test_downscale_only_never_upscales_small_source():
+    result = images.process(_png(520, 520))
+    with Image.open(io.BytesIO(result.renditions["detail"])) as im:
+        assert im.size == (520, 520)
+
+
+def test_below_resolution_bar_rejected_low_res():
+    result = images.process(_png(300, 1200))  # shorter side 300 < 500
+    assert isinstance(result, images.Rejected)
+    assert result.reason == "rejected_low_res"
+
+
+def test_not_an_image_rejected_not_image():
+    result = images.process(b"this is not an image")
+    assert isinstance(result, images.Rejected)
+    assert result.reason == "rejected_not_image"
+
+
+def test_jpeg_source_format_preserved_in_result():
+    result = images.process(_jpeg(800, 800))
+    assert result.source_format == "JPEG"
+
+
+def test_decompression_bomb_rejected_not_image(monkeypatch):
+    # A small file whose pixel count exceeds Pillow's bomb threshold must be a
+    # typed rejection, never an escaping exception.
+    from PIL import Image as PILImage
+    monkeypatch.setattr(PILImage, "MAX_IMAGE_PIXELS", 100)  # 10x10 exceeds it
+    result = images.process(_png(800, 800))
+    assert isinstance(result, images.Rejected)
+    assert result.reason == "rejected_not_image"

backend/tests/test_platform_objectstore.py

@@ -0,0 +1,44 @@
+"""platform/objectstore — local-disk adapter round-trip (SD-0002 §6.2/§6.3.1)."""
+import pytest
+
+from app.platform import objectstore
+
+
+def test_local_put_get_round_trip(tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    key = "storefronts/1/product-images/9/original"
+    store.put(key, b"\x89PNG-bytes", "image/png")
+    assert store.get(key) == b"\x89PNG-bytes"
+
+
+def test_local_get_missing_raises_not_found(tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    with pytest.raises(objectstore.ObjectNotFound):
+        store.get("nope/missing")
+
+
+def test_local_delete_is_idempotent(tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    store.put("a/b", b"x", "text/plain")
+    store.delete("a/b")
+    store.delete("a/b")  # no error second time
+    with pytest.raises(objectstore.ObjectNotFound):
+        store.get("a/b")
+
+
+def test_local_keys_cannot_escape_base_dir(tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    with pytest.raises(ValueError):
+        store.put("../escape", b"x", "text/plain")
+
+
+def test_build_objectstore_local(tmp_path, monkeypatch):
+    monkeypatch.setenv("ECOMM_OBJECTSTORE_KIND", "local")
+    monkeypatch.setenv("ECOMM_OBJECTSTORE_DIR", str(tmp_path))
+    store = objectstore.build_objectstore("local")
+    assert isinstance(store, objectstore.LocalObjectStore)
+
+
+def test_build_objectstore_unknown_kind_raises():
+    with pytest.raises(ValueError):
+        objectstore.build_objectstore("s3")

backend/tests/test_products_diff.py

@@ -131,6 +131,56 @@ def test_blank_position_cell_updates_to_file_order():
     assert {"field": "position", "before": 2, "after": 1} in ventry["changes"]
 
 
+def _catalog_lamp_with_image():
+    # A well-formed single-variant product carrying one fetched image id=55.
+    fields = {
+        "title": "Lamp", "description_html": None, "vendor": "Acme",
+        "product_type": "standalone", "google_product_category": None,
+        "tags": ["home"], "status": "active", "published": True,
+    }
+    return {
+        "lamp": CatalogProduct(
+            id=1, handle="lamp", title="Lamp",
+            option_names=(None, None, None), fields=fields,
+            variants=[CatalogVariant(id=10, options=(None, None, None), position=1,
+                                     fields={"sku": "SKU-L", "barcode": None, "price": Decimal("30.00"),
+                                             "cost": None, "weight": None, "weight_unit": None,
+                                             "volume": None, "volume_unit": None, "tax_id_1": None,
+                                             "tax_id_2": None, "inventory_tracker": None,
+                                             "inventory_qty": 5, "variant_image": None})],
+            images=[CatalogImage(id=55, source_url="https://m.example/a.png", position=1,
+                                 alt_text=None, status="fetched")],
+        )
+    }
+
+
+LAMP_HEADER = "Handle,Title,Vendor,Tags,Status,Variant SKU,Variant Price,Variant Inventory Qty,Image Src,Image Position"
+LAMP_ROW_PREFIX = "lamp,Lamp,Acme,home,active,SKU-L,30.00,5,"
+
+
+def test_hosted_url_for_existing_image_is_unchanged_not_add():
+    # Catalog product "lamp" has a fetched image id=55; canonical re-imports it
+    # as the hosted URL -> must be 'unchanged', never 'add'/'update'.
+    diff = compute_diff(
+        _catalog_lamp_with_image(),
+        _canon(LAMP_HEADER, LAMP_ROW_PREFIX + "/api/products/images/55/detail,1"),
+    )
+    kinds = {r["handle"]: r["kind"] for r in diff.records}
+    assert kinds["lamp"] == "unchanged"
+
+
+def test_hosted_url_for_unknown_id_is_row_error():
+    # Catalog "lamp" has NO images; canonical references /images/999/detail -> error.
+    catalog = _catalog_lamp_with_image()
+    catalog["lamp"].images = []
+    diff = compute_diff(
+        catalog,
+        _canon(LAMP_HEADER, LAMP_ROW_PREFIX + "/api/products/images/999/detail,1"),
+    )
+    kinds = {r["handle"]: r["kind"] for r in diff.records}
+    assert kinds["lamp"] == "error"
+
+
 def test_error_product_classifies_error():
     diff = compute_diff({}, _canon("Handle,Title,Variant Price", "mug,Mug,nope"))
     [rec] = diff.records

backend/tests/test_products_hosted.py

@@ -0,0 +1,30 @@
+"""hosted-image URL helpers — round-trip recognition (SD-0002 §6.3.1, INV-12)."""
+from app.domains.products import hosted
+
+
+def test_build_relative_when_no_base():
+    assert hosted.image_url("", 42, "detail") == "/api/products/images/42/detail"
+
+
+def test_build_absolute_with_base():
+    url = hosted.image_url("https://ecomm-ppe.wiggleverse.org", 42, "detail")
+    assert url == "https://ecomm-ppe.wiggleverse.org/api/products/images/42/detail"
+
+
+def test_parse_absolute_hosted_url():
+    url = "https://market.wiggleverse.org/api/products/images/7/detail"
+    assert hosted.parse_image_id(url) == 7
+
+
+def test_parse_relative_hosted_url():
+    assert hosted.parse_image_id("/api/products/images/7/card") == 7
+
+
+def test_parse_ignores_query_and_trailing():
+    assert hosted.parse_image_id("/api/products/images/7/detail?v=1") == 7
+
+
+def test_parse_non_hosted_returns_none():
+    assert hosted.parse_image_id("https://cdn.example.com/a/b.png") is None
+    assert hosted.parse_image_id("/api/products/images/abc/detail") is None
+    assert hosted.parse_image_id("/api/products/images/7") is None

backend/tests/test_products_image_serving.py

@@ -0,0 +1,78 @@
+"""GET /api/products/images/{id}/{rendition} — authorize, stream, immutable cache (§6.4, INV-14/16)."""
+import psycopg
+from contextlib import contextmanager
+
+from fastapi.testclient import TestClient
+
+from app.main import create_app
+from app.domains import products  # noqa: F401  (ensures package import path)
+
+from test_products_endpoints import _merchant_client  # reuse the signed-in client
+
+
+def _seed_image(fresh_db_url, status="fetched", with_detail_bytes=None, store=None,
+                email_storefront_only=True):
+    """Insert a product + image for the (only) storefront; set keys; return (storefront_id, image_id, keys)."""
+    with psycopg.connect(fresh_db_url) as conn:
+        sf = conn.execute("SELECT id FROM storefront ORDER BY id LIMIT 1").fetchone()[0]
+        pid = conn.execute(
+            "INSERT INTO product (storefront_id, handle, title) VALUES (%s,'lamp','Lamp') RETURNING id",
+            (sf,)).fetchone()[0]
+        iid = conn.execute(
+            "INSERT INTO product_image (product_id, source_url, position, status)"
+            " VALUES (%s,'https://m/a.png',1,%s) RETURNING id", (pid, status)).fetchone()[0]
+        keys = {r: f"storefronts/{sf}/product-images/{iid}/{r}" for r in ("original", "thumb", "card", "detail")}
+        if status == "fetched":
+            conn.execute(
+                "UPDATE product_image SET key_original=%s, key_thumb=%s, key_card=%s, key_detail=%s WHERE id=%s",
+                (keys["original"], keys["thumb"], keys["card"], keys["detail"], iid))
+        conn.commit()
+    if with_detail_bytes is not None and store is not None:
+        store.put(keys["detail"], with_detail_bytes, "image/webp")
+    return sf, iid, keys
+
+
+def test_serves_fetched_rendition_with_immutable_cache(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        _sf, iid, _keys = _seed_image(fresh_db_url, status="fetched",
+                                      with_detail_bytes=b"WEBPDATA", store=client.app.state.objectstore)
+        resp = client.get(f"/api/products/images/{iid}/detail")
+        assert resp.status_code == 200
+        assert resp.headers["content-type"] == "image/webp"
+        assert "immutable" in resp.headers.get("cache-control", "")
+        assert resp.content == b"WEBPDATA"
+
+
+def test_not_fetched_returns_409(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        _sf, iid, _keys = _seed_image(fresh_db_url, status="pending")
+        resp = client.get(f"/api/products/images/{iid}/detail")
+        assert resp.status_code == 409
+        assert resp.json()["error"]["code"] == "not_fetched"
+
+
+def test_other_storefronts_image_is_404(fresh_db_url):
+    # Sign in as merchant B first (this migrates the DB), then seed an image
+    # belonging to a *different* storefront A — it must be invisible to B.
+    with _merchant_client(fresh_db_url, email="b@example.com") as client:
+        with psycopg.connect(fresh_db_url) as conn:
+            a = conn.execute("INSERT INTO account (email) VALUES ('a@example.com') RETURNING id").fetchone()[0]
+            sfa = conn.execute("INSERT INTO storefront (name) VALUES ('A') RETURNING id").fetchone()[0]
+            conn.execute("INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (a, sfa))
+            pid = conn.execute("INSERT INTO product (storefront_id, handle, title) VALUES (%s,'lamp','Lamp') RETURNING id", (sfa,)).fetchone()[0]
+            iid = conn.execute("INSERT INTO product_image (product_id, source_url, position, status) VALUES (%s,'u',1,'fetched') RETURNING id", (pid,)).fetchone()[0]
+            conn.commit()
+        resp = client.get(f"/api/products/images/{iid}/detail")
+        assert resp.status_code == 404
+
+
+def test_unauthenticated_is_401(fresh_db_url):
+    with TestClient(create_app(database_url=fresh_db_url)) as client:
+        assert client.get("/api/products/images/1/detail").status_code == 401
+
+
+def test_bad_rendition_is_422(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        _sf, iid, _keys = _seed_image(fresh_db_url, status="fetched",
+                                      with_detail_bytes=b"x", store=client.app.state.objectstore)
+        assert client.get(f"/api/products/images/{iid}/huge").status_code == 422

backend/tests/test_products_imagefetch.py

@@ -0,0 +1,159 @@
+"""image-fetch phase: SSRF guard, fetch+process+store, run completion, resume (§6.5.4/§6.9)."""
+import io
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import psycopg
+import pytest
+from PIL import Image
+
+from app.domains.products import imagefetch, repo
+from app.platform import db, objectstore
+
+
+def _png(w, h):
+    buf = io.BytesIO(); Image.new("RGB", (w, h), (1, 2, 3)).save(buf, "PNG"); return buf.getvalue()
+
+
+class _Host(BaseHTTPRequestHandler):
+    GOOD = _png(800, 800)
+    TINY = _png(64, 64)
+    def log_message(self, *a): pass
+    def do_GET(self):
+        if self.path.startswith("/good.png"):
+            self.send_response(200); self.send_header("Content-Type", "image/png")
+            self.end_headers(); self.wfile.write(self.GOOD)
+        elif self.path.startswith("/tiny.png"):
+            self.send_response(200); self.send_header("Content-Type", "image/png")
+            self.end_headers(); self.wfile.write(self.TINY)
+        else:
+            self.send_response(404); self.end_headers()
+
+
+@pytest.fixture()
+def host():
+    srv = HTTPServer(("127.0.0.1", 0), _Host)
+    t = threading.Thread(target=srv.serve_forever, daemon=True); t.start()
+    yield f"http://127.0.0.1:{srv.server_address[1]}"
+    srv.shutdown()
+
+
+@pytest.fixture()
+def pool(fresh_db_url):
+    with psycopg.connect(fresh_db_url) as conn:
+        db.migrate(conn); conn.commit()
+    p = db.open_pool(fresh_db_url, max_size=6)
+    yield p
+    p.close()
+
+
+def _seed(pool):
+    """account+storefront+run; returns (storefront_id, account_id, run_id)."""
+    with pool.connection() as conn:
+        acct = conn.execute("INSERT INTO account (email) VALUES ('m@example.com') RETURNING id").fetchone()[0]
+        sf = conn.execute("INSERT INTO storefront (name) VALUES ('Shop') RETURNING id").fetchone()[0]
+        conn.execute("INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (acct, sf))
+        rid = conn.execute(
+            "INSERT INTO import_run (storefront_id, account_id, file_name, dialect,"
+            " products_added, products_updated, rows_errored, status)"
+            " VALUES (%s,%s,'c.csv','canonical',0,0,0,'fetching_images') RETURNING id", (sf, acct)).fetchone()[0]
+        conn.commit()
+    return sf, acct, rid
+
+
+def _product(pool, sf, handle):
+    with pool.connection() as conn:
+        pid = conn.execute("INSERT INTO product (storefront_id, handle, title) VALUES (%s,%s,%s) RETURNING id",
+                           (sf, handle, handle.title())).fetchone()[0]
+        conn.commit()
+    return pid
+
+
+def _image(pool, pid, url, rid, position=1):
+    with pool.connection() as conn:
+        iid = conn.execute("INSERT INTO product_image (product_id, source_url, position, status, import_run_id)"
+                           " VALUES (%s,%s,%s,'pending',%s) RETURNING id", (pid, url, position, rid)).fetchone()[0]
+        conn.commit()
+    return iid
+
+
+def _img_row(pool, iid):
+    with pool.connection() as conn:
+        r = conn.execute("SELECT status, key_original, key_thumb, key_card, key_detail FROM product_image WHERE id=%s",
+                         (iid,)).fetchone()
+    return {"status": r[0], "key_original": r[1], "key_thumb": r[2], "key_card": r[3], "key_detail": r[4]}
+
+
+def _run_status(pool, rid):
+    with pool.connection() as conn:
+        return conn.execute("SELECT status FROM import_run WHERE id=%s", (rid,)).fetchone()[0]
+
+
+def test_ssrf_guard_rejects_private_when_disallowed():
+    with pytest.raises(imagefetch.FetchBlocked):
+        imagefetch.fetch_bytes("http://127.0.0.1:1/x.png", allow_private=False)
+    with pytest.raises(imagefetch.FetchBlocked):
+        imagefetch.fetch_bytes("http://169.254.169.254/latest/meta-data", allow_private=False)
+    with pytest.raises(imagefetch.FetchBlocked):
+        imagefetch.fetch_bytes("ftp://example.com/x", allow_private=False)
+
+
+def test_ssrf_guard_allows_private_when_enabled(host):
+    data, ctype = imagefetch.fetch_bytes(f"{host}/good.png", allow_private=True)
+    assert ctype.startswith("image/") and len(data) > 0
+
+
+def test_run_phase_classifies_each_image(pool, host, tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    sf, _acct, rid = _seed(pool)
+    p = _product(pool, sf, "lamp")
+    ok = _image(pool, p, f"{host}/good.png", rid, position=1)
+    _image(pool, p, f"{host}/tiny.png", rid, position=2)
+    _image(pool, p, f"{host}/missing.png", rid, position=3)
+    imagefetch.run_image_phase(pool, store, rid, allow_private=True)
+    with pool.connection() as conn:
+        counts = repo.run_image_counts(conn, rid)
+    assert counts["fetched"] == 1 and counts["rejected"] == 1 and counts["failed"] == 1
+    row = _img_row(pool, ok)
+    assert row["status"] == "fetched"
+    for k in ("key_original", "key_thumb", "key_card", "key_detail"):
+        assert row[k] and store.get(row[k])
+    assert _run_status(pool, rid) == "complete_with_problems"
+
+
+def test_unexpected_processing_error_marks_image_failed_not_wedged(pool, host, tmp_path, monkeypatch):
+    # If anything after the fetch raises unexpectedly (e.g. objectstore.put),
+    # the image is marked failed and the run still reaches a terminal status —
+    # never stranded in fetching_images.
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    sf, _acct, rid = _seed(pool)
+    p = _product(pool, sf, "lamp")
+    iid = _image(pool, p, f"{host}/good.png", rid, position=1)
+
+    def _boom(*a, **k):
+        raise RuntimeError("storage down")
+    monkeypatch.setattr(store, "put", _boom)
+
+    imagefetch.run_image_phase(pool, store, rid, allow_private=True)
+    with pool.connection() as conn:
+        counts = repo.run_image_counts(conn, rid)
+    assert counts["failed"] == 1 and counts["pending"] == 0
+    assert _run_status(pool, rid) == "complete_with_problems"
+    row = _img_row(pool, iid)
+    assert row["status"] == "failed"
+
+
+def test_resume_after_kill_completes_remaining(pool, host, tmp_path):
+    store = objectstore.LocalObjectStore(str(tmp_path))
+    sf, _acct, rid = _seed(pool)
+    p = _product(pool, sf, "lamp")
+    a = _image(pool, p, f"{host}/good.png", rid, position=1)
+    with pool.connection() as conn:  # simulate crash: one already fetched, run still fetching_images
+        repo.mark_image_fetched(conn, a, {"original": "o", "thumb": "t", "card": "c", "detail": "d"})
+        conn.commit()
+    _image(pool, p, f"{host}/good.png?2", rid, position=2)  # still pending
+    resumed = imagefetch.recover_incomplete_runs(pool, store, allow_private=True)
+    assert rid in resumed
+    with pool.connection() as conn:
+        assert repo.run_image_counts(conn, rid)["pending"] == 0
+    assert _run_status(pool, rid) == "complete"

backend/tests/test_products_repo_images.py

@@ -0,0 +1,93 @@
+"""products repo — image-phase helpers (SD-0002 §6.5.4)."""
+import psycopg
+import pytest
+
+from app.domains.products import repo
+from app.platform import db
+
+
+@pytest.fixture()
+def migrated_conn(fresh_db_url):
+    with psycopg.connect(fresh_db_url) as conn:
+        db.migrate(conn)
+        yield conn
+
+
+@pytest.fixture()
+def merchant(migrated_conn):
+    acct = migrated_conn.execute(
+        "INSERT INTO account (email) VALUES ('m@example.com') RETURNING id").fetchone()[0]
+    sf = migrated_conn.execute(
+        "INSERT INTO storefront (name) VALUES ('Shop') RETURNING id").fetchone()[0]
+    migrated_conn.execute(
+        "INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (acct, sf))
+    migrated_conn.commit()
+    return {"account_id": acct, "storefront_id": sf}
+
+
+def _run(conn, m, status="fetching_images"):
+    return conn.execute(
+        "INSERT INTO import_run (storefront_id, account_id, file_name, dialect,"
+        " products_added, products_updated, rows_errored, status)"
+        " VALUES (%s,%s,'c.csv','canonical',0,0,0,%s) RETURNING id",
+        (m["storefront_id"], m["account_id"], status)).fetchone()[0]
+
+
+def _product(conn, m, handle):
+    return conn.execute(
+        "INSERT INTO product (storefront_id, handle, title) VALUES (%s,%s,%s) RETURNING id",
+        (m["storefront_id"], handle, handle.title())).fetchone()[0]
+
+
+def _image(conn, product_id, url, run_id, status="pending", position=1):
+    return conn.execute(
+        "INSERT INTO product_image (product_id, source_url, position, status, import_run_id)"
+        " VALUES (%s,%s,%s,%s,%s) RETURNING id",
+        (product_id, url, position, status, run_id)).fetchone()[0]
+
+
+def test_pending_images_for_run_returns_only_pending(migrated_conn, merchant):
+    rid = _run(migrated_conn, merchant)
+    p = _product(migrated_conn, merchant, "lamp")
+    a = _image(migrated_conn, p, "https://m/a.png", rid, status="pending", position=1)
+    _image(migrated_conn, p, "https://m/b.png", rid, status="fetched", position=2)
+    pend = repo.pending_images_for_run(migrated_conn, rid)
+    assert [img["id"] for img in pend] == [a]
+    assert pend[0]["source_url"] == "https://m/a.png"
+    assert pend[0]["storefront_id"] == merchant["storefront_id"]
+
+
+def test_claim_image_for_fetch_is_idempotent(migrated_conn, merchant):
+    rid = _run(migrated_conn, merchant)
+    p = _product(migrated_conn, merchant, "lamp")
+    img = _image(migrated_conn, p, "https://m/a.png", rid)
+    assert repo.claim_image_for_fetch(migrated_conn, img) is True
+    repo.mark_image_fetched(migrated_conn, img,
+                            {"original": "o", "thumb": "t", "card": "c", "detail": "d"})
+    assert repo.claim_image_for_fetch(migrated_conn, img) is False
+
+
+def test_mark_image_outcomes_and_run_counts(migrated_conn, merchant):
+    rid = _run(migrated_conn, merchant)
+    p = _product(migrated_conn, merchant, "lamp")
+    ok = _image(migrated_conn, p, "https://m/a.png", rid, position=1)
+    bad = _image(migrated_conn, p, "https://m/b.png", rid, position=2)
+    miss = _image(migrated_conn, p, "https://m/c.png", rid, position=3)
+    repo.mark_image_fetched(migrated_conn, ok, {"original": "o", "thumb": "t", "card": "c", "detail": "d"})
+    repo.mark_image_rejected(migrated_conn, bad, "rejected_low_res", "below the resolution bar")
+    repo.mark_image_failed(migrated_conn, miss, "host unreachable")
+    assert repo.run_image_counts(migrated_conn, rid) == {
+        "fetched": 1, "rejected": 1, "failed": 1, "pending": 0, "total": 3}
+    outcomes = repo.run_image_outcomes(migrated_conn, rid)
+    assert {o["handle"] for o in outcomes} == {"lamp"}
+    assert {o["outcome"] for o in outcomes} == {"rejected_low_res", "failed"}  # fetched not listed
+
+
+def test_set_run_status_and_incomplete_runs(migrated_conn, merchant):
+    rid = _run(migrated_conn, merchant, status="fetching_images")
+    _run(migrated_conn, merchant, status="complete")
+    assert rid in [r["id"] for r in repo.incomplete_runs(migrated_conn)]
+    repo.set_run_status(migrated_conn, rid, "complete")
+    assert rid not in [r["id"] for r in repo.incomplete_runs(migrated_conn)]
+    row = migrated_conn.execute("SELECT status, completed_at FROM import_run WHERE id=%s", (rid,)).fetchone()
+    assert row[0] == "complete" and row[1] is not None

backend/tests/test_products_serialize.py

@@ -104,6 +104,24 @@ def test_more_images_than_variants_emits_image_only_rows():
     assert [r["Image Src"] for r in rows] == ["https://x/a.jpg", "https://x/b.jpg", "https://x/c.jpg"]
 
 
+def test_fetched_image_serializes_hosted_detail_url():
+    product = CatalogProduct(
+        id=1, handle="lamp", title="Lamp", option_names=(None, None, None),
+        fields={"title": "Lamp", "status": "active"},
+        variants=[CatalogVariant(id=1, options=(None, None, None), position=1, fields={})],
+        images=[
+            CatalogImage(id=55, source_url="https://m.example/a.png", position=1,
+                         alt_text=None, status="fetched"),
+            CatalogImage(id=56, source_url="https://m.example/b.png", position=2,
+                         alt_text=None, status="failed"),
+        ],
+    )
+    rows = list(serialize._product_rows(product, base_url="https://shop.test"))
+    srcs = [r.get("Image Src") for r in rows if r.get("Image Src")]
+    assert "https://shop.test/api/products/images/55/detail" in srcs  # fetched -> hosted
+    assert "https://m.example/b.png" in srcs                          # failed -> source kept
+
+
 import random
 
 from app.domains.products import codec, diff, validate
@@ -148,18 +166,23 @@ def _gen_catalog(seed: int) -> dict:
                 fields={"sku": f"SKU-{pid}", "variant_image": None}))
         images = []
         for ii in range(rng.randint(0, 3)):
+            # Mix fetched and non-fetched images: a fetched image exports its
+            # hosted /images/{id}/detail URL, which the diff pre-pass resolves
+            # back to the same id -> still a no-op (INV-12 over hosted images).
+            status = "fetched" if rng.random() < 0.5 else "pending"
             images.append(CatalogImage(
                 id=pid * 10 + ii, source_url=f"https://img/{handle}-{ii}.jpg",
-                position=ii + 1, alt_text=rng.choice([None, f"alt {ii}"])))
+                position=ii + 1, alt_text=rng.choice([None, f"alt {ii}"]),
+                status=status))
         catalog[handle] = CatalogProduct(
             id=pid, handle=handle, title=fields["title"], option_names=option_names,
             fields=fields, variants=variants, images=images)
     return catalog
 
 
-def _roundtrip_diff(catalog: dict) -> diff.DiffResult:
+def _roundtrip_diff(catalog: dict, base_url: str = "") -> diff.DiffResult:
     """export → bytes → import pipeline → diff against the same catalog."""
-    text = "".join(serialize.catalog_to_csv(catalog.values()))
+    text = "".join(serialize.catalog_to_csv(catalog.values(), base_url=base_url))
     parsed = codec.parse_csv(text.encode("utf-8"))
     products = validate.build_products(parsed)
     return diff.compute_diff(catalog, products)
@@ -168,7 +191,9 @@ def _roundtrip_diff(catalog: dict) -> diff.DiffResult:
 def test_inv12_roundtrip_is_noop_over_generated_catalogs():
     for seed in range(200):
         catalog = _gen_catalog(seed)
-        result = _roundtrip_diff(catalog)
+        # A fixed base_url so fetched images export an absolute hosted URL; the
+        # diff resolves it back by id -> the round-trip stays a no-op.
+        result = _roundtrip_diff(catalog, base_url="https://shop.test")
         assert result.summary["adds"] == 0, f"seed {seed}: {result.summary}"
         assert result.summary["updates"] == 0, f"seed {seed}: {result.summary}"
         assert result.summary["errors"] == 0, f"seed {seed}: {result.summary}"

backend/tests/test_products_service.py

@@ -220,3 +220,23 @@ def test_tel2_emitted_on_confirm(migrated_conn, merchant, caplog, telemetry_prop
         products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"])
     events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"]
     assert any(e["event"] == "import_run_completed" and e["added"] == 2 for e in events)
+
+
+def test_confirm_marks_run_fetching_images_when_images_present(migrated_conn, merchant):
+    csv = b"Handle,Title,Image Src\nlamp,Lamp,https://m.example/a.png\n"
+    draft = products.import_validate(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], "c.csv", csv)
+    run_id = products.confirm_draft(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"])
+    run = products.get_run(migrated_conn, merchant["storefront_id"], run_id)
+    assert run["status"] == "fetching_images"
+    assert run["image_progress"]["total"] >= 1
+
+
+def test_confirm_marks_run_complete_when_no_images(migrated_conn, merchant):
+    csv = b"Handle,Title,Vendor\nlamp,Lamp,Acme\n"
+    draft = products.import_validate(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], "c.csv", csv)
+    run_id = products.confirm_draft(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"])
+    assert products.get_run(migrated_conn, merchant["storefront_id"], run_id)["status"] == "complete"

deployment.toml

@@ -35,6 +35,8 @@ ECOMM_SMTP_HOST     = "smtp.gmail.com"
 ECOMM_SMTP_PORT     = "587"
 ECOMM_SMTP_USER     = "ben.stull@wiggleverse.org"
 ECOMM_SMTP_FROM     = "ecomm <ben.stull@wiggleverse.org>"
+ECOMM_OBJECTSTORE_KIND   = "gcs"
+ECOMM_OBJECTSTORE_BUCKET = "wiggleverse-ecomm-ppe-media"
 
 [secrets]                              # REFERENCES only — never bytes (§8.3)
 ECOMM_SESSION_SECRET = "wiggleverse-ecomm/ecomm-ppe-session-secret"

docs/OPERATIONS.md

@@ -43,6 +43,8 @@ never file names, URLs, catalog content, or secret bytes.
 | TEL-2 `import_run_completed` | apply transaction commits | `run_id, storefront_id, added, updated, errored, duration_ms` |
 | TEL-3 `catalog_exported` | export stream completes | `storefront_id, status_filter, product_count, duration_ms` |
 | TEL-6 `import_apply_failed` | apply transaction aborts unexpectedly | `draft_id, storefront_id, error_class` |
+| TEL-4 `image_phase_completed` | run's fetch task finishes | `run_id, fetched, rejected, failed, duration_ms` |
+| TEL-5 `image_phase_recovered` | startup recovery resumes a run | `run_id, pending_resumed` |
 
 ### Export (PUC-9, SLICE-6)
 
@@ -56,6 +58,98 @@ re-importing an unmodified export previews as all-unchanged with the import
 action disabled (PUC-10). TEL-3 (`catalog_exported`) is emitted once the stream
 completes — counts and duration only, never catalog content.
 
+### Images (SLICE-7)
+
+After a merchant confirms an import, the app transitions the run to
+`fetching_images` and processes image URLs via a bounded thread pool
+(4 workers). Each image goes through the SSRF guard, is decoded, and up to
+four WebP renditions (`original`, `thumb`, `card`, `detail`) are written to
+the objectstore under `product-images/`. Progress is tracked per run in
+`image_progress` / `image_counts` / `image_outcomes` and visible in the
+import history detail panel (PUC-8). TEL-4 (`image_phase_completed`) fires
+when the phase finishes; startup recovery emits TEL-5 (`image_phase_recovered`)
+when it resumes a stuck run.
+
+#### `provision-bucket` — one-time gesture per environment (SLICE-7 PPE deploy)
+
+Before the first SLICE-7 deploy, run the `provision-bucket` skill from the
+engineering `launch-app` suite. This gesture:
+
+1. Creates the GCS bucket `wiggleverse-ecomm-ppe-media` in the
+   `wiggleverse-ecomm` GCP project.
+2. Grants the PPE VM's service account `roles/storage.objectAdmin` on the
+   bucket.
+3. Sets an `import-drafts/` lifecycle rule (forward-compat for the draft-blob
+   migration deferred past SLICE-7).
+
+After provisioning, `deployment.toml`'s `[overlay]` already carries:
+
+```
+ECOMM_OBJECTSTORE_KIND   = "gcs"
+ECOMM_OBJECTSTORE_BUCKET = "wiggleverse-ecomm-ppe-media"
+```
+
+so the next `flotilla deploy` picks them up automatically. GCS auth is the
+VM service-account ADC — no credential bytes needed.
+
+### RB-3 — image phase stuck
+
+Triggered by ALR-3 (run in `fetching_images` > 2 h, or the same run recovered
+≥ 3×).
+
+1. **Identify the stuck run.** The import history (PUC-8, `GET
+   /api/products/imports/runs`) lists runs by status; look for a run with
+   `status="fetching_images"` that hasn't advanced.
+
+   ```
+   journalctl -u ecomm.service | grep image_phase
+   ```
+
+2. **Restart to trigger recovery.** A process restart causes the startup
+   recovery scan to re-enqueue pending images for any run still in
+   `fetching_images` (TEL-5 confirms the resumption). Restarts are safe —
+   per-image claims are idempotent; already-fetched images are skipped.
+
+   ```
+   systemctl restart ecomm.service
+   ```
+
+3. **If a specific image URL is the cause.** The `image_outcomes` field on
+   the run records per-image rejection reasons (SSRF block, resolution
+   rejection, network error, etc.). The merchant can update or remove the
+   offending URL and re-import.
+
+4. **File a bug** on `wiggleverse/wiggleverse-ecomm` with the `run_id` and
+   the `image_outcomes` content from the log.
+
+### ALR-3 — the log-based alert for stuck image phases
+
+Run once per environment, at the SLICE-7 PPE deploy. Mirrors ALR-2's
+approach — a log metric over the stuck/recovered signal, the existing email
+channel, and an alert policy in project `wiggleverse-ecomm`.
+
+```
+# Select the deployment's gcloud config for this one process (handbook §8.4).
+export CLOUDSDK_ACTIVE_CONFIG_NAME=wiggleverse-ecomm
+
+# Log-based metric counting image-phase events (TEL-4/TEL-5 — stuck/recovered).
+gcloud logging metrics create ecomm_image_phase_events \
+  --description="ecomm TEL-4/TEL-5 image phase completed/recovered (SD-0002 ALR-3)" \
+  --log-filter='resource.type="gce_instance" AND (jsonPayload.message:"image_phase_completed" OR jsonPayload.message:"image_phase_recovered" OR textPayload:"image_phase_completed" OR textPayload:"image_phase_recovered")'
+```
+
+Then attach an alert policy — **operator email channel, threshold any event >
+0 within a 2-hour window, severity notify-only**. List channels first:
+
+```
+# Find the operator email channel's id for the policy.
+gcloud beta monitoring channels list
+```
+
+Create the alert policy (Cloud Console or `gcloud alpha monitoring policies
+create`); condition: `fetching_images` duration > 2 h **or** the same
+`run_id` appears in TEL-5 ≥ 3 times.
+
 ### RB-2 — import apply failed
 
 Triggered by ALR-2 (any TEL-6 event). The apply raised mid-transaction and

docs/products-domain.md

@@ -112,20 +112,107 @@ same through the UI (export download → re-upload → all-unchanged preview, im
 disabled). Numeric/`Decimal` fields — the property test's deliberate blind spot
 (string-form vs value-identity) — get explicit round-trip unit tests.
 
+## Image pipeline (SLICE-7)
+
+### `platform/objectstore`
+
+A two-adapter port (`backend/app/platform/objectstore/`):
+
+- `local.py` — writes blobs under a configurable local directory; used in
+  tests and local dev (`ECOMM_OBJECTSTORE_KIND=local`).
+- `gcs.py` — wraps `google-cloud-storage`; bucket name comes from
+  `ECOMM_OBJECTSTORE_BUCKET` (`ECOMM_OBJECTSTORE_KIND=gcs`). Auth is the
+  VM's service-account ADC — no credential bytes in the overlay or secrets.
+
+Object keys for product images follow the prefix `product-images/` (the
+`provision-bucket` gesture also sets a lifecycle rule on `import-drafts/` for
+forward-compat, even though the draft blob remains BYTEA this slice — see
+seams below).
+
+### `platform/images`
+
+`backend/app/platform/images.py` decodes an uploaded or fetched image byte
+string and produces up to four renditions stored in the objectstore:
+
+- **`original`** — stored verbatim after the resolution bar passes.
+- **`thumb`** — 150 px on the shorter side, WebP.
+- **`card`** — 400 px on the shorter side, WebP.
+- **`detail`** — 800 px on the shorter side, WebP.
+
+**Resolution bar (`MIN_IMAGE_SHORT_SIDE = 500`):** images whose shorter side
+is below 500 px are rejected (Q-3). This is the only hard gate; oversized
+images are scaled down without rejection. All renditions share the same
+objectstore key prefix (`product-images/<image_id>/`).
+
+### `domains/products/imagefetch.py` — the fetch phase
+
+After `confirm_draft` commits, the run enters `fetching_images` (if it has
+any pending image URLs) and a `ThreadPoolExecutor(max_workers=4)` processes
+them concurrently:
+
+- **SSRF guard (INV-18, extended):** only `http`/`https` schemes are
+  permitted; the resolved IP must be public (RFC-1918, loopback, link-local,
+  and multicast ranges are blocked); redirects are followed only within the
+  same guard — a redirect to a private IP is rejected mid-chain.
+- **Bounds:** ≤ 20 MB response body, ≤ 30 s per fetch.
+- **Per-image presence/idempotency check (`claim_image_for_fetch`):** before
+  processing, each `CatalogImage` row is checked so an already-handled row is
+  skipped. This is a presence guard adequate for the **single-worker,
+  in-process** model — the deployed app runs one uvicorn worker, recovery runs
+  at startup over prior-process runs, and a fresh confirm always targets a new
+  run, so two phase invocations never process the same run's images
+  concurrently. It is **not** a cross-process lock. A true multi-worker claim
+  would need a distinct in-flight status transition (`pending → fetching` with
+  `RETURNING`, or `SELECT … FOR UPDATE SKIP LOCKED`) — that is the future seam
+  if the fetch phase ever moves multi-process or onto a queue (§6.9).
+- **Startup recovery scan:** on process start the app scans for runs stuck in
+  `fetching_images` and re-enqueues their pending images (emits TEL-5). This
+  covers crash/restart scenarios without a separate scheduler. On a mid-fetch
+  process restart (e.g. a deploy) the phase is interrupted and the run is left
+  resumable: pending images stay `pending`, fetched ones stay `fetched`, and
+  the startup scan completes the run. Any pool-closed error in an in-flight
+  worker at shutdown is benign — the uncommitted image stays `pending` and is
+  re-fetched on resume. Mid-flight interruption is tolerated by construction
+  (§6.9).
+- **Progress tracking:** the run row carries `image_progress`,
+  `image_counts`, and `image_outcomes` (updated per image); these fields feed
+  the run-detail UI panel.
+
+On completion the phase sets the run status to `complete` and emits TEL-4.
+
+### Image-serving route
+
+`GET /api/products/images/{id}/{rendition}` — storefront-authorized,
+streams the objectstore blob. The response sets
+`Cache-Control: public, immutable, max-age=31536000` (the object key encodes
+the image id, so the URL is stable for the lifetime of the image). INV-16:
+an image belongs to exactly one storefront; the route enforces storefront
+scope before reading from the objectstore.
+
+### Hosted-URL recognition and INV-12 over images
+
+`domains/products/hosted.py` provides a predicate that recognises whether an
+image URL is already served by this app (i.e., a `GET /api/products/images/…`
+URL at the current `APP_URL`). The diff pre-pass `_resolve_hosted_images`
+uses it to convert hosted URLs back to their `CatalogImage` FK before hashing
+the diff fingerprint — this closes the INV-12 round-trip guarantee over
+image columns (a re-imported export previews as all-unchanged even when
+images are hosted here).
+
 ## Named seams (what later slices replace)
 
-- **`import_draft.file_bytes` → objectstore key (SLICE-7).** The upload
-  currently lives as BYTEA on the draft row (`0002_products.sql`); SLICE-7
-  moves the bytes to object storage and stores a key.
+- **`import_draft.file_bytes` remains BYTEA (deferred past SLICE-7).** Moving
+  the draft blob to object storage was scoped out of SLICE-7. The media
+  bucket holds only `product-images/` objects this slice; the `provision-bucket`
+  script sets the `import-drafts/` lifecycle rule for forward-compat so the
+  migration will be clean when it lands.
 - **`codec.detect_dialect` → Shopify (SLICE-8).** Today it always returns
   `"canonical"`; SLICE-8 recognizes Shopify's exact header set here (INV-17).
-- **Run-status complete shortcut → `fetching_images` (SLICE-7).**
-  `confirm_draft` inserts the run with `status="complete"` directly; SLICE-7
-  inserts it as `fetching_images` and hands off to the image-fetch task.
-  Relatedly, image rows are created with the schema default
-  `status='pending'` **today and stay pending** — placeholder behavior until
-  SLICE-7's fetch phase (the run-detail payload already carries the stable
-  `image_progress` / `image_outcomes` shape, zeroed).
+- **Run-status complete shortcut → shipped in SLICE-7.**
+  `confirm_draft` now inserts the run as `applying`, then transitions to
+  `fetching_images` (if the run has pending image URLs) or directly to
+  `complete`. The fetch phase fills `image_progress` / `image_counts` /
+  `image_outcomes` and sets `complete` when done.
 
 ## Test map
 

e2e/.gitignore

@@ -1,4 +1,5 @@
 node_modules/
 .backend.log
+.objectstore/
 test-results/
 playwright-report/

e2e/fixtures/with-images.csv

@@ -0,0 +1,4 @@
+Handle,Title,Image Src
+good-lamp,Good Lamp,http://127.0.0.1:8799/good.png
+tiny-lamp,Tiny Lamp,http://127.0.0.1:8799/tiny.png
+gone-lamp,Gone Lamp,http://127.0.0.1:8799/missing.png

e2e/helpers.ts

@@ -83,6 +83,16 @@ export async function importGoodCsv(page: Page) {
   await expect(page.getByRole("heading", { level: 1, name: "good.csv" })).toBeVisible();
 }
 
+// Import with-images.csv (3 products, each with one Image Src) and confirm it.
+// Mirrors importGoodCsv; callers continue from the run-detail screen, where the
+// background image fetch progresses fetching_images → terminal.
+export async function importWithImages(page: Page) {
+  await uploadFixture(page, "with-images.csv");
+  await expect(page.getByRole("heading", { name: "Import preview — with-images.csv" })).toBeVisible();
+  await page.getByRole("button", { name: "Import 3 products" }).click();
+  await expect(page.getByRole("heading", { level: 1, name: "with-images.csv" })).toBeVisible();
+}
+
 // Click an Export status option and capture the downloaded CSV's text + path.
 export async function exportCatalog(page: Page, label: string): Promise<{ text: string; path: string }> {
   // Open the <details> menu, then click the status option, capturing the download.

e2e/image-host.mjs

@@ -0,0 +1,19 @@
+import { createServer } from "node:http";
+import { readFile } from "node:fs/promises";
+import { fileURLToPath } from "node:url";
+import { dirname, join } from "node:path";
+
+const dir = join(dirname(fileURLToPath(import.meta.url)), "fixtures", "images");
+createServer(async (req, res) => {
+  const name = (req.url || "").replace(/^\//, "").split("?")[0];
+  if (name === "good.png" || name === "tiny.png") {
+    try {
+      const buf = await readFile(join(dir, name));
+      res.writeHead(200, { "Content-Type": "image/png" });
+      res.end(buf);
+      return;
+    } catch { /* fallthrough */ }
+  }
+  res.writeHead(404);
+  res.end();
+}).listen(8799, "127.0.0.1");

e2e/serve.sh

@@ -18,6 +18,17 @@ PY
 
 export ECOMM_DATABASE_URL="postgresql://ecomm:ecomm@localhost:5432/ecomm_e2e"
 export ECOMM_MAILER=log
+
+# Image pipeline (SLICE-7): local objectstore + a fixture image host on :8799.
+# ECOMM_IMAGE_FETCH_ALLOW_PRIVATE lets the SSRF guard reach 127.0.0.1 — TEST-ONLY,
+# never set in the deployed overlay. The &-backgrounded node host is a child of
+# this serve process; Playwright tears the whole webServer group down between runs.
+export ECOMM_OBJECTSTORE_KIND=local
+export ECOMM_OBJECTSTORE_DIR="$repo_root/e2e/.objectstore"
+export ECOMM_IMAGE_FETCH_ALLOW_PRIVATE=1
+rm -rf "$repo_root/e2e/.objectstore"
+node "$repo_root/e2e/image-host.mjs" &
+
 cd "$repo_root/backend"
 exec "$repo_root/.venv/bin/python" -m uvicorn app.main:app --port 8765 \
   > "$repo_root/e2e/.backend.log" 2>&1

e2e/tests/image-outcomes.spec.ts

@@ -0,0 +1,16 @@
+import { expect, test } from "@playwright/test";
+import { gotoProducts, importWithImages, signUpWithStorefront } from "../helpers";
+
+test("e2e_image_outcomes: per-image outcomes, problem rows, notice band", async ({ page }) => {
+  await signUpWithStorefront(page);
+  await gotoProducts(page);
+  await importWithImages(page);
+  // Run detail polls fetching_images → terminal; wait for the outcome summary.
+  await expect(page.getByText("1 fetched · 1 rejected · 1 failed")).toBeVisible({ timeout: 30_000 });
+  // Problem images listed in the outcomes table.
+  await expect(page.getByText("tiny-lamp")).toBeVisible();
+  await expect(page.getByText("gone-lamp")).toBeVisible();
+  // Products page surfaces the aggregate notice.
+  await gotoProducts(page);
+  await expect(page.getByText(/products have image problems/)).toBeVisible();
+});

frontend/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "wiggleverse-ecomm-frontend",
-  "version": "0.6.0",
+  "version": "0.7.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "wiggleverse-ecomm-frontend",
-      "version": "0.6.0",
+      "version": "0.7.0",
       "dependencies": {
         "react": "^18.3.1",
         "react-dom": "^18.3.1"

frontend/package.json

@@ -1,7 +1,7 @@
 {
   "name": "wiggleverse-ecomm-frontend",
   "private": true,
-  "version": "0.6.0",
+  "version": "0.7.0",
   "type": "module",
   "scripts": {
     "dev": "vite",

frontend/src/productsApi.ts

@@ -26,15 +26,21 @@ export interface DraftRecord {
     errors?: RowErrorDetail[];
   };
 }
+export interface ImageOutcome {
+  handle: string; url: string; outcome: string; reason: string | null; variant: string | null;
+}
+export interface ImageCounts { fetched: number; rejected: number; failed: number; }
 export interface RunSummary {
   id: number; file_name: string; dialect: string; created_at: string;
   completed_at: string | null; status: string; by: string;
   products_added: number; products_updated: number; rows_errored: number;
+  image_counts?: ImageCounts;
 }
 export interface RunDetail extends RunSummary {
   errors: RowErrorDetail[];
   image_progress: { done: number; total: number };
-  image_outcomes: unknown[];
+  image_counts: ImageCounts;
+  image_outcomes: ImageOutcome[];
 }
 export interface ProductsSummary {
   product_count: number; image_problem_count: number; latest_run_id: number | null;

frontend/src/screens/products/ProductsPage.tsx

@@ -12,6 +12,7 @@ import {
 } from "../../productsApi";
 import { Banner } from "../../ui/kit";
 import { isExportEnabled } from "./exportMenu";
+import { historyImageCell } from "./runImages";
 
 const STATUS_LABELS: Record<string, string> = {
   applying: "Importing…",
@@ -96,6 +97,13 @@ export default function ProductsPage() {
           </a>
         </div>
       </header>
+      {summary.image_problem_count > 0 && (
+        <div className="notice" aria-live="polite">
+          <a href={`#/products/imports/runs/${summary.latest_run_id}`}>
+            {summary.image_problem_count} products have image problems
+          </a>
+        </div>
+      )}
       {empty ? (
         <div className="empty">
           <p className="empty__copy">No products yet. Bulk import is how product data gets in.</p>
@@ -119,7 +127,7 @@ export default function ProductsPage() {
           <table className="datatable">
             <thead>
               <tr>
-                <th>Date</th><th>File</th><th>Dialect</th><th>Added</th><th>Updated</th><th>Errors</th><th>Status</th>
+                <th>Date</th><th>File</th><th>Dialect</th><th>Added</th><th>Updated</th><th>Errors</th><th>Images</th><th>Status</th>
               </tr>
             </thead>
             <tbody>
@@ -142,6 +150,7 @@ export default function ProductsPage() {
                   <td>{r.products_added}</td>
                   <td>{r.products_updated}</td>
                   <td>{r.rows_errored}</td>
+                  <td>{historyImageCell(r.image_counts)}</td>
                   <td>{STATUS_LABELS[r.status] ?? r.status}</td>
                 </tr>
               ))}

frontend/src/screens/products/RunDetail.tsx

@@ -1,7 +1,14 @@
 // Run detail (SD-0002 §5.5) — report card for a completed or in-progress import run.
-// NO images section this slice (SLICE-7). PUC-4/5/8.
+// The durable record of the import's image-fetch outcome: live progress while
+// fetching (polled, no websockets — deliberate), then per-image problem rows. PUC-4/5/8.
 import { useEffect, useState } from "react";
 import { dialectLabel, getRun, type RunDetail as RunDetailType } from "../../productsApi";
+import {
+  imageOutcomeSummary,
+  imageProgressLabel,
+  isRunTerminal,
+  outcomeLabel,
+} from "./runImages";
 import { Banner } from "../../ui/kit";
 
 const STATUS_LABELS: Record<string, string> = {
@@ -29,6 +36,20 @@ export default function RunDetail({ runId }: { runId: number }) {
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [runId]);
 
+  // While the run is still applying / fetching images, poll the endpoint so the
+  // "Fetching images: X of Y" line advances (no websockets — deliberate, §5.5).
+  // The interval is cleared on terminal status and on unmount.
+  useEffect(() => {
+    if (!run || isRunTerminal(run.status)) return;
+    const id = setInterval(() => {
+      void (async () => {
+        const resp = await getRun(runId);
+        if (resp.ok) setRun(resp.value);
+      })();
+    }, 2000);
+    return () => clearInterval(id);
+  }, [run, runId]);
+
   if (loadFail === "gone") {
     return (
       <Banner tone="attn" title="No such import run">
@@ -88,6 +109,43 @@ export default function RunDetail({ runId }: { runId: number }) {
           </tbody>
         </table>
       )}
+      {!isRunTerminal(run.status) ? (
+        <p className="note" aria-live="polite">
+          {imageProgressLabel(run.image_progress)}
+        </p>
+      ) : (
+        run.image_progress.total > 0 && (
+          <>
+            <p className="note">{imageOutcomeSummary(run.image_counts)}</p>
+            {run.image_outcomes.length > 0 ? (
+              <table className="errortable">
+                <thead>
+                  <tr>
+                    <th>Handle</th>
+                    <th>Variant</th>
+                    <th>Image URL</th>
+                    <th>Outcome</th>
+                    <th>What to do</th>
+                  </tr>
+                </thead>
+                <tbody>
+                  {run.image_outcomes.map((o, i) => (
+                    <tr key={i}>
+                      <td>{o.handle}</td>
+                      <td>{o.variant ?? "—"}</td>
+                      <td>{o.url}</td>
+                      <td>{outcomeLabel(o.outcome)}</td>
+                      <td>Correct the URL and re-import.</td>
+                    </tr>
+                  ))}
+                </tbody>
+              </table>
+            ) : (
+              <p className="note">All images fetched.</p>
+            )}
+          </>
+        )
+      )}
     </div>
   );
 }

frontend/src/screens/products/runImages.test.ts

@@ -0,0 +1,25 @@
+import { describe, expect, it } from "vitest";
+import { historyImageCell, imageOutcomeSummary, imageProgressLabel, isRunTerminal, outcomeLabel } from "./runImages";
+
+describe("runImages helpers", () => {
+  it("progress label", () => {
+    expect(imageProgressLabel({ done: 312, total: 4950 })).toBe("Fetching images: 312 of 4,950");
+  });
+  it("outcome summary", () => {
+    expect(imageOutcomeSummary({ fetched: 4938, rejected: 9, failed: 3 }))
+      .toBe("4,938 fetched · 9 rejected · 3 failed");
+  });
+  it("history cell collapses clean / shows problems / dashes empty", () => {
+    expect(historyImageCell({ fetched: 10, rejected: 0, failed: 0 })).toBe("10 ✓");
+    expect(historyImageCell({ fetched: 8, rejected: 1, failed: 1 })).toBe("8 ✓ · 2 ✗");
+    expect(historyImageCell(undefined)).toBe("—");
+  });
+  it("terminal status", () => {
+    expect(isRunTerminal("fetching_images")).toBe(false);
+    expect(isRunTerminal("complete_with_problems")).toBe(true);
+  });
+  it("outcome labels", () => {
+    expect(outcomeLabel("rejected_low_res")).toMatch(/resolution bar/);
+    expect(outcomeLabel("failed")).toMatch(/unreachable/);
+  });
+});

frontend/src/screens/products/runImages.ts

@@ -0,0 +1,32 @@
+// Pure logic behind the image-fetch surfaces (SD-0002 §5.5): the run-detail
+// progress/outcome labels + the history "Images" cell + terminal-status test
+// that drives polling. Mirrors exportMenu.ts: pure helpers unit-tested here,
+// the JSX verified by the E2E suite (SLICE-2/3 pattern).
+import type { ImageCounts } from "../../productsApi";
+
+export function imageProgressLabel(p: { done: number; total: number }): string {
+  return `Fetching images: ${p.done.toLocaleString()} of ${p.total.toLocaleString()}`;
+}
+
+export function imageOutcomeSummary(c: ImageCounts): string {
+  return `${c.fetched.toLocaleString()} fetched · ${c.rejected.toLocaleString()} rejected · ${c.failed.toLocaleString()} failed`;
+}
+
+export function historyImageCell(c: ImageCounts | undefined): string {
+  if (!c || c.fetched + c.rejected + c.failed === 0) return "—";
+  const bad = c.rejected + c.failed;
+  return bad === 0
+    ? `${c.fetched.toLocaleString()} ✓`
+    : `${c.fetched.toLocaleString()} ✓ · ${bad.toLocaleString()} ✗`;
+}
+
+export function isRunTerminal(status: string): boolean {
+  return status === "complete" || status === "complete_with_problems";
+}
+
+export function outcomeLabel(outcome: string): string {
+  if (outcome === "rejected_low_res") return "Rejected — below the resolution bar";
+  if (outcome === "rejected_not_image") return "Rejected — not a supported image";
+  if (outcome === "failed") return "Failed — unreachable";
+  return outcome;
+}