wiggleverse-ecomm-content/plans/2026-06-11-slice-7-images.md

SLICE-7 — Images pipeline end-to-end (SD-0002 §7.2) 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: Fetch, validate, host, and serve merchant product images — closing PUC-7 — so imported catalogs land with hosted renditions, image outcomes are reported per image, and the export → re-import round-trip stays a no-op over hosted image URLs (INV-12 extended over images).

Architecture: Two new model-free platform modules (platform/images = decode + resolution bar + WebP renditions; platform/objectstore = a put/get/delete port with local-disk and GCS adapters, mirroring the SD-0001 mailer port). A post-commit, in-process image-fetch phase in the products domain (domains/products/imagefetch.py) walks a run's pending images with bounded concurrency (~4 worker threads), fetches each URL behind an SSRF guard with INV-18 bounds, runs platform/images, stores renditions via objectstore, and moves each image pending → fetched | rejected_* | failed — idempotent per image, so it resumes after a restart. An app-served, storefront-authorized route streams renditions from the bucket with immutable caching (INV-16); exported Image Src carries the app's own hosted URL for fetched images and the original source_url otherwise, and the importer recognizes a hosted URL by path-parse and resolves it to the existing image record (never re-fetched), keeping the round-trip lossless. The per-environment private media bucket is provisioned by a new flotilla/launch-app provision-bucket operator gesture at the PPE deploy.

Tech Stack: Python 3 / FastAPI / psycopg 3 (sync) · Pillow (decode + WebP renditions) · httpx (sync, SSRF-guarded fetch) · concurrent.futures.ThreadPoolExecutor (bounded per-image concurrency) · google-cloud-storage (GCS adapter; VM service-account ADC) · React/Vite SPA (polling run detail) · Playwright E2E (local fixture image host) · launch-app skill + gcloud storage (bucket provisioning).

Anchor: SD-0002 wiggleverse-ecomm-content/specs/SD-0002-products-bulk-csv-import-export.md §7.2 SLICE-7 (graduated Solution Design, R2a) · issue wiggleverse/wiggleverse-ecomm#13.

---

Decisions locked for this slice (read before starting)

• Q-3 resolution bar = shorter side ≥ 500 px. Below it → rejected_low_res. A single constant MIN_IMAGE_SHORT_SIDE = 500 in platform/images.py; rationale recorded in SD-0002 §13. Consistent with the downscale-only rendition ladder (detail ≤1600 / card ~480 / thumb ~160 never upscale a ≥500 source).
• Fetch phase = background thread + inner ThreadPoolExecutor(max_workers=4), not asyncio. The stack is sync (sync psycopg pool, sync httpx); a thread runner satisfies the binding invariants (in-process, post-commit, ~4 concurrent, resumable, per-image idempotent) without an async/sync DB bridge. The §6.9 queue-portability seam is unchanged.
• Draft upload blobs stay BYTEA this slice. The §7.2 Ships list is images-only; moving draft blobs to the bucket (the softer products-domain.md seam) is deferred to a later hardening pass. The media bucket holds only product-images/ objects this slice. (The provision-bucket script still sets the import-drafts/ lifecycle rule for forward-compat — harmless when no such objects exist.)
• No new migration. 0002_products.sql already carries every needed column: product_image.status (enum incl. fetched/rejected_low_res/rejected_not_image/failed), failure_reason, key_original/key_thumb/key_card/key_detail, import_run_id, fetched_at; import_run.status enum incl. fetching_images/complete_with_problems.
• Hosted-URL recognition is host-agnostic and DB-free. A hosted URL is any URL whose path matches /api/products/images/{image_id}/{rendition}. Recognition is a pure path-parse (survives the #23 rebrand and works across localhost/PPE). The diff resolves a recognized id against the (already storefront-scoped) catalog snapshot — an id not in this storefront's snapshot is a row error, never a fetch.
• Public base URL for export = APP_URL (already in deployment.toml [overlay]). When unset (local/dev), the serializer emits a relative /api/products/images/{id}/detail path — still recognized on re-import.

---

File structure

Backend — new files

• backend/app/platform/images.py — pure: decode (jpeg/png/webp), resolution bar, WebP renditions. No I/O.
• backend/app/platform/objectstore.py — ObjectStore Protocol + LocalObjectStore + GcsObjectStore + build_objectstore(kind).
• backend/app/domains/products/hosted.py — hosted-image URL build/parse helpers (DB-free).
• backend/app/domains/products/imagefetch.py — the image-fetch phase + SSRF guard + startup recovery (uses repo + platform/images + platform/objectstore).
• backend/tests/test_platform_images.py, backend/tests/test_platform_objectstore.py, backend/tests/test_products_hosted.py, backend/tests/test_products_imagefetch.py, backend/tests/test_products_image_serving.py.

Backend — modified

• backend/app/platform/config.py — objectstore_kind, objectstore_bucket, objectstore_local_dir, public_base_url, image_fetch_allow_private.
• backend/app/domains/products/diff.py — CatalogImage gains status; image matching resolves hosted URLs.
• backend/app/domains/products/repo.py — load_catalog loads status + key_detail; image-phase repo helpers; get_run/list_runs image progress/outcomes/counts.
• backend/app/domains/products/serialize.py — _write_image emits hosted detail URL for fetched images.
• backend/app/domains/products/service.py — confirm_draft sets fetching_images when the run has pending images; export_catalog passes the base URL; serialize signature threaded.
• backend/app/domains/products/__init__.py — export the new image-phase entrypoints.
• backend/app/main.py — build objectstore in lifespan; startup recovery scan; app.state.image_runner; schedule the fetch phase post-confirm; GET /api/products/images/{image_id}/{rendition}.
• backend/requirements.txt — add Pillow, google-cloud-storage.
• existing backend/tests/test_products_serialize.py, test_products_diff.py, test_products_service.py — extend for hosted images.

Frontend — modified

• frontend/src/productsApi.ts — typed ImageOutcome, image counts, image-serving URL builder.
• frontend/src/screens/products/RunDetail.tsx — images section (live progress + outcomes table; poll while non-terminal).
• frontend/src/screens/products/ProductsPage.tsx — image-problems notice band; history "image outcomes" column.
• frontend/src/screens/products/runImages.ts (new) — pure helpers (progress label, outcome label) + vitest.

E2E — new/modified

• e2e/fixtures/images/ (good.png 800×800, tiny.png 64×64, served by a fixture host), e2e/fixtures/with-images.csv, e2e/image-host.mjs (or python), e2e/serve.sh (start fixture host + set env), e2e/tests/image-outcomes.spec.ts, e2e/helpers.ts (importWithImages helper).

Cross-repo (engineering) — new

• engineering/launch-app/skills/provision-bucket/SKILL.md + assets/provision-bucket.sh + assets/read reuse; engineering/launch-app/SPEC.md §6.6b note.

Deploy/docs — modified

• deployment.toml [overlay] — ECOMM_OBJECTSTORE_KIND, ECOMM_OBJECTSTORE_BUCKET.
• docs/products-domain.md, docs/OPERATIONS.md, VERSION, frontend/package.json (+lockfile).
• SD-0002 §13 Q-3 marked resolved (content repo, at finalize alongside plan archive).

---

Task 1: platform/images — decode, resolution bar, WebP renditions

Files:

• Create: backend/app/platform/images.py

• Create: backend/tests/test_platform_images.py

• Modify: backend/requirements.txt

• Step 1: Add Pillow to requirements

Add to backend/requirements.txt (keep alphabetical-ish grouping with the other runtime deps):

Pillow>=11.0

Then install into the worktree venv: ./.venv/bin/pip install -r backend/requirements.txt (the worktree's venv is created in Task 0 of execution / by scripts/dev.sh; if absent, run python3 -m venv .venv && ./.venv/bin/pip install -r backend/requirements.txt).

• Step 2: Write the failing tests

backend/tests/test_platform_images.py:

"""platform/images — pure decode + resolution bar + WebP renditions (SD-0002 §6.2)."""
import io

from PIL import Image

from app.platform import images


def _png(width: int, height: int, color=(120, 80, 200)) -> bytes:
    buf = io.BytesIO()
    Image.new("RGB", (width, height), color).save(buf, format="PNG")
    return buf.getvalue()


def _jpeg(width: int, height: int) -> bytes:
    buf = io.BytesIO()
    Image.new("RGB", (width, height), (10, 200, 90)).save(buf, format="JPEG")
    return buf.getvalue()


def test_good_image_produces_three_webp_renditions():
    result = images.process(_png(1200, 900))
    assert isinstance(result, images.Processed)
    assert result.source_format == "PNG"
    assert set(result.renditions) == {"thumb", "card", "detail"}
    for name, blob in result.renditions.items():
        with Image.open(io.BytesIO(blob)) as im:
            assert im.format == "WEBP"
        # downscale-only: longest side never exceeds the rendition cap
    with Image.open(io.BytesIO(result.renditions["thumb"])) as im:
        assert max(im.size) <= images.RENDITIONS["thumb"]
    with Image.open(io.BytesIO(result.renditions["detail"])) as im:
        assert max(im.size) <= images.RENDITIONS["detail"]


def test_downscale_only_never_upscales_small_source():
    # 520x520 passes the bar; detail cap is 1600 but must not upscale.
    result = images.process(_png(520, 520))
    with Image.open(io.BytesIO(result.renditions["detail"])) as im:
        assert im.size == (520, 520)


def test_below_resolution_bar_rejected_low_res():
    result = images.process(_png(300, 1200))  # shorter side 300 < 500
    assert isinstance(result, images.Rejected)
    assert result.reason == "rejected_low_res"


def test_not_an_image_rejected_not_image():
    result = images.process(b"this is not an image")
    assert isinstance(result, images.Rejected)
    assert result.reason == "rejected_not_image"


def test_jpeg_source_format_preserved_in_result():
    result = images.process(_jpeg(800, 800))
    assert result.source_format == "JPEG"

• Step 3: Run to verify failure

Run: ./.venv/bin/python -m pytest backend/tests/test_platform_images.py -q Expected: FAIL — ModuleNotFoundError: app.platform.images.

• Step 4: Implement platform/images.py

"""platform/images — pure image processing (SD-0002 §6.2).

Model-free, deterministic, no I/O of its own: decode bytes, enforce the
resolution bar (INV-18 / Q-3), and emit downscale-only WebP renditions. The
caller (the image-fetch phase) owns fetching, storage, and status.
"""
from __future__ import annotations

import io
from dataclasses import dataclass

from PIL import Image, UnidentifiedImageError

# Q-3 (SD-0002 §13): the minimum shorter-side dimension; below it an image is
# rejected_low_res. Rejects icons/thumbnails, accepts normal product photos.
MIN_IMAGE_SHORT_SIDE = 500

# Rendition longest-side caps (px); downscale-only — a smaller source is kept
# at its own size, never upscaled. Encoded WebP.
RENDITIONS = {"thumb": 160, "card": 480, "detail": 1600}

# Source formats we accept (Pillow format names).
_ACCEPTED = {"JPEG", "PNG", "WEBP"}

_WEBP_QUALITY = 82


@dataclass(frozen=True)
class Processed:
    source_format: str
    width: int
    height: int
    renditions: dict[str, bytes]  # name -> WebP bytes


@dataclass(frozen=True)
class Rejected:
    reason: str  # "rejected_low_res" | "rejected_not_image"


def process(data: bytes) -> Processed | Rejected:
    """Decode + bar-check + renditions, or a typed rejection. Never raises on
    bad input — undecodable/unsupported bytes are Rejected('rejected_not_image')."""
    try:
        with Image.open(io.BytesIO(data)) as im:
            im.load()
            source_format = im.format or ""
            if source_format not in _ACCEPTED:
                return Rejected(reason="rejected_not_image")
            rgb = im.convert("RGB")
            width, height = rgb.size
            if min(width, height) < MIN_IMAGE_SHORT_SIDE:
                return Rejected(reason="rejected_low_res")
            renditions = {
                name: _rendition(rgb, cap) for name, cap in RENDITIONS.items()
            }
            return Processed(
                source_format=source_format,
                width=width,
                height=height,
                renditions=renditions,
            )
    except (UnidentifiedImageError, OSError, ValueError):
        return Rejected(reason="rejected_not_image")


def _rendition(rgb: Image.Image, cap: int) -> bytes:
    longest = max(rgb.size)
    if longest > cap:
        scale = cap / longest
        size = (max(1, round(rgb.width * scale)), max(1, round(rgb.height * scale)))
        resized = rgb.resize(size, Image.LANCZOS)
    else:
        resized = rgb
    buf = io.BytesIO()
    resized.save(buf, format="WEBP", quality=_WEBP_QUALITY, method=6)
    return buf.getvalue()

• Step 5: Run to verify pass

Run: ./.venv/bin/python -m pytest backend/tests/test_platform_images.py -q Expected: PASS (5 tests).

• Step 6: Commit

git add backend/app/platform/images.py backend/tests/test_platform_images.py backend/requirements.txt
git commit -m "feat(platform): images port — decode, resolution bar (Q-3), WebP renditions (SD-0002 §6.2)"

---

Task 2: platform/objectstore — port + local & GCS adapters

Files:

• Create: backend/app/platform/objectstore.py

• Create: backend/tests/test_platform_objectstore.py

• Modify: backend/app/platform/config.py, backend/requirements.txt

• Step 1: Add config accessors

Append to backend/app/platform/config.py (mirror the existing os.environ.get style):

def objectstore_kind() -> str:
    """Which objectstore adapter to build: 'local' (dev/tests) or 'gcs' (deployed)."""
    return os.environ.get("ECOMM_OBJECTSTORE_KIND") or "local"


def objectstore_bucket() -> str:
    """The GCS bucket name for the media objects (gcs adapter; deployment overlay)."""
    return os.environ.get("ECOMM_OBJECTSTORE_BUCKET", "")


def objectstore_local_dir() -> str:
    """Base directory for the local-disk objectstore adapter (dev/tests)."""
    return os.environ.get("ECOMM_OBJECTSTORE_DIR") or "/tmp/ecomm-objectstore"


def public_base_url() -> str:
    """Absolute origin for hosted image URLs in export (e.g. APP_URL); '' → relative."""
    return (os.environ.get("APP_URL") or "").rstrip("/")


def image_fetch_allow_private() -> bool:
    """Allow image fetch from private/loopback hosts (dev/E2E fixture host only).
    Default off — the SSRF guard rejects private ranges in deployed envs."""
    return os.environ.get("ECOMM_IMAGE_FETCH_ALLOW_PRIVATE", "").strip().lower() in {
        "1", "true", "yes", "on",
    }

• Step 2: Add google-cloud-storage to requirements

Add to backend/requirements.txt:

google-cloud-storage>=2.18

Install: ./.venv/bin/pip install -r backend/requirements.txt.

• Step 3: Write the failing tests (local adapter)

backend/tests/test_platform_objectstore.py:

"""platform/objectstore — local-disk adapter round-trip (SD-0002 §6.2/§6.3.1)."""
import pytest

from app.platform import objectstore


def test_local_put_get_round_trip(tmp_path):
    store = objectstore.LocalObjectStore(str(tmp_path))
    key = "storefronts/1/product-images/9/original"
    store.put(key, b"\x89PNG-bytes", "image/png")
    assert store.get(key) == b"\x89PNG-bytes"


def test_local_get_missing_raises_not_found(tmp_path):
    store = objectstore.LocalObjectStore(str(tmp_path))
    with pytest.raises(objectstore.ObjectNotFound):
        store.get("nope/missing")


def test_local_delete_is_idempotent(tmp_path):
    store = objectstore.LocalObjectStore(str(tmp_path))
    store.put("a/b", b"x", "text/plain")
    store.delete("a/b")
    store.delete("a/b")  # no error second time
    with pytest.raises(objectstore.ObjectNotFound):
        store.get("a/b")


def test_local_keys_cannot_escape_base_dir(tmp_path):
    store = objectstore.LocalObjectStore(str(tmp_path))
    with pytest.raises(ValueError):
        store.put("../escape", b"x", "text/plain")


def test_build_objectstore_local(tmp_path, monkeypatch):
    monkeypatch.setenv("ECOMM_OBJECTSTORE_KIND", "local")
    monkeypatch.setenv("ECOMM_OBJECTSTORE_DIR", str(tmp_path))
    store = objectstore.build_objectstore("local")
    assert isinstance(store, objectstore.LocalObjectStore)


def test_build_objectstore_unknown_kind_raises():
    with pytest.raises(ValueError):
        objectstore.build_objectstore("s3")

• Step 4: Run to verify failure

Run: ./.venv/bin/python -m pytest backend/tests/test_platform_objectstore.py -q Expected: FAIL — module missing.

• Step 5: Implement platform/objectstore.py

"""platform/objectstore — the media-blob port (SD-0002 §6.2, §6.3.1).

A tiny put/get/delete port with two adapters, mirroring the SD-0001 mailer
pattern: local-disk (dev/tests) and GCS (deployed). Owns no semantics — keys
and content types come from the caller (the image-fetch phase). Objects are
written once, never mutated (immutable, image-id-addressed).
"""
from __future__ import annotations

import os
from pathlib import Path
from typing import Protocol

from app.platform import config


class ObjectNotFound(Exception):
    """Raised by get() when the key has no object."""


class ObjectStore(Protocol):
    def put(self, key: str, data: bytes, content_type: str) -> None: ...
    def get(self, key: str) -> bytes: ...
    def delete(self, key: str) -> None: ...


def build_objectstore(kind: str) -> ObjectStore:
    """Select the objectstore adapter by configured kind (config.objectstore_kind, INV-8)."""
    if kind == "local":
        return LocalObjectStore(config.objectstore_local_dir())
    if kind == "gcs":
        return GcsObjectStore(config.objectstore_bucket())
    raise ValueError(f"unknown objectstore kind: {kind!r}")


def _safe_segments(key: str) -> tuple[str, ...]:
    parts = tuple(p for p in key.split("/") if p)
    if not parts or any(p in {".", ".."} for p in parts):
        raise ValueError(f"unsafe objectstore key: {key!r}")
    return parts


class LocalObjectStore:
    """Disk-backed adapter under a base dir; key path-segments become subdirs."""

    def __init__(self, base_dir: str) -> None:
        self._base = Path(base_dir)

    def _path(self, key: str) -> Path:
        return self._base.joinpath(*_safe_segments(key))

    def put(self, key: str, data: bytes, content_type: str) -> None:
        path = self._path(key)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(data)

    def get(self, key: str) -> bytes:
        path = self._path(key)
        try:
            return path.read_bytes()
        except FileNotFoundError as exc:
            raise ObjectNotFound(key) from exc

    def delete(self, key: str) -> None:
        try:
            os.remove(self._path(key))
        except FileNotFoundError:
            pass


class GcsObjectStore:
    """GCS adapter; auth via the VM service account ADC (no secret bytes).

    google-cloud-storage is imported lazily so dev/test runs (local adapter)
    don't require the package at import time.
    """

    def __init__(self, bucket_name: str) -> None:
        if not bucket_name:
            raise ValueError("gcs objectstore requires ECOMM_OBJECTSTORE_BUCKET")
        from google.cloud import storage  # lazy

        self._bucket = storage.Client().bucket(bucket_name)

    def put(self, key: str, data: bytes, content_type: str) -> None:
        _safe_segments(key)
        self._bucket.blob(key).upload_from_string(data, content_type=content_type)

    def get(self, key: str) -> bytes:
        from google.cloud.exceptions import NotFound  # lazy

        try:
            return self._bucket.blob(key).download_as_bytes()
        except NotFound as exc:
            raise ObjectNotFound(key) from exc

    def delete(self, key: str) -> None:
        from google.cloud.exceptions import NotFound  # lazy

        try:
            self._bucket.blob(key).delete()
        except NotFound:
            pass

• Step 6: Run to verify pass

Run: ./.venv/bin/python -m pytest backend/tests/test_platform_objectstore.py -q Expected: PASS (6 tests).

• Step 7: Verify layering still holds

Run: cd backend && ../.venv/bin/lint-imports (the main > domains > platform contract; platform imports only stdlib/third-party). Expected: contract kept.

• Step 8: Commit

git add backend/app/platform/objectstore.py backend/app/platform/config.py backend/tests/test_platform_objectstore.py backend/requirements.txt
git commit -m "feat(platform): objectstore port — local + GCS adapters, config (SD-0002 §6.2)"

---

Task 3: hosted-image URL build/parse helpers

Files:

• Create: backend/app/domains/products/hosted.py
• Create: backend/tests/test_products_hosted.py

The contract: build a hosted Image Src for a fetched image; recognize one on re-import. Path is /api/products/images/{image_id}/{rendition} (§6.4); recognition is host-agnostic (path-parse).

• Step 1: Write the failing tests

backend/tests/test_products_hosted.py:

"""hosted-image URL helpers — round-trip recognition (SD-0002 §6.3.1, INV-12)."""
from app.domains.products import hosted


def test_build_relative_when_no_base():
    assert hosted.image_url("", 42, "detail") == "/api/products/images/42/detail"


def test_build_absolute_with_base():
    url = hosted.image_url("https://ecomm-ppe.wiggleverse.org", 42, "detail")
    assert url == "https://ecomm-ppe.wiggleverse.org/api/products/images/42/detail"


def test_parse_absolute_hosted_url():
    url = "https://market.wiggleverse.org/api/products/images/7/detail"
    assert hosted.parse_image_id(url) == 7


def test_parse_relative_hosted_url():
    assert hosted.parse_image_id("/api/products/images/7/card") == 7


def test_parse_ignores_query_and_trailing():
    assert hosted.parse_image_id("/api/products/images/7/detail?v=1") == 7


def test_parse_non_hosted_returns_none():
    assert hosted.parse_image_id("https://cdn.example.com/a/b.png") is None
    assert hosted.parse_image_id("/api/products/images/abc/detail") is None
    assert hosted.parse_image_id("/api/products/images/7") is None

• Step 2: Run to verify failure

Run: ./.venv/bin/python -m pytest backend/tests/test_products_hosted.py -q Expected: FAIL — module missing.

• Step 3: Implement hosted.py

"""Hosted-image URL helpers (SD-0002 §6.3.1, INV-12).

Build the canonical hosted Image Src for a fetched image, and recognize one on
re-import. DB-free and host-agnostic: recognition is a path-parse, so an export
round-trips across localhost / PPE / the #23 rebrand. The diff resolves a parsed
id against the storefront-scoped catalog snapshot.
"""
from __future__ import annotations

import re
from urllib.parse import urlsplit

RENDITIONS = ("original", "thumb", "card", "detail")
EXPORT_RENDITION = "detail"

_PATH = re.compile(r"^/api/products/images/(\d+)/(original|thumb|card|detail)$")


def image_url(base_url: str, image_id: int, rendition: str = EXPORT_RENDITION) -> str:
    """Absolute when base_url is set, else a root-relative path."""
    path = f"/api/products/images/{image_id}/{rendition}"
    return f"{base_url.rstrip('/')}{path}" if base_url else path


def parse_image_id(url: str) -> int | None:
    """The image id if url is one of our hosted-image URLs (any host), else None."""
    if not url:
        return None
    m = _PATH.match(urlsplit(url).path)
    return int(m.group(1)) if m else None

• Step 4: Run to verify pass

Run: ./.venv/bin/python -m pytest backend/tests/test_products_hosted.py -q Expected: PASS (6 tests).

• Step 5: Commit

git add backend/app/domains/products/hosted.py backend/tests/test_products_hosted.py
git commit -m "feat(products): hosted-image URL build/parse helpers (SD-0002 §6.3.1)"

---

Task 4: catalog snapshot carries image status; serializer emits hosted URLs

Files:

• Modify: backend/app/domains/products/diff.py (CatalogImage dataclass)

• Modify: backend/app/domains/products/repo.py (load_catalog image query)

• Modify: backend/app/domains/products/serialize.py (_write_image, catalog_to_csv signature)

• Modify: backend/app/domains/products/service.py (export_catalog passes base url)

• Modify: backend/tests/test_products_serialize.py

• Step 1: Write the failing test (serializer emits hosted URL for fetched, source for others)

Add to backend/tests/test_products_serialize.py (reuse its existing CatalogProduct/CatalogImage/CatalogVariant builders; pass status to images):

def test_fetched_image_serializes_hosted_detail_url():
    from app.domains.products import serialize
    from app.domains.products.diff import CatalogImage, CatalogProduct, CatalogVariant

    product = CatalogProduct(
        id=1, handle="lamp", title="Lamp", option_names=(None, None, None),
        fields={"title": "Lamp", "status": "active"},
        variants=[CatalogVariant(id=1, options=(None, None, None), position=1, fields={})],
        images=[
            CatalogImage(id=55, source_url="https://m.example/a.png", position=1,
                         alt_text=None, status="fetched"),
            CatalogImage(id=56, source_url="https://m.example/b.png", position=2,
                         alt_text=None, status="failed"),
        ],
    )
    rows = list(serialize._product_rows(product, base_url="https://shop.test"))
    srcs = [r.get("Image Src") for r in rows if r.get("Image Src")]
    assert "https://shop.test/api/products/images/55/detail" in srcs  # fetched -> hosted
    assert "https://m.example/b.png" in srcs                          # failed -> source kept

(Adjust the helper call to however the existing test constructs products; the key assertions are the two Image Src values.)

• Step 2: Run to verify failure

Run: ./.venv/bin/python -m pytest backend/tests/test_products_serialize.py -q Expected: FAIL — CatalogImage has no status; _product_rows/_write_image take no base_url.

• Step 3: Extend CatalogImage in diff.py

In backend/app/domains/products/diff.py, add status (default keeps existing call sites valid):

@dataclass
class CatalogImage:
    id: int
    source_url: str
    position: int
    alt_text: str | None
    status: str = "pending"

• Step 4: Load status in repo.load_catalog

In repo.py, the image query — add i.status and pass it:

    for row in conn.execute(
        "SELECT i.product_id, i.id, i.source_url, i.position, i.alt_text, i.status"
        " FROM product_image i"
        " JOIN product p ON p.id = i.product_id"
        " WHERE p.storefront_id = %s"
        " ORDER BY i.product_id, i.position, i.id",
        (storefront_id,),
    ):
        by_id[row[0]].images.append(
            CatalogImage(id=row[1], source_url=row[2], position=row[3],
                         alt_text=row[4], status=row[5])
        )

• Step 5: Thread base_url through the serializer

In serialize.py: catalog_to_csv(snapshot, base_url="") → _product_rows(product, base_url) → _write_image(row, image, base_url):

from .hosted import EXPORT_RENDITION, image_url


def _write_image(row: dict[str, str], image, base_url: str) -> None:
    # Fetched images export the platform-hosted URL (INV-12/16); everything
    # else keeps the original source_url so nothing is lost (§6.5.5).
    if image.status == "fetched":
        row["Image Src"] = image_url(base_url, image.id, EXPORT_RENDITION)
    else:
        row["Image Src"] = image.source_url
    row["Image Position"] = str(image.position)
    if image.alt_text is not None:
        row["Image Alt Text"] = image.alt_text

Update catalog_to_csv and _product_rows signatures to take and forward base_url (default "").

• Step 6: Pass the base url from service.export_catalog

In service.py, import config and pass it into the stream:

from app.platform import config
...
        yield from serialize.catalog_to_csv(snapshot, base_url=config.public_base_url())

• Step 7: Run to verify pass

Run: ./.venv/bin/python -m pytest backend/tests/test_products_serialize.py -q Expected: PASS (incl. the new test). Then full export test: ./.venv/bin/python -m pytest backend/tests/test_products_export.py -q (existing snapshots use never-fetched images → still emit source_url → unchanged).

• Step 8: Commit

git add backend/app/domains/products/diff.py backend/app/domains/products/repo.py backend/app/domains/products/serialize.py backend/app/domains/products/service.py backend/tests/test_products_serialize.py
git commit -m "feat(products): export hosted detail URL for fetched images; snapshot carries status (SD-0002 §6.5.5, INV-16)"

---

Task 5: diff resolves hosted URLs (INV-12 over images)

Files:

• Modify: backend/app/domains/products/diff.py (image matching in _update_plan; add-path hosted check)
• Modify: backend/tests/test_products_diff.py
• Modify: backend/tests/test_products_serialize.py (extend the INV-12 property test to fetched images)

The rule:

1. A canonical image whose source_url parses as a hosted id (hosted.parse_image_id) is not a new URL. Resolve it to the catalog image with that id within the same product: matched → treat as the existing image (position/alt diff only; never an add → never a fetch). If the id is not a catalog image of this product (cross-storefront, deleted, or another product) → row error ("image refers to an unknown hosted id") — never a fetch attempt on our own domain.
2. A non-hosted (original) URL keeps today's behavior: match by source_url, else add.

• Step 1: Write the failing tests

Add to backend/tests/test_products_diff.py:

def test_hosted_url_for_existing_image_is_unchanged_not_add():
    # Catalog has a fetched image id=55; canonical re-imports it as the hosted URL.
    from app.domains.products import diff
    from app.domains.products.diff import CatalogImage, CatalogProduct, CatalogVariant
    from app.domains.products.models import CanonicalImage, CanonicalProduct, CanonicalVariant

    catalog = {
        "lamp": CatalogProduct(
            id=1, handle="lamp", title="Lamp", option_names=(None, None, None),
            fields={"title": "Lamp", "status": "active", "description_html": None,
                    "vendor": None, "product_type": "standalone",
                    "google_product_category": None, "tags": [], "published": True},
            variants=[CatalogVariant(id=1, options=(None, None, None), position=1, fields={
                "sku": None, "barcode": None, "price": None, "cost": None,
                "weight": None, "weight_unit": None, "volume": None, "volume_unit": None,
                "tax_id_1": None, "tax_id_2": None, "inventory_tracker": None,
                "inventory_qty": None, "variant_image": None})],
            images=[CatalogImage(id=55, source_url="https://m.example/a.png",
                                 position=1, alt_text=None, status="fetched")],
        )
    }
    canonical = [CanonicalProduct(
        first_line=2, handle="lamp", title="Lamp",
        variants=[CanonicalVariant(line_number=2, options=(None, None, None), fields={})],
        images=[CanonicalImage(line_number=2,
                               source_url="/api/products/images/55/detail",
                               position=1, alt_text=None)],
    )]
    result = diff.compute_diff(catalog, canonical)
    handle_kind = {r["handle"]: r["kind"] for r in result.records}
    assert handle_kind["lamp"] == "unchanged"   # not "update", not "add"


def test_hosted_url_for_unknown_id_is_row_error():
    from app.domains.products import diff
    from app.domains.products.diff import CatalogProduct, CatalogVariant
    from app.domains.products.models import CanonicalImage, CanonicalProduct, CanonicalVariant

    catalog = {
        "lamp": CatalogProduct(
            id=1, handle="lamp", title="Lamp", option_names=(None, None, None),
            fields={"title": "Lamp", "status": "active", "description_html": None,
                    "vendor": None, "product_type": "standalone",
                    "google_product_category": None, "tags": [], "published": True},
            variants=[CatalogVariant(id=1, options=(None, None, None), position=1, fields={
                "sku": None, "barcode": None, "price": None, "cost": None,
                "weight": None, "weight_unit": None, "volume": None, "volume_unit": None,
                "tax_id_1": None, "tax_id_2": None, "inventory_tracker": None,
                "inventory_qty": None, "variant_image": None})],
            images=[],
        )
    }
    canonical = [CanonicalProduct(
        first_line=2, handle="lamp", title="Lamp",
        variants=[CanonicalVariant(line_number=2, options=(None, None, None), fields={})],
        images=[CanonicalImage(line_number=2,
                               source_url="/api/products/images/999/detail",
                               position=1, alt_text=None)],
    )]
    result = diff.compute_diff(catalog, canonical)
    kinds = {r["handle"]: r["kind"] for r in result.records}
    assert kinds["lamp"] == "error"

• Step 2: Run to verify failure

Run: ./.venv/bin/python -m pytest backend/tests/test_products_diff.py -k hosted -q Expected: FAIL — hosted URLs currently treated as a new add.

• Step 3: Implement hosted resolution in _update_plan (and the add path)

In diff.py, before the existing by_src matching loop in the update path, build an id index and resolve hosted canonical images. Concretely, in the image loop replace the match lookup:

from . import hosted
...
    by_src = {i.source_url: i for i in current.images}
    by_id = {i.id: i for i in current.images}
    for image in product.images:
        hosted_id = hosted.parse_image_id(image.source_url)
        if hosted_id is not None:
            match = by_id.get(hosted_id)
            if match is None:
                # Hosted URL whose id is not one of this product's images:
                # unknown/cross-storefront — a row error, never a fetch.
                product.errors.append(
                    RowError(image.line_number, "Image Src",
                             f"image '{image.source_url}' refers to an unknown hosted id")
                )
                continue
            # Resolve to the existing image; only position/alt can differ. A
            # hosted re-import of the same image at the same slot is unchanged.
            image = _resolved_hosted(image, match)  # see helper below
        else:
            match = by_src.get(image.source_url)
            if match is None:
                # ... existing add branch unchanged ...
                continue
        # ... existing position/alt update-compare branch, using `match` ...

Add a small helper so the position/alt compare reuses the matched image's identity:

def _resolved_hosted(image, match):
    """A hosted-URL canonical image, re-keyed onto the matched catalog image's
    source_url so the downstream position/alt compare and apply behave as an
    existing-image update (never an add)."""
    return replace(image, source_url=match.source_url)

(from dataclasses import replace. CanonicalImage is frozen — replace returns a new instance.) For the add path (a non-hosted brand-new URL) nothing changes — it stays an add (→ pending → fetched by the task).

Important: when the product is an add (plan.kind == "add", no current), a hosted URL cannot resolve to any existing image → it must be a row error too. In the add branch, before emitting image add-plans, run the same hosted.parse_image_id check: a hosted URL on a brand-new product is an unknown id → row error (downgrades the product to error, consistent with the update path). Add a guard at the top of the add-image handling.

• Step 4: Run to verify pass

Run: ./.venv/bin/python -m pytest backend/tests/test_products_diff.py -q Expected: PASS (incl. the two new tests + all existing).

• Step 5: Extend the INV-12 property test over fetched images

In backend/tests/test_products_serialize.py, the existing 200-catalog property generator (diff(catalog, import(export(catalog))) == ∅) — extend it so a fraction of images have status="fetched" (with stable ids), build the export with a fixed base_url="https://shop.test", and assert the re-import diffs to all-unchanged. The serializer emits hosted URLs for fetched images; the parse-back in parse_csv/validate produces canonical images carrying those hosted URLs; the diff resolves them to the same ids → unchanged. Concretely, where the generator creates each CatalogImage, set status to "fetched" for, say, every other image, and pass base_url into serialize.catalog_to_csv(...). Keep never-fetched images too (they export source_url and round-trip as before).

Run: ./.venv/bin/python -m pytest backend/tests/test_products_serialize.py -q Expected: PASS — round-trip no-op holds with fetched (hosted) and non-fetched images mixed.

• Step 6: Commit

git add backend/app/domains/products/diff.py backend/tests/test_products_diff.py backend/tests/test_products_serialize.py
git commit -m "feat(products): diff resolves hosted image URLs to existing records — INV-12 over images (SD-0002 §6.3)"

---

Task 6: image-phase repo helpers + run progress/outcomes in payloads

Files:

• Modify: backend/app/domains/products/repo.py

• Modify: backend/tests/test_products_service.py (or a new test_products_repo_images.py)

• Step 1: Write the failing tests

backend/tests/test_products_repo_images.py (uses the existing pytest Postgres fixtures — copy the conftest connection fixture pattern other repo tests use; insert a storefront/product/run/images directly, then assert helper behavior):

"""products repo — image-phase helpers (SD-0002 §6.5.4)."""
# Use the project's existing DB test fixtures (see conftest.py): a connection
# bound to a migrated test database, plus helpers to seed an account/storefront.

def test_pending_images_for_run_returns_only_pending(db, seed_storefront):
    sf = seed_storefront(db)
    run_id = _seed_run(db, sf, status="fetching_images")
    p = _seed_product(db, sf, "lamp")
    a = _seed_image(db, p, "https://m/a.png", run_id, status="pending")
    _seed_image(db, p, "https://m/b.png", run_id, status="fetched")
    pend = repo.pending_images_for_run(db, run_id)
    assert [img["id"] for img in pend] == [a]
    assert pend[0]["source_url"] == "https://m/a.png"
    assert pend[0]["storefront_id"] == sf


def test_claim_image_for_fetch_is_idempotent(db, seed_storefront):
    sf = seed_storefront(db); run_id = _seed_run(db, sf, status="fetching_images")
    p = _seed_product(db, sf, "lamp")
    img = _seed_image(db, p, "https://m/a.png", run_id, status="pending")
    assert repo.claim_image_for_fetch(db, img) is True
    # second claim returns False (already claimed/terminal) — restart safety
    repo.mark_image_fetched(db, img, {"original": "k/o", "thumb": "k/t",
                                      "card": "k/c", "detail": "k/d"})
    assert repo.claim_image_for_fetch(db, img) is False


def test_mark_image_outcomes_and_run_progress(db, seed_storefront):
    sf = seed_storefront(db); run_id = _seed_run(db, sf, status="fetching_images")
    p = _seed_product(db, sf, "lamp")
    ok = _seed_image(db, p, "https://m/a.png", run_id, status="pending")
    bad = _seed_image(db, p, "https://m/b.png", run_id, status="pending")
    miss = _seed_image(db, p, "https://m/c.png", run_id, status="pending")
    repo.mark_image_fetched(db, ok, {"original": "o", "thumb": "t", "card": "c", "detail": "d"})
    repo.mark_image_rejected(db, bad, "rejected_low_res", "below the resolution bar")
    repo.mark_image_failed(db, miss, "host unreachable")
    counts = repo.run_image_counts(db, run_id)
    assert counts == {"fetched": 1, "rejected": 1, "failed": 1, "pending": 0, "total": 3}


def test_incomplete_runs_lists_fetching_images_runs(db, seed_storefront):
    sf = seed_storefront(db)
    rid = _seed_run(db, sf, status="fetching_images")
    _seed_run(db, sf, status="complete")
    assert rid in [r["id"] for r in repo.incomplete_runs(db)]

(Define the small _seed_* helpers inline in the test using the existing insert primitives; db/seed_storefront mirror whatever conftest already provides — check backend/tests/conftest.py and reuse its names.)

• Step 2: Run to verify failure — Run: ./.venv/bin/python -m pytest backend/tests/test_products_repo_images.py -q. Expected: FAIL (helpers missing).

• Step 3: Implement the repo helpers

Add to repo.py:

# ---------------------------------------------------------------------------
# Image-fetch phase (SD-0002 §6.5.4) — called by domains/products/imagefetch.py
# ---------------------------------------------------------------------------

def pending_images_for_run(conn: psycopg.Connection, run_id: int) -> list[dict]:
    """The run's images still awaiting fetch, with the storefront for key layout."""
    rows = conn.execute(
        "SELECT i.id, i.source_url, p.storefront_id, i.product_id"
        " FROM product_image i JOIN product p ON p.id = i.product_id"
        " WHERE i.import_run_id = %s AND i.status = 'pending'"
        " ORDER BY i.id",
        (run_id,),
    ).fetchall()
    return [
        {"id": r[0], "source_url": r[1], "storefront_id": r[2], "product_id": r[3]}
        for r in rows
    ]


def claim_image_for_fetch(conn: psycopg.Connection, image_id: int) -> bool:
    """Atomically claim a pending image (idempotent restart guard). Returns True
    iff this caller won the claim (it was still 'pending')."""
    row = conn.execute(
        "UPDATE product_image SET status = 'pending' WHERE id = %s AND status = 'pending'"
        " RETURNING id",
        (image_id,),
    ).fetchone()
    return row is not None


def mark_image_fetched(conn: psycopg.Connection, image_id: int, keys: dict[str, str]) -> None:
    conn.execute(
        "UPDATE product_image SET status = 'fetched', failure_reason = NULL,"
        " key_original = %s, key_thumb = %s, key_card = %s, key_detail = %s,"
        " fetched_at = now() WHERE id = %s",
        (keys["original"], keys["thumb"], keys["card"], keys["detail"], image_id),
    )


def mark_image_rejected(conn: psycopg.Connection, image_id: int, status: str, reason: str) -> None:
    conn.execute(
        "UPDATE product_image SET status = %s, failure_reason = %s, fetched_at = now()"
        " WHERE id = %s",
        (status, reason, image_id),
    )


def mark_image_failed(conn: psycopg.Connection, image_id: int, reason: str) -> None:
    conn.execute(
        "UPDATE product_image SET status = 'failed', failure_reason = %s, fetched_at = now()"
        " WHERE id = %s",
        (reason, image_id),
    )


def run_image_counts(conn: psycopg.Connection, run_id: int) -> dict:
    row = conn.execute(
        "SELECT"
        " count(*) FILTER (WHERE status = 'fetched'),"
        " count(*) FILTER (WHERE status IN ('rejected_low_res', 'rejected_not_image')),"
        " count(*) FILTER (WHERE status = 'failed'),"
        " count(*) FILTER (WHERE status = 'pending'),"
        " count(*)"
        " FROM product_image WHERE import_run_id = %s",
        (run_id,),
    ).fetchone()
    return {"fetched": row[0], "rejected": row[1], "failed": row[2],
            "pending": row[3], "total": row[4]}


def set_run_status(conn: psycopg.Connection, run_id: int, status: str) -> None:
    conn.execute(
        "UPDATE import_run SET status = %s,"
        " completed_at = CASE WHEN %s THEN now() ELSE completed_at END"
        " WHERE id = %s",
        (status, status in _TERMINAL_RUN_STATUSES, run_id),
    )


def incomplete_runs(conn: psycopg.Connection) -> list[dict]:
    """Runs still in the image phase — the startup recovery scan input (§6.9)."""
    rows = conn.execute(
        "SELECT id, storefront_id FROM import_run WHERE status = 'fetching_images'"
        " ORDER BY id",
    ).fetchall()
    return [{"id": r[0], "storefront_id": r[1]} for r in rows]


def run_image_outcomes(conn: psycopg.Connection, run_id: int) -> list[dict]:
    """Problem images for the run-detail outcomes table (only non-fetched)."""
    rows = conn.execute(
        "SELECT p.handle, i.source_url, i.status, i.failure_reason,"
        " v.option1_value, v.option2_value, v.option3_value"
        " FROM product_image i JOIN product p ON p.id = i.product_id"
        " LEFT JOIN variant v ON v.image_id = i.id"
        " WHERE i.import_run_id = %s"
        " AND i.status IN ('rejected_low_res', 'rejected_not_image', 'failed')"
        " ORDER BY p.handle, i.position, i.id",
        (run_id,),
    ).fetchall()
    return [
        {"handle": r[0], "url": r[1], "outcome": r[2], "reason": r[3],
         "variant": " / ".join(x for x in (r[4], r[5], r[6]) if x) or None}
        for r in rows
    ]

• Step 4: Wire image progress/outcomes/counts into get_run and list_runs

Replace the SLICE-5 stub in get_run:

    counts = run_image_counts(conn, run_id)
    run["image_progress"] = {"done": counts["total"] - counts["pending"], "total": counts["total"]}
    run["image_counts"] = {"fetched": counts["fetched"], "rejected": counts["rejected"],
                           "failed": counts["failed"]}
    run["image_outcomes"] = run_image_outcomes(conn, run_id)

For list_runs, add per-run image counts so the history table can show "✓ · ✗". Simplest: in _run_dict leave as-is, and have list_runs attach image_counts per run via one grouped query:

def list_runs(conn, storefront_id, limit, offset):
    rows = conn.execute(_RUN_SELECT + " WHERE r.storefront_id = %s"
                        " ORDER BY r.created_at DESC, r.id DESC LIMIT %s OFFSET %s",
                        (storefront_id, limit, offset)).fetchall()
    runs = [_run_dict(row) for row in rows]
    if not runs:
        return runs
    ids = [r["id"] for r in runs]
    by_run = {rid: {"fetched": 0, "rejected": 0, "failed": 0, "total": 0} for rid in ids}
    for rid, fetched, rejected, failed, total in conn.execute(
        "SELECT import_run_id,"
        " count(*) FILTER (WHERE status='fetched'),"
        " count(*) FILTER (WHERE status IN ('rejected_low_res','rejected_not_image')),"
        " count(*) FILTER (WHERE status='failed'), count(*)"
        " FROM product_image WHERE import_run_id = ANY(%s) GROUP BY import_run_id",
        (ids,),
    ):
        by_run[rid] = {"fetched": fetched, "rejected": rejected, "failed": failed, "total": total}
    for r in runs:
        r["image_counts"] = by_run[r["id"]]
    return runs

• Step 5: Run to verify pass

Run: ./.venv/bin/python -m pytest backend/tests/test_products_repo_images.py backend/tests/test_products_service.py -q Expected: PASS. Note: test_products_service.py's existing assertion that image_progress is {done:0,total:0} for a no-image run still holds (total 0).

• Step 6: Commit

git add backend/app/domains/products/repo.py backend/tests/test_products_repo_images.py
git commit -m "feat(products): image-phase repo helpers + run progress/outcomes/counts (SD-0002 §6.5.4)"

---

Task 7: the image-fetch phase + SSRF guard + recovery

Files:

• Create: backend/app/domains/products/imagefetch.py

• Create: backend/tests/test_products_imagefetch.py

• Modify: backend/app/domains/products/__init__.py

• Step 1: Write the failing tests (SSRF guard unit; fetch against a local fixture server; resume)

backend/tests/test_products_imagefetch.py:

"""image-fetch phase: SSRF guard, fetch+process+store, run completion, resume (§6.5.4/§6.9)."""
import io
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import pytest
from PIL import Image

from app.domains.products import imagefetch
from app.platform import objectstore


def _png(w, h):
    buf = io.BytesIO(); Image.new("RGB", (w, h), (1, 2, 3)).save(buf, "PNG"); return buf.getvalue()


class _Host(BaseHTTPRequestHandler):
    GOOD = _png(800, 800)
    TINY = _png(64, 64)
    def log_message(self, *a): pass
    def do_GET(self):
        if self.path == "/good.png":
            self.send_response(200); self.send_header("Content-Type", "image/png")
            self.end_headers(); self.wfile.write(self.GOOD)
        elif self.path == "/tiny.png":
            self.send_response(200); self.send_header("Content-Type", "image/png")
            self.end_headers(); self.wfile.write(self.TINY)
        else:
            self.send_response(404); self.end_headers()


@pytest.fixture
def host():
    srv = HTTPServer(("127.0.0.1", 0), _Host)
    t = threading.Thread(target=srv.serve_forever, daemon=True); t.start()
    yield f"http://127.0.0.1:{srv.server_address[1]}"
    srv.shutdown()


def test_ssrf_guard_rejects_private_when_disallowed():
    with pytest.raises(imagefetch.FetchBlocked):
        imagefetch.fetch_bytes("http://127.0.0.1:1/x.png", allow_private=False)
    with pytest.raises(imagefetch.FetchBlocked):
        imagefetch.fetch_bytes("http://169.254.169.254/latest/meta-data", allow_private=False)
    with pytest.raises(imagefetch.FetchBlocked):
        imagefetch.fetch_bytes("ftp://example.com/x", allow_private=False)


def test_ssrf_guard_allows_private_when_enabled(host):
    data, ctype = imagefetch.fetch_bytes(f"{host}/good.png", allow_private=True)
    assert ctype.startswith("image/")
    assert len(data) > 0


def test_run_phase_classifies_each_image(db, seed_storefront, host, tmp_path):
    store = objectstore.LocalObjectStore(str(tmp_path))
    sf = seed_storefront(db); run_id = _seed_run(db, sf, status="fetching_images")
    p = _seed_product(db, sf, "lamp")
    ok = _seed_image(db, p, f"{host}/good.png", run_id)
    tiny = _seed_image(db, p, f"{host}/tiny.png", run_id)
    miss = _seed_image(db, p, f"{host}/missing.png", run_id)
    imagefetch.run_image_phase(_pool_for(db), store, run_id, allow_private=True)
    counts = repo.run_image_counts(db, run_id)
    assert counts["fetched"] == 1 and counts["rejected"] == 1 and counts["failed"] == 1
    # fetched image got four objects stored
    keys = _image_keys(db, ok)
    for k in ("key_original", "key_thumb", "key_card", "key_detail"):
        assert keys[k] and store.get(keys[k])
    # run reached a terminal status
    assert _run_status(db, run_id) == "complete_with_problems"


def test_resume_after_kill_completes_remaining(db, seed_storefront, host, tmp_path):
    store = objectstore.LocalObjectStore(str(tmp_path))
    sf = seed_storefront(db); run_id = _seed_run(db, sf, status="fetching_images")
    p = _seed_product(db, sf, "lamp")
    a = _seed_image(db, p, f"{host}/good.png", run_id)
    # simulate a crash: one image already fetched, run still fetching_images
    repo.mark_image_fetched(db, a, {"original": "o", "thumb": "t", "card": "c", "detail": "d"})
    b = _seed_image(db, p, f"{host}/good.png?2", run_id)  # still pending
    resumed = imagefetch.recover_incomplete_runs(_pool_for(db), store, allow_private=True)
    assert run_id in resumed
    assert repo.run_image_counts(db, run_id)["pending"] == 0
    assert _run_status(db, run_id) == "complete"

(_pool_for(db) returns a minimal object exposing a connection() context manager yielding db — or use the real test pool fixture if conftest exposes one. _seed_*, _image_keys, _run_status are inline helpers over raw SQL.)

• Step 2: Run to verify failure — ./.venv/bin/python -m pytest backend/tests/test_products_imagefetch.py -q. Expected: FAIL (module missing).

• Step 3: Implement imagefetch.py

"""The post-commit image-fetch phase (SD-0002 §6.5.4, §6.9).

In-process, bounded-concurrency, resumable. For a run's pending images: fetch
each URL behind an SSRF guard with INV-18 bounds, run platform/images, store
renditions via the objectstore, and move the image pending -> fetched |
rejected_* | failed. Per-image idempotent (claim guard), so the phase can die
and resume. Lives in the domains layer (imports repo + platform); main.py
schedules it post-confirm and runs recovery at startup.
"""
from __future__ import annotations

import ipaddress
import socket
import time
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urlsplit

import httpx

from app.platform import images as images_mod
from app.platform import objectstore as objectstore_mod
from app.platform import telemetry

from . import repo

MAX_FETCH_BYTES = 20 * 1024 * 1024  # INV-18
FETCH_TIMEOUT_S = 30                # INV-18
FETCH_CONCURRENCY = 4              # §6.6
_CONTENT_PREFIX = "image/"


class FetchBlocked(Exception):
    """SSRF guard or bounds refusal — maps to image status 'failed' with reason."""


def _guard_host(url: str, allow_private: bool) -> None:
    parts = urlsplit(url)
    if parts.scheme not in {"http", "https"}:
        raise FetchBlocked(f"unsupported scheme: {parts.scheme!r}")
    host = parts.hostname
    if not host:
        raise FetchBlocked("no host")
    if allow_private:
        return
    # Resolve and reject private / loopback / link-local / reserved / multicast.
    try:
        infos = socket.getaddrinfo(host, parts.port or (443 if parts.scheme == "https" else 80))
    except socket.gaierror as exc:
        raise FetchBlocked(f"dns failure: {exc}") from exc
    for info in infos:
        ip = ipaddress.ip_address(info[4][0])
        if (ip.is_private or ip.is_loopback or ip.is_link_local
                or ip.is_reserved or ip.is_multicast or ip.is_unspecified):
            raise FetchBlocked(f"non-public address: {ip}")


def fetch_bytes(url: str, allow_private: bool) -> tuple[bytes, str]:
    """SSRF-guarded, bounded GET. Returns (bytes, content_type). Raises FetchBlocked."""
    _guard_host(url, allow_private)
    # No redirects across the guard (§6.6): a 3xx is treated as a failure.
    with httpx.Client(follow_redirects=False, timeout=FETCH_TIMEOUT_S) as client:
        with client.stream("GET", url) as resp:
            if resp.status_code != 200:
                raise FetchBlocked(f"http {resp.status_code}")
            ctype = resp.headers.get("content-type", "").split(";")[0].strip().lower()
            if not ctype.startswith(_CONTENT_PREFIX):
                raise FetchBlocked(f"not an image content-type: {ctype!r}")
            chunks: list[bytes] = []
            total = 0
            for chunk in resp.iter_bytes():
                total += len(chunk)
                if total > MAX_FETCH_BYTES:
                    raise FetchBlocked("oversize")
                chunks.append(chunk)
    return b"".join(chunks), ctype


def _key(storefront_id: int, image_id: int, name: str) -> str:
    return f"storefronts/{storefront_id}/product-images/{image_id}/{name}"


def _process_one(pool, store, image: dict, allow_private: bool) -> None:
    """One image, its own short transaction(s). Idempotent via the claim guard."""
    with pool.connection() as conn:
        claimed = repo.claim_image_for_fetch(conn, image["id"])
        conn.commit()
        if not claimed:
            return
    image_id, sf = image["id"], image["storefront_id"]
    try:
        data, ctype = fetch_bytes(image["source_url"], allow_private)
    except (FetchBlocked, httpx.HTTPError) as exc:
        with pool.connection() as conn:
            repo.mark_image_failed(conn, image_id, str(exc)[:200]); conn.commit()
        return
    result = images_mod.process(data)
    if isinstance(result, images_mod.Rejected):
        reason = ("below the resolution bar" if result.reason == "rejected_low_res"
                  else "not a supported image")
        with pool.connection() as conn:
            repo.mark_image_rejected(conn, image_id, result.reason, reason); conn.commit()
        return
    keys = {"original": _key(sf, image_id, "original"),
            "thumb": _key(sf, image_id, "thumb.webp"),
            "card": _key(sf, image_id, "card.webp"),
            "detail": _key(sf, image_id, "detail.webp")}
    store.put(keys["original"], data, ctype)
    store.put(keys["thumb"], result.renditions["thumb"], "image/webp")
    store.put(keys["card"], result.renditions["card"], "image/webp")
    store.put(keys["detail"], result.renditions["detail"], "image/webp")
    with pool.connection() as conn:
        repo.mark_image_fetched(conn, image_id, keys); conn.commit()


def run_image_phase(pool, store, run_id: int, allow_private: bool) -> None:
    """Fetch all pending images of one run, then move the run to a terminal status."""
    started = time.monotonic()
    with pool.connection() as conn:
        pending = repo.pending_images_for_run(conn, run_id)
    if pending:
        with ThreadPoolExecutor(max_workers=FETCH_CONCURRENCY) as pool_x:
            list(pool_x.map(lambda img: _process_one(pool, store, img, allow_private), pending))
    with pool.connection() as conn:
        counts = repo.run_image_counts(conn, run_id)
        status = ("complete_with_problems"
                  if counts["rejected"] + counts["failed"] > 0 else "complete")
        repo.set_run_status(conn, run_id, status)
        conn.commit()
    telemetry.emit("image_phase_completed", run_id=run_id, fetched=counts["fetched"],
                   rejected=counts["rejected"], failed=counts["failed"],
                   duration_ms=int((time.monotonic() - started) * 1000))


def recover_incomplete_runs(pool, store, allow_private: bool) -> list[int]:
    """Startup scan (§6.9): resume any run stuck in fetching_images. Returns the ids."""
    with pool.connection() as conn:
        runs = repo.incomplete_runs(conn)
    resumed: list[int] = []
    for run in runs:
        with pool.connection() as conn:
            pending = len(repo.pending_images_for_run(conn, run["id"]))
        telemetry.emit("image_phase_recovered", run_id=run["id"], pending_resumed=pending)
        run_image_phase(pool, store, run["id"], allow_private)
        resumed.append(run["id"])
    return resumed

• Step 4: Export the entrypoints — add to domains/products/__init__.py __all__ and imports: run_image_phase, recover_incomplete_runs (re-export from .imagefetch). Keep the existing exports.

• Step 5: Run to verify pass

Run: ./.venv/bin/python -m pytest backend/tests/test_products_imagefetch.py -q Expected: PASS (SSRF unit + fixture-host classify + resume).

• Step 6: Verify layering — cd backend && ../.venv/bin/lint-imports. Expected: kept (imagefetch is in domains, imports platform only).

• Step 7: Commit

git add backend/app/domains/products/imagefetch.py backend/app/domains/products/__init__.py backend/tests/test_products_imagefetch.py
git commit -m "feat(products): image-fetch phase — SSRF guard, bounds, renditions, resume (SD-0002 §6.5.4, §6.9, INV-18)"

---

Task 8: confirm sets fetching_images; lifespan wiring + post-commit scheduling

Files:

• Modify: backend/app/domains/products/service.py (confirm_draft run status)

• Modify: backend/app/main.py (lifespan, scheduler, recovery)

• Modify: backend/tests/test_products_service.py

• Step 1: Write the failing test (confirm marks the run fetching_images when it has pending images; complete otherwise)

Add to backend/tests/test_products_service.py:

def test_confirm_marks_run_fetching_images_when_images_present(db, seed_storefront):
    # Import a product whose CSV references an image URL → run should be
    # 'fetching_images' (the post-commit task is scheduled by the BFF, not here).
    ...  # build a draft from a CSV with an Image Src, confirm, assert run status
    run = service.get_run(db, sf, run_id)
    assert run["status"] == "fetching_images"
    assert run["image_progress"]["total"] >= 1


def test_confirm_marks_run_complete_when_no_images(db, seed_storefront):
    ...  # CSV with no Image Src → run status 'complete' immediately
    assert service.get_run(db, sf, run_id)["status"] == "complete"

• Step 2: Run to verify failure — Expected: FAIL (confirm always writes 'complete').

• Step 3: Make confirm_draft choose the status by pending-image presence

In service.py confirm_draft, after applying plans and before/at commit, decide the run status. Since images are inserted as pending during apply, count them on the just-applied run; insert the run as applying first, then set the terminal/transitional status at the end of the transaction:

        run_id = repo.insert_run(
            conn, storefront_id, account_id, row["file_name"], row["dialect"],
            added=summary_counts["adds"], updated=summary_counts["updates"],
            errored=len(error_rows), status="applying",
        )
        for plan in diff_result.plan:
            _apply_product_plan(conn, storefront_id, plan, run_id)
        repo.insert_run_errors(conn, run_id, error_rows)
        pending = repo.run_image_counts(conn, run_id)["pending"]
        repo.set_run_status(conn, run_id, "fetching_images" if pending else "complete")
        repo.delete_draft(conn, storefront_id, draft_id)
        conn.commit()

insert_run with status="applying" writes no completed_at (it's not terminal); set_run_status("complete") then stamps it. TEL-2 (import_run_completed) still fires after commit as today — it marks the catalog apply, independent of the image phase.

• Step 4: Wire lifespan, recovery, scheduler in main.py

from app.platform import objectstore as objectstore_mod
from app.domains import products
from concurrent.futures import ThreadPoolExecutor

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = db.open_pool(dsn)
    with app.state.pool.connection() as conn:
        db.migrate(conn)
    app.state.mailer = mailer_mod.build_mailer(config.mailer_kind())
    app.state.objectstore = objectstore_mod.build_objectstore(config.objectstore_kind())
    app.state.image_runner = ThreadPoolExecutor(max_workers=2, thread_name_prefix="img-run")
    allow_private = config.image_fetch_allow_private()
    app.state.image_allow_private = allow_private
    # §6.9 startup recovery — resume runs stuck mid image-fetch (off the request path)
    app.state.image_runner.submit(
        products.recover_incomplete_runs, app.state.pool, app.state.objectstore, allow_private
    )
    try:
        yield
    finally:
        app.state.image_runner.shutdown(wait=False)
        app.state.pool.close()

In the confirm endpoint, after confirm_draft returns run_id, schedule the phase off the request thread (only if the run has work — cheap to always submit; the phase no-ops when there are no pending images):

    run_id = products.confirm_draft(conn, sf.id, account.id, draft_id)
    request.app.state.image_runner.submit(
        products.run_image_phase, request.app.state.pool,
        request.app.state.objectstore, run_id, request.app.state.image_allow_private,
    )
    return JSONResponse({"run_id": run_id}, status_code=201)

(Add request: Request to the confirm handler signature if not present.)

• Step 5: Run to verify pass — ./.venv/bin/python -m pytest backend/tests/test_products_service.py -q. Expected: PASS.

• Step 6: Commit

git add backend/app/domains/products/service.py backend/app/main.py backend/tests/test_products_service.py
git commit -m "feat(products): confirm → fetching_images + post-commit fetch scheduling + startup recovery (SD-0002 §6.5.3/§6.9)"

---

Task 9: app-served image route (INV-16)

Files:

• Modify: backend/app/main.py

• Create: backend/tests/test_products_image_serving.py

• Step 1: Write the failing endpoint tests

backend/tests/test_products_image_serving.py (use the existing TestClient @contextmanager app fixture; the test app uses the local objectstore; seed a fetched image with stored bytes and a second storefront for isolation):

"""GET /api/products/images/{id}/{rendition} — authorize, stream, immutable cache (§6.4, INV-14/16)."""

def test_serves_fetched_rendition_with_immutable_cache(client, signed_in_with_image):
    image_id = signed_in_with_image  # a fetched image owned by the session's storefront
    resp = client.get(f"/api/products/images/{image_id}/detail")
    assert resp.status_code == 200
    assert resp.headers["content-type"] == "image/webp"
    assert "immutable" in resp.headers.get("cache-control", "")
    assert resp.content  # bytes streamed


def test_not_fetched_returns_409(client, signed_in_with_pending_image):
    image_id = signed_in_with_pending_image
    resp = client.get(f"/api/products/images/{image_id}/detail")
    assert resp.status_code == 409  # not_fetched


def test_other_storefronts_image_is_404(client, image_of_other_storefront):
    resp = client.get(f"/api/products/images/{image_of_other_storefront}/detail")
    assert resp.status_code == 404


def test_unauthenticated_is_401(client_no_session, any_image_id):
    resp = client_no_session.get(f"/api/products/images/{any_image_id}/detail")
    assert resp.status_code == 401


def test_bad_rendition_is_422(client, signed_in_with_image):
    resp = client.get(f"/api/products/images/{signed_in_with_image}/huge")
    assert resp.status_code == 422  # path pattern rejects unknown rendition

• Step 2: Run to verify failure — route missing → 404/405 mismatches. Expected: FAIL.

• Step 3: Implement the route + repo lookup

Add a repo helper image_storage(conn, storefront_id, image_id) returning the storage key + status for a rendition, storefront-scoped:

def image_for_serving(conn, storefront_id, image_id):
    row = conn.execute(
        "SELECT i.status, i.key_original, i.key_thumb, i.key_card, i.key_detail"
        " FROM product_image i JOIN product p ON p.id = i.product_id"
        " WHERE i.id = %s AND p.storefront_id = %s",
        (image_id, storefront_id),
    ).fetchone()
    if row is None:
        return None
    return {"status": row[0], "original": row[1], "thumb": row[2],
            "card": row[3], "detail": row[4]}

In main.py:

from fastapi import Response
from app.platform import objectstore as objectstore_mod

_RENDITION_CT = {"original": None, "thumb": "image/webp", "card": "image/webp", "detail": "image/webp"}

@app.get("/api/products/images/{image_id}/{rendition}")
def serve_product_image(
    image_id: int,
    rendition: str = Path(..., pattern="^(original|thumb|card|detail)$"),
    request: Request = None,
    conn: psycopg.Connection = Depends(get_conn),
    sess: dict | None = Depends(get_session),
):
    gate = _merchant_gate(conn, sess)
    if isinstance(gate, JSONResponse):
        return gate
    _account, sf = gate
    rec = products.image_for_serving(conn, sf.id, image_id)
    if rec is None:
        return _error(404, "not_found", "No such image.")
    if rec["status"] != "fetched":
        return _error(409, "not_fetched", "This image has not been fetched yet.")
    key = rec[rendition]
    if not key:
        return _error(404, "not_found", "No such rendition.")
    try:
        data = request.app.state.objectstore.get(key)
    except objectstore_mod.ObjectNotFound:
        return _error(404, "not_found", "No such image.")
    media_type = _RENDITION_CT[rendition] or "application/octet-stream"
    return Response(content=data, media_type=media_type,
                    headers={"Cache-Control": "public, max-age=31536000, immutable"})

Export image_for_serving from domains/products/__init__.py. (FastAPI returns 422 automatically for a path that fails the pattern — covers the bad-rendition test.)

• Step 4: Run to verify pass — ./.venv/bin/python -m pytest backend/tests/test_products_image_serving.py -q. Expected: PASS.

• Step 5: Commit

git add backend/app/main.py backend/app/domains/products/repo.py backend/app/domains/products/__init__.py backend/tests/test_products_image_serving.py
git commit -m "feat(products): app-served image route — storefront-authorized, immutable cache (SD-0002 §6.4, INV-16)"

---

Task 10: frontend — run-detail images section, notice band, history column

Files:

• Modify: frontend/src/productsApi.ts

• Modify: frontend/src/screens/products/RunDetail.tsx

• Modify: frontend/src/screens/products/ProductsPage.tsx

• Create: frontend/src/screens/products/runImages.ts + frontend/src/screens/products/runImages.test.ts

• Step 1: Type the payloads + pure helpers (vitest first)

In productsApi.ts:

export interface ImageOutcome {
  handle: string; url: string; outcome: string; reason: string | null; variant: string | null;
}
export interface ImageCounts { fetched: number; rejected: number; failed: number; }
export interface RunDetail extends RunSummary {
  errors: RowErrorDetail[];
  image_progress: { done: number; total: number };
  image_counts: ImageCounts;
  image_outcomes: ImageOutcome[];
}
export interface RunSummary {
  /* ...existing... */ image_counts?: ImageCounts;
}

runImages.ts:

import type { ImageCounts } from "../../productsApi";

export function imageProgressLabel(p: { done: number; total: number }): string {
  return `Fetching images: ${p.done.toLocaleString()} of ${p.total.toLocaleString()}`;
}

export function imageOutcomeSummary(c: ImageCounts): string {
  return `${c.fetched.toLocaleString()} fetched · ${c.rejected.toLocaleString()} rejected · ${c.failed.toLocaleString()} failed`;
}

export function historyImageCell(c: ImageCounts | undefined): string {
  if (!c || c.fetched + c.rejected + c.failed === 0) return "—";
  const bad = c.rejected + c.failed;
  return bad === 0 ? `${c.fetched.toLocaleString()} ✓` : `${c.fetched.toLocaleString()} ✓ · ${bad.toLocaleString()} ✗`;
}

export function isRunTerminal(status: string): boolean {
  return status === "complete" || status === "complete_with_problems";
}

runImages.test.ts:

import { describe, expect, it } from "vitest";
import { historyImageCell, imageOutcomeSummary, imageProgressLabel, isRunTerminal } from "./runImages";

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

Run: cd frontend && npm test -- runImages → FAIL (helpers missing), then PASS after creating runImages.ts.

• Step 2: RunDetail — images section + polling

In RunDetail.tsx: remove the // NO images section this slice (SLICE-7) comment; add a useEffect that re-fetches the run on an interval (e.g. 2 s) while !isRunTerminal(run.status), clearing on terminal/unmount. Render:

• while !isRunTerminal: <p aria-live="polite">{imageProgressLabel(run.image_progress)}</p> (status line "Fetching images…").

• when terminal and image_progress.total > 0: <p>{imageOutcomeSummary(run.image_counts)}</p> and, if image_outcomes.length, an outcomes table: columns Handle · Variant · Image URL · Outcome · What to do. Map outcome to label ("rejected: below the resolution bar / not an image", "failed: unreachable") and a fixed "Correct the URL and re-import." action. All-clean (image_outcomes.length === 0) collapses to one line "All images fetched."

• Step 3: ProductsPage — notice band + history image column

In ProductsPage.tsx:

• Notice band (after header, before content): when summary.image_problem_count > 0, render <div className="notice"><a href={#/products/imports/runs/${summary.latest_run_id}}>{summary.image_problem_count} products have image problems</a></div> (aria-live). Uses the existing getProductsSummary payload (already returns image_problem_count, latest_run_id).

• History table: add an "Images" column header and a cell {historyImageCell(r.image_counts)} between Errors and Status.

• Step 4: Build + unit gate

Run: cd frontend && npm run build && npm test Expected: typecheck + build clean; vitest green (incl. runImages).

• Step 5: Commit

git add frontend/src/productsApi.ts frontend/src/screens/products/RunDetail.tsx frontend/src/screens/products/ProductsPage.tsx frontend/src/screens/products/runImages.ts frontend/src/screens/products/runImages.test.ts
git commit -m "feat(products-ui): run-detail images section + progress poll, image-problems notice band, history images column (SD-0002 §5.2/§5.5)"

---

Task 11: E2E — e2e_image_outcomes against a fixture image host

Files:

• Create: e2e/image-host.mjs, e2e/fixtures/images/good.png, e2e/fixtures/images/tiny.png, e2e/fixtures/with-images.csv

• Modify: e2e/serve.sh, e2e/helpers.ts, create e2e/tests/image-outcomes.spec.ts

• Step 1: Create the fixture images + CSV

Generate the PNGs (one ≥500px good, one tiny) — use Python/Pillow once, commit the bytes:

./.venv/bin/python - <<'PY'
from PIL import Image
Image.new("RGB",(800,800),(40,120,200)).save("e2e/fixtures/images/good.png")
Image.new("RGB",(64,64),(200,40,40)).save("e2e/fixtures/images/tiny.png")
PY

e2e/fixtures/with-images.csv (canonical; three image URLs against the fixture host on :8799 — good / tiny(low-res) / missing(404)):

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

• Step 2: Fixture image host

e2e/image-host.mjs — a tiny static server on :8799 serving fixtures/images/* and 404 for anything else:

import { createServer } from "node:http";
import { readFile } from "node:fs/promises";
import { fileURLToPath } from "node:url";
import { dirname, join } from "node:path";

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

• Step 3: serve.sh — start the host + set image env

In e2e/serve.sh, before launching uvicorn, add the objectstore + fetch env and start the host in the background:

export ECOMM_OBJECTSTORE_KIND=local
export ECOMM_OBJECTSTORE_DIR="$repo_root/e2e/.objectstore"
export ECOMM_IMAGE_FETCH_ALLOW_PRIVATE=1
rm -rf "$repo_root/e2e/.objectstore"
node "$repo_root/e2e/image-host.mjs" &

(Add .objectstore to e2e/.gitignore.) The host is a daemon of the serve process; Playwright tears the whole webServer down between runs.

• Step 4: helper + spec

e2e/helpers.ts — add importWithImages(page) (upload with-images.csv, confirm, land on run detail) mirroring importGoodCsv. e2e/tests/image-outcomes.spec.ts:

import { expect, test } from "@playwright/test";
import { gotoProducts, importWithImages, signUpWithStorefront } from "../helpers";

test("e2e_image_outcomes: per-image outcomes, placeholder, notice band", async ({ page }) => {
  await signUpWithStorefront(page);
  await importWithImages(page);
  // Run detail polls fetching_images → terminal. Wait for the outcome summary.
  await expect(page.getByText(/fetched ·/)).toBeVisible({ timeout: 30_000 });
  // one good (fetched), one tiny (rejected), one missing (failed)
  await expect(page.getByText("1 fetched · 1 rejected · 1 failed")).toBeVisible();
  // problems listed in the outcomes table
  await expect(page.getByText("tiny-lamp")).toBeVisible();
  await expect(page.getByText("gone-lamp")).toBeVisible();
  // Products page surfaces the aggregate notice
  await gotoProducts(page);
  await expect(page.getByText(/products have image problems/)).toBeVisible();
});

• Step 5: Run the full E2E suite

Run: bash scripts/e2e.sh Expected: 7/7 green (the 6 existing + e2e_image_outcomes). If the SPA poll timing flakes, raise the expect(...).toBeVisible timeout; the fetch of two local images completes in <2 s.

• Step 6: Commit

git add e2e/
git commit -m "test(e2e): e2e_image_outcomes — fixture image host, per-image outcomes + notice band (SD-0002 §6.8)"

---

Task 12: flotilla/launch-app provision-bucket gesture (engineering repo)

Files (in ~/git/wiggleverse.org/wiggleverse/engineering):

• Create: launch-app/skills/provision-bucket/SKILL.md
• Create: launch-app/skills/provision-bucket/assets/provision-bucket.sh
• Modify: launch-app/SPEC.md (add §6.6b)

This is built by this session (dev work), mirroring provision-datastore; the operator runs it at the PPE deploy (Task 14). Work on a branch in the engineering repo, PR, merge.

• Step 1: Branch the engineering repo

cd ~/git/wiggleverse.org/wiggleverse/engineering
git fetch origin --quiet && git switch -c provision-bucket-gesture origin/main

• Step 2: Write assets/provision-bucket.sh (idempotent; probes before create; selects the dedicated gcloud config; never touches the shared active pointer)

#!/usr/bin/env bash
# launch-app §6.6b — provision the per-environment private media bucket
# (SD-0002 §6.3.1). Idempotent. Reads params from read-deployment-params.py.
set -euo pipefail

: "${NAME:?}"; : "${PROJECT:?}"; : "${GCLOUD_CONFIG:?}"
REGION="${REGION:-us-central1}"
ENVNAME="${ENVNAME:-ppe}"
BUCKET="${BUCKET:-${PROJECT}-${ENVNAME}-media}"
VM_SA="${VM_SA:-$(CLOUDSDK_ACTIVE_CONFIG_NAME=$GCLOUD_CONFIG gcloud compute project-info describe \
  --project "$PROJECT" --format='value(defaultServiceAccount)')}"

run() { CLOUDSDK_ACTIVE_CONFIG_NAME="$GCLOUD_CONFIG" "$@"; }

echo "==> enabling storage API"
run gcloud services enable storage.googleapis.com --project "$PROJECT"

if run gcloud storage buckets describe "gs://$BUCKET" --project "$PROJECT" >/dev/null 2>&1; then
  echo "==> bucket gs://$BUCKET exists"
else
  echo "==> creating gs://$BUCKET (Standard, uniform access, no public, region $REGION)"
  run gcloud storage buckets create "gs://$BUCKET" \
    --project "$PROJECT" --location "$REGION" \
    --uniform-bucket-level-access --public-access-prevention --default-storage-class STANDARD
fi

echo "==> lifecycle: delete import-drafts/ objects after 1 day (§6.3.1 safety net)"
tmp=$(mktemp)
cat > "$tmp" <<'JSON'
{"rule":[{"action":{"type":"Delete"},"condition":{"age":1,"matchesPrefix":["import-drafts/"]}}]}
JSON
run gcloud storage buckets update "gs://$BUCKET" --project "$PROJECT" --lifecycle-file="$tmp"
rm -f "$tmp"

echo "==> granting objectAdmin on the bucket to the VM service account"
run gcloud storage buckets add-iam-policy-binding "gs://$BUCKET" --project "$PROJECT" \
  --member="serviceAccount:${VM_SA}" --role="roles/storage.objectAdmin"

echo
echo "bucket: $BUCKET"
echo "Bind on the deployment overlay (non-secret):"
echo "  ECOMM_OBJECTSTORE_KIND=gcs"
echo "  ECOMM_OBJECTSTORE_BUCKET=$BUCKET"

• Step 3: Write SKILL.md — frontmatter (name provision-bucket, description mirroring provision-datastore's: stand up the per-env private media bucket SD-0002 §6.3.1; idempotent; never public; grants the VM SA objectAdmin scoped to the bucket; bucket name is non-secret config) + the same Step-0 gate (scaffold-gcp-project + deployment.toml validate), Step-1 params reuse (read-deployment-params.py → NAME/PROJECT/GCLOUD_CONFIG/ZONE; derive REGION from ZONE), Step-2 run the script, Step-3 bind the overlay (ECOMM_OBJECTSTORE_KIND + ECOMM_OBJECTSTORE_BUCKET in deployment.toml [overlay], re-import), Step-4 teaching contract (what exists: a private regional bucket, the VM SA can read/write it, no public endpoint).

• Step 4: SPEC §6.6b note — add a short section in launch-app/SPEC.md after §6.6a documenting provision-bucket as the media-bucket sibling of provision-datastore (trigger: ecomm SD-0002 SLICE-7), same three invariants (engine needs no awareness — the bucket name is plain overlay config, not even a secret; idempotent; private by default).

• Step 5: Commit + PR + merge (engineering repo)

git add launch-app/skills/provision-bucket launch-app/SPEC.md
git commit -m "feat(launch-app): provision-bucket gesture — per-env private media bucket (§6.6b; ecomm SD-0002 SLICE-7)"
git push -u origin provision-bucket-gesture
# open + merge the PR via the Gitea API (issues/PR token), then return to the ecomm worktree

---

Task 13: telemetry/docs/config glue + version bump

Files:

• Modify: deployment.toml ([overlay])

• Modify: docs/products-domain.md, docs/OPERATIONS.md

• Modify: VERSION, frontend/package.json (+ lockfile)

• Step 1: deployment.toml overlay — add under [overlay]:

ECOMM_OBJECTSTORE_KIND   = "gcs"
ECOMM_OBJECTSTORE_BUCKET = "wiggleverse-ecomm-ppe-media"

(No secret — GCS auth is the VM service-account ADC. public_base_url() reuses the existing APP_URL.)

• Step 2: products-domain.md — document the image pipeline (objectstore port, platform/images bar = 500px, the fetch phase + SSRF guard + bounds + thread model + resume), hosted-URL recognition (INV-12 over images), and re-target the draft-blob seam note: "BYTEA draft blob → objectstore" is deferred past SLICE-7 (drafts remain BYTEA; the media bucket holds only product-images this slice). Update the "complete shortcut" seam: confirm now inserts applying → sets fetching_images/complete; the fetch task fills image_progress/image_counts/image_outcomes.

• Step 3: OPERATIONS.md — add: the provision-bucket operator gesture (run launch-app provision-bucket before the first SLICE-7 PPE deploy; bind the overlay), the SLICE-7 telemetry events (TEL-4 image_phase_completed, TEL-5 image_phase_recovered), and RB-3 (image phase stuck): how to read it — runs stuck in fetching_images are visible in the import history; a restart triggers the recovery scan (TEL-5); ALR-3 fires on >2 h stuck or ≥3 recoveries. Note ALR-3 is provisioned as a gcloud log-based alert at deploy (Task 14), mirroring ALR-2.

• Step 4: Version bump — VERSION → 0.7.0; frontend/package.json version → 0.7.0; refresh lockfile (cd frontend && npm install).

• Step 5: Commit

git add deployment.toml docs/products-domain.md docs/OPERATIONS.md VERSION frontend/package.json frontend/package-lock.json
git commit -m "docs(products): image pipeline ops + seams (DOC-1/DOC-4); overlay bucket config; v0.7.0"

---

Task 14: integration gate, ship through the pipeline

Files: none new — verification, merge, deploy.

• Step 1: Extend the bootstrap test (INV-1 over images) — in backend/tests/test_bootstrap.py (or the existing bootstrap test), add: fresh DB → sign-up → storefront → import sample.csv → export → re-import = all-unchanged still holds with the image columns present. (No image host needed; sample.csv's images stay pending/never-fetched → export keeps source_url → round-trips.) Run that test green.

• Step 2: Full local gate

Run: bash scripts/check.sh Expected: lint-imports (layering kept), ~180+ backend pytest green, frontend typecheck+build, vitest green.

Run: bash scripts/e2e.sh Expected: 7/7 E2E green.

• Step 3: Open the PR (ecomm), self-review, merge

git push -u origin worktree-slice-7-images
# open PR via Gitea API; body cites SD-0002 §7.2 SLICE-7, PUC-7, INV-12/16/18, closes nothing
# (issue #13 stays open until SLICE-8). Merge after green local gates.

Tag the merge commit: v0.7.0 and release/<YYYY-MM-DDTHH-MM> (PST) on the deployed ref (per ppe-before-prod §9.1).

• Step 4: Provision the bucket (operator gesture) + ALR-3

Hand the operator the provision-bucket command to run (it creates wiggleverse-ecomm-ppe-media, grants the VM SA objectAdmin, sets the lifecycle rule). Then provision ALR-3 as a gcloud log-based alert in project wiggleverse-ecomm, mirroring SLICE-5's ALR-2: a log metric over image_phase_recovered/stuck-run signal + the existing email channel + an alert policy (condition: run in fetching_images > 2 h, or image_phase_recovered ≥ 3× for one run). These are operator-run/infra gestures (flotilla-only-provisioning + §8.4 dedicated config).

• Step 5: Deploy to PPE + verify (§9)

CLOUDSDK_ACTIVE_CONFIG_NAME=wiggleverse-ecomm flotilla-core deploy ecomm

Expected: 9/9 green. Then verify on https://ecomm-ppe.wiggleverse.org: /healthz reports 0.7.0; sign in; import a small CSV with a public image URL; watch the run detail progress to a terminal status; confirm the served rendition loads (/api/products/images/{id}/detail) and the round-trip (export → re-import) is all-unchanged. Re-run bash scripts/e2e.sh is localhost-only; PPE verification is the manual walk (E2E-against-PPE is the §10.6 gap).

• Step 6: Finalize — invoke wgl-session-finalize (archives this plan to the content repo plans/, marks SD-0002 §13 Q-3 resolved, updates memory with the SLICE-7 ship + the SLICE-8 next-goal, publishes the transcript, tears down the worktree).

---

Self-review notes (coverage vs §7.2 SLICE-7 DoD)

• platform/objectstore (local + GCS) → Task 2. platform/images → Task 1.
• Per-env media bucket via provision-bucket → Task 12 (built) + Task 14 (operator runs).
• Fetch task: SSRF guard, bounds, restart recovery → Task 7 (+ wiring Task 8).
• Image-serving endpoint → Task 9. INV-16 in force → Task 9 (render paths take keys; route authorizes).
• Hosted-URL resolution in diff + export (INV-12 over images) → Tasks 3/4/5 + extended property test (Task 5).
• Run-detail images section; Products-page notice band → Task 10.
• PUC-7 acceptance; E2E e2e_image_outcomes → Task 11. fetch-resume + SSRF integration → Task 7.
• TEL-4/5 → Task 7 (image_phase_completed / image_phase_recovered). ALR-3 → Task 14.
• DOC-1 updated (+ bucket gesture in operator guide) → Task 13. Q-3 pinned → decisions block + Task 13/14.
• Version 0.7.0 shipped through merge + PPE → Tasks 13/14.