VERSION

@@ -1 +1 @@
-0.5.0
+0.6.0

backend/app/domains/products/__init__.py

@@ -11,6 +11,7 @@ from pathlib import Path
 from .errors import (
     DraftExpired,
     DraftNotFound,
+    EmptyCatalog,
     FileRejected,
     NothingToApply,
     PreviewStale,
@@ -21,6 +22,7 @@ from .models import MAX_DATA_ROWS, MAX_FILE_BYTES
 from .service import (
     confirm_draft,
     discard_draft,
+    export_catalog,
     get_draft,
     get_draft_records,
     get_run,
@@ -34,8 +36,8 @@ 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",
 ]

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/repo.py

@@ -104,6 +104,23 @@ def load_catalog(conn: psycopg.Connection, storefront_id: int) -> dict[str, Cata
     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,)

backend/app/domains/products/serialize.py

@@ -0,0 +1,119 @@
+"""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 .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]) -> 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):
+            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) -> 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])
+        rows.append(row)
+    return rows
+
+
+def _write_image(row: dict[str, str], image) -> None:
+    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 . 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,
@@ -218,6 +226,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)
+        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

@@ -17,7 +17,7 @@ from typing import Any
 
 import psycopg
 from fastapi import Depends, FastAPI, File, Query, Response, UploadFile
-from fastapi.responses import JSONResponse, PlainTextResponse
+from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel
 
@@ -374,6 +374,27 @@ 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"'},
+        )
+
     @app.get("/api/products/sample.csv")
     def products_sample_csv():
         """The DOC-3 worked-example CSV. Documentation, so no auth gate (§6.4)."""

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_serialize.py

@@ -0,0 +1,224 @@
+"""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"]
+
+
+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)):
+            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}"])))
+        catalog[handle] = CatalogProduct(
+            id=pid, handle=handle, title=fields["title"], option_names=option_names,
+            fields=fields, variants=variants, images=images)
+    return catalog
+
+
+def _roundtrip_diff(catalog: dict) -> diff.DiffResult:
+    """export → bytes → import pipeline → diff against the same catalog."""
+    text = "".join(serialize.catalog_to_csv(catalog.values()))
+    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)
+        result = _roundtrip_diff(catalog)
+        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

docs/OPERATIONS.md

@@ -41,8 +41,21 @@ 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` |
 
+### 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.
+
 ### RB-2 — import apply failed
 
 Triggered by ALR-2 (any TEL-6 event). The apply raised mid-transaction and
@@ -94,8 +107,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,6 +87,31 @@ 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.
+
 ## Named seams (what later slices replace)
 
 - **`import_draft.file_bytes` → objectstore key (SLICE-7).** The upload
@@ -107,6 +134,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 +143,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/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,38 @@ 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();
+}
+
+// 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/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/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.6.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "wiggleverse-ecomm-frontend",
-      "version": "0.4.0",
+      "version": "0.6.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.6.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

@@ -47,6 +47,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,17 @@
-// 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";
 
 const STATUS_LABELS: Record<string, string> = {
   applying: "Importing…",
@@ -54,7 +57,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,12 +66,31 @@ 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>

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