wiggleverse-ecomm/docs/superpowers/plans/2026-06-11-slice-5-import-spine.md

SLICE-5 — Import Spine (SD-0002) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Ship SD-0002 SLICE-5 — the products import spine: products domain + tables, canonical CSV codec, validation, diff engine, draft/confirm/run endpoints, and the admin Products section (upload → preview → confirm → run detail + history), images stubbed, export deferred to SLICE-6.

Spec: ~/git/wiggleverse.org/wiggleverse/wiggleverse-ecomm-content/specs/SD-0002-products-bulk-csv-import-export.md (§5 UX, §6 technical design, §7.2 SLICE-5 DoD). Anchor: wiggleverse/wiggleverse-ecomm#13. Invariants in force this slice: INV-10/11/13/14/15/17/18 (+ SD-0001's INV-5/6/7/8).

Architecture: New products domain under the existing 3-layer FastAPI app (main → domains → platform, import-linter enforced). Raw psycopg3 SQL, forward-only migration 0002_products.sql. The CSV file is parsed → canonical rows → validated per-product → diffed against the storefront's catalog (read-only) → stored as an import_draft (raw bytes + precomputed records JSONB + fingerprint). Confirm re-derives the diff from the stored bytes, fingerprint-checks it (INV-11), and applies upserts in one transaction (INV-10). Image rows are written status='pending' but the fetch phase is a no-op stub — runs go straight to complete (SLICE-7 adds the real phase). Frontend: hash-routed Products section inside the SD-0001 admin shell. E2E: first Playwright suite, run locally (not yet in check.sh/CI — no browsers on the runner).

Tech stack: FastAPI + psycopg3 (existing) + nh3 (HTML sanitizer, INV-15) + python-multipart (upload). React 18 + plain CSS tokens (existing). @playwright/test (new, repo-root e2e/).

Working directory: the session worktree …/wiggleverse-ecomm/.claude/worktrees/slice-5-import-spine (branch worktree-slice-5-import-spine). Backend venv at <worktree>/.venv. All pytest commands: cd backend && ../.venv/bin/python -m pytest …. Postgres: the dev compose container is already running on localhost:5432 (shared — do not restart it).

Locked decisions (do not re-litigate in tasks):

• Draft file bytes live in import_draft.file_bytes BYTEA this slice (objectstore is SLICE-7's; cap is 10 MB so Postgres is fine). SLICE-7 swaps to an objectstore key.
• Dialect is always "canonical" this slice; detect_dialect() exists as the seam, Shopify arrives SLICE-8.
• Error granularity: any row error marks its whole product kind="error" (excluded from apply); the error table is per-row.
• Product-level fields are read from a product's first row only; on later rows of the same product they are ignored (Shopify grammar).
• Blank-vs-absent (§6.5.1): a column absent from the header never touches that field; a present-but-empty cell clears the field to its default (status→'active', published→TRUE, type→'standalone', tags→{}, everything else → NULL).
• nothing_to_apply ⇔ adds + updates == 0 (errors/unchanged alone never produce a run).
• The Products page shows Export disabled with an honest "arrives in a coming release" note (endpoint is SLICE-6); the column-reference link is SLICE-8 (only the sample-CSV link ships now).
• Run detail keeps the §6.4 image fields in its payload (image_progress: {done:0,total:0}, image_outcomes: []) so SLICE-7 doesn't change the API shape.

---

File map

File	Role
backend/requirements.txt	+ nh3, python-multipart
backend/migrations/0002_products.sql	§6.3 tables (new)
backend/app/domains/products/__init__.py	public surface
backend/app/domains/products/errors.py	FileRejected, DraftNotFound, DraftExpired, PreviewStale, NothingToApply, RunNotFound
backend/app/domains/products/models.py	column registry, canonical row model, row errors
backend/app/domains/products/codec.py	bytes → ParsedFile (file-level checks, INV-18)
backend/app/domains/products/validate.py	rows → CanonicalProducts + RowErrors (INV-15 sanitize)
backend/app/domains/products/diff.py	catalog × canonical → records + summary + fingerprint (INV-11)
backend/app/domains/products/repo.py	all SQL: catalog load, draft/run CRUD, upserts
backend/app/domains/products/service.py	orchestration: validate/records/discard/confirm/runs/summary + TEL
backend/app/domains/products/sample.csv	DOC-3 asset
backend/app/platform/telemetry.py	structured JSON log events (TEL-*)
backend/app/main.py	+ /api/products/* endpoints
backend/tests/test_products_codec.py · test_products_validate.py · test_products_diff.py · test_products_service.py · test_products_invariants.py · test_products_endpoints.py	unit/integration
frontend/src/productsApi.ts	typed fetch wrappers
frontend/src/adminRouting.ts (+ .test.ts)	hash → admin view
frontend/src/screens/Admin.tsx	nav + section dispatch (modify)
frontend/src/screens/products/ProductsPage.tsx · ImportUpload.tsx · ImportPreview.tsx · RunDetail.tsx	the four §5 surfaces
frontend/src/styles/products.css (+ import in styles/index.css)	section styles
e2e/package.json · playwright.config.ts · serve.sh · helpers.ts · fixtures/*.csv · tests/*.spec.ts	Playwright suite
docs/OPERATIONS.md	DOC-1: RB-2, caps, ALR-2 gesture (new)
docs/products-domain.md	DOC-4 dev notes (new)
VERSION, frontend/package.json	0.4.0 → 0.5.0

---

Task 1: Dependencies + migration 0002 (the §6.3 tables)

Files:

• Modify: backend/requirements.txt

• Create: backend/migrations/0002_products.sql

• Test: backend/tests/test_migrations.py (extend)

• Step 1: Add deps and install

Append to backend/requirements.txt:

nh3>=0.2
python-multipart>=0.0.9

Run: .venv/bin/python -m pip install -r backend/requirements.txt → installs cleanly.

• Step 2: Write a failing migration test

Append to backend/tests/test_migrations.py:

def test_0002_products_tables_exist(fresh_db_url):
    with psycopg.connect(fresh_db_url) as conn:
        db.migrate(conn)
        for table in ("product", "variant", "product_image", "import_draft", "import_run", "import_run_error"):
            assert conn.execute("SELECT to_regclass(%s)", (f"public.{table}",)).fetchone()[0] == table


def test_0002_variant_option_combo_unique_treats_nulls_as_equal(fresh_db_url):
    # INV-13: the no-option product's single variant has NULL option values; a second
    # all-NULL combo must collide (NULLS NOT DISTINCT).
    with psycopg.connect(fresh_db_url) as conn:
        db.migrate(conn)
        sf = conn.execute("INSERT INTO storefront (name) VALUES ('s') RETURNING id").fetchone()[0]
        pid = conn.execute(
            "INSERT INTO product (storefront_id, handle, title) VALUES (%s,'h','T') RETURNING id", (sf,)
        ).fetchone()[0]
        conn.execute("INSERT INTO variant (product_id, position) VALUES (%s, 1)", (pid,))
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute("INSERT INTO variant (product_id, position) VALUES (%s, 2)", (pid,))

(Ensure import pytest is present in that file.)

Run: cd backend && ../.venv/bin/python -m pytest tests/test_migrations.py -q → the two new tests FAIL (tables absent).

• Step 3: Write the migration

backend/migrations/0002_products.sql — complete content:

-- 0002_products.sql — SD-0002 §6.3 data model (SLICE-5). Forward-only (INV-7):
-- never edit once merged; add a new numbered migration.

-- product — one catalog product per (storefront, handle) (INV-13/14). Option *names*
-- live here; product_type's kit values are schema-open but a service-layer rule
-- rejects non-'standalone' until #15 (same pattern as INV-4).
CREATE TABLE product (
    id            BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    storefront_id BIGINT NOT NULL REFERENCES storefront (id),
    handle        TEXT NOT NULL,
    title         TEXT NOT NULL,
    description_html TEXT,
    vendor        TEXT,
    product_type  TEXT NOT NULL DEFAULT 'standalone'
                  CHECK (product_type IN ('standalone', 'kit_virtual', 'kit_assembled')),
    google_product_category TEXT,
    tags          TEXT[] NOT NULL DEFAULT '{}',
    status        TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('draft', 'active', 'archived')),
    published     BOOLEAN NOT NULL DEFAULT TRUE,
    option1_name  TEXT,
    option2_name  TEXT,
    option3_name  TEXT,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE UNIQUE INDEX product_handle_key ON product (storefront_id, handle);  -- INV-13

-- variant — one purchasable form, identified by its option-value combo (INV-13).
-- SKU is indexed data, never identity. image_id FK is added after product_image.
CREATE TABLE variant (
    id            BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    product_id    BIGINT NOT NULL REFERENCES product (id),
    position      INTEGER NOT NULL,
    option1_value TEXT,
    option2_value TEXT,
    option3_value TEXT,
    sku           TEXT,
    barcode       TEXT,
    price         NUMERIC,
    cost          NUMERIC,
    weight        NUMERIC,
    weight_unit   TEXT,
    volume        NUMERIC,
    volume_unit   TEXT,
    tax_id_1      TEXT,
    tax_id_2      TEXT,
    inventory_tracker TEXT,
    inventory_qty INTEGER,
    image_id      BIGINT,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- Postgres 16 (compose + Cloud SQL pin): NULLS NOT DISTINCT makes the all-NULL
-- no-option combo unique too (INV-13).
CREATE UNIQUE INDEX variant_option_combo_key
    ON variant (product_id, option1_value, option2_value, option3_value)
    NULLS NOT DISTINCT;
CREATE INDEX variant_sku_idx ON variant (sku);

-- product_image — identity within a product is source_url (§6.3); bytes live in
-- object storage from SLICE-7 (keys nullable until fetched). status starts 'pending';
-- SLICE-5 stubs the fetch phase so rows simply stay pending.
CREATE TABLE product_image (
    id            BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    product_id    BIGINT NOT NULL REFERENCES product (id),
    position      INTEGER NOT NULL,
    source_url    TEXT NOT NULL,
    alt_text      TEXT,
    status        TEXT NOT NULL DEFAULT 'pending'
                  CHECK (status IN ('pending', 'fetched', 'rejected_low_res', 'rejected_not_image', 'failed')),
    failure_reason TEXT,
    key_original  TEXT,
    key_thumb     TEXT,
    key_card      TEXT,
    key_detail    TEXT,
    import_run_id BIGINT,
    fetched_at    TIMESTAMPTZ,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE UNIQUE INDEX product_image_src_key ON product_image (product_id, source_url);
ALTER TABLE variant
    ADD CONSTRAINT variant_image_fk FOREIGN KEY (image_id) REFERENCES product_image (id);

-- import_draft — the preview's server side (INV-11). file_bytes is the SLICE-5
-- interim home for the upload (objectstore key from SLICE-7). Deleted outright on
-- cancel/expiry — drafts never appear in history (PUC-3a).
CREATE TABLE import_draft (
    id            BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    storefront_id BIGINT NOT NULL REFERENCES storefront (id),
    account_id    BIGINT NOT NULL REFERENCES account (id),
    file_name     TEXT NOT NULL,
    dialect       TEXT NOT NULL,
    file_bytes    BYTEA NOT NULL,
    summary       JSONB NOT NULL,
    records       JSONB NOT NULL,
    fingerprint   TEXT NOT NULL,
    unknown_columns TEXT[] NOT NULL DEFAULT '{}',
    expires_at    TIMESTAMPTZ NOT NULL,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- import_run — the durable record of one confirmed import (PUC-8); created only at
-- confirm (PUC-4).
CREATE TABLE import_run (
    id            BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    storefront_id BIGINT NOT NULL REFERENCES storefront (id),
    account_id    BIGINT NOT NULL REFERENCES account (id),
    file_name     TEXT NOT NULL,
    dialect       TEXT NOT NULL,
    products_added   INTEGER NOT NULL,
    products_updated INTEGER NOT NULL,
    rows_errored     INTEGER NOT NULL,
    status        TEXT NOT NULL
                  CHECK (status IN ('applying', 'fetching_images', 'complete', 'complete_with_problems')),
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    completed_at  TIMESTAMPTZ
);
CREATE INDEX import_run_history_idx ON import_run (storefront_id, created_at DESC);

-- import_run_error — one row per rejected CSV row (PUC-5), merchant-language message.
CREATE TABLE import_run_error (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    run_id      BIGINT NOT NULL REFERENCES import_run (id),
    line_number INTEGER NOT NULL,
    column_name TEXT,
    message     TEXT NOT NULL
);

ALTER TABLE product_image
    ADD CONSTRAINT product_image_run_fk FOREIGN KEY (import_run_id) REFERENCES import_run (id);

• Step 4: Run the tests

Run: cd backend && ../.venv/bin/python -m pytest tests/test_migrations.py -q → PASS (all, including the pre-existing ones).

• Step 5: Commit

git add backend/requirements.txt backend/migrations/0002_products.sql backend/tests/test_migrations.py
git commit -m "feat(products): SD-0002 §6.3 schema — migration 0002 + nh3/multipart deps (SLICE-5)"

---

Task 2: Domain skeleton — errors, column registry, canonical model

Files:

• Create: backend/app/domains/products/__init__.py, errors.py, models.py

No behavior yet (pure data structures) — covered by Task 3+ tests; just typecheck by import.

• Step 1: errors.py

"""products domain errors (SD-0002 §6.4 error envelope codes)."""
from __future__ import annotations


class ProductsError(Exception):
    """Base for products-domain errors."""


class FileRejected(ProductsError):
    """PUC-5a: the whole file is unusable; no draft is created. `code` is the §6.4
    error code (not_csv | missing_required_column | unknown_dialect | too_many_rows |
    file_too_large)."""

    def __init__(self, code: str, message: str):
        super().__init__(message)
        self.code = code
        self.message = message


class DraftNotFound(ProductsError):
    """No such draft for this storefront (or already discarded)."""


class DraftExpired(ProductsError):
    """The draft's validity window passed (§6.3 ~1 h)."""


class PreviewStale(ProductsError):
    """INV-11: the catalog changed since validation — the previewed diff no longer holds."""


class NothingToApply(ProductsError):
    """PUC-10: no adds and no updates — confirming would be a no-op."""


class RunNotFound(ProductsError):
    """No such import run for this storefront."""

• Step 2: models.py

"""Canonical row model + column registry — the one model every dialect maps to (INV-17)."""
from __future__ import annotations

from dataclasses import dataclass, field

# §6.5.1 canonical columns, by level. Header detection, unknown-column warnings, and
# validation all read from this registry.
PRODUCT_COLUMNS: dict[str, str] = {
    # column -> product field name
    "Title": "title",
    "Description": "description_html",
    "Vendor": "vendor",
    "Type": "product_type",
    "Google Product Category": "google_product_category",
    "Tags": "tags",
    "Status": "status",
    "Published": "published",
    "Option1 Name": "option1_name",
    "Option2 Name": "option2_name",
    "Option3 Name": "option3_name",
}
VARIANT_COLUMNS: dict[str, str] = {
    "Variant SKU": "sku",
    "Variant Barcode": "barcode",
    "Variant Price": "price",
    "Variant Cost": "cost",
    "Variant Weight": "weight",
    "Variant Weight Unit": "weight_unit",
    "Variant Volume": "volume",
    "Variant Volume Unit": "volume_unit",
    "Variant Tax ID 1": "tax_id_1",
    "Variant Tax ID 2": "tax_id_2",
    "Variant Inventory Tracker": "inventory_tracker",
    "Variant Inventory Qty": "inventory_qty",
    "Variant Position": "position",
    "Variant Image": "variant_image",
}
OPTION_VALUE_COLUMNS = ("Option1 Value", "Option2 Value", "Option3 Value")
IMAGE_COLUMNS = ("Image Src", "Image Position", "Image Alt Text")
COMPONENT_COLUMNS = tuple(
    f"Component {i} {kind}" for i in range(1, 11) for kind in ("SKU", "Quantity")
)
KNOWN_COLUMNS = (
    {"Handle"}
    | set(PRODUCT_COLUMNS)
    | set(VARIANT_COLUMNS)
    | set(OPTION_VALUE_COLUMNS)
    | set(IMAGE_COLUMNS)
    | set(COMPONENT_COLUMNS)
)

# Clearing a field (present-but-empty cell, §6.5.1) resets it to its default.
CLEAR_DEFAULTS: dict[str, object] = {
    "status": "active",
    "published": True,
    "product_type": "standalone",
    "tags": [],
}

MAX_DATA_ROWS = 5_000  # INV-18
MAX_FILE_BYTES = 10 * 1024 * 1024  # INV-18


@dataclass(frozen=True)
class Row:
    """One CSV data row: 1-based file line number + the cells of known columns
    present in the header (column name -> raw string, possibly empty)."""

    line_number: int
    cells: dict[str, str]


@dataclass(frozen=True)
class ParsedFile:
    dialect: str
    header: list[str]
    unknown_columns: list[str]
    rows: list[Row]


@dataclass(frozen=True)
class RowError:
    line_number: int
    column: str | None
    message: str

    def as_json(self) -> dict:
        return {"line": self.line_number, "column": self.column, "message": self.message}


@dataclass
class CanonicalVariant:
    line_number: int
    options: tuple[str | None, str | None, str | None]
    # field name -> normalized value; present only for columns in the file.
    # value None == clear (reset to default/NULL).
    fields: dict[str, object] = field(default_factory=dict)


@dataclass
class CanonicalImage:
    line_number: int
    source_url: str
    position: int
    alt_text: str | None


@dataclass
class CanonicalProduct:
    first_line: int
    handle: str
    title: str  # "" when missing (the block then carries an error)
    option_names: tuple[str | None, str | None, str | None] = (None, None, None)
    fields: dict[str, object] = field(default_factory=dict)  # product-level, same semantics
    variants: list[CanonicalVariant] = field(default_factory=list)
    images: list[CanonicalImage] = field(default_factory=list)
    errors: list[RowError] = field(default_factory=list)

    @property
    def valid(self) -> bool:
        return not self.errors

• Step 3: __init__.py (initial; service symbols join in Tasks 6–8)

"""products domain — catalog + bulk CSV import/export (SD-0002 §6.2).

Owns the canonical row model, codec, validation, diff engine, and import
drafts/runs. Storefront-scoped throughout (INV-14); upsert is the only mutation
(INV-10). Imported via this package surface only.
"""
from __future__ import annotations

from .errors import (
    DraftExpired,
    DraftNotFound,
    FileRejected,
    NothingToApply,
    PreviewStale,
    ProductsError,
    RunNotFound,
)
from .models import MAX_DATA_ROWS, MAX_FILE_BYTES

__all__ = [
    "ProductsError", "FileRejected", "DraftNotFound", "DraftExpired",
    "PreviewStale", "NothingToApply", "RunNotFound",
    "MAX_DATA_ROWS", "MAX_FILE_BYTES",
]

• Step 4: Verify import + lint layers

Run: cd backend && ../.venv/bin/python -c "from app.domains import products" && ../.venv/bin/lint-imports → both clean.

• Step 5: Commit — git add backend/app/domains/products && git commit -m "feat(products): domain skeleton — errors, column registry, canonical row model"

---

Task 3: codec.py — bytes → ParsedFile (file-level gates, INV-18)

Files:

• Create: backend/app/domains/products/codec.py
• Test: backend/tests/test_products_codec.py

Behavior contract:

• UTF-8 (utf-8-sig — BOM tolerated), csv.reader (RFC 4180 quoting).

• Decode failure / no rows at all / a csv.Error → FileRejected("not_csv", "This file isn't readable as CSV.").

• Header row = first row. Handle or Title missing from header → FileRejected("missing_required_column", "This file is missing the required column '<name>'.") (check Handle first).

• Duplicate known column in header → first occurrence wins (no error).

• More than MAX_DATA_ROWS data rows → FileRejected("too_many_rows", f"This file has more than {MAX_DATA_ROWS:,} rows — split it and import in parts.").

• len(data) > MAX_FILE_BYTES → FileRejected("file_too_large", "This file is larger than 10 MB.") (also enforced at the HTTP layer with 413).

• Fully-empty rows are skipped. Short rows are padded with ""; long rows' extra cells ignored.

• unknown_columns = header names not in KNOWN_COLUMNS, original order, deduped, "" headers ignored.

• dialect = detect_dialect(header) → always "canonical" for now (the INV-17/SLICE-8 seam; a header without Handle never reaches it).

• Row.cells contains an entry for every known column present in the header (value may be "").

• Line numbers are 1-based physical file lines (header = line 1, first data row = 2) — csv.reader.line_num handles multi-line quoted cells.

• Step 1: Write the failing tests — backend/tests/test_products_codec.py:

"""§6.5.1 file-level codec — parse, caps, required columns (PUC-5a fixtures)."""
import pytest

from app.domains.products import FileRejected
from app.domains.products.codec import parse_csv


def _csv(*lines: str) -> bytes:
    return ("\n".join(lines) + "\n").encode()


GOOD = _csv(
    "Handle,Title,Variant Price,Bogus Column",
    "moon-mug,Moon Mug,18.00,x",
    "star-tee,Star Tee,24.00,y",
)


def test_parses_header_rows_and_unknown_columns():
    parsed = parse_csv(GOOD)
    assert parsed.dialect == "canonical"
    assert parsed.unknown_columns == ["Bogus Column"]
    assert [r.line_number for r in parsed.rows] == [2, 3]
    assert parsed.rows[0].cells["Handle"] == "moon-mug"
    assert parsed.rows[0].cells["Variant Price"] == "18.00"
    assert "Bogus Column" not in parsed.rows[0].cells


def test_bom_tolerated():
    parsed = parse_csv(b"\xef\xbb\xbf" + GOOD)
    assert parsed.rows[0].cells["Handle"] == "moon-mug"


def test_quoted_cells_rfc4180():
    parsed = parse_csv(_csv("Handle,Title,Tags", 'mug,"The ""Best"" Mug","a, b"'))
    assert parsed.rows[0].cells["Title"] == 'The "Best" Mug'
    assert parsed.rows[0].cells["Tags"] == "a, b"


def test_empty_rows_skipped_short_rows_padded():
    parsed = parse_csv(_csv("Handle,Title,Vendor", "mug,Mug", "", ",,", "tee,Tee,Acme"))
    assert [r.cells["Handle"] for r in parsed.rows] == ["mug", "tee"]
    assert parsed.rows[0].cells["Vendor"] == ""


@pytest.mark.parametrize(
    "data,code",
    [
        (b"\xff\xfe\x00garbage\x00", "not_csv"),
        (b"", "not_csv"),
        (_csv("Title,Vendor", "Mug,Acme"), "missing_required_column"),
        (_csv("Handle,Vendor", "mug,Acme"), "missing_required_column"),
        (
            _csv("Handle,Title", *(f"h{i},T{i}" for i in range(5001))),
            "too_many_rows",
        ),
        (b"Handle,Title\n" + b"x" * (10 * 1024 * 1024), "file_too_large"),
    ],
)
def test_file_level_rejections(data, code):
    with pytest.raises(FileRejected) as exc:
        parse_csv(data)
    assert exc.value.code == code


def test_missing_column_message_names_the_column():
    with pytest.raises(FileRejected) as exc:
        parse_csv(_csv("Handle,Vendor", "mug,Acme"))
    assert "'Title'" in exc.value.message

Run: cd backend && ../.venv/bin/python -m pytest tests/test_products_codec.py -q → FAIL (module missing).

• Step 2: Implement codec.py

"""CSV codec — bytes → ParsedFile (SD-0002 §6.5.1). File-level gates only (PUC-5a);
row semantics live in validate.py. Dialect detection is the INV-17 seam (SLICE-8
adds Shopify)."""
from __future__ import annotations

import csv
import io

from .errors import FileRejected
from .models import KNOWN_COLUMNS, MAX_DATA_ROWS, MAX_FILE_BYTES, ParsedFile, Row

_REQUIRED_HEADER_COLUMNS = ("Handle", "Title")


def detect_dialect(header: list[str]) -> str:
    """The INV-17 seam: SLICE-8 recognizes Shopify's exact header set here."""
    return "canonical"


def parse_csv(data: bytes) -> ParsedFile:
    if len(data) > MAX_FILE_BYTES:
        raise FileRejected("file_too_large", "This file is larger than 10 MB.")
    try:
        text = data.decode("utf-8-sig")
    except UnicodeDecodeError:
        raise FileRejected("not_csv", "This file isn't readable as CSV.") from None
    reader = csv.reader(io.StringIO(text))
    try:
        try:
            raw_header = next(reader)
        except StopIteration:
            raise FileRejected("not_csv", "This file isn't readable as CSV.") from None
        header = [h.strip() for h in raw_header]
        for col in _REQUIRED_HEADER_COLUMNS:
            if col not in header:
                raise FileRejected(
                    "missing_required_column",
                    f"This file is missing the required column '{col}'.",
                )
        # First occurrence of a duplicated column wins.
        col_index: dict[str, int] = {}
        for i, name in enumerate(header):
            if name and name not in col_index:
                col_index[name] = i
        known_present = [c for c in col_index if c in KNOWN_COLUMNS]
        unknown = [c for c in col_index if c not in KNOWN_COLUMNS]
        rows: list[Row] = []
        for raw in reader:
            if not any(cell.strip() for cell in raw):
                continue
            if len(rows) >= MAX_DATA_ROWS:
                raise FileRejected(
                    "too_many_rows",
                    f"This file has more than {MAX_DATA_ROWS:,} rows — split it and import in parts.",
                )
            cells = {
                c: (raw[col_index[c]].strip() if col_index[c] < len(raw) else "")
                for c in known_present
            }
            rows.append(Row(line_number=reader.line_num, cells=cells))
    except csv.Error:
        raise FileRejected("not_csv", "This file isn't readable as CSV.") from None
    return ParsedFile(
        dialect=detect_dialect(header), header=header, unknown_columns=unknown, rows=rows
    )

Note: reader.line_num after consuming a row is the line the row ended on; for multi-line quoted cells the spec needs a stable, explainable number — capture line_number = reader.line_num as shown (matches the single-line common case; tests pin it).

• Step 3: Run cd backend && ../.venv/bin/python -m pytest tests/test_products_codec.py -q → PASS.

• Step 4: Commit — git commit -am "feat(products): CSV codec — file-level parse + INV-18 caps (PUC-5a)"

---

Task 4: validate.py — rows → canonical products + row errors

Files:

• Create: backend/app/domains/products/validate.py
• Test: backend/tests/test_products_validate.py

Behavior contract (every rule is a §6.5.1 row error; messages are merchant-language; each records the offending line_number + column):

Rule	Error message (format)
Handle empty	"a row needs a Handle"
Handle not ^[a-z0-9-]+$	"'{v}' isn't a valid handle — lowercase letters, numbers, and dashes only"
handle block reappears non-consecutively	"rows for '{handle}' must be consecutive — it already appeared earlier in the file"
Title empty on a product's first row	"'{handle}' is missing its Title" (column Title)
Type non-empty and not standalone	"kits arrive in a coming release — Type must be 'standalone'"
any Component n * cell non-empty	"kits arrive in a coming release — leave the Component columns empty"
Status not in draft/active/archived (case-insensitive)	"'{v}' is not a status — use draft, active, or archived"
Published not TRUE/FALSE (case-insensitive)	"'{v}' is not TRUE or FALSE"
Variant Price / Variant Cost / Variant Weight / Variant Volume not a non-negative decimal	"'{v}' is not a price" (price/cost) / "'{v}' is not a number" (weight/volume)
Variant Inventory Qty not an integer ≥ 0	"'{v}' is not a whole number"
Variant Position / Image Position not an integer ≥ 1	"'{v}' is not a position"
OptionN Value present but product has no OptionN Name	"Option{n} Value given but the product has no Option{n} Name"
product has OptionN Name but a variant row's OptionN Value is empty	"this variant is missing its Option{n} Value ('{name}')"
duplicate option-value combo within the file	"duplicate variant — '{handle}' already has a variant with these options"
>1 variant row on a no-option product	"a product without options can have only one variant"
a data row that is neither a product's first row, nor carries variant data, nor image data	"this row has no variant or image data"

Mechanics:

• Group consecutive rows by Handle. Rows whose Handle is empty/invalid are recorded as block-less errors → collected into a synthetic error product per row? No — attach them to a CanonicalProduct(handle=raw_handle or "(missing)", title="", errors=[...]) block of their own (one per offending row) so every error reaches the preview as a kind="error" record.

• Option names: from the first row's Option1–3 Name cells (empty → None).

• Product-level fields: read from the first row only, for the product-level columns present in the header. Normalization: empty cell → None (= clear); Tags → [t.strip() for t in cell.split(",") if t.strip()]; Status → lowercase; Published → bool; Description → nh3.clean(cell) (INV-15 — default nh3 allowlist already strips scripts/handlers/embeds); Type → validated standalone.

• A row carries a variant iff any OptionN Value is non-empty or any Variant * cell is non-empty; on a no-option product, the first row always carries its single variant (even with all variant cells empty). It carries an image iff Image Src is non-empty. First rows that carry neither are still product rows (not an error). Non-first rows carrying neither → the no-data error above.

• Variant fields: numeric columns → decimal.Decimal (reject negatives), qty → int, Variant Image → kept as URL string under field variant_image. Empty cell → None (clear).

• Images: dedupe by source_url within the product (first occurrence wins; later duplicates silently merged); Image Position defaults to encounter order (1-based); Variant Image URLs are appended to the product's images if not already present (alt None, position = next).

• Any error anywhere in a block → the whole product is invalid (.errors non-empty), but parsing continues to collect all errors (BUC-1a: complete accounting).

• Signature: def build_products(parsed: ParsedFile) -> list[CanonicalProduct] (errors ride on the products; helper def all_errors(products) -> list[RowError]).

• Step 1: Write the failing tests — backend/tests/test_products_validate.py. Test every table row above with a minimal fixture file built via the Task 3 _csv helper (import it or redefine). Full code:

"""§6.5.1 row validation — every row-error rule has a fixture (SD-0002 §6.8)."""
from decimal import Decimal

import pytest

from app.domains.products.codec import parse_csv
from app.domains.products.validate import build_products


def _products(*lines: str):
    return build_products(parse_csv(("\n".join(lines) + "\n").encode()))


def _errors(*lines: str):
    return [e for p in _products(*lines) for e in p.errors]


def test_simple_product_parses_clean():
    [p] = _products(
        "Handle,Title,Vendor,Tags,Status,Published,Variant Price,Variant SKU",
        "moon-mug,Moon Mug,Acme,\"kitchen, mugs\",active,TRUE,18.00,SKU-1",
    )
    assert p.valid and p.handle == "moon-mug" and p.title == "Moon Mug"
    assert p.fields["tags"] == ["kitchen", "mugs"]
    assert p.fields["status"] == "active" and p.fields["published"] is True
    [v] = p.variants
    assert v.options == (None, None, None)
    assert v.fields["price"] == Decimal("18.00") and v.fields["sku"] == "SKU-1"


def test_option_product_groups_consecutive_rows():
    [p] = _products(
        "Handle,Title,Option1 Name,Option1 Value,Variant Price",
        "tee,Tee,Size,S,24.00",
        "tee,,,M,24.00",
        "tee,,,L,26.00",
    )
    assert p.valid and p.option_names == ("Size", None, None)
    assert [v.options[0] for v in p.variants] == ["S", "M", "L"]


def test_image_only_rows_and_dedupe():
    [p] = _products(
        "Handle,Title,Image Src,Image Position,Image Alt Text",
        "mug,Mug,https://x/a.jpg,1,front",
        "mug,,https://x/b.jpg,2,back",
        "mug,,https://x/a.jpg,3,dupe",
    )
    assert p.valid
    assert [(i.source_url, i.position) for i in p.images] == [
        ("https://x/a.jpg", 1), ("https://x/b.jpg", 2),
    ]


def test_variant_image_joins_product_images():
    [p] = _products(
        "Handle,Title,Variant Image",
        "mug,Mug,https://x/v.jpg",
    )
    assert p.variants[0].fields["variant_image"] == "https://x/v.jpg"
    assert [i.source_url for i in p.images] == ["https://x/v.jpg"]


def test_description_sanitized_inv15():
    [p] = _products(
        "Handle,Title,Description",
        'mug,Mug,"<p onclick=\'x()\'>hi</p><script>evil()</script>"',
    )
    html = p.fields["description_html"]
    assert "<p>" in html and "script" not in html and "onclick" not in html


def test_blank_cell_clears_absent_column_missing():
    [p] = _products("Handle,Title,Vendor", "mug,Mug,")
    assert p.fields["vendor"] is None          # present-but-empty == clear
    assert "status" not in p.fields            # absent column == untouched


@pytest.mark.parametrize(
    "header,row,column,fragment",
    [
        ("Handle,Title", ",NoHandle", "Handle", "needs a Handle"),
        ("Handle,Title", "Bad_Handle!,T", "Handle", "isn't a valid handle"),
        ("Handle,Title", "mug,", "Title", "missing its Title"),
        ("Handle,Title,Type", "mug,Mug,kit_virtual", "Type", "kits arrive"),
        ("Handle,Title,Component 1 SKU", "mug,Mug,ABC", "Component 1 SKU", "kits arrive"),
        ("Handle,Title,Status", "mug,Mug,live", "Status", "is not a status"),
        ("Handle,Title,Published", "mug,Mug,YES", "Published", "is not TRUE or FALSE"),
        ("Handle,Title,Variant Price", 'mug,Mug,"12,50"', "Variant Price", "is not a price"),
        ("Handle,Title,Variant Cost", "mug,Mug,-3", "Variant Cost", "is not a price"),
        ("Handle,Title,Variant Weight", "mug,Mug,heavy", "Variant Weight", "is not a number"),
        ("Handle,Title,Variant Inventory Qty", "mug,Mug,3.5", "Variant Inventory Qty", "is not a whole number"),
        ("Handle,Title,Variant Position", "mug,Mug,0", "Variant Position", "is not a position"),
        ("Handle,Title,Option1 Value", "mug,Mug,Red", "Option1 Value", "no Option1 Name"),
    ],
)
def test_row_error_rules(header, row, column, fragment):
    errors = _errors(header, row)
    assert any(e.column == column and fragment in e.message for e in errors), errors


def test_missing_option_value_for_named_option():
    errors = _errors(
        "Handle,Title,Option1 Name,Option1 Value,Variant SKU",
        "tee,Tee,Size,S,A",
        "tee,,,,B",
    )
    assert any("missing its Option1 Value" in e.message and e.line_number == 3 for e in errors)


def test_duplicate_option_combo_is_error():
    errors = _errors(
        "Handle,Title,Option1 Name,Option1 Value",
        "tee,Tee,Size,S",
        "tee,,,S",
    )
    assert any("duplicate variant" in e.message for e in errors)


def test_second_variant_on_no_option_product_is_error():
    errors = _errors(
        "Handle,Title,Variant SKU",
        "mug,Mug,A",
        "mug,,B",
    )
    assert any("without options can have only one variant" in e.message for e in errors)


def test_non_consecutive_handle_is_error():
    errors = _errors(
        "Handle,Title",
        "mug,Mug",
        "tee,Tee",
        "mug,",
    )
    assert any("must be consecutive" in e.message and e.line_number == 4 for e in errors)


def test_no_data_row_is_error():
    errors = _errors(
        "Handle,Title,Variant SKU,Image Src",
        "mug,Mug,A,",
        "mug,,,",
    )
    assert any("no variant or image data" in e.message for e in errors)


def test_errors_poison_their_product_only():
    products = _products(
        "Handle,Title,Variant Price",
        "good-mug,Mug,10.00",
        "bad-tee,Tee,not-a-price",
    )
    by_handle = {p.handle: p for p in products}
    assert by_handle["good-mug"].valid
    assert not by_handle["bad-tee"].valid


def test_all_errors_collected_not_first_only():
    [p] = _products(
        "Handle,Title,Status,Variant Price",
        "mug,,bogus,abc",
    )
    assert len(p.errors) == 3  # missing Title + bad status + bad price

Run → FAIL (module missing).

• Step 2: Implement validate.py (~180 lines). Skeleton with all rules:

"""Row validation — ParsedFile → CanonicalProducts + RowErrors (SD-0002 §6.5.1).

Every rule here is a *row* error: it poisons its product block (kind=error in the
preview) but never the file (PUC-5 — good rows are not blocked). File-level
problems live in codec.py (PUC-5a). HTML is sanitized at this boundary (INV-15).
"""
from __future__ import annotations

import re
from decimal import Decimal, InvalidOperation

import nh3

from .models import (
    COMPONENT_COLUMNS, PRODUCT_COLUMNS, VARIANT_COLUMNS,
    CanonicalImage, CanonicalProduct, CanonicalVariant, ParsedFile, Row, RowError,
)

_HANDLE_RE = re.compile(r"^[a-z0-9-]+$")
_STATUSES = {"draft", "active", "archived"}


def build_products(parsed: ParsedFile) -> list[CanonicalProduct]:
    blocks = _group_blocks(parsed.rows)          # consecutive-handle grouping +
    return [_build_product(handle, rows) for handle, rows in blocks]  # per-row errors


def all_errors(products: list[CanonicalProduct]) -> list[RowError]:
    return [e for p in products for e in p.errors]

Implementation notes (bind exactly to the contract table):

• _group_blocks walks rows; invalid/empty handles become single-row blocks with the error pre-attached; a handle seen in an earlier closed block yields a single-row block carrying the non-consecutive error.

• _build_product reads option names + product fields from the first row (only columns present in row.cells), then walks every row: component-column check, image extraction (dedupe by URL), variant detection per the contract, variant field normalization, option-value rules, duplicate-combo and single-variant rules, no-data rule.

• Normalizers raise nothing — they append RowErrors and skip the field.

• nh3.clean(value) for Description.

• Price/cost message "is not a price", weight/volume "is not a number"; both reject negatives via Decimal(v) < 0.

• Step 3: Run …/pytest tests/test_products_validate.py -q → PASS (iterate until all 20+ green).

• Step 4: Commit — git commit -am "feat(products): row validation — §6.5.1 rules + INV-15 sanitization"

---

Task 5: diff.py — catalog × canonical → records + fingerprint (INV-11)

Files:

• Create: backend/app/domains/products/diff.py
• Test: backend/tests/test_products_diff.py

Inputs: catalog: dict[str, CatalogProduct] (DB snapshot, Task 6 shape — defined here so diff is DB-free and unit-testable) and products: list[CanonicalProduct].

@dataclass
class CatalogVariant:
    id: int
    options: tuple[str | None, str | None, str | None]
    position: int
    fields: dict[str, object]      # sku, barcode, price(Decimal), …, variant_image (source_url of linked image or None)

@dataclass
class CatalogImage:
    id: int
    source_url: str
    position: int
    alt_text: str | None

@dataclass
class CatalogProduct:
    id: int
    handle: str
    title: str
    option_names: tuple[str | None, str | None, str | None]
    fields: dict[str, object]      # title, description_html, vendor, product_type, google_product_category, tags(list), status, published(bool)
    variants: list[CatalogVariant]
    images: list[CatalogImage]

Classification per canonical product:

• errors non-empty → kind="error", detail={"errors": [e.as_json(), …]}.
• handle not in catalog → kind="add", detail={"set": {...product fields resolved with CLEAR_DEFAULTS…, "option_names": [...]}, "variants": [{"options": [...], "set": {...}}…], "images": [{"src", "position", "alt_text"}…]}.
• handle in catalog → field-by-field compare. Only fields present in the file are compared (absent = untouched). A present None resolves to its CLEAR_DEFAULTS default (or None). Changes collect as {"field", "before", "after"} (JSON-safe: Decimal→str, tuples→lists). Variants matched by option combo: new combo → variant add; matched → compare present fields (including variant_image vs the linked image's source_url). Images matched by source_url: new → image add; matched → compare position / alt_text only if the corresponding column is in the file (alt/position always parsed together with Image Src, so compare when the image row exists). DB variants/images not in the file → untouched (INV-10 — never a change).
• Option names present on first row → compared like product fields (field option1_name etc.).
• No changes anywhere → kind="unchanged", detail={}.

Output:

@dataclass(frozen=True)
class DiffResult:
    records: list[dict]   # {"handle","title","kind","variant_count","detail"} — JSONB-ready
    summary: dict         # {"adds":N,"updates":M,"unchanged":U,"errors":K}
    fingerprint: str      # sha256 of canonical-JSON of records

def compute_diff(catalog: dict[str, CatalogProduct], products: list[CanonicalProduct]) -> DiffResult

title on a record: canonical title if non-empty else the catalog title else "". variant_count: len(canonical variants) (for adds/updates) — for unchanged, the catalog's count. Fingerprint: hashlib.sha256(json.dumps(records, sort_keys=True, separators=(",", ":")).encode()).hexdigest().

• Step 1: Write failing tests — backend/tests/test_products_diff.py:

"""Diff engine — classification, blank-vs-absent, option matching, fingerprint (§6.8)."""
from decimal import Decimal

from app.domains.products.codec import parse_csv
from app.domains.products.diff import (
    CatalogImage, CatalogProduct, CatalogVariant, compute_diff,
)
from app.domains.products.validate import build_products


def _canon(*lines: str):
    return build_products(parse_csv(("\n".join(lines) + "\n").encode()))


def _catalog_mug(**overrides):
    fields = {
        "title": "Moon Mug", "description_html": None, "vendor": "Acme",
        "product_type": "standalone", "google_product_category": None,
        "tags": ["kitchen"], "status": "active", "published": True,
    } | overrides
    return {
        "moon-mug": CatalogProduct(
            id=1, handle="moon-mug", title=fields["title"],
            option_names=(None, None, None), fields=fields,
            variants=[CatalogVariant(id=10, options=(None, None, None), position=1,
                                     fields={"sku": "SKU-1", "barcode": None, "price": Decimal("18.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": 40, "variant_image": None})],
            images=[CatalogImage(id=100, source_url="https://x/a.jpg", position=1, alt_text=None)],
        )
    }


HEADER = "Handle,Title,Vendor,Tags,Status,Variant SKU,Variant Price,Variant Inventory Qty,Image Src"
MUG_ROW = 'moon-mug,Moon Mug,Acme,kitchen,active,SKU-1,18.00,40,https://x/a.jpg'


def test_new_handle_classifies_add():
    diff = compute_diff({}, _canon(HEADER, MUG_ROW))
    [rec] = diff.records
    assert rec["kind"] == "add" and rec["handle"] == "moon-mug" and rec["variant_count"] == 1
    assert diff.summary == {"adds": 1, "updates": 0, "unchanged": 0, "errors": 0}
    assert rec["detail"]["set"]["status"] == "active"


def test_identical_file_classifies_unchanged():
    diff = compute_diff(_catalog_mug(), _canon(HEADER, MUG_ROW))
    assert diff.records[0]["kind"] == "unchanged"
    assert diff.summary["unchanged"] == 1


def test_changed_price_classifies_update_with_before_after():
    row = MUG_ROW.replace("18.00", "21.00")
    diff = compute_diff(_catalog_mug(), _canon(HEADER, row))
    [rec] = diff.records
    assert rec["kind"] == "update"
    [vchange] = rec["detail"]["variants"]
    assert {"field": "price", "before": "18.00", "after": "21.00"} in vchange["changes"]


def test_absent_column_untouched_empty_cell_clears():
    # Vendor column absent: vendor stays Acme. Status present-but-empty: clears to default 'active' (already active → no change).
    diff = compute_diff(
        _catalog_mug(),
        _canon("Handle,Title,Status,Variant SKU,Variant Price,Variant Inventory Qty,Image Src",
               "moon-mug,Moon Mug,,SKU-1,18.00,40,https://x/a.jpg"),
    )
    assert diff.records[0]["kind"] == "unchanged"


def test_empty_cell_clear_shows_in_diff():
    # Vendor present-but-empty clears Acme -> None: an explicit, previewable change (§6.5.1).
    diff = compute_diff(
        _catalog_mug(),
        _canon("Handle,Title,Vendor,Variant SKU,Variant Price,Variant Inventory Qty,Image Src",
               "moon-mug,Moon Mug,,SKU-1,18.00,40,https://x/a.jpg"),
    )
    [rec] = diff.records
    assert rec["kind"] == "update"
    assert {"field": "vendor", "before": "Acme", "after": None} in rec["detail"]["changes"]


def test_new_option_combo_is_variant_add_existing_untouched():
    catalog = _catalog_mug()
    diff = compute_diff(
        catalog,
        _canon("Handle,Title,Option1 Name,Option1 Value,Variant Price",
               "moon-mug,Moon Mug,Size,Large,25.00"),
    )
    [rec] = diff.records
    assert rec["kind"] == "update"
    kinds = [v.get("kind") for v in rec["detail"]["variants"]]
    assert "add" in kinds


def test_error_product_classifies_error():
    diff = compute_diff({}, _canon("Handle,Title,Variant Price", "mug,Mug,nope"))
    [rec] = diff.records
    assert rec["kind"] == "error"
    assert rec["detail"]["errors"][0]["column"] == "Variant Price"


def test_fingerprint_stable_and_drift_sensitive():
    d1 = compute_diff(_catalog_mug(), _canon(HEADER, MUG_ROW))
    d2 = compute_diff(_catalog_mug(), _canon(HEADER, MUG_ROW))
    d3 = compute_diff(_catalog_mug(title="Renamed"), _canon(HEADER, MUG_ROW))
    assert d1.fingerprint == d2.fingerprint
    assert d1.fingerprint != d3.fingerprint

Note for test_fingerprint_stable_and_drift_sensitive: _catalog_mug(title="Renamed") must change both title field and CatalogProduct.title. (The helper builds both from fields.)

Run → FAIL.

• Step 2: Implement diff.py per the contract. JSON-safety helper:

def _json_safe(v: object) -> object:
    if isinstance(v, Decimal):
        return str(v)
    if isinstance(v, tuple):
        return list(v)
    return v

Decimal comparison nuance: compare Decimal("18.00") == Decimal("18.0") → True (fine). But fingerprint serializes str(v) — "18.00" vs DB round-trip "18.00" (NUMERIC preserves scale) — stable.

• Step 3: Run → PASS. Step 4: Commit — git commit -am "feat(products): diff engine — add/update/unchanged/error + INV-11 fingerprint"

---

Task 6: repo.py — catalog snapshot + draft/run SQL

Files:

• Create: backend/app/domains/products/repo.py
• Tested through Tasks 7–8 (service tests) — keep functions thin/SQL-only.

Complete function list (all take conn: psycopg.Connection first; all catalog queries storefront-scoped, INV-14):

def load_catalog(conn, storefront_id) -> dict[str, CatalogProduct]
    # 3 queries (products, variants, images for the storefront), assembled in Python.
    # variant.fields["variant_image"] = the linked product_image.source_url (LEFT JOIN).

def product_count(conn, storefront_id) -> int
def image_problem_count(conn, storefront_id) -> int   # status IN rejected_*, failed (0 this slice)
def latest_run_id(conn, storefront_id) -> int | None

def insert_draft(conn, storefront_id, account_id, file_name, dialect, file_bytes,
                 summary: dict, records: list, fingerprint, unknown_columns) -> dict
    # INSERT … RETURNING id, expires_at(now()+interval '1 hour'), created_at; returns the §6.4 draft payload dict
def get_draft_row(conn, storefront_id, draft_id) -> Row | None   # full row incl. file_bytes, expires_at
def draft_records(conn, storefront_id, draft_id, kind: str | None, limit, offset) -> list[dict]
    # jsonb_array_elements over records with optional kind filter + LIMIT/OFFSET, preserving order (WITH ORDINALITY)
def delete_draft(conn, storefront_id, draft_id) -> None
def sweep_expired_drafts(conn) -> None                 # DELETE WHERE expires_at < now()

def insert_run(conn, storefront_id, account_id, file_name, dialect,
               added, updated, errored, status) -> int
def insert_run_errors(conn, run_id, errors: list[dict]) -> None   # executemany
def list_runs(conn, storefront_id, limit, offset) -> list[dict]
    # newest first; each: id, file_name, dialect, created_at(iso), completed_at(iso|None), status, counts, by (account email JOIN)
def get_run(conn, storefront_id, run_id) -> dict | None
    # run + its errors array + by-email; plus image_progress {done:0,total:0} and image_outcomes [] (SLICE-7 fills)

# Apply primitives (used inside the confirm transaction):
def insert_product(conn, storefront_id, handle, resolved_fields: dict, option_names) -> int
def update_product(conn, product_id, changed_fields: dict) -> None     # SET …, updated_at=now()
def insert_variant(conn, product_id, position, options, resolved_fields: dict, image_id) -> int
def update_variant(conn, variant_id, changed_fields: dict, image_id: int | None | EllipsisType) -> None
def get_or_create_image(conn, product_id, source_url, position, alt_text, run_id) -> int
def update_image(conn, image_id, position=None, alt_text=None) -> None

Notes:

• Dynamic SET clauses: build from a dict with psycopg.sql.SQL/Identifier composition (never f-strings of values).
• tags binds as Python list → TEXT[].
• Timestamps serialize ISO-8601 (.isoformat()).
• draft_records SQL:

SELECT rec FROM import_draft d,
       jsonb_array_elements(d.records) WITH ORDINALITY AS r(rec, ord)
WHERE d.id = %s AND d.storefront_id = %s
  AND (%s::text IS NULL OR rec->>'kind' = %s)
ORDER BY ord LIMIT %s OFFSET %s

• Step 1: Write repo.py per the list. Step 2: lint-imports + import check. Step 3: Commit — git commit -am "feat(products): repo — catalog snapshot, draft/run SQL, upsert primitives"

---

Task 7: service.py part 1 — validate → draft, records, discard + TEL-1 + platform/telemetry.py

Files:

• Create: backend/app/platform/telemetry.py
• Create: backend/app/domains/products/service.py (part 1)
• Modify: backend/app/domains/products/__init__.py (export service functions)
• Test: backend/tests/test_products_service.py

telemetry.py:

"""Structured log-event telemetry (SD-0002 §9.1). One JSON object per event on the
`ecomm.telemetry` logger — counts and durations only; never file names, URLs,
catalog content, or secret bytes (§6.3-handbook)."""
from __future__ import annotations

import json
import logging

_logger = logging.getLogger("ecomm.telemetry")


def emit(event: str, **fields: object) -> None:
    _logger.info(json.dumps({"event": event, **fields}, sort_keys=True, default=str))

service.py part 1:

def import_validate(conn, storefront_id: int, account_id: int, file_name: str, data: bytes) -> dict:
    """Upload → validate → diff → persist draft (PUC-2/3; INV-11 read-only). Raises FileRejected."""
    started = time.monotonic()
    repo.sweep_expired_drafts(conn)
    parsed = codec.parse_csv(data)                      # PUC-5a gates
    products = validate.build_products(parsed)
    catalog = repo.load_catalog(conn, storefront_id)    # read-only
    diff_result = diff.compute_diff(catalog, products)
    draft = repo.insert_draft(conn, storefront_id, account_id, file_name, parsed.dialect,
                              data, diff_result.summary, diff_result.records,
                              diff_result.fingerprint, parsed.unknown_columns)
    conn.commit()
    telemetry.emit("import_draft_created", storefront_id=storefront_id,
                   dialect=parsed.dialect, row_count=len(parsed.rows),
                   adds=diff_result.summary["adds"], updates=diff_result.summary["updates"],
                   unchanged=diff_result.summary["unchanged"], errors=diff_result.summary["errors"],
                   unknown_columns_count=len(parsed.unknown_columns),
                   duration_ms=int((time.monotonic() - started) * 1000))      # TEL-1
    return draft

def get_draft(conn, storefront_id, draft_id) -> dict:        # raises DraftNotFound | DraftExpired
def get_draft_records(conn, storefront_id, draft_id, kind=None, limit=100, offset=0) -> list[dict]
def discard_draft(conn, storefront_id, draft_id) -> None:    # PUC-3a; idempotent (absent == fine); commits

get_draft returns the §6.4 payload: {"id", "file_name", "dialect", "summary", "unknown_columns", "expires_at"} — never file_bytes/records. Expired → raise DraftExpired (and delete the row + commit — lazy expiry).

• Step 1: Write failing tests — backend/tests/test_products_service.py (fixtures reused by Task 8; uses Task 1's migrated_conn pattern):

"""products service — drafts: validate/preview/discard (PUC-2/3/3a/5a; INV-11)."""
import psycopg
import pytest

from app.domains import products
from app.platform import db

GOOD_CSV = b"Handle,Title,Vendor,Variant Price\nmoon-mug,Moon Mug,Acme,18.00\nstar-tee,Star Tee,Acme,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 test_import_validate_creates_draft_with_summary(migrated_conn, merchant):
    draft = products.import_validate(
        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
    assert draft["dialect"] == "canonical"
    assert draft["summary"] == {"adds": 2, "updates": 0, "unchanged": 0, "errors": 0}
    assert draft["expires_at"]


def test_validate_writes_nothing_to_catalog_inv11(migrated_conn, merchant):
    products.import_validate(
        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
    assert migrated_conn.execute("SELECT count(*) FROM product").fetchone()[0] == 0
    assert migrated_conn.execute("SELECT count(*) FROM variant").fetchone()[0] == 0


def test_file_rejection_leaves_no_draft(migrated_conn, merchant):
    with pytest.raises(products.FileRejected):
        products.import_validate(
            migrated_conn, merchant["storefront_id"], merchant["account_id"], "bad.csv",
            b"Vendor,Price\nAcme,1\n")
    assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0


def test_records_paging_and_kind_filter(migrated_conn, merchant):
    draft = products.import_validate(
        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
    recs = products.get_draft_records(migrated_conn, merchant["storefront_id"], draft["id"])
    assert [r["handle"] for r in recs] == ["moon-mug", "star-tee"]
    adds = products.get_draft_records(
        migrated_conn, merchant["storefront_id"], draft["id"], kind="add", limit=1)
    assert len(adds) == 1 and adds[0]["kind"] == "add"


def test_discard_deletes_no_trace_puc3a(migrated_conn, merchant):
    draft = products.import_validate(
        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
    products.discard_draft(migrated_conn, merchant["storefront_id"], draft["id"])
    assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0
    products.discard_draft(migrated_conn, merchant["storefront_id"], draft["id"])  # idempotent


def test_draft_scoped_to_storefront_inv14(migrated_conn, merchant):
    draft = products.import_validate(
        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
    other_sf = migrated_conn.execute(
        "INSERT INTO storefront (name) VALUES ('Other') RETURNING id").fetchone()[0]
    migrated_conn.commit()
    with pytest.raises(products.DraftNotFound):
        products.get_draft(migrated_conn, other_sf, draft["id"])


def test_expired_draft_raises_and_lazily_deletes(migrated_conn, merchant):
    draft = products.import_validate(
        migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
    migrated_conn.execute(
        "UPDATE import_draft SET expires_at = now() - interval '1 minute' WHERE id = %s",
        (draft["id"],))
    migrated_conn.commit()
    with pytest.raises(products.DraftExpired):
        products.get_draft(migrated_conn, merchant["storefront_id"], draft["id"])
    assert migrated_conn.execute("SELECT count(*) FROM import_draft").fetchone()[0] == 0


def test_tel1_emitted(migrated_conn, merchant, caplog):
    import logging
    with caplog.at_level(logging.INFO, logger="ecomm.telemetry"):
        products.import_validate(
            migrated_conn, merchant["storefront_id"], merchant["account_id"], "cat.csv", GOOD_CSV)
    assert any('"event": "import_draft_created"' in r.message.replace("'", '"')
               or '"event":"import_draft_created"' in r.message.replace(" ", "")
               for r in caplog.records)

(For the TEL assertion, simplest robust form: json.loads(record.message)["event"] == "import_draft_created" over caplog.records — use that.)

• Step 2: Implement telemetry + service part 1 + exports in __init__.py (import_validate, get_draft, get_draft_records, discard_draft). Step 3: Run the file → PASS. Also lint-imports. Step 4: Commit — git commit -am "feat(products): import_validate → draft, preview records, discard + TEL-1"

---

Task 8: service.py part 2 — confirm/apply, runs, summary + TEL-2/6 + invariant tests

Files:

• Modify: backend/app/domains/products/service.py, __init__.py
• Test: append to backend/tests/test_products_service.py; create backend/tests/test_products_invariants.py

confirm_draft(conn, storefront_id, account_id, draft_id) -> int:

1. get_draft_row → DraftNotFound / expired → DraftExpired (lazy delete + commit first).
2. Re-derive: parse_csv(file_bytes) → build_products → load_catalog → compute_diff. Fingerprint ≠ stored → raise PreviewStale (draft kept; rollback any sweep work first).
3. adds + updates == 0 → raise NothingToApply.
4. In ONE transaction (the connection is already non-autocommit; just don't commit until done):
   • run_id = repo.insert_run(…, status='complete') — completed_at = now() (image phase is the SLICE-5 no-op stub; SLICE-7 changes this to applying→fetching_images).
   • For each record kind add: insert_product (resolved fields = file fields with None→CLEAR_DEFAULTS/NULL; title required, status default active, published default TRUE, type standalone, tags []), then variants (insert_variant, position = Variant Position or file order 1-based; variant_image → get_or_create_image first, link image_id), then images (get_or_create_image each, import_run_id=run_id).
   • For each update: apply ONLY the diffed changes — update_product with changed product fields; per variant-diff entry: add → insert_variant, update → update_variant; per image-diff entry: add → get_or_create_image, update → update_image. (Re-walk the freshly recomputed diff records' details, not the stored JSONB — same content by fingerprint equality, but typed values come from the recomputation. Implement by having compute_diff also return an apply plan: DiffResult.plan: list[ProductPlan] where ProductPlan carries the canonical product + its classification + matched catalog ids. This avoids re-parsing JSON detail. Add plan to DiffResult in this task — a dataclass ProductPlan: kind: str; canonical: CanonicalProduct; catalog: CatalogProduct | None; product_changes: dict[str, object]; variant_plans: list[VariantPlan]; image_plans: list[ImagePlan] with VariantPlan(kind, canonical: CanonicalVariant, catalog_id: int | None, changes: dict), ImagePlan(kind, src, position, alt_text, image_id: int | None, changes: dict). The JSON records are derived from the plan so preview and apply can never diverge.)
   • insert_run_errors(run_id, [e.as_json() for e in all error rows]).
   • delete_draft.
   • conn.commit().
5. On any unexpected exception inside step 4: conn.rollback(), telemetry.emit("import_apply_failed", draft_id=…, storefront_id=…, error_class=type(exc).__name__) (TEL-6), re-raise.
6. After commit: telemetry.emit("import_run_completed", run_id=…, storefront_id=…, added=…, updated=…, errored=…, duration_ms=…) (TEL-2). Return run_id.

Also: list_runs(conn, storefront_id, limit=50, offset=0), get_run(conn, storefront_id, run_id) (→ RunNotFound), summary(conn, storefront_id) → {"product_count", "image_problem_count", "latest_run_id"} — thin repo passthroughs.

Refactor note: Task 5's compute_diff gains the plan field here — update diff.py + keep Task 5 tests green (records unchanged).

• Step 1: Write failing tests. Append to test_products_service.py:

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)   # second identical draft
    products.confirm_draft(migrated_conn, merchant["storefront_id"], merchant["account_id"], other["id"])
    # catalog changed since `draft` was validated -> its diff no longer holds
    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)      # identical file again
    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):
    import json, logging
    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)

And backend/tests/test_products_invariants.py:

"""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)   # a file naming only one product
    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) == []
    # same handles import independently into the second storefront
    _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)

    real = repo.insert_run_errors
    def boom(*a, **k):
        raise RuntimeError("mid-apply crash")
    monkeypatch.setattr(service.repo, "insert_run_errors", boom)
    with caplog.at_level(logging.INFO, logger="ecomm.telemetry"):
        with pytest.raises(RuntimeError):
            products.confirm_draft(migrated_conn, sf, acct, d["id"])
    monkeypatch.setattr(service.repo, "insert_run_errors", real)
    # zero catalog change, run not recorded, draft intact (§6.9 failure mode 1)
    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)

(insert_run_errors is called unconditionally in apply — with an empty list it's a no-op executemany — so the monkeypatch always fires. Implement apply that way.)

• Step 2: Implement (diff plan refactor + service part 2 + exports: confirm_draft, list_runs, get_run, summary).
• Step 3: Run …/pytest tests/test_products_service.py tests/test_products_invariants.py tests/test_products_diff.py -q → PASS.
• Step 4: Full backend suite …/pytest -q → PASS. Step 5: Commit — git commit -am "feat(products): confirm/apply in one transaction, runs, summary — INV-10/11/14 + TEL-2/6"

---

Task 9: BFF endpoints + sample CSV (DOC-3)

Files:

• Modify: backend/app/main.py
• Create: backend/app/domains/products/sample.csv (+ tiny loader export in __init__.py: SAMPLE_CSV_PATH = Path(__file__).parent / "sample.csv")
• Test: backend/tests/test_products_endpoints.py

Endpoint code to add inside create_app() (after the storefronts endpoint; imports at top: from app.domains import products, from fastapi import File, Query, UploadFile, from fastapi.responses import PlainTextResponse):

    def _merchant_gate(conn, sess):
        """Session + storefront membership (SD-0002 §6.4 gate). Returns (account, sf) or an error response."""
        if sess is None:
            return _error(401, "unauthenticated", "You are not signed in.")
        account = accounts.get_account(conn, sess["account_id"])
        if account is None:
            return _error(401, "unauthenticated", "You are not signed in.")
        sf = storefronts.storefront_for(conn, account.id)
        if sf is None:
            return _error(404, "no_storefront", "Create your storefront first.")
        return account, sf

    @app.post("/api/products/imports")
    async def create_import(
        file: UploadFile = File(...),
        conn: psycopg.Connection = Depends(get_conn),
        sess: dict | None = Depends(get_session),
    ):
        """Upload a CSV → validated draft (PUC-2; §6.5.2). No draft on rejection (PUC-5a)."""
        gate = _merchant_gate(conn, sess)
        if isinstance(gate, JSONResponse):
            return gate
        account, sf = gate
        data = await file.read()
        if len(data) > products.MAX_FILE_BYTES:
            return _error(413, "file_too_large", "This file is larger than 10 MB.")
        try:
            draft = products.import_validate(conn, sf.id, account.id, file.filename or "import.csv", data)
        except products.FileRejected as exc:
            return _error(400, exc.code, exc.message)
        return JSONResponse(status_code=201, content=draft)

    @app.get("/api/products/imports/drafts/{draft_id}")
    def get_import_draft(draft_id: int, conn=Depends(get_conn), sess=Depends(get_session)):
        gate = _merchant_gate(conn, sess)
        if isinstance(gate, JSONResponse):
            return gate
        _account, sf = gate
        try:
            return products.get_draft(conn, sf.id, draft_id)
        except products.DraftNotFound:
            return _error(404, "not_found", "No such import preview.")
        except products.DraftExpired:
            return _error(410, "draft_expired", "This preview expired — upload the file again.")

    @app.get("/api/products/imports/drafts/{draft_id}/records")
    def get_import_draft_records(
        draft_id: int,
        kind: str | None = Query(default=None, pattern="^(add|update|unchanged|error)$"),
        limit: int = Query(default=100, ge=1, le=500),
        offset: int = Query(default=0, ge=0),
        conn=Depends(get_conn), sess=Depends(get_session),
    ):
        # gate + same 404/410 mapping; body: {"records": products.get_draft_records(...)}

    @app.post("/api/products/imports/drafts/{draft_id}/confirm")
    def confirm_import(draft_id: int, conn=Depends(get_conn), sess=Depends(get_session)):
        # gate; try confirm_draft → 201 {"run_id": run_id}
        # DraftNotFound→404 · DraftExpired→410 · PreviewStale→409 "preview_stale",
        #   "Your catalog changed since this preview — upload the file again."
        # NothingToApply→409 "nothing_to_apply", "Nothing to change — your catalog already matches this file."

    @app.delete("/api/products/imports/drafts/{draft_id}")
    def cancel_import(draft_id: int, conn=Depends(get_conn), sess=Depends(get_session)):
        # gate; discard_draft (idempotent); 204

    @app.get("/api/products/imports/runs")
    def list_import_runs(limit: int = Query(default=50, ge=1, le=200), offset: int = Query(default=0, ge=0),
                         conn=Depends(get_conn), sess=Depends(get_session)):
        # gate; {"runs": products.list_runs(...)}

    @app.get("/api/products/imports/runs/{run_id}")
    def get_import_run(run_id: int, conn=Depends(get_conn), sess=Depends(get_session)):
        # gate; RunNotFound→404; body: the run dict

    @app.get("/api/products/summary")
    def products_summary(conn=Depends(get_conn), sess=Depends(get_session)):
        # gate; products.summary(...)

    @app.get("/api/products/sample.csv")
    def products_sample():
        """DOC-3 (PUC-11): the documented sample — no auth gate (it's documentation)."""
        return PlainTextResponse(
            products.SAMPLE_CSV_PATH.read_text(),
            media_type="text/csv",
            headers={"content-disposition": 'attachment; filename="ecomm-products-sample.csv"'},
        )

sample.csv content (3 worked examples: simple product, 3-variant tee with two options, image-only row):

Handle,Title,Description,Vendor,Type,Google Product Category,Tags,Status,Published,Option1 Name,Option1 Value,Option2 Name,Option2 Value,Variant SKU,Variant Price,Variant Inventory Qty,Image Src,Image Position,Image Alt Text
moon-mug,Moon Mug,"<p>A ceramic mug glazed in moonlight grey.</p>",Wiggle Goods,standalone,Home & Garden > Kitchen & Dining,"kitchen, mugs",active,TRUE,,,,,WG-MUG-001,18.00,40,https://images.example.com/moon-mug.jpg,1,Moon Mug on a desk
star-tee,Star Tee,"<p>Soft cotton tee with a hand-printed star.</p>",Wiggle Goods,standalone,Apparel & Accessories > Clothing,"apparel, tees",active,TRUE,Size,S,Color,Indigo,WG-TEE-S,24.00,12,https://images.example.com/star-tee.jpg,1,Star Tee flat lay
star-tee,,,,,,,,,,M,,Indigo,WG-TEE-M,24.00,18,,,
star-tee,,,,,,,,,,L,,Indigo,WG-TEE-L,26.00,9,,,
star-tee,,,,,,,,,,,,,,,,https://images.example.com/star-tee-back.jpg,2,Star Tee back print

• Step 1: Failing tests — backend/tests/test_products_endpoints.py, reusing _signed_in_client style from test_storefronts_create.py (copy the helper; add storefront creation):

"""§6.4 /api/products/* endpoint scenarios (PUC-2/3/3a/4/5/5a/8 + gates)."""
import io
import re
from contextlib import contextmanager

from fastapi.testclient import TestClient

from app.main import create_app

GOOD_CSV = b"Handle,Title,Vendor,Variant Price\nmoon-mug,Moon Mug,Acme,18.00\n"


@contextmanager
def _merchant_client(fresh_db_url, email="m@example.com"):
    with TestClient(create_app(database_url=fresh_db_url)) as client:
        client.post("/api/auth/request-code", json={"email": email})
        code = re.search(r"\b(\d{6})\b", client.app.state.mailer.outbox[-1].body).group(1)
        client.post("/api/auth/verify", json={"email": email, "code": code})
        client.post("/api/storefronts", json={})
        yield client


def _upload(client, data=GOOD_CSV, name="cat.csv"):
    return client.post("/api/products/imports", files={"file": (name, io.BytesIO(data), "text/csv")})


def test_upload_returns_201_draft(fresh_db_url):
    with _merchant_client(fresh_db_url) as client:
        resp = _upload(client)
        assert resp.status_code == 201
        body = resp.json()
        assert body["summary"]["adds"] == 1 and body["dialect"] == "canonical"


def test_upload_rejections_carry_codes(fresh_db_url):
    with _merchant_client(fresh_db_url) as client:
        resp = _upload(client, b"Vendor\nAcme\n")
        assert resp.status_code == 400
        assert resp.json()["error"]["code"] == "missing_required_column"
        resp = _upload(client, b"Handle,Title\n" + b"x" * (10 * 1024 * 1024 + 1))
        assert resp.status_code == 413


def test_unauthenticated_401_and_no_storefront_404(fresh_db_url):
    with TestClient(create_app(database_url=fresh_db_url)) as client:
        assert _upload(client).status_code == 401
        client.post("/api/auth/request-code", json={"email": "x@example.com"})
        code = re.search(r"\b(\d{6})\b", client.app.state.mailer.outbox[-1].body).group(1)
        client.post("/api/auth/verify", json={"email": "x@example.com", "code": code})
        assert _upload(client).status_code == 404


def test_preview_confirm_run_flow(fresh_db_url):
    with _merchant_client(fresh_db_url) as client:
        draft = _upload(client).json()
        recs = client.get(f"/api/products/imports/drafts/{draft['id']}/records").json()["records"]
        assert recs[0]["kind"] == "add"
        run_id = client.post(f"/api/products/imports/drafts/{draft['id']}/confirm").json()["run_id"]
        run = client.get(f"/api/products/imports/runs/{run_id}").json()
        assert run["products_added"] == 1 and run["by"] == "m@example.com"
        assert client.get("/api/products/summary").json()["product_count"] == 1
        assert client.get("/api/products/imports/runs").json()["runs"][0]["id"] == run_id


def test_cancel_no_trace_puc3a(fresh_db_url):
    with _merchant_client(fresh_db_url) as client:
        draft = _upload(client).json()
        assert client.delete(f"/api/products/imports/drafts/{draft['id']}").status_code == 204
        assert client.get(f"/api/products/imports/drafts/{draft['id']}").status_code == 404
        assert client.get("/api/products/imports/runs").json()["runs"] == []


def test_confirm_conflicts(fresh_db_url):
    with _merchant_client(fresh_db_url) as client:
        d1 = _upload(client).json()
        client.post(f"/api/products/imports/drafts/{d1['id']}/confirm")
        d2 = _upload(client).json()   # identical -> all unchanged
        resp = client.post(f"/api/products/imports/drafts/{d2['id']}/confirm")
        assert resp.status_code == 409 and resp.json()["error"]["code"] == "nothing_to_apply"


def test_sample_csv_served(fresh_db_url):
    with TestClient(create_app(database_url=fresh_db_url)) as client:
        resp = client.get("/api/products/sample.csv")
        assert resp.status_code == 200
        assert resp.headers["content-type"].startswith("text/csv")
        assert resp.text.startswith("Handle,Title,")


def test_sample_csv_imports_clean(fresh_db_url):
    """DOC-3 honesty: our own sample must validate with zero errors."""
    with _merchant_client(fresh_db_url) as client:
        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

• Step 2: Implement endpoints + sample.csv + SAMPLE_CSV_PATH export. Step 3: Run the file, then the full backend suite + lint-imports → PASS. Step 4: Commit — git commit -am "feat(products): /api/products/* BFF endpoints + sample.csv (DOC-3)"

---

Task 10: Frontend — productsApi.ts + adminRouting.ts (+ tests)

Files:

• Create: frontend/src/productsApi.ts, frontend/src/adminRouting.ts, frontend/src/adminRouting.test.ts

adminRouting.ts:

// Hash routing for admin sections (SD-0002 §5). The URL is the durable handle on a
// section (PUC-8: run detail can be left and returned to); the SD-0001 entry-routing
// rule (routing.ts) still decides whether the admin renders at all.
export type AdminView =
  | { view: "home" }
  | { view: "products" }
  | { view: "import-upload" }
  | { view: "import-preview"; draftId: number }
  | { view: "run-detail"; runId: number };

export function adminViewFor(hash: string): AdminView {
  const parts = hash.replace(/^#\/?/, "").split("/").filter(Boolean);
  if (parts[0] !== "products") return { view: "home" };
  if (parts.length === 1) return { view: "products" };
  if (parts[1] === "import" && parts.length === 2) return { view: "import-upload" };
  if (parts[1] === "imports" && parts[2] === "drafts" && /^\d+$/.test(parts[3] ?? ""))
    return { view: "import-preview", draftId: Number(parts[3]) };
  if (parts[1] === "imports" && parts[2] === "runs" && /^\d+$/.test(parts[3] ?? ""))
    return { view: "run-detail", runId: Number(parts[3]) };
  return { view: "products" };
}

export function hashFor(v: AdminView): string {
  switch (v.view) {
    case "home": return "#/";
    case "products": return "#/products";
    case "import-upload": return "#/products/import";
    case "import-preview": return `#/products/imports/drafts/${v.draftId}`;
    case "run-detail": return `#/products/imports/runs/${v.runId}`;
  }
}

adminRouting.test.ts (vitest): round-trip every view through hashFor→adminViewFor; junk hashes ("", "#/x", "#/products/imports/drafts/abc") land on home/products safely:

import { describe, expect, it } from "vitest";
import { adminViewFor, hashFor, type AdminView } from "./adminRouting";

const VIEWS: AdminView[] = [
  { view: "home" }, { view: "products" }, { view: "import-upload" },
  { view: "import-preview", draftId: 7 }, { view: "run-detail", runId: 12 },
];

describe("adminRouting", () => {
  it("round-trips every view", () => {
    for (const v of VIEWS) expect(adminViewFor(hashFor(v))).toEqual(v);
  });
  it("defaults junk to home/products", () => {
    expect(adminViewFor("")).toEqual({ view: "home" });
    expect(adminViewFor("#/nonsense")).toEqual({ view: "home" });
    expect(adminViewFor("#/products/imports/drafts/abc")).toEqual({ view: "products" });
  });
});

productsApi.ts — typed wrappers, same conventions as api.ts (credentials: "include", errorOf-style envelope; export errorOf from api.ts instead of duplicating):

import { type ApiError } from "./api";

export interface DiffSummary { adds: number; updates: number; unchanged: number; errors: number; }
export interface Draft {
  id: number; file_name: string; dialect: string;
  summary: DiffSummary; unknown_columns: string[]; expires_at: string;
}
export type RecordKind = "add" | "update" | "unchanged" | "error";
export interface RowErrorDetail { line: number; column: string | null; message: string; }
export interface DraftRecord {
  handle: string; title: string; kind: RecordKind; variant_count: number;
  detail: {
    set?: Record<string, unknown>;
    changes?: { field: string; before: unknown; after: unknown }[];
    variants?: { options: (string | null)[]; kind?: string; set?: Record<string, unknown>;
                 changes?: { field: string; before: unknown; after: unknown }[] }[];
    images?: { src: string; kind?: string;
               changes?: { field: string; before: unknown; after: unknown }[] }[];
    errors?: RowErrorDetail[];
  };
}
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;
}
export interface RunDetail extends RunSummary {
  errors: RowErrorDetail[];
  image_progress: { done: number; total: number };
  image_outcomes: unknown[];
}
export interface ProductsSummary {
  product_count: number; image_problem_count: number; latest_run_id: number | null;
}

type Result<T> = { ok: true; value: T } | { ok: false; error: ApiError; status: number };

Functions (each ~6 lines, standard fetch + envelope): getProductsSummary(), uploadImport(file: File) (FormData, no content-type header — the browser sets the boundary), getDraft(id), getDraftRecords(id, kind?, limit?, offset?) → DraftRecord[], confirmDraft(id) → {run_id}, cancelDraft(id), listRuns() → RunSummary[], getRun(id) → RunDetail. Export errorOf from api.ts (one-line change: export async function errorOf…).

• Step 1: Write adminRouting.test.ts → cd frontend && npm test → FAIL. Step 2: Implement both modules + the api.ts export tweak. Step 3: npm test → PASS; npm run build → clean. Step 4: Commit — git commit -am "feat(products-ui): typed products API client + admin hash routing"

---

Task 11: Admin nav + ProductsPage + styles

Files:

• Modify: frontend/src/screens/Admin.tsx
• Create: frontend/src/screens/products/ProductsPage.tsx, frontend/src/styles/products.css
• Modify: frontend/src/styles/index.css (add @import "./products.css";)

Admin.tsx changes: keep TopBar exactly as-is; below it add a nav strip and dispatch on the hash view:

import { useEffect, useState } from "react";
import { adminViewFor, hashFor, type AdminView } from "../adminRouting";
import ProductsPage from "./products/ProductsPage";
import ImportUpload from "./products/ImportUpload";
import ImportPreview from "./products/ImportPreview";
import RunDetail from "./products/RunDetail";
// existing imports stay

export function navigate(view: AdminView) {
  window.location.hash = hashFor(view);
}

export default function Admin({ storefrontName, email, welcome, onSignedOut }: Props) {
  const [view, setView] = useState<AdminView>(adminViewFor(window.location.hash));
  useEffect(() => {
    const onHash = () => setView(adminViewFor(window.location.hash));
    window.addEventListener("hashchange", onHash);
    return () => window.removeEventListener("hashchange", onHash);
  }, []);
  // … signOut unchanged …
  return (
    <Screen plain>
      <TopBar … unchanged … />
      <nav className="adminnav" aria-label="Admin sections">
        <a className={`adminnav__item${view.view === "home" ? " adminnav__item--active" : ""}`} href="#/">Overview</a>
        <a className={`adminnav__item${view.view !== "home" ? " adminnav__item--active" : ""}`} href="#/products">Products</a>
      </nav>
      <main className="screen__main">
        {view.view === "home" && (/* the existing empty-state block, unchanged, welcome banner included */)}
        {view.view === "products" && <ProductsPage />}
        {view.view === "import-upload" && <ImportUpload />}
        {view.view === "import-preview" && <ImportPreview draftId={view.draftId} />}
        {view.view === "run-detail" && <RunDetail runId={view.runId} email={email} />}
      </main>
    </Screen>
  );
}

(RunDetail doesn't actually need email — the run payload carries by; do not pass it. Signature: RunDetail({ runId }: { runId: number }).)

ProductsPage.tsx (§5.2, SLICE-5 shape — no notices band yet, Export disabled, honest pre-#14 body):

// Products page (SD-0002 §5.2) — the catalog's home: where imports start and history
// lives. SLICE-5: export is disabled (SLICE-6); the browsable list is #14's.
import { useEffect, useState } from "react";
import { getProductsSummary, listRuns, type ProductsSummary, type RunSummary } from "../../productsApi";
import { Banner } from "../../ui/kit";

export default function ProductsPage() {
  const [summary, setSummary] = useState<ProductsSummary | null>(null);
  const [runs, setRuns] = useState<RunSummary[] | null>(null);
  const [failed, setFailed] = useState(false);

  async function load() {
    setFailed(false);
    const [s, r] = await Promise.all([getProductsSummary(), listRuns()]);
    if (!s.ok || !r.ok) { setFailed(true); return; }
    setSummary(s.value); setRuns(r.value);
  }
  useEffect(() => { void load(); }, []);

  if (failed) return (
    <Banner tone="attn" title="Couldn't load your products">
      Something went wrong on our side. <button className="linklike" onClick={() => void load()}>Retry</button>
    </Banner>
  );
  if (!summary || !runs) return <p className="note">Loading…</p>;

  const empty = summary.product_count === 0;
  return (
    <div className="products">
      <header className="products__header">
        <h1>Products{!empty && <span className="products__count"> · {summary.product_count.toLocaleString()}</span>}</h1>
        <div className="products__actions">
          <button className="btn-secondary" disabled title="Export arrives in a coming release">Export</button>
          <a className="btn-primary" href="#/products/import">Import products</a>
        </div>
      </header>
      {empty ? (
        <div className="empty">
          <p className="empty__copy">No products yet. Bulk import is how product data gets in.</p>
          <a className="btn-primary" href="#/products/import">Import products</a>
          <p className="note"><a href="/api/products/sample.csv" download>Download sample CSV</a></p>
        </div>
      ) : (
        <p className="note">Your catalog is loaded. The browsable product list arrives with an upcoming release.</p>
      )}
      <section className="products__history">
        <h2>Import history</h2>
        {runs.length === 0 ? <p className="note">No imports yet.</p> : (
          <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></tr></thead>
            <tbody>
              {runs.map((r) => (
                <tr key={r.id} className="datatable__rowlink" onClick={() => { window.location.hash = `#/products/imports/runs/${r.id}`; }}>
                  <td>{new Date(r.created_at).toLocaleString()}</td>
                  <td>{r.file_name}</td><td>{r.dialect}</td>
                  <td>{r.products_added}</td><td>{r.products_updated}</td><td>{r.rows_errored}</td>
                  <td>{r.status === "complete" ? "Complete" : r.status}</td>
                </tr>
              ))}
            </tbody>
          </table>
        )}
      </section>
    </div>
  );
}

products.css — tokens-based; cover: .adminnav (horizontal strip under the topbar), .products__header (flex, actions right), .products__count (muted), .btn-secondary (outline twin of .btn-primary — define here, generic), .linklike, .datatable (+ __rowlink hover/cursor), .tiles/.tile/.tile--active + per-kind tile accents (--st-add #1F8A5B, --st-update #B5830F, --st-error #C2513E from the design bundle; define as --st-* custom props at the top of this file), .difflist row + expand area, .diffchange (before→after with +/− glyphs — not color-only, §6.6 accessibility), .errortable, .sticky-footer. ~120 lines.

• Step 1: Implement (stub the three not-yet-written screens as export default function ImportUpload() { return null; } etc. so the build stays green — they're replaced in Tasks 12–14). Step 2: npm run build && npm test → green. Step 3: Commit — git commit -am "feat(products-ui): admin nav + Products page with import history (§5.2)"

---

Task 12: ImportUpload screen (§5.3)

Files: Create (replace stub): frontend/src/screens/products/ImportUpload.tsx

Contract: header "Import products" + back link to #/products; a file picker (<input type="file" accept=".csv,text/csv"> behind a styled label) with caption "CSV, up to 5,000 rows"; help line "Works with the canonical format." + sample-CSV link; selecting a file uploads immediately (PUC-2 — no extra click); uploading state shows "Validating…" and disables the picker; file-level rejection (400/413) renders an attn Banner with the server's error.message, picker stays for retry (PUC-5a); success navigates to #/products/imports/drafts/{id}.

// Import — upload (SD-0002 §5.3). Selecting a file starts upload + validation
// immediately; file-level rejections (PUC-5a) render in place with the picker live.
import { useRef, useState } from "react";
import { uploadImport } from "../../productsApi";
import { Banner } from "../../ui/kit";

export default function ImportUpload() {
  const [busy, setBusy] = useState(false);
  const [error, setError] = useState<string | null>(null);
  const inputRef = useRef<HTMLInputElement>(null);

  async function onPick(files: FileList | null) {
    const file = files?.[0];
    if (!file) return;
    setBusy(true); setError(null);
    const resp = await uploadImport(file);
    setBusy(false);
    if (inputRef.current) inputRef.current.value = "";
    if (!resp.ok) { setError(resp.error.message); return; }
    window.location.hash = `#/products/imports/drafts/${resp.value.id}`;
  }

  return (
    <div className="products products--narrow">
      <p><a href="#/products">← Products</a></p>
      <h1>Import products</h1>
      {error && <Banner tone="attn" title="That file can't be imported">{error}</Banner>}
      <label className={`dropzone${busy ? " dropzone--busy" : ""}`}>
        <input ref={inputRef} type="file" accept=".csv,text/csv" disabled={busy}
               onChange={(e) => void onPick(e.target.files)} />
        <span className="dropzone__title">{busy ? "Validating…" : "Choose a CSV file"}</span>
        <span className="note">CSV, up to 5,000 rows</span>
      </label>
      <p className="note">
        Works with the canonical format. <a href="/api/products/sample.csv" download>Download sample CSV</a>
      </p>
    </div>
  );
}

Add .dropzone styles to products.css (bordered dashed block, hidden native input).

• Steps: implement → npm run build && npm test green → commit git commit -am "feat(products-ui): import upload screen (§5.3, PUC-2/5a)".

---

Task 13: ImportPreview screen (§5.4 — the consent gate)

Files: Create (replace stub): frontend/src/screens/products/ImportPreview.tsx

Contract:

• Load getDraft(draftId); 404 → banner "This preview is gone" + link back; 410 → banner "This preview expired — upload the file again" + link to #/products/import.
• Header: "Import preview — {file_name}"; dialect line "Canonical format".
• Warnings band when unknown_columns.length > 0: "Columns not imported: a, b, c".
• Four summary tiles (N to add / M to update / U unchanged / K errors); clicking a tile sets the kind filter (active tile highlighted); default filter: null (all).
• Detail list via getDraftRecords(draftId, kind, 100, offset) with a "Show more" button when a full page came back; each row: handle · title · kind chip · variant count; expandable (<details>) drill-in:
  • add → detail.set as a field list (field: value), plus per-variant sets and image srcs;
  • update → detail.changes rows rendered field: before → after (+/− glyphs for set/clear), variants/images subsections the same;
  • error → its detail.errors rows.
• Under the error tile filter additionally render the flat error table: line · column · message.
• Sticky footer: primary Import N products (N = adds + updates; disabled when N === 0 with the note "Nothing to change — your catalog already matches this file", PUC-10) → confirmDraft → on 201 navigate #/products/imports/runs/{run_id}; on 409 preview_stale/410 → attn banner prompting re-upload; on 409 nothing_to_apply → same disabled note. Secondary Cancel → cancelDraft then #/products (PUC-3a).

Implementation: one component + two small local helpers (KindChip({kind}), ChangeRow({field, before, after}) rendering String(before ?? "—") → String(after ?? "—")). Full code follows the same patterns as Tasks 11–12 (useState/useEffect, productsApi, Banner); ~170 lines.

• Steps: implement → build + vitest green → commit git commit -am "feat(products-ui): import preview — tiles, drill-in diffs, confirm/cancel (§5.4)".

---

Task 14: RunDetail screen (§5.5, no images section)

Files: Create (replace stub): frontend/src/screens/products/RunDetail.tsx

Contract: load getRun(runId) (404 → banner + back link). Header: file name · "imported {localized created_at} by {by}" · dialect. Result summary line: added / updated / errors (the §5.5 mirror of the preview's numbers). Error table when errors.length > 0 (line · column · message — exactly the preview's error table, PUC-5); else nothing. No images section this slice (SLICE-7). Status renders "Complete" / "Complete with problems" (rows_errored > 0 runs still say what the server says — display status mapped: complete → "Complete", complete_with_problems → "Complete with problems", others verbatim). Back link to #/products.

~70 lines, same patterns.

• Steps: implement → npm run build && npm test → green → full backend suite still green → commit git commit -am "feat(products-ui): import run detail (§5.5, PUC-4/5/8)".

---

Task 15: Playwright stand-up (first E2E suite)

Files:

• Create: e2e/package.json, e2e/playwright.config.ts, e2e/serve.sh, e2e/helpers.ts, e2e/.gitignore (node_modules/, .backend.log, test-results/, playwright-report/), e2e/fixtures/{good.csv,mixed-errors.csv,missing-title.csv}
• Modify: scripts/check.sh — no (E2E stays out of the CI gate this slice: the Gitea runner has no browsers; §10.6 names the gap). Instead create scripts/e2e.sh.

e2e/package.json:

{
  "name": "wiggleverse-ecomm-e2e",
  "private": true,
  "scripts": { "test": "playwright test" },
  "devDependencies": { "@playwright/test": "^1.48.0" }
}

e2e/serve.sh (the webServer command — fresh DB, log-mailer, backend serving the built SPA):

#!/usr/bin/env bash
# E2E server: fresh ecomm_e2e database, LogMailer (codes land in .backend.log),
# backend on :8765 serving the built SPA (the deployed topology, SD-0001 §6.2).
set -euo pipefail
repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"

if [ ! -f "$repo_root/frontend/dist/index.html" ]; then
  ( cd "$repo_root/frontend" && npm run build )
fi

"$repo_root/.venv/bin/python" - <<'PY'
import psycopg
admin = psycopg.connect("postgresql://ecomm:ecomm@localhost:5432/postgres", autocommit=True)
admin.execute("SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname='ecomm_e2e' AND pid <> pg_backend_pid()")
admin.execute("DROP DATABASE IF EXISTS ecomm_e2e")
admin.execute("CREATE DATABASE ecomm_e2e")
PY

export ECOMM_DATABASE_URL="postgresql://ecomm:ecomm@localhost:5432/ecomm_e2e"
export ECOMM_MAILER=log
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/playwright.config.ts:

import { defineConfig } from "@playwright/test";

export default defineConfig({
  testDir: "./tests",
  timeout: 60_000,
  retries: 0,
  workers: 1, // one shared backend + log file; sign-ins interleave otherwise
  use: { baseURL: "http://localhost:8765" },
  webServer: {
    command: "bash ./serve.sh",
    url: "http://localhost:8765/healthz",
    reuseExistingServer: false,
    timeout: 120_000,
  },
});

e2e/helpers.ts:

import { expect, type Page } from "@playwright/test";
import { readFile } from "node:fs/promises";
import { join } from "node:path";

const LOG = join(__dirname, ".backend.log");
let seq = 0;

export function freshEmail(): string {
  return `merchant${Date.now()}-${seq++}@example.com`;
}

async function codeFor(email: string): Promise<string> {
  // LogMailer writes "LogMailer -> <email> | <subject>\n<body with the 6-digit code>".
  for (let i = 0; i < 50; i++) {
    const log = await readFile(LOG, "utf8").catch(() => "");
    const at = log.lastIndexOf(`LogMailer -> ${email}`);
    if (at >= 0) {
      const m = log.slice(at).match(/\b(\d{6})\b/);
      if (m) return m[1];
    }
    await new Promise((r) => setTimeout(r, 200));
  }
  throw new Error(`no code logged for ${email}`);
}

export async function signUpWithStorefront(page: Page, email = freshEmail()): Promise<string> {
  await page.goto("/");
  await page.getByRole("button", { name: /get started/i }).click();
  await page.getByLabel(/email/i).fill(email);
  await page.getByRole("button", { name: /code/i }).click();
  const code = await codeFor(email);
  await page.getByLabel(/code/i).fill(code);
  await page.getByRole("button", { name: /verify|continue|sign/i }).click();
  await page.getByRole("button", { name: /create/i }).click();
  await expect(page.locator(".topbar")).toBeVisible();
  return email;
}

export async function gotoProducts(page: Page) {
  await page.getByRole("link", { name: "Products" }).click();
  await expect(page.getByRole("heading", { name: /products/i })).toBeVisible();
}

export async function uploadFixture(page: Page, fixture: string) {
  await page.getByRole("link", { name: /import products/i }).first().click();
  await page.setInputFiles('input[type="file"]', join(__dirname, "fixtures", fixture));
}

Calibrate the selectors against the real screens while writing Task 16 (SignIn/CreateStorefront labels/buttons are SD-0001's — read frontend/src/screens/SignIn.tsx / CreateStorefront.tsx and use their actual accessible names; the helper above is the shape, the selectors must be verified, not trusted).

Fixtures:

# good.csv
Handle,Title,Vendor,Tags,Status,Published,Option1 Name,Option1 Value,Variant SKU,Variant Price,Variant Inventory Qty
moon-mug,Moon Mug,Wiggle Goods,"kitchen, mugs",active,TRUE,,,WG-MUG-001,18.00,40
star-tee,Star Tee,Wiggle Goods,apparel,active,TRUE,Size,S,WG-TEE-S,24.00,12
star-tee,,,,,,,M,WG-TEE-M,24.00,18

# mixed-errors.csv
Handle,Title,Variant Price
good-mug,Good Mug,10.00
bad-tee,Bad Tee,not-a-price
also-good,Also Good,5.00

# missing-title.csv
Handle,Vendor
mug,Acme

(Strip the # name comment lines — three separate files.)

scripts/e2e.sh:

#!/usr/bin/env bash
# E2E browser gate (SD-0002 §6.8) — Playwright against a fresh local stack.
# Not yet part of check.sh/CI: the CI runner has no browsers (§10.6 gap).
set -euo pipefail
repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$repo_root/e2e"
if [ ! -d node_modules ]; then
  npm install
  npx playwright install chromium
fi
npx playwright test "$@"

chmod +x e2e/serve.sh scripts/e2e.sh.

• Steps: scaffold; cd e2e && npm install && npx playwright install chromium; write a smoke spec tests/smoke.spec.ts (sign up → admin topbar visible → Products nav exists) to prove the harness; bash scripts/e2e.sh → smoke PASSES; commit git commit -am "test(e2e): stand up Playwright — fresh-DB server harness + sign-up helper".

---

Task 16: The four SLICE-5 E2E scenarios

Files: Create e2e/tests/import-preview-confirm.spec.ts, import-errors.spec.ts, import-file-rejected.spec.ts, import-cancel.spec.ts; delete smoke.spec.ts (subsumed).

Scenario contracts (DoD names, §6.8):

e2e_import_preview_confirm:

test("e2e_import_preview_confirm", async ({ page }) => {
  await signUpWithStorefront(page);
  await gotoProducts(page);
  await uploadFixture(page, "good.csv");
  await expect(page.getByText(/2 to add/i)).toBeVisible();        // tiles
  await page.getByText("star-tee").click();                        // drill-in
  await expect(page.getByText(/WG-TEE-S/)).toBeVisible();          // field detail
  await page.getByRole("button", { name: /import 2 products/i }).click();
  await expect(page.getByText(/2.*added|added.*2/i)).toBeVisible();  // run detail
  await page.getByRole("link", { name: /products/i }).first().click();
  await expect(page.getByText(/Products · 2/)).toBeVisible();      // summary updated
  await expect(page.locator(".datatable tbody tr")).toHaveCount(1); // history row
});

e2e_import_errors_actionable: upload mixed-errors.csv → error tile shows 1 → error table lists line 3, column Variant Price, message contains "not a price" → confirm ("Import 2 products") → run detail still lists the error row → Products shows count 2.

e2e_import_file_rejected: upload missing-title.csv → banner contains "missing the required column 'Title'" → navigate to Products → history shows "No imports yet." (no run recorded).

e2e_import_cancel_no_trace: upload good.csv → preview visible → Cancel → back on Products → count 0 (still "No products yet") → "No imports yet."

Exact assertion text must be calibrated to the implemented screens (Tasks 11–14) — adjust selectors to what actually renders, keeping each scenario's observable contract (the Gherkin in spec §4) intact.

• Steps: write specs → bash scripts/e2e.sh → all 4 PASS → commit git commit -am "test(e2e): SLICE-5 scenarios — preview/confirm, errors, rejection, cancel (§6.8)".

---

Task 17: Docs (DOC-1/DOC-4), VERSION bump, full gate

Files:

• Create: docs/OPERATIONS.md — operator guide home (DOC-1): import/export ops section — the INV-18 caps (5,000 rows / 10 MB, where enforced), draft expiry (~1 h, lazy sweep), RB-2 (import apply failed: locate by draft_id/run_id in logs via TEL-6, confirm rollback left catalog unchanged, advise re-upload, file bug), ALR-2 gesture: the gcloud log-based alert on import_apply_failed (give the exact two commands: gcloud logging metrics create ecomm_import_apply_failed --description=… --log-filter='jsonPayload.event="import_apply_failed" OR textPayload:"import_apply_failed"' + the gcloud alpha monitoring policies create / console step; note CLOUDSDK_ACTIVE_CONFIG_NAME=wiggleverse-ecomm per §8.4-handbook; mark it "run at PPE deploy of this slice"). Also note E2E suite exists at e2e/ + scripts/e2e.sh, not yet in CI (no browsers on the runner).

• Create: docs/products-domain.md — DOC-4 dev notes: the canonical row model + column registry, codec/validate/diff pipeline (one diagram), blank-vs-absent semantics, fingerprint/INV-11 mechanics, the apply plan, the three named seams (objectstore ← SLICE-7 replaces import_draft.file_bytes; dialect detection ← SLICE-8; image fetch task ← SLICE-7 replaces the complete shortcut), error-granularity decision (product-level), test map.

• Modify: VERSION → 0.5.0; frontend/package.json "version": "0.5.0".

• Modify: README.md — only if it lists capabilities (check; if it's a stub, skip).

• Step 1: Write both docs (~120 lines each). Step 2: Bump versions. Step 3: Run the full gate exactly as CI will: bash scripts/check.sh → all four stages green. Step 4: bash scripts/e2e.sh → 4 scenarios green. Step 5: Commit — git commit -am "docs(products): operator guide (DOC-1) + domain notes (DOC-4); version 0.5.0"

---

Task 18: Ship — PR, merge, PPE deploy, tags (§9.1)

Session-level (driver does this directly, not a subagent):

• Push branch; open PR against main titled SLICE-5: products import spine — canonical CSV → preview → confirm (SD-0002 §7.2); body cites SD-0002 §7.2 DoD items + the four E2E scenarios; Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> trailers on commits already.
• Self-review the diff (autonomous posture) — run /code-review-grade pass; fix findings; merge when check.sh + E2E are green.
• Read …/skills/wgl-planning-and-executing/DEPLOY-FLOTILLA.md, then deploy to PPE per its pointers (flotilla deploy ecomm with CLOUDSDK_ACTIVE_CONFIG_NAME), verify https://ecomm-ppe.wiggleverse.org/healthz reports 0.5.0, smoke the Products page on PPE.
• Run the ALR-2 gesture from docs/OPERATIONS.md against the PPE project (ad hoc op on existing env — allowed autonomously; log it).
• Tag: annotated v0.5.0 on the merge commit + release/<YYYY-MM-DDTHH-MM> (PST) per §9.1. Push tags.
• Prod: does not exist yet — PPE-green is this slice's pipeline terminus (spec §7.3); prod promotion rides the operator's market.wiggleverse.org stand-up.

---

Self-review checklist (ran at authoring)

• Spec coverage: PUC-1 (Products page incl. empty state + sample link) T11 · PUC-2 T12 · PUC-3/3a T13 · PUC-4 T8/T13/T14 · PUC-5 T4/T13/T14 · PUC-5a T3/T12 · PUC-8 T11/T14 · PUC-11 sample T9 · INV-10 T8(inv tests) · INV-11 T7/T8 (zero-write + fingerprint) · INV-13 T1 (indexes) + T4/T5 (combo matching) · INV-14 T7/T8 · INV-15 T4 · INV-17 seam T3 · INV-18 T3/T9 · TEL-1/2/6 T7/T8 · ALR-2 T17/T18 · DOC-1/3/4 T17/T9 · 4 E2E T16 · version/tag/PPE T17/T18.
• Deliberately out (later slices): export endpoint + INV-12 property test (SLICE-6), objectstore/images/SSRF/notice band (SLICE-7), Shopify dialect + DOC-2 (SLICE-8), bootstrap-test extension (SLICE-6 with export).
• Type consistency: DiffResult.records JSON shape == DraftRecord TS interface == repo.draft_records output; FileRejected.code values == §6.4 error codes == endpoint tests; CatalogProduct defined in diff.py, built by repo.load_catalog.