backend/app/domains/products/__init__.py

@@ -16,11 +16,21 @@ from .errors import (
     RunNotFound,
 )
 from .models import MAX_DATA_ROWS, MAX_FILE_BYTES
-from .service import discard_draft, get_draft, get_draft_records, import_validate
+from .service import (
+    confirm_draft,
+    discard_draft,
+    get_draft,
+    get_draft_records,
+    get_run,
+    import_validate,
+    list_runs,
+    summary,
+)
 
 __all__ = [
     "ProductsError", "FileRejected", "DraftNotFound", "DraftExpired",
     "PreviewStale", "NothingToApply", "RunNotFound",
     "MAX_DATA_ROWS", "MAX_FILE_BYTES",
     "import_validate", "get_draft", "get_draft_records", "discard_draft",
+    "confirm_draft", "list_runs", "get_run", "summary",
 ]

backend/app/domains/products/diff.py

@@ -1,9 +1,12 @@
-"""Diff engine — catalog × canonical products → preview records (SD-0002 §6.5.2).
+"""Diff engine — catalog × canonical products → apply plan + preview records (SD-0002 §6.5.2).
 
 Classifies each canonical product against the storefront's current catalog as
-add / update / unchanged / error, producing JSON-ready preview records (stored
-as draft JSONB, served verbatim to the SPA) plus a summary and a deterministic
-fingerprint that detects catalog drift between preview and confirm (INV-11).
+add / update / unchanged / error. One walk produces two views of the same
+computation: a typed apply *plan* carrying resolved native values (Decimal etc.)
+for the confirm transaction, and JSON-ready preview records derived from that
+walk (stored as draft JSONB, served verbatim to the SPA) — so what confirm
+applies is exactly what preview showed (INV-11). A summary and a deterministic
+fingerprint over the records detect catalog drift between preview and confirm.
 
 Deliberately DB-free: the catalog snapshot dataclasses are defined here and the
 repo layer builds them. Only fields present in the file's canonical fields{}
@@ -16,7 +19,7 @@ from __future__ import annotations
 
 import hashlib
 import json
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from decimal import Decimal
 
 from .models import CLEAR_DEFAULTS, CanonicalProduct, CanonicalVariant
@@ -54,11 +57,49 @@ class CatalogProduct:
     images: list[CatalogImage]
 
 
+# ---------------------------------------------------------------------------
+# Apply plan — the typed twin of the preview records. confirm_draft executes
+# these; the values are resolved natives (Decimal, bool, list), never the
+# json-safe strings the records carry for display.
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class VariantPlan:
+    kind: str  # "add" | "update"
+    canonical: CanonicalVariant
+    catalog_id: int | None  # None for add
+    # field -> resolved after-value (update only); adds resolve from canonical.fields
+    changes: dict[str, object] = field(default_factory=dict)
+
+
+@dataclass
+class ImagePlan:
+    kind: str  # "add" | "update"
+    source_url: str
+    position: int
+    alt_text: str | None
+    image_id: int | None  # None for add
+    changes: dict[str, object] = field(default_factory=dict)  # subset of {"position","alt_text"}
+
+
+@dataclass
+class ProductPlan:
+    kind: str  # "add" | "update" | "unchanged" | "error"
+    canonical: CanonicalProduct
+    catalog: CatalogProduct | None
+    # field -> resolved after-value (update only; may include title/option*_name)
+    product_changes: dict[str, object] = field(default_factory=dict)
+    variant_plans: list[VariantPlan] = field(default_factory=list)
+    image_plans: list[ImagePlan] = field(default_factory=list)
+
+
 @dataclass(frozen=True)
 class DiffResult:
     records: list[dict]
     summary: dict
     fingerprint: str
+    plan: list[ProductPlan]
 
 
 # Option-name presence markers in fields{} (see validate.py); the values are
@@ -69,16 +110,20 @@ _SUMMARY_KEY = {"add": "adds", "update": "updates", "unchanged": "unchanged", "e
 
 def compute_diff(catalog: dict[str, CatalogProduct], products: list[CanonicalProduct]) -> DiffResult:
     records: list[dict] = []
+    plan: list[ProductPlan] = []
     summary = {"adds": 0, "updates": 0, "unchanged": 0, "errors": 0}
     for product in products:
         current = catalog.get(product.handle)
         if product.errors:
-            kind, detail = "error", {"errors": [e.as_json() for e in product.errors]}
+            product_plan = ProductPlan(kind="error", canonical=product, catalog=current)
+            detail: dict = {"errors": [e.as_json() for e in product.errors]}
         elif current is None:
-            kind, detail = "add", _add_detail(product)
+            product_plan = _add_plan(product)
+            detail = _add_detail(product_plan)
         else:
-            detail = _update_detail(product, current)
-            kind = "update" if detail else "unchanged"
+            product_plan, detail = _update_plan(product, current)
+        kind = product_plan.kind
+        plan.append(product_plan)
         records.append(
             {
                 "handle": product.handle,
@@ -92,86 +137,125 @@ def compute_diff(catalog: dict[str, CatalogProduct], products: list[CanonicalPro
     fingerprint = hashlib.sha256(
         json.dumps(records, sort_keys=True, separators=(",", ":")).encode()
     ).hexdigest()
-    return DiffResult(records=records, summary=summary, fingerprint=fingerprint)
+    return DiffResult(records=records, summary=summary, fingerprint=fingerprint, plan=plan)
 
 
-def _add_detail(product: CanonicalProduct) -> dict:
+def _add_plan(product: CanonicalProduct) -> ProductPlan:
+    return ProductPlan(
+        kind="add",
+        canonical=product,
+        catalog=None,
+        variant_plans=[VariantPlan(kind="add", canonical=v, catalog_id=None) for v in product.variants],
+        image_plans=[
+            ImagePlan(kind="add", source_url=i.source_url, position=i.position,
+                      alt_text=i.alt_text, image_id=None)
+            for i in product.images
+        ],
+    )
+
+
+def _add_detail(plan: ProductPlan) -> dict:
     # On add, absent fields fall back to their defaults for display where one
     # exists — the detail shows what will actually be set.
-    set_fields = _resolved_product_fields(product)
+    set_fields = resolved_product_fields(plan.canonical)
     for field_name, default in CLEAR_DEFAULTS.items():
         set_fields.setdefault(field_name, default)
     return {
         "set": {f: _json_safe(v) for f, v in set_fields.items()},
-        "option_names": list(product.option_names),
+        "option_names": list(plan.canonical.option_names),
         "variants": [
-            {"options": list(v.options), "set": _resolved_variant_fields(v)}
-            for v in product.variants
+            {"options": list(vp.canonical.options), "set": _resolved_variant_fields(vp.canonical)}
+            for vp in plan.variant_plans
         ],
         "images": [
-            {"src": i.source_url, "position": i.position, "alt_text": i.alt_text}
-            for i in product.images
+            {"src": ip.source_url, "position": ip.position, "alt_text": ip.alt_text}
+            for ip in plan.image_plans
         ],
     }
 
 
-def _update_detail(product: CanonicalProduct, current: CatalogProduct) -> dict:
+def _update_plan(product: CanonicalProduct, current: CatalogProduct) -> tuple[ProductPlan, dict]:
+    """One walk, two outputs: the resolved-value plan entries and the json-safe
+    record detail entries are appended side by side, so they can never diverge."""
+    product_changes: dict[str, object] = {}
     changes: list[dict] = []
     # Title is always file-present (required header column); "" means the block
     # already carries an error and never reaches here.
     if product.title and product.title != current.title:
+        product_changes["title"] = product.title
         changes.append(_change("title", current.title, product.title))
-    for field_name, resolved in _resolved_product_fields(product).items():
+    for field_name, resolved in resolved_product_fields(product).items():
         before = current.fields.get(field_name)
         if resolved != before:
+            product_changes[field_name] = resolved
             changes.append(_change(field_name, before, resolved))
     for slot in (1, 2, 3):
         if f"option{slot}_name" not in product.fields:
             continue
         before, after = current.option_names[slot - 1], product.option_names[slot - 1]
         if after != before:
+            product_changes[f"option{slot}_name"] = after
             changes.append(_change(f"option{slot}_name", before, after))
 
+    variant_plans: list[VariantPlan] = []
     variant_entries: list[dict] = []
     by_options = {v.options: v for v in current.variants}
     for variant in product.variants:
         match = by_options.get(variant.options)
         if match is None:
+            variant_plans.append(VariantPlan(kind="add", canonical=variant, catalog_id=None))
             variant_entries.append(
                 {"options": list(variant.options), "kind": "add", "set": _resolved_variant_fields(variant)}
             )
             continue
-        variant_changes = []
-        for f, resolved in _resolved_fields(variant.fields).items():
+        variant_changes: dict[str, object] = {}
+        variant_change_entries: list[dict] = []
+        for f, resolved in resolved_fields(variant.fields).items():
             # position lives on the catalog variant as an attribute, not in
             # fields{} — compare it explicitly (as images do for theirs).
             before = match.position if f == "position" else match.fields.get(f)
             if resolved != before:
-                variant_changes.append(_change(f, before, resolved))
+                variant_changes[f] = resolved
+                variant_change_entries.append(_change(f, before, resolved))
         if variant_changes:
+            variant_plans.append(
+                VariantPlan(kind="update", canonical=variant, catalog_id=match.id, changes=variant_changes)
+            )
             variant_entries.append(
-                {"options": list(variant.options), "kind": "update", "changes": variant_changes}
+                {"options": list(variant.options), "kind": "update", "changes": variant_change_entries}
             )
 
+    image_plans: list[ImagePlan] = []
     image_entries: list[dict] = []
     by_src = {i.source_url: i for i in current.images}
     for image in product.images:
         match = by_src.get(image.source_url)
         if match is None:
+            image_plans.append(
+                ImagePlan(kind="add", source_url=image.source_url, position=image.position,
+                          alt_text=image.alt_text, image_id=None)
+            )
             image_entries.append(
                 {"src": image.source_url, "kind": "add", "position": image.position, "alt_text": image.alt_text}
             )
             continue
-        image_changes = [
-            _change(f, before, after)
-            for f, before, after in (
-                ("position", match.position, image.position),
-                ("alt_text", match.alt_text, image.alt_text),
-            )
-            if after != before
-        ]
+        image_changes: dict[str, object] = {}
+        image_change_entries: list[dict] = []
+        for f, before, after in (
+            ("position", match.position, image.position),
+            ("alt_text", match.alt_text, image.alt_text),
+        ):
+            if after != before:
+                image_changes[f] = after
+                image_change_entries.append(_change(f, before, after))
         if image_changes:
-            image_entries.append({"src": image.source_url, "kind": "update", "changes": image_changes})
+            image_plans.append(
+                ImagePlan(kind="update", source_url=image.source_url, position=image.position,
+                          alt_text=image.alt_text, image_id=match.id, changes=image_changes)
+            )
+            image_entries.append(
+                {"src": image.source_url, "kind": "update", "changes": image_change_entries}
+            )
 
     detail: dict = {}
     if changes:
@@ -180,22 +264,28 @@ def _update_detail(product: CanonicalProduct, current: CatalogProduct) -> dict:
         detail["variants"] = variant_entries
     if image_entries:
         detail["images"] = image_entries
-    return detail
+    kind = "update" if detail else "unchanged"
+    return (
+        ProductPlan(kind=kind, canonical=product, catalog=current, product_changes=product_changes,
+                    variant_plans=variant_plans, image_plans=image_plans),
+        detail,
+    )
 
 
-def _resolved_fields(fields: dict[str, object]) -> dict[str, object]:
+def resolved_fields(fields: dict[str, object]) -> dict[str, object]:
     """File-present fields with None (an explicit clear) resolved to the default."""
     return {f: (v if v is not None else CLEAR_DEFAULTS.get(f)) for f, v in fields.items()}
 
 
-def _resolved_product_fields(product: CanonicalProduct) -> dict[str, object]:
+def resolved_product_fields(product: CanonicalProduct) -> dict[str, object]:
+    """Resolved product-level fields, minus the option-name presence markers."""
     return {
-        f: v for f, v in _resolved_fields(product.fields).items() if f not in _OPTION_NAME_FIELDS
+        f: v for f, v in resolved_fields(product.fields).items() if f not in _OPTION_NAME_FIELDS
     }
 
 
 def _resolved_variant_fields(variant: CanonicalVariant) -> dict[str, object]:
-    return {f: _json_safe(v) for f, v in _resolved_fields(variant.fields).items()}
+    return {f: _json_safe(v) for f, v in resolved_fields(variant.fields).items()}
 
 
 def _change(field_name: str, before: object, after: object) -> dict:

backend/app/domains/products/service.py

@@ -14,7 +14,7 @@ import psycopg
 from app.platform import telemetry
 
 from . import codec, diff, repo, validate
-from .errors import DraftExpired, DraftNotFound
+from .errors import DraftExpired, DraftNotFound, NothingToApply, PreviewStale, RunNotFound
 
 
 def import_validate(conn: psycopg.Connection, storefront_id: int, account_id: int,
@@ -86,3 +86,138 @@ def discard_draft(conn: psycopg.Connection, storefront_id: int, draft_id: int) -
     """Delete the draft, no trace kept; idempotent — an absent draft is fine (PUC-3a)."""
     repo.delete_draft(conn, storefront_id, draft_id)
     conn.commit()
+
+
+def confirm_draft(conn: psycopg.Connection, storefront_id: int, account_id: int,
+                  draft_id: int) -> int:
+    """Apply the previewed diff in one transaction (PUC-4; INV-10/11).
+
+    Everything is re-derived from the draft's stored file bytes against the live
+    catalog; a fingerprint mismatch means the catalog drifted since preview
+    (PreviewStale — the draft is kept so the merchant can re-validate). The apply
+    executes the typed plan compute_diff built alongside the preview records, so
+    what lands is exactly what the preview showed. rows_errored counts the
+    import_run_error rows recorded (one per RowError), which is what the run
+    detail's error table shows; the preview's errors tile counts error *products*.
+    """
+    started = time.monotonic()
+    row = _live_draft_row(conn, storefront_id, draft_id)
+    parsed = codec.parse_csv(row["file_bytes"])
+    products = validate.build_products(parsed)
+    catalog = repo.load_catalog(conn, storefront_id)
+    diff_result = diff.compute_diff(catalog, products)
+    if diff_result.fingerprint != row["fingerprint"]:
+        raise PreviewStale()
+    summary_counts = diff_result.summary
+    if summary_counts["adds"] + summary_counts["updates"] == 0:
+        raise NothingToApply()
+    error_rows = [
+        error.as_json()
+        for plan in diff_result.plan if plan.kind == "error"
+        for error in plan.canonical.errors
+    ]
+    try:
+        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",
+        )
+        for plan in diff_result.plan:
+            _apply_product_plan(conn, storefront_id, plan, run_id)
+        repo.insert_run_errors(conn, run_id, error_rows)
+        repo.delete_draft(conn, storefront_id, draft_id)
+        conn.commit()
+    except Exception as exc:
+        conn.rollback()
+        telemetry.emit(
+            "import_apply_failed",
+            draft_id=draft_id,
+            storefront_id=storefront_id,
+            error_class=type(exc).__name__,
+        )
+        raise
+    telemetry.emit(
+        "import_run_completed",
+        run_id=run_id,
+        storefront_id=storefront_id,
+        added=summary_counts["adds"],
+        updated=summary_counts["updates"],
+        errored=len(error_rows),
+        duration_ms=int((time.monotonic() - started) * 1000),
+    )
+    return run_id
+
+
+def _apply_product_plan(conn: psycopg.Connection, storefront_id: int,
+                        plan: diff.ProductPlan, run_id: int) -> None:
+    """Execute one product's plan inside the confirm transaction (no commits here)."""
+    if plan.kind == "add":
+        # Title is a canonical attribute, not a fields{} entry — non-error
+        # products always carry one (validate guarantees it).
+        product_fields = {"title": plan.canonical.title}
+        product_fields.update(diff.resolved_product_fields(plan.canonical))
+        product_id = repo.insert_product(
+            conn, storefront_id, plan.canonical.handle, product_fields, plan.canonical.option_names
+        )
+        image_ids: dict[str, int] = {}
+    elif plan.kind == "update":
+        product_id = plan.catalog.id
+        repo.update_product(conn, product_id, plan.product_changes)
+        image_ids = {image.source_url: image.id for image in plan.catalog.images}
+    else:
+        return
+
+    # Images first, so variants' variant_image URLs resolve to ids: validate puts
+    # every variant_image URL into canonical.images, so each URL is in either the
+    # catalog map (existing image) or the adds below.
+    for image_plan in plan.image_plans:
+        if image_plan.kind == "add":
+            image_ids[image_plan.source_url] = repo.get_or_create_image(
+                conn, product_id, image_plan.source_url, image_plan.position,
+                image_plan.alt_text, run_id,
+            )
+        else:
+            repo.update_image(conn, image_plan.image_id, image_plan.changes)
+
+    # File order within the product, for variant adds without an explicit position.
+    file_positions = {id(v): i for i, v in enumerate(plan.canonical.variants, start=1)}
+    for variant_plan in plan.variant_plans:
+        if variant_plan.kind == "add":
+            fields = diff.resolved_fields(variant_plan.canonical.fields)
+            position = fields.get("position") or file_positions[id(variant_plan.canonical)]
+            url = fields.get("variant_image")
+            image_id = image_ids[url] if url else None
+            repo.insert_variant(
+                conn, product_id, position, variant_plan.canonical.options, fields, image_id
+            )
+        elif "variant_image" in variant_plan.changes:
+            url = variant_plan.changes["variant_image"]
+            repo.update_variant(
+                conn, variant_plan.catalog_id, variant_plan.changes,
+                image_id=image_ids[url] if url else None,
+            )
+        else:
+            repo.update_variant(conn, variant_plan.catalog_id, variant_plan.changes)
+
+
+def list_runs(conn: psycopg.Connection, storefront_id: int,
+              limit: int = 50, offset: int = 0) -> list[dict]:
+    """The storefront's import history, newest first (PUC-8)."""
+    return repo.list_runs(conn, storefront_id, limit, offset)
+
+
+def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict:
+    """One run's §6.4 detail payload, errors included."""
+    run = repo.get_run(conn, storefront_id, run_id)
+    if run is None:
+        raise RunNotFound()
+    return run
+
+
+def summary(conn: psycopg.Connection, storefront_id: int) -> dict:
+    """The products dashboard counts (§6.4)."""
+    return {
+        "product_count": repo.product_count(conn, storefront_id),
+        "image_problem_count": repo.image_problem_count(conn, storefront_id),
+        "latest_run_id": repo.latest_run_id(conn, storefront_id),
+    }

backend/tests/test_products_invariants.py

@@ -0,0 +1,79 @@
+"""SD-0002 invariants: INV-10 (never deletes), INV-14 (two-storefront zero bleed),
+apply transactionality (§6.8), TEL-6."""
+import json
+import logging
+
+import psycopg
+import pytest
+
+from app.domains import products
+from app.domains.products import repo, service
+from app.platform import db
+
+CSV_A = b"Handle,Title,Variant Price\nmug,Mug,10.00\ntee,Tee,20.00\n"
+CSV_PARTIAL = b"Handle,Title,Variant Price\nmug,Mug,12.00\n"
+
+
+@pytest.fixture()
+def migrated_conn(fresh_db_url):
+    with psycopg.connect(fresh_db_url) as conn:
+        db.migrate(conn)
+        yield conn
+
+
+def _merchant(conn, email="m@example.com", shop="Shop"):
+    acct = conn.execute("INSERT INTO account (email) VALUES (%s) RETURNING id", (email,)).fetchone()[0]
+    sf = conn.execute("INSERT INTO storefront (name) VALUES (%s) RETURNING id", (shop,)).fetchone()[0]
+    conn.execute("INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (acct, sf))
+    conn.commit()
+    return acct, sf
+
+
+def _import(conn, acct, sf, data):
+    d = products.import_validate(conn, sf, acct, "f.csv", data)
+    return products.confirm_draft(conn, sf, acct, d["id"])
+
+
+def test_inv10_partial_file_never_deletes(migrated_conn):
+    acct, sf = _merchant(migrated_conn)
+    _import(migrated_conn, acct, sf, CSV_A)
+    before = migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0]
+    _import(migrated_conn, acct, sf, CSV_PARTIAL)
+    after = migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0]
+    assert after >= before == 2
+
+
+def test_inv14_two_storefronts_zero_bleed(migrated_conn):
+    acct1, sf1 = _merchant(migrated_conn)
+    acct2, sf2 = _merchant(migrated_conn, "n@example.com", "Other")
+    _import(migrated_conn, acct1, sf1, CSV_A)
+    assert products.summary(migrated_conn, sf2)["product_count"] == 0
+    assert products.list_runs(migrated_conn, sf2) == []
+    _import(migrated_conn, acct2, sf2, CSV_A)
+    assert products.summary(migrated_conn, sf2)["product_count"] == 2
+    run1 = products.list_runs(migrated_conn, sf1)[0]
+    with pytest.raises(products.RunNotFound):
+        products.get_run(migrated_conn, sf2, run1["id"])
+
+
+def test_apply_failure_rolls_back_whole_transaction_tel6(migrated_conn, monkeypatch, caplog):
+    acct, sf = _merchant(migrated_conn)
+    d = products.import_validate(migrated_conn, sf, acct, "f.csv", CSV_A)
+
+    def boom(*a, **k):
+        raise RuntimeError("mid-apply crash")
+    monkeypatch.setattr(service.repo, "insert_run_errors", boom)
+    lg = logging.getLogger("ecomm")
+    prior = lg.propagate
+    lg.propagate = True
+    try:
+        with caplog.at_level(logging.INFO, logger="ecomm.telemetry"):
+            with pytest.raises(RuntimeError):
+                products.confirm_draft(migrated_conn, sf, acct, d["id"])
+    finally:
+        lg.propagate = prior
+    assert migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0] == 0
+    assert migrated_conn.execute("SELECT count(*) FROM import_run").fetchone()[0] == 0
+    assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 1
+    events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"]
+    assert any(e["event"] == "import_apply_failed" and e["error_class"] == "RuntimeError" for e in events)

backend/tests/test_products_service.py

@@ -115,3 +115,89 @@ def test_tel1_emitted(migrated_conn, merchant, caplog, telemetry_propagation):
         and "duration_ms" in e and e["unknown_columns_count"] == 0
         for e in events
     )
+
+
+UPDATE_CSV = b"Handle,Title,Vendor,Variant Price\nmoon-mug,Moon Mug,Acme,21.00\nstar-tee,Star Tee,Acme,24.00\n"
+MIXED_CSV = b"Handle,Title,Variant Price\ngood-mug,Mug,10.00\nbad-tee,Tee,not-a-price\n"
+
+
+def _validate(conn, m, data=GOOD_CSV):
+    return products.import_validate(conn, m["storefront_id"], m["account_id"], "cat.csv", data)
+
+
+def test_confirm_applies_adds_and_records_run(migrated_conn, merchant):
+    draft = _validate(migrated_conn, merchant)
+    run_id = products.confirm_draft(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"])
+    assert migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0] == 2
+    run = products.get_run(migrated_conn, merchant["storefront_id"], run_id)
+    assert run["products_added"] == 2 and run["status"] == "complete"
+    assert run["by"] == "m@example.com"
+    assert run["image_progress"] == {"done": 0, "total": 0} and run["image_outcomes"] == []
+    assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0
+
+
+def test_confirm_update_changes_only_diffed_fields(migrated_conn, merchant):
+    d1 = _validate(migrated_conn, merchant)
+    products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], d1["id"])
+    d2 = _validate(migrated_conn, merchant, UPDATE_CSV)
+    assert d2["summary"] == {"adds": 0, "updates": 1, "unchanged": 1, "errors": 0}
+    products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], d2["id"])
+    price = migrated_conn.execute(
+        "SELECT v.price FROM variant v JOIN product p ON p.id = v.product_id WHERE p.handle='moon-mug'"
+    ).fetchone()[0]
+    assert str(price) == "21.00"
+    assert migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0] == 2  # no dupes (BUC-3)
+
+
+def test_confirm_mixed_applies_valid_records_errors(migrated_conn, merchant):
+    draft = _validate(migrated_conn, merchant, MIXED_CSV)
+    run_id = products.confirm_draft(
+        migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"])
+    assert migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0] == 1
+    run = products.get_run(migrated_conn, merchant["storefront_id"], run_id)
+    assert run["rows_errored"] == 1
+    assert run["errors"][0]["column"] == "Variant Price"
+
+
+def test_confirm_stale_fingerprint_409_inv11(migrated_conn, merchant):
+    draft = _validate(migrated_conn, merchant)
+    other = _validate(migrated_conn, merchant)
+    products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], other["id"])
+    with pytest.raises(products.PreviewStale):
+        products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], draft["id"])
+
+
+def test_confirm_nothing_to_apply_puc10(migrated_conn, merchant):
+    d1 = _validate(migrated_conn, merchant)
+    products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], d1["id"])
+    d2 = _validate(migrated_conn, merchant)
+    assert d2["summary"]["unchanged"] == 2
+    with pytest.raises(products.NothingToApply):
+        products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], d2["id"])
+
+
+def test_runs_history_newest_first(migrated_conn, merchant):
+    d1 = _validate(migrated_conn, merchant)
+    products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], d1["id"])
+    d2 = _validate(migrated_conn, merchant, UPDATE_CSV)
+    r2 = products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], d2["id"])
+    runs = products.list_runs(migrated_conn, merchant["storefront_id"])
+    assert [r["id"] for r in runs][0] == r2
+
+
+def test_summary_counts(migrated_conn, merchant):
+    assert products.summary(migrated_conn, merchant["storefront_id"]) == {
+        "product_count": 0, "image_problem_count": 0, "latest_run_id": None}
+    d = _validate(migrated_conn, merchant)
+    rid = products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], d["id"])
+    s = products.summary(migrated_conn, merchant["storefront_id"])
+    assert s == {"product_count": 2, "image_problem_count": 0, "latest_run_id": rid}
+
+
+def test_tel2_emitted_on_confirm(migrated_conn, merchant, caplog, telemetry_propagation):
+    draft = _validate(migrated_conn, merchant)
+    with caplog.at_level(logging.INFO, logger="ecomm.telemetry"):
+        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)