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.5.0
+0.7.0

backend/app/domains/products/__init__.py

@@ -8,9 +8,11 @@ from __future__ import annotations
 
 from pathlib import Path
 
+from .imagefetch import recover_incomplete_runs, run_image_phase
 from .errors import (
     DraftExpired,
     DraftNotFound,
+    EmptyCatalog,
     FileRejected,
     NothingToApply,
     PreviewStale,
@@ -18,9 +20,11 @@ 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,
+    export_catalog,
     get_draft,
     get_draft_records,
     get_run,
@@ -34,8 +38,9 @@ SAMPLE_CSV_PATH = Path(__file__).parent / "sample.csv"
 
 __all__ = [
     "ProductsError", "FileRejected", "DraftNotFound", "DraftExpired",
-    "PreviewStale", "NothingToApply", "RunNotFound",
+    "PreviewStale", "NothingToApply", "RunNotFound", "EmptyCatalog",
     "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",
+    "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/errors.py

@@ -35,3 +35,7 @@ class NothingToApply(ProductsError):
 
 class RunNotFound(ProductsError):
     """No such import run for this storefront."""
+
+
+class EmptyCatalog(ProductsError):
+    """PUC-9: nothing to export (no products, or none matching the status filter)."""

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,11 +99,29 @@ 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
 
 
+_EXPORT_STATUSES = ("all", "active", "draft", "archived")
+
+
+def export_catalog(
+    conn: psycopg.Connection, storefront_id: int, status_filter: str
+) -> list[CatalogProduct]:
+    """The storefront's catalog as an ordered snapshot list, optionally filtered
+    by product status (PUC-9). Reuses load_catalog's snapshot builder; the
+    catalog fits in memory (≤5k rows, INV-18). Ordered by handle for a stable,
+    deterministic export."""
+    catalog = load_catalog(conn, storefront_id)
+    products = sorted(catalog.values(), key=lambda p: p.handle)
+    if status_filter and status_filter != "all":
+        products = [p for p in products if p.fields.get("status") == status_filter]
+    return products
+
+
 def product_count(conn: psycopg.Connection, storefront_id: int) -> int:
     return conn.execute(
         "SELECT count(*) FROM product WHERE storefront_id = %s", (storefront_id,)
@@ -313,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:
@@ -341,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

@@ -0,0 +1,125 @@
+"""Canonical serializer — CatalogProduct snapshot → canonical CSV (SD-0002 §6.5.5).
+
+The export half of "one codec, two directions": this writes exactly the columns
+codec.py/validate.py parse, in a row grammar (§6.5.1) the validator regroups
+identically — so re-importing an unmodified export diffs to nothing (INV-12).
+DB-free, like diff.py: it consumes the same CatalogProduct snapshot the diff
+engine reads (repo.load_catalog), and the service streams the result.
+"""
+from __future__ import annotations
+
+import csv
+import io
+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,
+    PRODUCT_COLUMNS,
+    VARIANT_COLUMNS,
+)
+
+# Canonical column order: Handle, then product, option-value, variant, image
+# columns. A superset of everything the parser knows (models.KNOWN_COLUMNS minus
+# the #15-reserved Component columns, which export never emits).
+HEADER: list[str] = [
+    "Handle",
+    *PRODUCT_COLUMNS,
+    *OPTION_VALUE_COLUMNS,
+    *VARIANT_COLUMNS,
+    *IMAGE_COLUMNS,
+]
+
+# field name -> column name, inverting the model registry (the parser reads
+# column->field; the serializer writes field->column).
+_PRODUCT_FIELD_TO_COL = {field: col for col, field in PRODUCT_COLUMNS.items()}
+_VARIANT_FIELD_TO_COL = {field: col for col, field in VARIANT_COLUMNS.items()}
+
+
+def _cell(value: object) -> str:
+    """Serialize one value to its canonical cell text (the parse inverse)."""
+    if value is None:
+        return ""
+    if isinstance(value, bool):
+        return "TRUE" if value else "FALSE"
+    if isinstance(value, Decimal):
+        return str(value)
+    if isinstance(value, (list, tuple)):
+        return ", ".join(str(v) for v in value)
+    return str(value)
+
+
+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, base_url):
+            writer.writerow([row.get(col, "") for col in HEADER])
+        yield _drain(buf)
+
+
+def _drain(buf: io.StringIO) -> str:
+    text = buf.getvalue()
+    buf.seek(0)
+    buf.truncate(0)
+    return text
+
+
+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."""
+    base = {"Handle": product.handle, "Title": product.title}
+    for field, col in _PRODUCT_FIELD_TO_COL.items():
+        if field in product.fields:
+            base[col] = _cell(product.fields[field])
+    for slot, name in enumerate(product.option_names, start=1):
+        if name:
+            base[f"Option{slot} Name"] = name
+
+    count = max(len(product.variants), len(product.images))
+    rows: list[dict[str, str]] = []
+    for i in range(count):
+        # Row 0 carries the product-level fields; later rows carry only Handle.
+        row = dict(base) if i == 0 else {"Handle": product.handle}
+        if i < len(product.variants):
+            _write_variant(row, product, product.variants[i])
+        if i < len(product.images):
+            _write_image(row, product.images[i], base_url)
+        rows.append(row)
+    return rows
+
+
+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
+
+
+def _write_variant(row: dict[str, str], product: CatalogProduct, variant) -> None:
+    """Fill a row's option-value + variant columns for one variant."""
+    for slot, value in enumerate(variant.options, start=1):
+        # Only emit an option value when the product actually has that option
+        # (a no-option product's single variant carries all-NULL options).
+        if product.option_names[slot - 1] and value is not None:
+            row[f"Option{slot} Value"] = value
+    # position is a CatalogVariant attribute, not a fields{} entry — emit it
+    # explicitly. An empty Variant Position cell re-imports as "reset to file
+    # order", so a non-sequential stored position would round-trip to an update
+    # (INV-12). (Like _write_image, which emits its position attribute.)
+    row["Variant Position"] = str(variant.position)
+    for field, col in _VARIANT_FIELD_TO_COL.items():
+        if field in variant.fields:
+            row[col] = _cell(variant.fields[field])

backend/app/domains/products/service.py

@@ -7,14 +7,22 @@ writes exactly one row — the import_draft. TEL events per §9.1.
 from __future__ import annotations
 
 import time
+from collections.abc import Iterator
 from datetime import datetime, timezone
 
 import psycopg
 
-from app.platform import telemetry
+from app.platform import config, telemetry
 
-from . import codec, diff, repo, validate
-from .errors import DraftExpired, DraftNotFound, NothingToApply, PreviewStale, RunNotFound
+from . import codec, diff, repo, serialize, validate
+from .errors import (
+    DraftExpired,
+    DraftNotFound,
+    EmptyCatalog,
+    NothingToApply,
+    PreviewStale,
+    RunNotFound,
+)
 
 
 def import_validate(conn: psycopg.Connection, storefront_id: int, account_id: int,
@@ -124,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:
@@ -218,6 +228,34 @@ def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict:
     return run
 
 
+def export_catalog(
+    conn: psycopg.Connection, storefront_id: int, status_filter: str
+) -> Iterator[str]:
+    """Stream the storefront's catalog as canonical CSV (PUC-9; INV-12 codec).
+
+    Read-only: builds the snapshot, then streams the serializer over it. TEL-3
+    is emitted once the stream is exhausted, with the product count and elapsed
+    time. Raises EmptyCatalog before yielding anything if the (filtered) catalog
+    is empty, so the BFF can answer 409 cleanly with no partial body.
+    """
+    started = time.monotonic()
+    snapshot = repo.export_catalog(conn, storefront_id, status_filter)
+    if not snapshot:
+        raise EmptyCatalog()
+
+    def _stream() -> Iterator[str]:
+        yield from serialize.catalog_to_csv(snapshot, base_url=config.public_base_url())
+        telemetry.emit(
+            "catalog_exported",
+            storefront_id=storefront_id,
+            status_filter=status_filter,
+            product_count=len(snapshot),
+            duration_ms=int((time.monotonic() - started) * 1000),
+        )
+
+    return _stream()
+
+
 def summary(conn: psycopg.Connection, storefront_id: int) -> dict:
     """The products dashboard counts (§6.4)."""
     return {

backend/app/main.py

@@ -11,19 +11,22 @@ 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.responses import JSONResponse, PlainTextResponse
+from fastapi import Path as ApiPath
+from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 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}")
@@ -374,6 +390,57 @@ def create_app(database_url: str | None = None, static_dir: str | Path | None =
         _account, sf = gate
         return products.summary(conn, sf.id)
 
+    @app.get("/api/products/export")
+    def export_products(
+        status: str = Query(default="all", pattern="^(all|active|draft|archived)$"),
+        conn: psycopg.Connection = Depends(get_conn),
+        sess: dict | None = Depends(get_session),
+    ):
+        """Stream the catalog as canonical CSV, optionally status-filtered (§6.4; PUC-9)."""
+        gate = _merchant_gate(conn, sess)
+        if isinstance(gate, JSONResponse):
+            return gate
+        _account, sf = gate
+        try:
+            stream = products.export_catalog(conn, sf.id, status)
+        except products.EmptyCatalog:
+            return _error(409, "empty_catalog", "There are no products to export.")
+        return StreamingResponse(
+            stream,
+            media_type="text/csv",
+            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_endpoints.py

@@ -93,3 +93,43 @@ def test_sample_csv_imports_clean(fresh_db_url):
         sample = client.get("/api/products/sample.csv").content
         body = _upload(client, sample, "sample.csv").json()
         assert body["summary"]["errors"] == 0 and body["summary"]["adds"] == 2
+
+
+def test_export_returns_canonical_csv(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        draft = _upload(client).json()
+        client.post(f"/api/products/imports/drafts/{draft['id']}/confirm")
+        resp = client.get("/api/products/export")
+        assert resp.status_code == 200
+        assert resp.headers["content-type"].startswith("text/csv")
+        assert "attachment" in resp.headers["content-disposition"]
+        body = resp.text
+        assert body.splitlines()[0].startswith("Handle,")
+        assert "moon-mug" in body
+
+
+def test_export_status_filter_respected(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        # GOOD_CSV's moon-mug has no Status column → defaults to active.
+        draft = _upload(client).json()
+        client.post(f"/api/products/imports/drafts/{draft['id']}/confirm")
+        assert "moon-mug" in client.get("/api/products/export?status=active").text
+        # No archived products → 409.
+        assert client.get("/api/products/export?status=archived").status_code == 409
+
+
+def test_export_empty_catalog_409(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        resp = client.get("/api/products/export")
+        assert resp.status_code == 409
+        assert resp.json()["error"]["code"] == "empty_catalog"
+
+
+def test_export_requires_merchant(fresh_db_url):
+    with TestClient(create_app(database_url=fresh_db_url)) as client:
+        assert client.get("/api/products/export").status_code == 401
+
+
+def test_export_bad_status_422(fresh_db_url):
+    with _merchant_client(fresh_db_url) as client:
+        assert client.get("/api/products/export?status=bogus").status_code == 422

backend/tests/test_products_export.py

@@ -0,0 +1,100 @@
+"""Export: status-filtered catalog snapshot + the streamed service (PUC-9, TEL-3)."""
+import csv
+import io
+import json
+import logging
+
+import psycopg
+import pytest
+
+from app.domains import products
+from app.domains.products import repo
+from app.platform import db
+
+CSV = (
+    b"Handle,Title,Vendor,Status,Variant Price\n"
+    b"active-mug,Active Mug,Acme,active,18.00\n"
+    b"draft-tee,Draft Tee,Acme,draft,24.00\n"
+)
+
+
+@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 _seed(conn, merchant):
+    draft = products.import_validate(conn, merchant["storefront_id"], merchant["account_id"], "c.csv", CSV)
+    products.confirm_draft(conn, merchant["storefront_id"], merchant["account_id"], draft["id"])
+
+
+def test_export_all_returns_both(migrated_conn, merchant):
+    _seed(migrated_conn, merchant)
+    snap = repo.export_catalog(migrated_conn, merchant["storefront_id"], "all")
+    assert {p.handle for p in snap} == {"active-mug", "draft-tee"}
+
+
+def test_export_status_filter(migrated_conn, merchant):
+    _seed(migrated_conn, merchant)
+    active = repo.export_catalog(migrated_conn, merchant["storefront_id"], "active")
+    assert [p.handle for p in active] == ["active-mug"]
+    draft = repo.export_catalog(migrated_conn, merchant["storefront_id"], "draft")
+    assert [p.handle for p in draft] == ["draft-tee"]
+
+
+@pytest.fixture()
+def telemetry_propagation():
+    """create_app() sets propagate=False on the parent "ecomm" logger
+    (main._ensure_app_logging), which hides ecomm.telemetry records from caplog's
+    root-logger handler whenever an API test ran first. Restore propagation here."""
+    lg = logging.getLogger("ecomm")
+    prior = lg.propagate
+    lg.propagate = True
+    yield
+    lg.propagate = prior
+
+
+def test_export_streams_canonical_csv(migrated_conn, merchant):
+    _seed(migrated_conn, merchant)
+    text = "".join(products.export_catalog(migrated_conn, merchant["storefront_id"], "all"))
+    rows = list(csv.DictReader(io.StringIO(text)))
+    assert {r["Handle"] for r in rows} == {"active-mug", "draft-tee"}
+    assert rows[0]["Handle"] == "active-mug"  # sorted by handle
+
+
+def test_export_empty_raises(migrated_conn, merchant):
+    # No catalog at all → empty.
+    with pytest.raises(products.EmptyCatalog):
+        list(products.export_catalog(migrated_conn, merchant["storefront_id"], "all"))
+
+
+def test_export_empty_after_filter_raises(migrated_conn, merchant):
+    _seed(migrated_conn, merchant)  # only active + draft exist
+    with pytest.raises(products.EmptyCatalog):
+        list(products.export_catalog(migrated_conn, merchant["storefront_id"], "archived"))
+
+
+def test_tel3_emitted(migrated_conn, merchant, caplog, telemetry_propagation):
+    _seed(migrated_conn, merchant)
+    with caplog.at_level(logging.INFO, logger="ecomm.telemetry"):
+        list(products.export_catalog(migrated_conn, merchant["storefront_id"], "all"))
+    events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"]
+    exported = [e for e in events if e["event"] == "catalog_exported"]
+    assert len(exported) == 1
+    assert exported[0]["product_count"] == 2
+    assert exported[0]["status_filter"] == "all"
+    assert "duration_ms" in exported[0]

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

@@ -0,0 +1,249 @@
+"""Canonical serializer + INV-12 round-trip lock (SD-0002 §6.5.5, §6.8)."""
+import csv
+import io
+
+from app.domains.products import serialize
+from app.domains.products.diff import CatalogImage, CatalogProduct, CatalogVariant
+
+
+def _product(**kw) -> CatalogProduct:
+    """A minimal no-option catalog product (one all-NULL variant)."""
+    base = dict(
+        id=1, handle="moon-mug", title="Moon Mug",
+        option_names=(None, None, None),
+        fields={
+            "title": "Moon Mug", "description_html": None, "vendor": "Acme",
+            "product_type": "standalone", "google_product_category": None,
+            "tags": [], "status": "active", "published": True,
+        },
+        variants=[CatalogVariant(
+            id=1, options=(None, None, None), position=1,
+            fields={"sku": "WG-MUG", "barcode": None, "price": None, "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": None,
+                    "variant_image": None})],
+        images=[],
+    )
+    base.update(kw)
+    return CatalogProduct(**base)
+
+
+def _rows(products) -> list[dict]:
+    text = "".join(serialize.catalog_to_csv(products))
+    return list(csv.DictReader(io.StringIO(text)))
+
+
+def test_header_is_full_canonical_set():
+    text = "".join(serialize.catalog_to_csv([_product()]))
+    header = next(csv.reader(io.StringIO(text)))
+    # Handle + Title first; every known column present exactly once.
+    assert header[0] == "Handle"
+    assert "Title" in header and "Variant SKU" in header and "Image Src" in header
+    assert len(header) == len(set(header))
+
+
+def test_single_no_option_product_one_row():
+    rows = _rows([_product()])
+    assert len(rows) == 1
+    assert rows[0]["Handle"] == "moon-mug"
+    assert rows[0]["Title"] == "Moon Mug"
+    assert rows[0]["Variant SKU"] == "WG-MUG"
+    # A no-option product emits no option values.
+    assert rows[0]["Option1 Value"] == ""
+
+
+def _star_tee() -> CatalogProduct:
+    """A 3-variant, 2-image product (the sample.csv shape)."""
+    return CatalogProduct(
+        id=2, handle="star-tee", title="Star Tee",
+        option_names=("Size", "Color", None),
+        fields={"title": "Star Tee", "description_html": None, "vendor": "Acme",
+                "product_type": "standalone", "google_product_category": None,
+                "tags": ["apparel", "tees"], "status": "active", "published": True},
+        variants=[
+            CatalogVariant(id=10, options=("S", "Indigo", None), position=1,
+                           fields={"sku": "WG-TEE-S", "variant_image": None}),
+            CatalogVariant(id=11, options=("M", "Indigo", None), position=2,
+                           fields={"sku": "WG-TEE-M", "variant_image": None}),
+            CatalogVariant(id=12, options=("L", "Indigo", None), position=3,
+                           fields={"sku": "WG-TEE-L", "variant_image": None}),
+        ],
+        images=[
+            CatalogImage(id=1, source_url="https://x/a.jpg", position=1, alt_text="front"),
+            CatalogImage(id=2, source_url="https://x/b.jpg", position=2, alt_text="back"),
+        ],
+    )
+
+
+def test_multivariant_with_images_interleaves():
+    rows = _rows([_star_tee()])
+    assert len(rows) == 3  # max(3 variants, 2 images)
+    # Product-level fields only on the first row.
+    assert rows[0]["Title"] == "Star Tee" and rows[1]["Title"] == ""
+    assert rows[0]["Tags"] == "apparel, tees" and rows[1]["Tags"] == ""
+    # Option names on row 0; option values on every variant row.
+    assert rows[0]["Option1 Name"] == "Size" and rows[1]["Option1 Name"] == ""
+    assert [r["Option1 Value"] for r in rows] == ["S", "M", "L"]
+    assert [r["Variant SKU"] for r in rows] == ["WG-TEE-S", "WG-TEE-M", "WG-TEE-L"]
+    # Two images on the first two rows; third row has no image.
+    assert [r["Image Src"] for r in rows] == ["https://x/a.jpg", "https://x/b.jpg", ""]
+    assert [r["Image Position"] for r in rows] == ["1", "2", ""]
+
+
+def test_more_images_than_variants_emits_image_only_rows():
+    p = _product(images=[
+        CatalogImage(id=1, source_url="https://x/a.jpg", position=1, alt_text=None),
+        CatalogImage(id=2, source_url="https://x/b.jpg", position=2, alt_text=None),
+        CatalogImage(id=3, source_url="https://x/c.jpg", position=3, alt_text=None),
+    ])
+    rows = _rows([p])
+    assert len(rows) == 3  # 1 variant, 3 images
+    assert rows[0]["Variant SKU"] == "WG-MUG"
+    assert rows[1]["Variant SKU"] == "" and rows[1]["Handle"] == "moon-mug"
+    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
+
+
+def _gen_catalog(seed: int) -> dict:
+    """A deterministic random text-field catalog, in load_catalog()'s shape."""
+    rng = random.Random(seed)
+    words = ["moon", "star", "river", "cloud", "ember", "fern", "slate", "wave"]
+    catalog: dict[str, CatalogProduct] = {}
+    n_products = rng.randint(1, 6)
+    for pid in range(1, n_products + 1):
+        handle = "-".join(rng.sample(words, rng.randint(1, 3))) + f"-{pid}"
+        if handle in catalog:
+            continue
+        has_opts = rng.random() < 0.6
+        option_names = ("Size", "Color", None) if has_opts else (None, None, None)
+        tags = rng.sample(["apparel", "kitchen", "sale", "new"], rng.randint(0, 3))
+        fields = {
+            "title": f"{handle.title()} Thing",
+            "description_html": rng.choice([None, "<p>hi</p>"]),
+            "vendor": rng.choice([None, "Acme", "Wiggle Goods"]),
+            "product_type": "standalone",
+            "google_product_category": rng.choice([None, "Home & Garden"]),
+            "tags": tags,
+            "status": rng.choice(["draft", "active", "archived"]),
+            "published": rng.choice([True, False]),
+        }
+        variants = []
+        if has_opts:
+            sizes = rng.sample(["S", "M", "L", "XL"], rng.randint(1, 4))
+            for vi, size in enumerate(sizes, start=1):
+                # Non-sequential positions (×10) so a serializer that drops the
+                # stored position and lets re-import default to file order is
+                # caught — a merchant can import explicit positions like 10/20.
+                variants.append(CatalogVariant(
+                    id=pid * 100 + vi, options=(size, "Indigo", None), position=vi * 10,
+                    fields={"sku": f"SKU-{pid}-{vi}", "variant_image": None}))
+        else:
+            variants.append(CatalogVariant(
+                id=pid * 100 + 1, options=(None, None, None), position=rng.choice([1, 5, 9]),
+                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}"]),
+                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, base_url: str = "") -> diff.DiffResult:
+    """export → bytes → import pipeline → diff against the same catalog."""
+    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)
+
+
+def test_inv12_roundtrip_is_noop_over_generated_catalogs():
+    for seed in range(200):
+        catalog = _gen_catalog(seed)
+        # 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}"
+        assert result.summary["unchanged"] == len(catalog), f"seed {seed}"
+
+
+from decimal import Decimal
+
+
+def test_decimal_and_int_variant_fields_roundtrip():
+    catalog = {
+        "priced": CatalogProduct(
+            id=1, handle="priced", title="Priced",
+            option_names=(None, None, None),
+            fields={"title": "Priced", "description_html": None, "vendor": None,
+                    "product_type": "standalone", "google_product_category": None,
+                    "tags": [], "status": "active", "published": True},
+            variants=[CatalogVariant(
+                id=1, options=(None, None, None), position=1,
+                fields={"sku": "P1", "price": Decimal("18.00"),
+                        "cost": Decimal("9.5"), "weight": Decimal("0.250"),
+                        "inventory_qty": 40, "variant_image": None})],
+            images=[],
+        )
+    }
+    result = _roundtrip_diff(catalog)
+    assert result.summary["unchanged"] == 1, result.records
+    assert result.summary["updates"] == 0
+
+
+def test_nonsequential_variant_position_roundtrips():
+    """A stored variant position that isn't its file order must survive export —
+    else re-import reads the empty cell as 'reset to file order' and the round-trip
+    spuriously updates (INV-12 regression: position is a CatalogVariant attribute,
+    not a fields{} entry, so the serializer must emit it explicitly)."""
+    catalog = {
+        "tee": CatalogProduct(
+            id=1, handle="tee", title="Tee", option_names=("Size", None, None),
+            fields={"title": "Tee", "description_html": None, "vendor": None,
+                    "product_type": "standalone", "google_product_category": None,
+                    "tags": [], "status": "active", "published": True},
+            variants=[
+                CatalogVariant(id=1, options=("S", None, None), position=10,
+                               fields={"sku": "T-S", "variant_image": None}),
+                CatalogVariant(id=2, options=("M", None, None), position=20,
+                               fields={"sku": "T-M", "variant_image": None}),
+            ],
+            images=[],
+        )
+    }
+    result = _roundtrip_diff(catalog)
+    assert result.summary["unchanged"] == 1, result.records
+    assert result.summary["updates"] == 0

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

@@ -41,7 +41,114 @@ never file names, URLs, catalog content, or secret bytes.
 | --- | --- | --- |
 | TEL-1 `import_draft_created` | validation completes, draft stored | `storefront_id, dialect, row_count, adds, updates, unchanged, errors, unknown_columns_count, duration_ms` |
 | 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)
+
+`GET /api/products/export?status=all|active|draft|archived` streams the
+storefront's catalog as a canonical-format CSV (one codec, two directions — the
+same format the importer parses). It is **read-only** (no draft, no run) and
+storefront-scoped (INV-14). An empty catalog — no products, or none matching the
+status filter — returns `409 empty_catalog`; the Products page disables the
+Export action with a note in that case. The round-trip is lossless (INV-12):
+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
 
@@ -94,8 +201,9 @@ gcloud beta monitoring channels list
 
 ### E2E browser suite
 
-- Lives at `e2e/` — Playwright, Chromium, four SLICE-5 scenarios
-  (preview/confirm happy path, actionable errors, file rejection, cancel).
+- Lives at `e2e/` — Playwright, Chromium, six scenarios (SLICE-5:
+  preview/confirm happy path, actionable errors, file rejection, cancel;
+  SLICE-6: `e2e_export_download`, `e2e_roundtrip_noop`).
 - Run with `bash scripts/e2e.sh`. The harness boots a **fresh `ecomm_e2e`
   database** against the local compose Postgres and serves the built SPA from
   the backend on **:8765** (the deployed topology), so it needs the dev

docs/products-domain.md

@@ -13,6 +13,8 @@ the spec is SD-0002 in the content repo.
 - `codec.py` — bytes → `ParsedFile`; file-level gates only.
 - `validate.py` — rows → `CanonicalProduct` blocks + per-row errors.
 - `diff.py` — catalog × canonical products → apply plan + preview records.
+- `serialize.py` — the export half: `CatalogProduct` snapshot → canonical CSV
+  (the inverse of `codec`/`validate`). DB-free, like `diff.py`.
 - `repo.py` — SQL only: catalog snapshot, draft/run CRUD, apply primitives.
   Never commits or rolls back.
 - `service.py` — use-case orchestration; owns every transaction boundary and
@@ -85,20 +87,132 @@ mismatch at confirm means the catalog drifted since preview → `PreviewStale`
 product/variant/image writes, error rows, draft delete — is one transaction;
 any exception rolls it back and emits TEL-6.
 
+## Export & the round-trip (SLICE-6)
+
+`serialize.py` is the export half of "one codec, two directions": it turns the
+`CatalogProduct` snapshot (the same one `repo.load_catalog` builds for the diff
+engine) back into canonical CSV, writing exactly the columns `codec.py` /
+`validate.py` parse, in the §6.5.1 row grammar — `HEADER` is the full canonical
+column set; product fields + option names sit on the first row; one variant per
+row; images interleave; a product with more images than variants emits
+image-only rows.
+
+`repo.export_catalog` returns the status-filtered snapshot list (sorted by
+handle, deterministic); the `service.export_catalog` generator streams
+`serialize.catalog_to_csv` over it and emits TEL-3 (`catalog_exported`) once the
+stream is exhausted. The BFF wraps it in a `StreamingResponse`; an empty
+(filtered) catalog raises `EmptyCatalog` **eagerly** → `409 empty_catalog`
+before any bytes stream.
+
+**INV-12** (`diff(catalog, import(export(catalog))) = ∅`) is locked two ways: a
+property test (`test_products_serialize.py`) runs the real
+export→parse→diff loop over 200 generated **text-field** catalogs and asserts
+every product is `unchanged`; the `e2e_roundtrip_noop` browser scenario does the
+same through the UI (export download → re-upload → all-unchanged preview, import
+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
 
@@ -107,6 +221,8 @@ any exception rolls it back and emits TEL-6.
 | `backend/tests/test_products_codec.py` | file-level gates: parse, caps (INV-18), required columns, dialect |
 | `backend/tests/test_products_validate.py` | every §6.5.1 row-error rule, one fixture each |
 | `backend/tests/test_products_diff.py` | classification, blank-vs-absent, option matching, fingerprint |
+| `backend/tests/test_products_serialize.py` | serializer grammar + INV-12 property test (round-trip no-op over generated catalogs) + decimal round-trip |
+| `backend/tests/test_products_export.py` | status-filtered snapshot, streamed export, EmptyCatalog, TEL-3 |
 | `backend/tests/test_products_service.py` | draft lifecycle: validate/preview/discard, expiry, TEL-1 |
 | `backend/tests/test_products_invariants.py` | INV-10 (never deletes), INV-14 (storefront isolation), apply transactionality, TEL-6 |
 | `backend/tests/test_products_endpoints.py` | §6.4 API scenarios + auth/storefront gates |
@@ -114,3 +230,5 @@ any exception rolls it back and emits TEL-6.
 | `e2e/tests/import-errors.spec.ts` | actionable row errors at preview and on the run report |
 | `e2e/tests/import-file-rejected.spec.ts` | file-level rejection, picker stays live, no trace |
 | `e2e/tests/import-cancel.spec.ts` | cancel at preview leaves no trace |
+| `e2e/tests/export-download.spec.ts` | export downloads canonical CSV, status filter respected |
+| `e2e/tests/roundtrip-noop.spec.ts` | export → re-import → all-unchanged, import disabled (PUC-10) |

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

@@ -2,7 +2,8 @@
 // navigation (SD-0002 §5.2). Selectors are role/label-based against the real screens
 // (Landing.tsx, SignIn.tsx, CreateStorefront.tsx, Admin.tsx, ProductsPage.tsx).
 import { expect, type Page } from "@playwright/test";
-import { readFile } from "node:fs/promises";
+import { readFile, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
 import { join } from "node:path";
 
 const LOG = join(__dirname, ".backend.log");
@@ -72,3 +73,48 @@ export async function uploadFixture(page: Page, fixture: string) {
   await expect(page.getByRole("heading", { name: "Import products" })).toBeVisible();
   await page.locator('input[type="file"]').setInputFiles(join(__dirname, "fixtures", fixture));
 }
+
+// Import good.csv and confirm it, leaving a 2-product catalog. Returns nothing;
+// callers continue from the run-detail screen.
+export async function importGoodCsv(page: Page) {
+  await uploadFixture(page, "good.csv");
+  await expect(page.getByRole("heading", { name: "Import preview — good.csv" })).toBeVisible();
+  await page.getByRole("button", { name: "Import 2 products" }).click();
+  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.
+  // The menu is a native <details> that *toggles* on each summary click and is
+  // NOT closed by clicking a download <a> (a download doesn't navigate). So on a
+  // second export the menu may already be open — deterministically open it
+  // rather than blind-toggling, and wait for the status link to be visible.
+  const details = page.locator("details.products__export");
+  if (!(await details.evaluate((el: HTMLDetailsElement) => el.open))) {
+    await details.locator("> summary").click();
+  }
+  const link = page.getByRole("link", { name: label, exact: true });
+  await expect(link).toBeVisible();
+  const [download] = await Promise.all([
+    page.waitForEvent("download"),
+    link.click(),
+  ]);
+  const stream = await download.createReadStream();
+  const chunks: Buffer[] = [];
+  for await (const c of stream) chunks.push(c as Buffer);
+  const text = Buffer.concat(chunks).toString("utf8");
+  const path = join(tmpdir(), `export-${Date.now()}.csv`);
+  await writeFile(path, text, "utf8");
+  return { text, path };
+}

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/export-download.spec.ts

@@ -0,0 +1,21 @@
+// DoD scenario e2e_export_download (SD-0002 §6.8): export downloads a canonical
+// CSV and the status filter is respected.
+import { expect, test } from "@playwright/test";
+import { exportCatalog, gotoProducts, importGoodCsv, signUpWithStorefront } from "../helpers";
+
+test("e2e_export_download", async ({ page }) => {
+  await signUpWithStorefront(page);
+  await gotoProducts(page);
+  await importGoodCsv(page);
+  await gotoProducts(page);
+
+  // Export "All products" → a canonical CSV with both handles.
+  const all = await exportCatalog(page, "All products");
+  expect(all.text.split("\n")[0]).toContain("Handle,");
+  expect(all.text).toContain("moon-mug");
+  expect(all.text).toContain("star-tee");
+
+  // good.csv's products are active → "Active" exports both, "Draft" is empty/disabled-path.
+  const active = await exportCatalog(page, "Active");
+  expect(active.text).toContain("moon-mug");
+});

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();
+});

e2e/tests/roundtrip-noop.spec.ts

@@ -0,0 +1,30 @@
+// DoD scenario e2e_roundtrip_noop (SD-0002 §6.8, PUC-10): exporting then
+// re-importing the unmodified file previews as all-unchanged with the import
+// action disabled and the "Nothing to change" note.
+import { expect, test } from "@playwright/test";
+import { exportCatalog, gotoProducts, importGoodCsv, signUpWithStorefront } from "../helpers";
+
+test("e2e_roundtrip_noop", async ({ page }) => {
+  await signUpWithStorefront(page);
+  await gotoProducts(page);
+  await importGoodCsv(page);
+  await gotoProducts(page);
+
+  // Export the catalog, then re-import the unmodified file.
+  const { path } = await exportCatalog(page, "All products");
+  await page.getByRole("link", { name: "Import products" }).first().click();
+  await expect(page.getByRole("heading", { name: "Import products" })).toBeVisible();
+  await page.locator('input[type="file"]').setInputFiles(path);
+
+  // Preview: everything unchanged, nothing to add/update (PUC-10).
+  await expect(page.getByRole("button", { name: "2 unchanged" })).toBeVisible();
+  await expect(page.getByRole("button", { name: "0 to add" })).toBeVisible();
+  await expect(page.getByRole("button", { name: "0 to update" })).toBeVisible();
+
+  // The import action is disabled, with the no-op note.
+  const importBtn = page.getByRole("button", { name: /^Import 0 products$/ });
+  await expect(importBtn).toBeDisabled();
+  await expect(
+    page.getByText("Nothing to change — your catalog already matches this file"),
+  ).toBeVisible();
+});

frontend/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "wiggleverse-ecomm-frontend",
-  "version": "0.4.0",
+  "version": "0.7.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "wiggleverse-ecomm-frontend",
-      "version": "0.4.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.5.0",
+  "version": "0.7.0",
   "type": "module",
   "scripts": {
     "dev": "vite",

frontend/src/productsApi.export.test.ts

@@ -0,0 +1,12 @@
+import { describe, expect, it } from "vitest";
+import { EXPORT_STATUSES, exportUrl } from "./productsApi";
+
+describe("export url", () => {
+  it("lists the four status filters with 'all' first", () => {
+    expect(EXPORT_STATUSES.map((s) => s.value)).toEqual(["all", "active", "draft", "archived"]);
+  });
+  it("builds the endpoint url with the status query", () => {
+    expect(exportUrl("all")).toBe("/api/products/export?status=all");
+    expect(exportUrl("archived")).toBe("/api/products/export?status=archived");
+  });
+});

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;
@@ -47,6 +53,22 @@ export function dialectLabel(d: string): string {
   return d === "canonical" ? "Canonical format" : d;
 }
 
+export type ExportStatus = "all" | "active" | "draft" | "archived";
+
+// PUC-9 status filter, 'all' first (the default). Labels drive the export menu.
+export const EXPORT_STATUSES: { value: ExportStatus; label: string }[] = [
+  { value: "all", label: "All products" },
+  { value: "active", label: "Active" },
+  { value: "draft", label: "Draft" },
+  { value: "archived", label: "Archived" },
+];
+
+// The export is a browser download (native save dialog + streaming), not a fetch
+// wrapper — so the API module contributes a URL builder, not a request().
+export function exportUrl(status: ExportStatus): string {
+  return `/api/products/export?status=${status}`;
+}
+
 async function request<T>(path: string, init?: RequestInit): Promise<Result<T>> {
   const resp = await fetch(path, { credentials: "include", ...init });
   if (!resp.ok) return { ok: false, error: await errorOf(resp), status: resp.status };

frontend/src/screens/products/ProductsPage.tsx

@@ -1,14 +1,18 @@
-// Products page (SD-0002 §5.2) — the catalog's home: where imports start and history
-// lives. SLICE-5: export is disabled (SLICE-6 ships it); the browsable list is #14's.
+// Products page (SD-0002 §5.2) — the catalog's home: where imports start, exports
+// download, and history lives. SLICE-6 ships the export menu; the browsable list is #14's.
 import { useEffect, useState } from "react";
 import {
   dialectLabel,
+  EXPORT_STATUSES,
+  exportUrl,
   getProductsSummary,
   listRuns,
   type ProductsSummary,
   type RunSummary,
 } from "../../productsApi";
 import { Banner } from "../../ui/kit";
+import { isExportEnabled } from "./exportMenu";
+import { historyImageCell } from "./runImages";
 
 const STATUS_LABELS: Record<string, string> = {
   applying: "Importing…",
@@ -54,7 +58,7 @@ export default function ProductsPage() {
     );
   }
 
-  const empty = summary.product_count === 0;
+  const empty = !isExportEnabled(summary.product_count);
   return (
     <div className="products">
       <header className="products__header">
@@ -63,17 +67,43 @@ export default function ProductsPage() {
           {!empty && <span className="products__count"> · {summary.product_count.toLocaleString()}</span>}
         </h1>
         <div className="products__actions">
+          {empty ? (
             <div className="products__export">
-            <button type="button" className="btn-secondary" disabled title="Export arrives in a coming release">
+              <button type="button" className="btn-secondary" disabled>
                 Export
               </button>
-            <span className="note">Export arrives in a coming release</span>
+              <span className="note">Export arrives when you have products</span>
             </div>
+          ) : (
+            <details className="products__export menu">
+              <summary className="btn-secondary" role="button">
+                Export
+              </summary>
+              <ul className="menu__list">
+                {EXPORT_STATUSES.map((s) => (
+                  <li key={s.value}>
+                    {/* A real download: the browser navigates to the streamed
+                        endpoint and saves the attachment (PUC-9). */}
+                    <a className="menu__item" href={exportUrl(s.value)} download>
+                      {s.label}
+                    </a>
+                  </li>
+                ))}
+              </ul>
+            </details>
+          )}
           <a className="btn-primary" href="#/products/import">
             Import products
           </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>
@@ -97,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>
@@ -120,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/exportMenu.test.ts

@@ -0,0 +1,19 @@
+import { describe, expect, it } from "vitest";
+import { EXPORT_STATUSES, exportUrl } from "../../productsApi";
+import { isExportEnabled } from "./exportMenu";
+
+describe("export menu", () => {
+  it("disables export for an empty catalog and enables it once there are products", () => {
+    expect(isExportEnabled(0)).toBe(false);
+    expect(isExportEnabled(3)).toBe(true);
+  });
+
+  it("builds the four status download URLs, 'all' first", () => {
+    expect(EXPORT_STATUSES.map((s) => exportUrl(s.value))).toEqual([
+      "/api/products/export?status=all",
+      "/api/products/export?status=active",
+      "/api/products/export?status=draft",
+      "/api/products/export?status=archived",
+    ]);
+  });
+});

frontend/src/screens/products/exportMenu.ts

@@ -0,0 +1,11 @@
+// Pure logic behind the Products page Export menu (SD-0002 §5.2, PUC-9). The
+// status list + URL builder live in productsApi.ts (re-used by the component);
+// this is the one decision the menu turns on — whether export is offered at all.
+// The disclosure JSX is verified by the E2E suite, matching SLICE-2/3's pattern
+// of pure-logic unit tests + E2E for screens.
+
+// Export is offered only when the catalog has products (PUC-9: an empty catalog
+// shows a disabled button + note instead).
+export function isExportEnabled(count: number): boolean {
+  return count > 0;
+}

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;
+}

frontend/src/styles/products.css

@@ -53,9 +53,52 @@
 }
 .products__count { color: var(--text-on-dark-mute); font-weight: var(--weight-medium); }
 .products__actions { display: flex; gap: 12px; align-items: center; }
-/* Disabled Export + its visible "coming release" caption, stacked. */
+/* Disabled Export + its visible "no products yet" caption, stacked (empty catalog). */
 .products__export { display: flex; flex-direction: column; gap: 4px; align-items: center; }
 .products__export .note { font-size: 11.5px; }
+
+/* Export status menu (SD-0002 §5.2 — PUC-9). A native <details> disclosure so
+   it's keyboard-accessible with no extra JS (§6.6). Tokens align with the design
+   bundle (--surface-raised / --border-card / --radius-panel); the fallbacks keep
+   it working regardless. */
+.products__export.menu {
+  position: relative;
+  display: inline-block;
+}
+.products__export.menu > summary {
+  list-style: none;
+  cursor: pointer;
+}
+.products__export.menu > summary::-webkit-details-marker {
+  display: none;
+}
+.menu__list {
+  position: absolute;
+  right: 0;
+  z-index: 10;
+  margin: 0.25rem 0 0;
+  padding: 0.25rem;
+  list-style: none;
+  background: var(--surface-raised, #fff);
+  border: 1px solid var(--border-card, #d8d3c8);
+  border-radius: var(--radius-panel, 8px);
+  box-shadow: var(--shadow-soft, 0 6px 20px rgba(0, 0, 0, 0.12));
+  min-width: 10rem;
+}
+.menu__item {
+  display: block;
+  padding: 0.5rem 0.75rem;
+  border-radius: var(--radius-sm, 6px);
+  text-decoration: none;
+  color: var(--text-on-dark-soft, inherit);
+  font-family: var(--wv-font-display);
+  font-size: 14px;
+}
+.menu__item:hover,
+.menu__item:focus {
+  background: var(--surface-raised-hi, #f3efe7);
+  color: var(--wv-starlight);
+}
 .products .btn-primary { width: auto; text-decoration: none; }
 .products .empty { margin: 24px auto 0; }