fix(m3): Option A must-fix #79 #82 #83 #84 #89 #98

- #79 publish SKU/create via WooClient (User-Agent + typed 403 unrecoverable)
- #82 stale_field current_options from live taxonomy + owner labels
- #83 create allowlist/extra-key safe; taxonomy+enum 4xx on create
- #84 health woo_reachable real probe (non-failing); version 0.5.0-m3
- #89 review persists assembled title/short after taxonomy lock; UI applies
- #98 live-edge from synced term; attrs resolved by slug then name

backend/app.py

@@ -25,7 +25,7 @@ from starlette.middleware.base import BaseHTTPMiddleware
 from .db import get_connection, init_db, probe_images_writable, probe_writable
 from . import settings as settings_mod
 
-__version__ = "0.5.0-m3-s1"
+__version__ = "0.5.0-m3"
 
 # --- Structured JSON logging (stdlib only) ---
 request_id_var: ContextVar[str] = ContextVar("request_id", default="")
@@ -156,35 +156,39 @@ def _error(code: str, message: str, detail: Any = None) -> dict:
 @app.get("/api/health")
 async def health() -> dict[str, Any]:
     """Health per APP-30/45: non-200 when db or image volume not writable (probe file).
-    #84: woo_reachable is a real check (HEAD /wp-json/ with a short timeout),
+
-    not a hardcoded stub.
+    #84: woo_reachable is a real authenticated probe (signal only — store outage
+    does not fail health).
     """
     db_ok = probe_writable()
     img_ok = probe_images_writable()
     if not (db_ok and img_ok):
         logger.warning("health: storage not writable", extra={"request_id": request_id_var.get("")})
         raise HTTPException(status_code=503, detail="storage not writable")
-    woo_reachable = False
-    try:
-        woo_base = os.environ.get("WOO_BASE_URL", "")
-        if woo_base:
-            import urllib.request
-            req = urllib.request.Request(
-                woo_base.rstrip("/") + "/wp-json/",
-                method="GET",
-                headers={"User-Agent": "SlabUploader/health"},
-            )
-            with urllib.request.urlopen(req, timeout=3) as resp:
-                woo_reachable = resp.status < 500
-    except Exception:
-        woo_reachable = False
     return {
         "status": "ok",
         "version": __version__,
-        "woo_reachable": woo_reachable,
+        "woo_reachable": _probe_woo_reachable(),
     }
 
 
+def _probe_woo_reachable() -> bool:
+    """Cheap non-failing Woo reachability probe for health (SETTINGS-40 style).
+
+    Storage health never depends on this result. Missing creds → False.
+    """
+    try:
+        base, username, enc_password = _woo_credentials()
+        if not base or not username or not enc_password:
+            return False
+        client = WooClient.from_settings(base, username, enc_password)
+        # One lightweight authenticated call (categories page 1).
+        code, _h, _b = client._get("/products/categories?per_page=1&page=1")
+        return code == 200
+    except Exception:
+        return False
+
+
 @app.get("/api/v1/meta")
 async def meta() -> dict[str, Any]:
     """Per API.md §3."""

backend/publish.py

@@ -16,7 +16,14 @@ from fastapi import APIRouter
 from fastapi.responses import JSONResponse
 
 from .db import get_connection
-from .woo import WooAuthError, WooClient, WooConnectionError, WooCredentialsMissing, WooError
+from .woo import (
+    WooAuthError,
+    WooClient,
+    WooConnectionError,
+    WooCredentialsMissing,
+    WooError,
+    WooForbiddenError,
+)
 from .review import compute_ready, _slab_dict
 
 router = APIRouter(prefix="/api/v1/slabs")
@@ -72,21 +79,24 @@ def _build_payload(slab: dict, conn, status: str) -> dict:
     tags = [t for t in (slab.get("fig_tag_ids") or [])]+[t for t in (slab.get("feat_tag_ids") or [])]
     if tags:
         payload["tags"] = [{"id": t} for t in tags]
-    # Five locked attributes by synced term name.
+    # Five locked attributes: resolve the attribute row from the synced cache by
+    # slug first (METADATA-5 / QUALITY-105 — never hardcode live Woo attribute ids),
+    # then fall back to case-insensitive name. Term *options* are always the
+    # names of the selected synced terms.
     terms = {r["id"]: dict(r) for r in conn.execute("SELECT * FROM woo_attribute_terms").fetchall()}
     attrs = {r["id"]: dict(r) for r in conn.execute("SELECT * FROM woo_attributes").fetchall()}
     built_attrs = []
-    for field, attr_name, pos in (
+    for field, attr_slug, attr_name, pos in (
-        ("edge_type_term_id", "Edge Type", 0),
+        ("edge_type_term_id", "edge-type", "Edge Type", 0),
-        ("figure_term_ids", "Figure", 1),
+        ("figure_term_ids", "pa_figure", "Figure", 1),
-        ("grade_term_id", "Grade", 2),
+        ("grade_term_id", "pa_grade", "Grade", 2),
-        ("thickness_term_id", "Thickness", 3),
+        ("thickness_term_id", "pa_thickness", "Thickness", 3),
-        ("moisture_term_id", "Moisture", 4),
+        ("moisture_term_id", "pa_moisture", "Moisture", 4),
     ):
         raw = slab.get(field)
         term_ids = raw if isinstance(raw, list) else ([raw] if raw else [])
         options = [terms[t]["name"] for t in term_ids if t in terms]
-        attr_id = _attr_id_by_name(attrs, attr_name)
+        attr_id = _attr_id_from_synced(attrs, attr_slug, attr_name)
         if attr_id and options:
             built_attrs.append({"id": attr_id, "position": pos, "visible": True, "variation": False, "options": options})
     if built_attrs:
@@ -94,19 +104,35 @@ def _build_payload(slab: dict, conn, status: str) -> dict:
     return payload
 
 
-def _attr_id_by_name(attrs: dict, name: str) -> int | None:
+def _attr_id_from_synced(attrs: dict, slug: str, name: str) -> int | None:
+    """Match a locked attribute against the synced Woo set by slug, then name."""
+    slug_l = slug.lower().lstrip()
+    # Accept both "figure" and "pa_figure" style slugs from Woo.
+    slug_alts = {slug_l, slug_l.removeprefix("pa_"), f"pa_{slug_l.removeprefix('pa_')}"}
     for a in attrs.values():
-        if a.get("name", "").lower() == name.lower():
+        s = (a.get("slug") or "").lower()
+        if s in slug_alts:
+            return a["id"]
+    name_l = name.lower()
+    for a in attrs.values():
+        if (a.get("name") or "").lower() == name_l:
             return a["id"]
     return None
 
 
 def _classify(e: Exception) -> tuple[str, str, bool]:
-    """Return (code, plain message, recoverable)."""
+    """Return (code, plain message, recoverable). PUBLISH-36/37/38."""
     if isinstance(e, WooCredentialsMissing):
         return "woo_unconfigured", "Woo is not configured. Set the store credentials in Settings.", False
     if isinstance(e, WooAuthError):
         return "woo_auth", "The store rejected the credentials. Check the app password.", False
+    if isinstance(e, WooForbiddenError):
+        # WAF / security plugin block — retrying the same request will not help.
+        return (
+            "woo_forbidden",
+            "The store blocked this request (security filter). Check the store firewall or User-Agent allowlist — retrying will not help.",
+            False,
+        )
     if isinstance(e, WooConnectionError):
         return "woo_unreachable", "The store is unreachable right now. You can retry.", True
     return "woo_failed", "The store create failed. You can retry.", True
@@ -209,26 +235,13 @@ def _create_status(conn) -> str:
 
 
 def _sku_exists(client: WooClient, sku: str) -> bool:
-    code, _h, body = client._request(
+    """SKU lookup via WooClient so User-Agent and 403 typing apply (#79)."""
-        f"{client.base_url}/wp-json/wc/v3/products?sku={sku}",
+    return len(client.get_products_by_sku(sku)) > 0
-        "GET", None,
-        {"Authorization": f"Basic {client._auth}", "Accept": "application/json"},
-    )
-    if code != 200:
-        raise WooError(f"sku lookup returned {code}")
-    return len(json.loads(body)) > 0
 
 
 def _create_product(client: WooClient, payload: dict) -> int:
-    body = json.dumps(payload).encode("utf-8")
+    """Product create via WooClient so User-Agent and 403 typing apply (#79)."""
-    code, _h, resp = client._request(
+    return client.create_product(payload)
-        f"{client.base_url}/wp-json/wc/v3/products",
-        "POST", body,
-        {"Authorization": f"Basic {client._auth}", "Content-Type": "application/json", "Accept": "application/json"},
-    )
-    if code not in (200, 201):
-        raise WooError(f"product create returned {code}")
-    return int(json.loads(resp)["id"])
 
 
 def _log(conn, slab_id: str, woo_product_id: int | None, status: str, message: str) -> None:

backend/review.py

@@ -71,14 +71,16 @@ def _dims(length_in: float, width_avg_in: float, thickness_in: float) -> str:
     return f"{_fmt1(length_in)} x {_fmt1(width_avg_in)} x {_fmt2(thickness_in)} in"
 
 
-def build_title(slab: dict, figure: str | None, species_name: str, primary_leaf: dict) -> str:
+def build_title(slab: dict, figure: str | None, species_name: str, primary_leaf: dict, terms: dict | None = None) -> str:
     pattern, noun, carries_dims = _LEAF_PATTERNS.get(
         primary_leaf.get("slug"), ("{figure}{species} Slab", "Slab", False)
     )
     figure = figure.strip() if figure else ""
     live_edge = ""
-    if primary_leaf["slug"] in ("dimensional-wood", "bowl-blanks"):
+    if primary_leaf.get("slug") in ("dimensional-wood", "bowl-blanks"):
-        live_edge = "Live Edge " if slab.get("edge_type_term_id") else ""
+        # "Live Edge" is a product-form flag driven by the *synced* edge term
+        # (METADATA-5 / #98) — never treat any non-null edge id as Live Edge.
+        live_edge = "Live Edge " if _edge_is_live(slab.get("edge_type_term_id"), terms or {}) else ""
     title = pattern.format(
         figure=figure + " " if figure else "",
         species=species_name,
@@ -89,8 +91,17 @@ def build_title(slab: dict, figure: str | None, species_name: str, primary_leaf:
     return re.sub(r"\s+", " ", title).strip()
 
 
-def build_short_description(slab: dict, figure: str | None, species_name: str, primary_leaf: dict) -> str:
+def _edge_is_live(edge_term_id: Any, terms: dict) -> bool:
-    title = build_title(slab, figure, species_name, primary_leaf)
+    """True when the synced edge term name/slug indicates live edge."""
+    if not edge_term_id:
+        return False
+    term = terms.get(edge_term_id) or {}
+    blob = f"{term.get('name', '')} {term.get('slug', '')}".lower()
+    return "live" in blob
+
+
+def build_short_description(slab: dict, figure: str | None, species_name: str, primary_leaf: dict, terms: dict | None = None) -> str:
+    title = build_title(slab, figure, species_name, primary_leaf, terms)
     ft, rem = _split_length(slab["length_in"])
     length_str = f"{ft}ft {_fmt1(rem)}in"
     bdft = round(slab["bdft"], 1)
@@ -147,6 +158,13 @@ def price_suggestion(conn, slab: dict) -> float | None:
 
 @router.post("/{slab_id}/review")
 async def review_slab(slab_id: str, payload: dict[str, Any] | None = None) -> dict:
+    """Compute ready + deterministic title/short_description from *locked* taxonomy.
+
+    METADATA-80 / REVIEW-10 / #89: title and short_description are assembled only
+    from the confirmed (persisted) listing facts. Owner-supplied title/short_desc
+    win (METADATA-50). Assembled values are written back to the slab so the lock
+    sticks for publish.
+    """
     payload = payload or {}
     conn = get_connection()
     row = conn.execute(
@@ -157,9 +175,9 @@ async def review_slab(slab_id: str, payload: dict[str, Any] | None = None) -> di
     slab = _slab_dict(row)
 
     # --- Deterministic review: report ready/missing + the deterministic
-    # title/short_description built from the current locked metadata. The client's
+    # title/short_description built from the current metadata. The client's
     # "Create title" / "Create short desc" buttons call this endpoint and write the
-    # returned strings into the fields (item 1: no lock action, no metadata_locked).
+    # returned strings into the fields (#89: no separate lock action).
     # #80: persist owner overrides (price/sku) from the payload so the review
     # endpoint is self-contained. title/short_description are NOT overridden here —
     # they are produced by the deterministic build below, not persisted from the
@@ -169,8 +187,8 @@ async def review_slab(slab_id: str, payload: dict[str, Any] | None = None) -> di
     wood_cats = slab.get("wood_category_ids") or []
     if not primary_id:
         primary_id = wood_cats[0] if wood_cats else None
-    primary_leaf = cats.get(primary_id, {})
+    primary_leaf = cats.get(primary_id, {}) if primary_id else {}
-    species_name = cats.get(slab.get("species_id"), {}).get("name", "")
+    species_name = cats.get(slab.get("species_id"), {}).get("name", "") if slab.get("species_id") else ""
     figure_term_ids = slab.get("figure_term_ids") or []
     figure = terms.get(figure_term_ids[0], {}).get("name") if figure_term_ids else None
 
@@ -190,8 +208,17 @@ async def review_slab(slab_id: str, payload: dict[str, Any] | None = None) -> di
 
     # Deterministic title / short_description, re-derived from the current
     # metadata on every review (the Create button's source of truth).
-    title = build_title(slab, figure, species_name, primary_leaf) if (primary_leaf and species_name) else slab.get("title") or None
+    # #98: pass synced terms so Live Edge comes from the edge term, not any id.
-    short_desc = build_short_description(slab, figure, species_name, primary_leaf) if (primary_leaf and species_name) else slab.get("short_description") or None
+    title = (
+        build_title(slab, figure, species_name, primary_leaf, terms)
+        if (primary_leaf and species_name)
+        else slab.get("title") or None
+    )
+    short_desc = (
+        build_short_description(slab, figure, species_name, primary_leaf, terms)
+        if (primary_leaf and species_name)
+        else slab.get("short_description") or None
+    )
 
     price = _resolve_price(conn, slab, payload)
     ready, missing = compute_ready({**slab, "price": price, "title": title, "short_description": short_desc})

backend/slabs.py

@@ -43,6 +43,20 @@ TAXONOMY_TABLES = {
     "fig_tag_ids": "woo_tags",
     "feat_tag_ids": "woo_tags",
 }
+# Owner-facing labels for stale_field messages (QUALITY-120 — never raw column names).
+FIELD_LABELS = {
+    "species_id": "Species",
+    "wood_category_ids": "Wood category",
+    "edge_type_term_id": "Edge type",
+    "figure_term_ids": "Figure",
+    "grade_term_id": "Grade",
+    "thickness_term_id": "Thickness",
+    "moisture_term_id": "Moisture",
+    "fig_tag_ids": "Figure tags",
+    "feat_tag_ids": "Feature tags",
+}
+PRICE_SOURCES = {"recommendation", "override", "manual"}
+INFERENCE_STATUSES = {"idle", "inferring", "done", "failed", "skipped"}
 
 
 def _now_iso() -> str:
@@ -102,6 +116,58 @@ def _check_taxonomy(conn, table: str, val: Any) -> bool:
     return conn.execute(f"SELECT 1 FROM {table} WHERE id=?", (val,)).fetchone() is not None
 
 
+def _current_options(conn, table: str) -> list[dict[str, Any]]:
+    """Live taxonomy options for a stale_field response (PUBLISH-20 / #82)."""
+    rows = conn.execute(f"SELECT id, name FROM {table} ORDER BY name, id").fetchall()
+    return [{"id": r["id"], "name": r["name"]} for r in rows]
+
+
+def _normalize_tax_val(val: Any) -> Any:
+    if isinstance(val, str):
+        try:
+            return json.loads(val)
+        except (ValueError, TypeError):
+            return []
+    return val
+
+
+def _validate_taxonomy_fields(conn, payload: Dict[str, Any]) -> JSONResponse | None:
+    """Return a 422 stale_field response, or None if all present taxonomy ids are live."""
+    for field, table in TAXONOMY_TABLES.items():
+        if field not in payload or payload[field] is None:
+            continue
+        val = _normalize_tax_val(payload[field])
+        if not _check_taxonomy(conn, table, val):
+            label = FIELD_LABELS.get(field, field)
+            return _err(
+                "stale_field",
+                f"{label} is no longer in the store taxonomy",
+                {"field": field, "current_options": _current_options(conn, table)},
+                status=422,
+            )
+    return None
+
+
+def _validate_enums(payload: Dict[str, Any]) -> JSONResponse | None:
+    if "price_source" in payload and payload["price_source"] is not None:
+        if payload["price_source"] not in PRICE_SOURCES:
+            return _err(
+                "validation",
+                "price_source is not a known value",
+                {"field": "price_source", "allowed": sorted(PRICE_SOURCES)},
+                status=400,
+            )
+    if "inference_status" in payload and payload["inference_status"] is not None:
+        if payload["inference_status"] not in INFERENCE_STATUSES:
+            return _err(
+                "validation",
+                "inference_status is not a known value",
+                {"field": "inference_status", "allowed": sorted(INFERENCE_STATUSES)},
+                status=400,
+            )
+    return None
+
+
 # Capture-only fields required to create a slab. The Review screen fills the
 # taxonomy/content fields after create (CREATE.md §1: "every other field stays
 # editable until create"), so create must accept a capture-math payload.
@@ -127,11 +193,17 @@ async def create_slab(payload: Dict[str, Any]) -> Dict[str, Any]:
     # Business rule (middle tier, not the DB): an empty/blank SKU at create means
     # "not set yet" (the owner fills it in Review). Store NULL — the DB CHECK only
     # governs present SKUs, and NULL is the honest "absent" representation.
-    sku = payload["sku"]
+    sku = payload.get("sku")
     if isinstance(sku, str) and sku.strip() == "":
         payload = {**payload, "sku": None}
+    enum_err = _validate_enums(payload)
+    if enum_err is not None:
+        return enum_err
     conn = get_connection()
-    if payload["sku"] is not None and _sku_taken(conn, payload["sku"]):
+    tax_err = _validate_taxonomy_fields(conn, payload)
+    if tax_err is not None:
+        return tax_err
+    if payload.get("sku") is not None and _sku_taken(conn, payload["sku"]):
         return _err("duplicate_sku", "SKU already exists", {"sku": payload["sku"]}, status=409)
     # METADATA-15 / QUALITY-95: the thickness band is derived, never owner-chosen.
     # Resolve it from the measured thickness against the live synced Woo bands and
@@ -148,8 +220,13 @@ async def create_slab(payload: Dict[str, Any]) -> Dict[str, Any]:
     # One value per col, in cols order: data values for the allowed fields, then the 6 server values.
     vals = [data[f] for f in allowed_cols] + server_vals
     ph = ",".join(["?"] * len(cols))
-    conn.execute(f"INSERT INTO slabs ({', '.join(cols)}) VALUES ({ph})", vals)
+    assert len(vals) == len(cols), f"create_slab binding mismatch: {len(vals)} vals vs {len(cols)} cols"
-    conn.commit()
+    try:
+        conn.execute(f"INSERT INTO slabs ({', '.join(cols)}) VALUES ({ph})", vals)
+        conn.commit()
+    except Exception:
+        # QUALITY-120: never surface a driver stack to the owner.
+        return _err("validation", "Could not create the slab. Check the fields and try again.", status=400)
     row = conn.execute("SELECT * FROM slabs WHERE id=?", (payload["id"],)).fetchone()
     return _row_to_slab(row)
 
@@ -174,11 +251,13 @@ async def get_slab(id: str) -> Dict[str, Any]:
 async def update_slab(id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
     if "client_rev" not in payload:
         return _err("validation", "client_rev required", status=400)
-    # #89: the metadata lock is server-managed via the explicit review lock
+    # #89: server-managed fields a plain PUT must not set.
-    # action; a plain PUT may not flip it.
     for field in CLIENT_FORBIDDEN_FIELDS:
         if field in payload:
             return _err("validation", f"{field} is set by the review lock action, not by update", status=400)
+    enum_err = _validate_enums(payload)
+    if enum_err is not None:
+        return enum_err
     conn = get_connection()
     current = conn.execute("SELECT * FROM slabs WHERE id=? AND deleted_at IS NULL", (id,)).fetchone()
     if current is None:
@@ -188,27 +267,12 @@ async def update_slab(id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
         return _err("revision_conflict", "client_rev mismatch",
                     {"current_server_rev": current["server_rev"]}, status=409)
     if "sku" in payload and payload["sku"] != current["sku"]:
-        if _sku_taken(conn, payload["sku"], exclude_id=id):
+        if payload["sku"] is not None and _sku_taken(conn, payload["sku"], exclude_id=id):
             return _err("duplicate_sku", "SKU already exists", {"sku": payload["sku"]}, status=409)
-    for field, table in TAXONOMY_TABLES.items():
+    tax_err = _validate_taxonomy_fields(conn, payload)
-        if field in payload and payload[field] is not None:
+    if tax_err is not None:
-            val = payload[field]
+        return tax_err
-            if isinstance(val, str):
+    # METADATA-15 / QUALITY-95: thickness band is derived, never owner-chosen.
-                try:
-                    val = json.loads(val)
-                except (ValueError, TypeError):
-                    val = []
-            if not _check_taxonomy(conn, table, val):
-                # #82: return the LIVE taxonomy options (valid ids in the table),
-                # not the submitted value. The client needs to know what's valid.
-                if isinstance(val, list):
-                    live = [r[0] for r in conn.execute(
-                        f"SELECT id FROM {table} ORDER BY id"
-                    ).fetchall()]
-                else:
-                    live = [val]  # single-id field: just show what was submitted
-                return _err("stale_field", f"{field} not in current taxonomy",
-                            {"field": field, "submitted": val, "current_options": live}, status=422)
     eff_thickness = payload.get("thickness_in", current["thickness_in"])
     band = resolve_thickness_band_id(eff_thickness, load_thickness_terms(conn))
     if band is not None:

backend/tests/test_publish.py

@@ -300,3 +300,63 @@ def test_partial_media_failure_502(env, monkeypatch):
     # sync_log written, status failed
     assert conn.execute("SELECT status FROM slabs WHERE id='s1'").fetchone()["status"] == "failed"
     assert conn.execute("SELECT COUNT(*) c FROM sync_log WHERE status='failed'").fetchone()["c"] >= 1
+
+
+def test_publish_sends_user_agent_on_sku_and_create(env, monkeypatch):
+    """#79: every publish Woo hop carries User-Agent (same transport as client)."""
+    client, conn = env
+    s = _full_slab()
+    _insert_slab(conn, s)
+    conn.commit()
+    seen = []
+
+    def fake_request(url, method, data, headers):
+        seen.append({"url": url, "method": method, "headers": dict(headers or {})})
+        if "/products?sku=" in url:
+            return 200, {}, b"[]"
+        if url.endswith("/wp-json/wc/v3/products") and method == "POST":
+            return 201, {}, json.dumps({"id": 42}).encode()
+        return 200, {}, b"[]"
+
+    from backend.woo import WooClient, _USER_AGENT
+
+    orig_init = WooClient.__init__
+
+    def patched_init(self, *a, **k):
+        orig_init(self, *a, **k)
+        self._request = fake_request
+
+    monkeypatch.setattr(WooClient, "__init__", patched_init)
+    r = client.post("/api/v1/slabs/s1/publish")
+    assert r.status_code == 200, r.text
+    sku_hops = [h for h in seen if "/products?sku=" in h["url"]]
+    create_hops = [h for h in seen if h["url"].endswith("/wp-json/wc/v3/products") and h["method"] == "POST"]
+    assert sku_hops and create_hops
+    for hop in sku_hops + create_hops:
+        assert hop["headers"].get("User-Agent") == _USER_AGENT
+
+
+def test_publish_403_is_unrecoverable(env, monkeypatch):
+    """#79: WAF 403 on SKU lookup is unrecoverable — do not invite retry."""
+    client, conn = env
+    s = _full_slab()
+    _insert_slab(conn, s)
+    conn.commit()
+    from backend.woo import WooClient, WooForbiddenError
+
+    def fake_request(url, method, data, headers):
+        raise WooForbiddenError("request forbidden (403)")
+
+    orig_init = WooClient.__init__
+
+    def patched_init(self, *a, **k):
+        orig_init(self, *a, **k)
+        self._request = fake_request
+
+    monkeypatch.setattr(WooClient, "__init__", patched_init)
+    r = client.post("/api/v1/slabs/s1/publish")
+    assert r.status_code == 502
+    body = r.json()
+    assert body["error"]["code"] == "woo_forbidden"
+    assert body["error"]["detail"]["recoverable"] is False
+    assert "retry" not in body["error"]["message"].lower() or "will not help" in body["error"]["message"].lower()

backend/tests/test_slabs_api.py

@@ -340,18 +340,45 @@ def test_create_rejects_unknown_fields(client, db):
 
 
 def test_update_stale_field_returns_live_options(client, db):
-    # #82: stale_field error returns the LIVE taxonomy ids, not the submitted value.
+    # #82: stale_field returns LIVE taxonomy options (id+name), not the submitted id.
     slab_id = str(uuid.uuid4())
-    payload = {"id": slab_id, "sku": "SKU001", "length_in":10,"thickness_in":2,"sqft":20,"bdft":5,"width_min_in":1,"width_max_in":2,"width_avg_in":1.5,"species_id":1,"wood_category_ids":json.dumps([2]),"edge_type_term_id":1,"figure_term_ids":json.dumps([1]),"grade_term_id":1,"thickness_term_id":1,"moisture_term_id":1,"fig_tag_ids":json.dumps([1]),"feat_tag_ids":json.dumps([2]),"price":100,"price_source":"recommendation","title":"Test","short_description":"short","description":"desc","content_source":"manual"}
+    payload = {
-    client.post("/api/v1/slabs", json=payload)
+        "id": slab_id, "sku": "SKU-STALE", "length_in": 10, "thickness_in": 2, "sqft": 20, "bdft": 5,
-    # Submit a wood_category_id that doesn't exist in the taxonomy (id=999).
+        "width_min_in": 1, "width_max_in": 2, "width_avg_in": 1.5,
-    resp = client.put(f"/api/v1/slabs/{slab_id}", json={"client_rev": 1, "wood_category_ids": json.dumps([999])})
+    }
+    assert client.post("/api/v1/slabs", json=payload).status_code == 201
+    resp = client.put(f"/api/v1/slabs/{slab_id}", json={"client_rev": 1, "edge_type_term_id": 999})
     assert resp.status_code == 422
     body = resp.json()
     assert body["error"]["code"] == "stale_field"
-    detail = body["error"]["detail"]
+    opts = body["error"]["detail"]["current_options"]
-    assert detail["submitted"] == [999]
+    assert isinstance(opts, list) and opts
-    # current_options must be the LIVE ids from woo_categories, not [999].
+    assert all(isinstance(o, dict) and "id" in o and "name" in o for o in opts)
-    assert 999 not in detail["current_options"]
+    assert 999 not in [o["id"] for o in opts]
-    # The seeded taxonomy has ids 1 and 2 in woo_categories.
+    assert "Edge type" in body["error"]["message"] or "edge" in body["error"]["message"].lower()
-    assert set(detail["current_options"]) == {1, 2}
+
+
+def test_create_extra_key_and_bad_taxonomy_are_4xx(client, db):
+    """#83: server-managed/extra keys and bad taxonomy/enums are 4xx, not 500."""
+    base = {
+        "id": str(uuid.uuid4()), "sku": "SKU-XTRA", "length_in": 10, "thickness_in": 2, "sqft": 20, "bdft": 5,
+        "width_min_in": 1, "width_max_in": 2, "width_avg_in": 1.5,
+        "status": "published",  # server-managed → unknown field 400
+    }
+    r = client.post("/api/v1/slabs", json=base)
+    assert r.status_code == 400, r.text
+    assert "status" in r.json()["error"]["detail"]["fields"]
+    # Bad taxonomy on create
+    r2 = client.post("/api/v1/slabs", json={
+        "id": str(uuid.uuid4()), "sku": "SKU-BADTAX", "length_in": 10, "thickness_in": 2, "sqft": 20, "bdft": 5,
+        "width_min_in": 1, "width_max_in": 2, "width_avg_in": 1.5, "species_id": 999,
+    })
+    assert r2.status_code == 422
+    assert r2.json()["error"]["code"] == "stale_field"
+    # Bad enum
+    r3 = client.post("/api/v1/slabs", json={
+        "id": str(uuid.uuid4()), "sku": "SKU-BADENUM", "length_in": 10, "thickness_in": 2, "sqft": 20, "bdft": 5,
+        "width_min_in": 1, "width_max_in": 2, "width_avg_in": 1.5, "price_source": "not-a-source",
+    })
+    assert r3.status_code == 400
+    assert r3.json()["error"]["code"] == "validation"

backend/woo.py

@@ -98,14 +98,7 @@ class WooClient:
 
     def _get(self, path: str) -> tuple[int, dict, bytes]:
         url = f"{self.base_url}/wp-json/wc/v3{path}"
-        return self._request(
+        return self._request(url, "GET", None, self._auth_headers())
-            url, "GET", None,
-            {
-                "Authorization": f"Basic {self._auth}",
-                "Accept": "application/json",
-                "User-Agent": _USER_AGENT,
-            },
-        )
 
     def _fetch_all(self, path: str) -> list[dict[str, Any]]:
         page = 1
@@ -132,6 +125,42 @@ class WooClient:
     def get_attribute_terms(self, attr_id: int) -> list[dict[str, Any]]:
         return self._fetch_all(f"/products/attributes/{attr_id}/terms")
 
+    def _auth_headers(self, extra: dict[str, str] | None = None) -> dict[str, str]:
+        """Shared headers for every outbound Woo/WP request (User-Agent + auth)."""
+        h = {
+            "Authorization": f"Basic {self._auth}",
+            "Accept": "application/json",
+            "User-Agent": _USER_AGENT,
+        }
+        if extra:
+            h.update(extra)
+        return h
+
+    def get_products_by_sku(self, sku: str) -> list[dict[str, Any]]:
+        """GET products filtered by SKU. Raises typed Woo* errors on non-200."""
+        code, _h, body = self._request(
+            f"{self.base_url}/wp-json/wc/v3/products?sku={sku}",
+            "GET",
+            None,
+            self._auth_headers(),
+        )
+        if code != 200:
+            raise WooServerError(f"sku lookup returned {code}")
+        return list(json.loads(body))
+
+    def create_product(self, payload: dict[str, Any]) -> int:
+        """POST /products. Raises typed Woo* errors on failure; returns product id."""
+        body = json.dumps(payload).encode("utf-8")
+        code, _h, resp = self._request(
+            f"{self.base_url}/wp-json/wc/v3/products",
+            "POST",
+            body,
+            self._auth_headers({"Content-Type": "application/json"}),
+        )
+        if code not in (200, 201):
+            raise WooServerError(f"product create returned {code}")
+        return int(json.loads(resp)["id"])
+
     def upload_media(self, png_bytes: bytes, filename: str) -> int:
         """POST a PNG to WordPress core media; return the attachment id."""
         boundary = "----slabuploader-media-boundary"
@@ -143,11 +172,7 @@ class WooClient:
         url = f"{self.base_url}/wp-json/wp/v2/media"
         code, _headers, resp = self._request(
             url, "POST", body,
-            {
+            self._auth_headers({"Content-Type": f"multipart/form-data; boundary={boundary}"}),
-                "Authorization": f"Basic {self._auth}",
-                "Content-Type": f"multipart/form-data; boundary={boundary}",
-                "User-Agent": _USER_AGENT,
-            },
         )
         if code not in (200, 201):
             raise WooServerError(f"media upload returned {code}")