"""Inference (S7): prompt revisions (REST), Call 1 vision classification, Call 2
content prose, test-inference, inference-off server description, inference_log.
Cites METADATA-35..115, SETTINGS-30/45/90.

overnight-whole-app: Call 1 POST is hard-disabled (410 call1_disabled). Call 2 stays.
"""

import base64
import hashlib
import html
import re
import io
import json
import time
import urllib.error
import urllib.request
from typing import Any

from fastapi import APIRouter
from fastapi.responses import JSONResponse
from PIL import Image

from .db import get_connection
from .crypto import decrypt

router = APIRouter(prefix="/api/v1")

# METADATA-35: Call 1 (and Call 2) send clean processed photos at 1024 px long edge or less.
_INFER_MAX_LONG_EDGE = 1024


def fit_long_edge_png(png_bytes: bytes, max_long_edge: int = _INFER_MAX_LONG_EDGE) -> bytes:
    """Downscale a PNG so max(width, height) <= max_long_edge. Never upscale. Keep alpha.

    Used only for the inference request payload; stored processed_blob / Woo media stay full size.
    Unreadable or non-image bytes are returned unchanged so a bad blob cannot crash Call 1.
    """
    if not png_bytes or max_long_edge < 1:
        return png_bytes
    try:
        im = Image.open(io.BytesIO(png_bytes))
        im.load()
    except Exception:
        return png_bytes
    w, h = im.size
    long_edge = max(w, h)
    if long_edge <= max_long_edge:
        # Already within budget; re-encode only if we need a stable PNG (caller always base64s).
        return png_bytes
    scale = max_long_edge / float(long_edge)
    nw = max(1, int(round(w * scale)))
    nh = max(1, int(round(h * scale)))
    # Preserve alpha: convert palette/etc. to RGBA when needed, LANCZOS downscale.
    if im.mode not in ("RGB", "RGBA", "L", "LA"):
        im = im.convert("RGBA")
    out = im.resize((nw, nh), Image.Resampling.LANCZOS)
    buf = io.BytesIO()
    out.save(buf, format="PNG", optimize=True)
    return buf.getvalue()


def _image_part_from_processed(png_bytes: bytes) -> dict:
    """Build one OpenAI-style image_url part from a processed PNG (METADATA-35 sized)."""
    fitted = fit_long_edge_png(png_bytes)
    b64 = base64.b64encode(fitted).decode()
    return {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}

# Locked Call 1 seed (#86 / METADATA-40): expert helper + closed Woo lists only.
CALL1_SEED = (
    "You are a wood identification expert for retail lumber slabs. Judge only from the photos "
    "and the LISTS below. Use list ids only — never invent a name or id.\n"
    "\n"
    "Work in this order: (1) species, (2) wood categories, (3) edge type, (4) figure, "
    "(5) feature tags. Decide wood categories before edge type — edge rules depend on product form.\n"
    "\n"
    "For each field, pick from its list and give a confidence 0–1 (1 = certain, 0 = no idea). "
    "Prefer null or [] over a weak guess.\n"
    "\n"
    "Species — which tree species is this slab. Match grain, color, and figure to the species "
    "list. If more than one entry is plausible, return only the single highest-confidence match "
    "(one id). Null only when none are credible.\n"
    "\n"
    "Wood categories — product form / how it will be sold (live edge, dimensional, cookies & "
    "burl slices, turning blanks, etc.). Choose every wood_categories id that clearly fits; "
    "often one, sometimes more. [] if none fit. Finish this field before edge type.\n"
    "\n"
    "Edge type — the long-edge mill profile. One id from edge_terms, or null only if none fit.\n"
    "  Meanings (match the closest list name):\n"
    "  • Square / square edge — both long edges milled straight (2 milled long edges).\n"
    "  • Live one side / Live Edge 1 side — one long edge milled, the other natural bark or "
    "waney (1 milled long edge).\n"
    "  • Live both / Live edge 2 sides — neither long edge milled; both natural "
    "(0 milled long edges).\n"
    "  • Wild edge — use when the three mill-count cases do not fit, including cookies and burl "
    "cuts/slices (no conventional slab long-edge pair). Prefer Wild for those forms rather than null.\n"
    "\n"
    "Figure — visible figure in the wood (curl, quilt, spalt, plain, burl, …). Zero or more ids "
    "from figure_terms that clearly apply.\n"
    "\n"
    "Feature tags — defects and character marks the store tags for shoppers (knots, checks, "
    "inclusions, …). Zero or more ids from feature_tags. If that list is empty, return [].\n"
    "\n"
    "Return one JSON object with exactly these keys:\n"
    '{"species_id": <id|null>, "species_confidence": <0-1>, '
    '"wood_category_ids": [<id>,...], "wood_category_confidence": <0-1>, '
    '"edge_type_term_id": <id|null>, "edge_confidence": <0-1>, '
    '"figure_term_ids": [<id>,...], "figure_confidence": <0-1>, '
    '"feature_tag_ids": [<id>,...], "feature_tag_confidence": <0-1>}'
)
CALL2_SEED = (
    "Write listing copy for this slab for woodworkers choosing stock for a project. "
    "Audience: makers matching wood to a tabletop, bar, bench, or similar build — practical, "
    "material/figure/spec-forward. No lifestyle or marketing hype "
    "(avoid empty phrases like elevate, stunning, dramatic visual contrast).\n"
    "You write BOTH the short description and the long description.\n"
    "\n"
    "PLAIN TEXT ONLY (short AND long):\n"
    "- Output plain text only. No HTML tags (<p>, <ul>, <li>, <strong>, <br>, etc.). "
    "No markdown bold/headers that look like markup (**bold**, # headers).\n"
    "- Structure the long description with newlines and \"- \" bullets when helpful.\n"
    "\n"
    "Rules that always apply:\n"
    "- Dimensions in the DIMENSION FACT BLOCK are exact: use those strings; invent no numbers.\n"
    "- Use taxonomy names from DECLARED ATTRIBUTES; do not invent species, edge, figure, grade, "
    "moisture, or features.\n"
    "- Use the Brand voice and Geography sections when provided. If Brand voice is empty, stay "
    "neutral. Never invent return, warranty, or shipping policies.\n"
    "- The product name may already appear in DECLARED ATTRIBUTES — do not invent a new title.\n"
    "- Lead with usable traits (species, figure, size, edge, moisture, grade) and what the piece "
    "suits (tabletop, bar, etc.) without ad-copy fluff.\n"
    "\n"
    "SHORT DESCRIPTION (~50 words — most important on-page copy; plain text only):\n"
    "1. Lead with a practical outcome for a maker (what they can build / how the stock helps), "
    "not lifestyle hype and not a bare species list.\n"
    "2. Add the key differentiator or most important spec (figure, edge, size, moisture, grade).\n"
    "3. Close with a trust signal or concrete reason to choose this piece.\n"
    "\n"
    "LONG DESCRIPTION (plain text only — newlines and \"- \" bullets; no HTML, no markdown "
    "markup). Four parts, in order:\n"
    "1. Hook — one sentence on what the piece suits and its standout usable trait.\n"
    "2. Benefits — 3–5 bullets; each maker-useful trait first, then the supporting spec.\n"
    "3. Full specs — clean list from DECLARED ATTRIBUTES + DIMENSION FACT BLOCK "
    "(dims, species, moisture, edge, figure, grade, etc.).\n"
    "4. Trust close — return/warranty/brand; use Brand voice when present; otherwise a short "
    "neutral close with no invented policies.\n"
    "\n"
    "Return ONE JSON object only (no prose outside JSON) with exactly these keys:\n"
    '{"short_description": "<plain text ~50 words>", "description": "<plain text long description>"}'
)
_SEEDS = {"call1": CALL1_SEED, "call2": CALL2_SEED}
_THRESHOLD = 0.7


def _error(code: str, message: str, detail: Any = None) -> dict:
    return {"error": {"code": code, "message": message, "detail": detail}}


def _now() -> str:
    from datetime import datetime, timezone

    return datetime.now(timezone.utc).replace(microsecond=0, tzinfo=None).isoformat()


def _hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _seed_prompts(conn) -> None:
    """Ensure each call has a prompt row and the shipped seed revision.

    If the shipped seed hash is new and the active revision is still seed-authored
    (or missing), activate the new seed so deploys pick up CALL1_SEED / CALL2_SEED
    updates without clobbering an owner-activated API revision.
    """
    for name, seed in _SEEDS.items():
        prow = conn.execute("SELECT id, active_revision_id FROM prompts WHERE name=?", (name,)).fetchone()
        if prow is None:
            cur = conn.execute(
                "INSERT INTO prompts(name, active_revision_id, created_at, updated_at) VALUES (?,?,?,?)",
                (name, None, _now(), _now()),
            )
            pid = cur.lastrowid
            active_id = None
        else:
            pid = prow[0]
            active_id = prow[1]
        h = _hash(seed)
        existing = conn.execute(
            "SELECT id FROM prompt_revisions WHERE prompt_id=? AND content_hash=?",
            (pid, h),
        ).fetchone()
        if existing is None:
            cur2 = conn.execute(
                "INSERT INTO prompt_revisions(prompt_id, content, content_hash, created_at, created_by) "
                "VALUES (?,?,?,?,?)",
                (pid, seed, h, _now(), "seed"),
            )
            seed_rev_id = cur2.lastrowid
        else:
            seed_rev_id = existing[0]
        # Activate shipped seed when nothing active, or active is still a seed revision.
        activate = False
        if active_id is None:
            activate = True
        else:
            act = conn.execute(
                "SELECT created_by FROM prompt_revisions WHERE id=?", (active_id,)
            ).fetchone()
            if act is None or act[0] == "seed":
                activate = True
        if activate and active_id != seed_rev_id:
            conn.execute(
                "UPDATE prompts SET active_revision_id=?, updated_at=? WHERE id=?",
                (seed_rev_id, _now(), pid),
            )


def _active_prompt(conn, call: str) -> tuple[int, str] | None:
    row = conn.execute(
        "SELECT p.id, pr.id, pr.content FROM prompts p LEFT JOIN prompt_revisions pr ON p.active_revision_id=pr.id WHERE p.name=?",
        (call,),
    ).fetchone()
    if row is None or row[2] is None:
        return None
    return row[1], row[2]


# ---------- Prompt revision REST (METADATA-106) ----------
@router.post("/prompts/{call}/revisions", status_code=201)
async def append_revision(call: str, payload: dict[str, Any]) -> dict:
    if call not in _SEEDS:
        return JSONResponse(status_code=404, content=_error("not_found", "unknown call"))
    content = payload.get("content")
    if not content:
        return JSONResponse(status_code=400, content=_error("validation", "content required"))
    conn = get_connection()
    _seed_prompts(conn)
    prow = conn.execute("SELECT id FROM prompts WHERE name=?", (call,)).fetchone()
    cur = conn.execute(
        "INSERT INTO prompt_revisions(prompt_id, content, content_hash, created_at, created_by) VALUES (?,?,?,?,?)",
        (prow[0], content, _hash(content), _now(), "api"),
    )
    conn.execute("UPDATE prompts SET active_revision_id=?, updated_at=? WHERE id=?", (cur.lastrowid, _now(), prow[0]))
    conn.commit()
    return {"id": cur.lastrowid, "prompt": call, "content_hash": _hash(content)}


@router.post("/prompts/{call}/activate")
async def activate_revision(call: str, payload: dict[str, Any]) -> dict:
    if call not in _SEEDS:
        return JSONResponse(status_code=404, content=_error("not_found", "unknown call"))
    rev_id = payload.get("revision_id")
    conn = get_connection()
    _seed_prompts(conn)
    prow = conn.execute("SELECT id FROM prompts WHERE name=?", (call,)).fetchone()
    row = conn.execute(
        "SELECT id FROM prompt_revisions WHERE id=? AND prompt_id=?", (rev_id, prow[0])
    ).fetchone()
    if row is None:
        return JSONResponse(status_code=404, content=_error("not_found", "revision not found"))
    conn.execute("UPDATE prompts SET active_revision_id=?, updated_at=? WHERE id=?", (rev_id, _now(), prow[0]))
    conn.commit()
    return {"ok": True, "active_revision_id": rev_id}


@router.get("/prompts/{call}")
async def get_prompt(call: str) -> dict:
    if call not in _SEEDS:
        return JSONResponse(status_code=404, content=_error("not_found", "unknown call"))
    conn = get_connection()
    _seed_prompts(conn)
    prow = conn.execute("SELECT * FROM prompts WHERE name=?", (call,)).fetchone()
    revs = conn.execute(
        "SELECT id, content_hash, created_at, created_by FROM prompt_revisions WHERE prompt_id=? ORDER BY id",
        (prow[0],),
    ).fetchall()
    active = conn.execute(
        "SELECT content FROM prompt_revisions WHERE id=?", (prow["active_revision_id"],)
    ).fetchone()
    return {
        "name": call,
        "active_revision_id": prow["active_revision_id"],
        "active_content": active[0] if active else None,
        "revisions": [dict(r) for r in revs],
    }


# ---------- Inference client (OpenAI-compatible, stdlib) ----------
class InferenceClient:
    def __init__(self, base_url: str, api_key: str, model: str, timeout: float = 60.0):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.model = model
        self.timeout = timeout

    def chat(self, messages: list[dict]) -> str:
        url = f"{self.base_url}/chat/completions"
        body = json.dumps({"model": self.model, "messages": messages}).encode("utf-8")
        req = urllib.request.Request(
            url, data=body,
            headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
            method="POST",
        )
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                data = json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            raise InferenceError(f"inference HTTP {e.code}") from e
        except urllib.error.URLError as e:
            raise InferenceConnectionError(f"inference unreachable: {e.reason}") from e
        return data["choices"][0]["message"]["content"]


class InferenceError(Exception):
    pass


class InferenceConnectionError(InferenceError):
    pass


def _inference_creds(conn) -> tuple[str, str, str]:
    b = conn.execute("SELECT value_enc FROM settings WHERE key='inference_base_url'").fetchone()
    k = conn.execute("SELECT value_enc FROM settings WHERE key='inference_api_key'").fetchone()
    m = conn.execute("SELECT value FROM settings WHERE key='inference_model'").fetchone()
    return (decrypt(b[0]) if b and b[0] else ""), (decrypt(k[0]) if k and k[0] else ""), (m[0] if m and m[0] else "")


def _enabled(conn, key: str) -> bool:
    row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
    return bool(row and row[0] in ("true", "True", "1"))


@router.post("/slabs/{slab_id}/inference/call1")
async def call1(slab_id: str) -> dict:
    """Call 1 OUT (overnight-whole-app): hard-disabled. Call 2 stays.

    Route retained so clients get an explicit 410 instead of a half-wired path.
    Prompt seed / revision APIs for call1 remain for docs & future re-enable.
    """
    _ = slab_id  # unused while disabled
    return JSONResponse(
        status_code=410,
        content=_error(
            "call1_disabled",
            "Call 1 (vision classification) is disabled on this branch. Use Call 2 for description text.",
            {"call": 1, "disabled": True},
        ),
    )



def _feat_display_name(name: str) -> str:
    """Strip leading feat- / feat_ for the model (#86); Woo id stays authoritative."""
    n = (name or "").strip()
    low = n.lower()
    for prefix in ("feat-", "feat_"):
        if low.startswith(prefix):
            return n[len(prefix):].strip() or n
    return n


def _taxonomy_lists(conn) -> dict:
    """Closed Call 1 lists only (#86 / METADATA-40).

    Keys match the seed prompt. No grade/thickness/moisture (owner fields).
    Feature tag names are shown without the feat- prefix; ids are Woo/local tag ids.
    Groupers resolved by slug only (QUALITY-105).
    """
    out: dict[str, list] = {
        "species": [],
        "wood_categories": [],
        "edge_terms": [],
        "figure_terms": [],
        "feature_tags": [],
    }
    # species = leaves under the 'species' grouper
    sp = conn.execute("SELECT id FROM woo_categories WHERE slug='species'").fetchone()
    if sp:
        for r in conn.execute(
            "SELECT id,name FROM woo_categories WHERE parent_id=?", (sp[0],)
        ).fetchall():
            out["species"].append({"id": r[0], "name": r[1]})
    # wood categories = leaves under the 'wcat' grouper (leaves may sit under sub-groupers).
    wc = conn.execute("SELECT id FROM woo_categories WHERE slug='wcat'").fetchone()
    if wc:
        kids = [r[0] for r in conn.execute(
            "SELECT id FROM woo_categories WHERE parent_id=?", (wc[0],)
        ).fetchall()]
        leaf_ids = []
        for cid in kids:
            has_kids = conn.execute(
                "SELECT 1 FROM woo_categories WHERE parent_id=? LIMIT 1", (cid,)
            ).fetchone()
            if has_kids:
                leaf_ids.extend(r[0] for r in conn.execute(
                    "SELECT id FROM woo_categories WHERE parent_id=?", (cid,)
                ).fetchall())
            else:
                leaf_ids.append(cid)
        if leaf_ids:
            q = ",".join("?" * len(leaf_ids))
            for r in conn.execute(
                "SELECT id,name FROM woo_categories WHERE id IN (%s)" % q, leaf_ids
            ).fetchall():
                out["wood_categories"].append({"id": r[0], "name": r[1]})
    for attr, key in (("Edge type", "edge_terms"), ("Figure", "figure_terms")):
        rows = conn.execute(
            "SELECT t.id,t.name FROM woo_attribute_terms t JOIN woo_attributes a ON t.attribute_id=a.id "
            "WHERE lower(a.name)=lower(?)",
            (attr,),
        ).fetchall()
        out[key] = [{"id": r[0], "name": r[1]} for r in rows]
    for r in conn.execute("SELECT id,name FROM woo_tags WHERE slug LIKE 'feat-%'").fetchall():
        out["feature_tags"].append({"id": r[0], "name": _feat_display_name(r[1])})
    return out


def _parse_json(raw: str) -> dict:
    try:
        s = raw.strip()
        if s.startswith("```"):
            s = s.split("```")[1]
            if s.startswith("json"):
                s = s[4:]
        return json.loads(s)
    except (ValueError, IndexError):
        return {}


def _gate(parsed: dict) -> dict:
    """Apply the 0.7 confidence gate; below-threshold fields are omitted (METADATA-45).

    Confidence is read from the flat '<field>_confidence' keys; as a safety net it also
    understands the nested 'confidence: {<field>: n}' shape some models emit, so a shape
    wobble cannot silently nuke every suggestion. Grade is never suggested (METADATA-40).
    Free-text observation bags are dropped — feature tags are ids only (#86).
    """
    conf_key = {
        "species_id": "species_confidence",
        "wood_category_ids": "wood_category_confidence",
        "edge_type_term_id": "edge_confidence",
        "figure_term_ids": "figure_confidence",
        "feature_tag_ids": "feature_tag_confidence",
    }
    conf_obj = parsed.get("confidence") if isinstance(parsed.get("confidence"), dict) else {}

    def _conf(field: str) -> Any:
        k = conf_key.get(field, f"{field}_confidence")
        if k in conf_obj:
            return conf_obj[k]
        return parsed.get(k)

    def keep(field: str) -> Any:
        conf = _conf(field)
        if isinstance(conf, (int, float)) and conf >= _THRESHOLD and field in parsed:
            return parsed[field]
        return None

    out: dict[str, Any] = {}
    if v := keep("species_id"):
        out["species_id"] = v
    if v := keep("wood_category_ids"):
        out["wood_category_ids"] = v
    if v := keep("edge_type_term_id"):
        out["edge_type_term_id"] = v
    if v := keep("figure_term_ids"):
        out["figure_term_ids"] = v
    if v := keep("feature_tag_ids"):
        out["feature_tag_ids"] = v
    # grade / free-text bags intentionally omitted (METADATA-40, #86).
    return out


def _json_list(val: Any) -> list:
    if val is None:
        return []
    if isinstance(val, list):
        return val
    if isinstance(val, str):
        try:
            parsed = json.loads(val)
            return parsed if isinstance(parsed, list) else []
        except (ValueError, TypeError):
            return []
    return []


def _term_name(terms: dict, term_id: Any) -> str:
    if not term_id:
        return ""
    t = terms.get(int(term_id) if not isinstance(term_id, int) and str(term_id).isdigit() else term_id) or {}
    try:
        from .taxonomy import _decode_name
        return _decode_name(t.get("name") or "")
    except Exception:
        return str(t.get("name") or "")


def _cat_name(cats: dict, cat_id: Any) -> str:
    if not cat_id:
        return ""
    c = cats.get(int(cat_id) if not isinstance(cat_id, int) and str(cat_id).isdigit() else cat_id) or {}
    try:
        from .taxonomy import _decode_name
        return _decode_name(c.get("name") or "")
    except Exception:
        return str(c.get("name") or "")


def _parent_path(cats: dict, cat_id: Any) -> str:
    """Parent-category path for a wood-category leaf (METADATA-60)."""
    if not cat_id:
        return ""
    cid = int(cat_id) if not isinstance(cat_id, int) and str(cat_id).isdigit() else cat_id
    leaf = cats.get(cid) or {}
    parts: list[str] = []
    pid = leaf.get("parent_id")
    seen = set()
    while pid and pid not in seen:
        seen.add(pid)
        parent = cats.get(pid) or {}
        name = _cat_name(cats, pid)
        if name:
            parts.append(name)
        pid = parent.get("parent_id")
    parts.reverse()
    leaf_name = _cat_name(cats, cid)
    if leaf_name:
        parts.append(leaf_name)
    return " > ".join(parts)


def _feat_names(conn, feat_ids: list) -> list[str]:
    if not feat_ids:
        return []
    q = ",".join("?" * len(feat_ids))
    rows = conn.execute(
        f"SELECT id, name FROM woo_tags WHERE id IN ({q})", feat_ids
    ).fetchall()
    by_id = {r[0]: _feat_display_name(r[1]) for r in rows}
    return [by_id[i] for i in feat_ids if i in by_id]


def build_declared_attributes(conn, slab: dict) -> str:
    """Structured DECLARED ATTRIBUTES from the slab for Call 2 (METADATA-60).

    Names only — no hardcoded Curl≠Curly bridges. Primary = first wood category
    when not otherwise known; secondary = the rest.
    """
    cats = {r["id"]: dict(r) for r in conn.execute("SELECT * FROM woo_categories").fetchall()}
    terms = {r["id"]: dict(r) for r in conn.execute("SELECT * FROM woo_attribute_terms").fetchall()}
    wood_ids = _json_list(slab.get("wood_category_ids"))
    primary_id = wood_ids[0] if wood_ids else None
    secondary_ids = wood_ids[1:] if len(wood_ids) > 1 else []
    figure_ids = _json_list(slab.get("figure_term_ids"))
    feat_ids = _json_list(slab.get("feat_tag_ids"))
    lines = [
        "DECLARED ATTRIBUTES (confirmed in review — use these names; do not invent taxonomy):",
        f"species: {_cat_name(cats, slab.get('species_id')) or '(none)'}",
        f"primary wood category: {_cat_name(cats, primary_id) or '(none)'}",
        f"primary path: {_parent_path(cats, primary_id) or '(none)'}",
        f"secondary wood categories: {', '.join(_cat_name(cats, i) for i in secondary_ids) or '(none)'}",
        f"edge: {_term_name(terms, slab.get('edge_type_term_id')) or '(none)'}",
        f"figure: {', '.join(_term_name(terms, i) for i in figure_ids) or '(none)'}",
        f"grade: {_term_name(terms, slab.get('grade_term_id')) or '(none)'}",
        f"moisture: {_term_name(terms, slab.get('moisture_term_id')) or '(none)'}",
        f"features: {', '.join(_feat_names(conn, feat_ids)) or '(none)'}",
    ]
    title = (slab.get("title") or "").strip()
    if title:
        lines.append(f"title: {title}")
    # short_description is Call 2 output — do not feed a prior value back into the prompt.
    return "\n".join(lines)


def build_dimension_fact_block(slab: dict) -> str:
    """Exact dimension strings — invent no numbers (METADATA-65/90)."""
    return (
        "DIMENSION FACT BLOCK (exact, invent no numbers):\n"
        f"length={slab.get('length_in')}in thickness={slab.get('thickness_in')}in "
        f"avg_width={slab.get('width_avg_in')}in width_min={slab.get('width_min_in')}in "
        f"width_max={slab.get('width_max_in')}in bdft={slab.get('bdft')}"
    )


def assemble_call2_text(conn, slab: dict, prompt_text: str | None = None) -> str:
    """Assemble the Call 2 text payload: seed + declared attrs + dims + brand/geo.

    Shared by View-prompt GET and POST .../inference/call2 so both show the same text.
    """
    seed = prompt_text
    if seed is None:
        act = _active_prompt(conn, "call2")
        seed = act[1] if act else CALL2_SEED
    brand = conn.execute("SELECT value FROM settings WHERE key='brand_voice'").fetchone()
    geo = conn.execute("SELECT value FROM settings WHERE key='geo_context'").fetchone()
    brand_s = brand[0] if brand and brand[0] else ""
    geo_s = geo[0] if geo and geo[0] else ""
    return (
        f"{seed}\n\n"
        f"{build_declared_attributes(conn, slab)}\n\n"
        f"{build_dimension_fact_block(slab)}\n\n"
        f"Brand voice: {brand_s}\n"
        f"Geo: {geo_s}"
    )


def assemble_call2_photo_slots(conn, slab_id: str, *, for_wire: bool = False) -> tuple[list[dict], list[dict]]:
    """Up to 8 processed PNGs for Call 2.

    Returns (image_parts for the model, photo_slots without base64 for View prompt / log).
    Slot list: seq, id, bytes, long_edge_px of the *fitted* infer payload.
    """
    photos = conn.execute(
        "SELECT id, seq, processed_blob FROM slab_photos "
        "WHERE slab_id=? AND processed_blob IS NOT NULL ORDER BY seq LIMIT 8",
        (slab_id,),
    ).fetchall()
    image_parts: list[dict] = []
    photo_slots: list[dict] = []
    for p in photos:
        blob = p["processed_blob"] if hasattr(p, "keys") else p[2]
        pid = p["id"] if hasattr(p, "keys") else p[0]
        seq = p["seq"] if hasattr(p, "keys") else p[1]
        fitted = fit_long_edge_png(blob)
        if for_wire:
            image_parts.append(_image_part_from_processed(blob))
        meta = _photo_meta_row(pid, seq, fitted)
        photo_slots.append({
            "seq": meta["seq"],
            "id": meta["photo_id"],
            "bytes": meta["bytes"],
            "long_edge_px": meta["infer_long_edge_px"],
        })
    return image_parts, photo_slots


def assemble_call2(conn, slab_id: str) -> dict[str, Any] | None:
    """Full Call 2 assembly used by View prompt and the live POST.

    Returns None if the slab is missing. Does not call the model.
    """
    row = conn.execute(
        "SELECT * FROM slabs WHERE id=? AND deleted_at IS NULL", (slab_id,)
    ).fetchone()
    if row is None:
        return None
    slab = dict(row)
    for f in ("wood_category_ids", "figure_term_ids", "fig_tag_ids", "feat_tag_ids"):
        if slab.get(f) is not None and isinstance(slab[f], str):
            try:
                slab[f] = json.loads(slab[f])
            except (ValueError, TypeError):
                slab[f] = []
    act = _active_prompt(conn, "call2")
    prompt_text = act[1] if act else CALL2_SEED
    rev_id = act[0] if act else None
    text = assemble_call2_text(conn, slab, prompt_text)
    image_parts, photo_slots = assemble_call2_photo_slots(conn, slab_id, for_wire=True)
    # photo_meta for the inference log (same order as image_parts)
    photo_meta = [
        {
            "photo_id": s["id"],
            "seq": s["seq"],
            "bytes": s["bytes"],
            "infer_long_edge_px": s["long_edge_px"],
            "label": f"photo seq={s['seq']} id={s['id']}",
        }
        for s in photo_slots
    ]
    return {
        "text": text,
        "photos": photo_slots,
        "image_parts": image_parts,
        "photo_meta": photo_meta,
        "prompt_revision_id": rev_id,
        "slab": slab,
    }


@router.get("/slabs/{slab_id}/inference/call2/prompt")
async def call2_prompt(slab_id: str) -> dict:
    """View prompt: exact Call 2 text + photo slot list (no base64, no model call)."""
    conn = get_connection()
    assembled = assemble_call2(conn, slab_id)
    if assembled is None:
        return JSONResponse(status_code=404, content=_error("not_found", "Slab not found"))
    return {
        "text": assembled["text"],
        "photos": assembled["photos"],
        "prompt_revision_id": assembled["prompt_revision_id"],
    }


def _html_to_plain(value: str) -> str:
    """Defensive: strip HTML/markup so Review fields never show raw tags.

    Converts common block tags to newlines and <li> to "- " bullets, then removes
    remaining tags and unescapes entities. Already-plain text is returned cleaned.
    """
    if not value:
        return ""
    t = value
    # Normalize common break / list / paragraph boundaries before dropping tags.
    t = re.sub(r"(?is)<br\s*/?>", "\n", t)
    t = re.sub(r"(?is)</\s*(p|div|h[1-6]|tr|table)\s*>", "\n", t)
    t = re.sub(r"(?is)<\s*li[^>]*>", "- ", t)
    t = re.sub(r"(?is)</\s*(ul|ol)\s*>", "\n", t)
    t = re.sub(r"(?is)<[^>]+>", "", t)
    t = html.unescape(t)
    # Drop leftover markdown-ish bold/headers that look like markup in Review.
    t = re.sub(r"(?m)^#{1,6}\s+", "", t)
    t = re.sub(r"\*\*([^*]+)\*\*", r"\1", t)
    t = re.sub(r"[ \t]+\n", "\n", t)
    t = re.sub(r"\n{3,}", "\n\n", t)
    return t.strip()


def parse_call2_response(raw: str) -> tuple[str | None, str]:
    """Split Call 2 model output into (short_description, description).

    Preferred shape is a JSON object with short_description + description.
    Both fields are forced to plain text (HTML stripped) before return.
    If JSON is missing or incomplete, treat the whole body as description
    (stripped) and leave short_description None (caller keeps any prior short).
    """
    text = (raw or "").strip()
    if not text:
        return None, ""
    # Strip common markdown fences around JSON.
    body = text
    if body.startswith("```"):
        lines = body.split("\n")
        if lines and lines[0].startswith("```"):
            lines = lines[1:]
        if lines and lines[-1].strip() == "```":
            lines = lines[:-1]
        body = "\n".join(lines).strip()
    try:
        obj = json.loads(body)
    except (ValueError, TypeError):
        # Try to find a JSON object embedded in prose.
        start = body.find("{")
        end = body.rfind("}")
        if start >= 0 and end > start:
            try:
                obj = json.loads(body[start : end + 1])
            except (ValueError, TypeError):
                return None, _html_to_plain(text)
        else:
            return None, _html_to_plain(text)
    if not isinstance(obj, dict):
        return None, _html_to_plain(text)
    short = obj.get("short_description")
    desc = obj.get("description")
    if isinstance(short, str):
        short = _html_to_plain(short) or None
    else:
        short = None
    if isinstance(desc, str) and desc.strip():
        return short, _html_to_plain(desc)
    # JSON without a usable description — fall back to raw (stripped).
    return short, _html_to_plain(text)


@router.post("/slabs/{slab_id}/inference/call2")
async def call2(slab_id: str) -> dict:
    conn = get_connection()
    if not _enabled(conn, "content_llm_enabled"):
        return JSONResponse(status_code=409, content=_error("content_disabled", "Content generation is off"))
    base, key, model = _inference_creds(conn)
    if not base or not key:
        return JSONResponse(status_code=409, content=_error("inference_unconfigured", "Inference is not configured"))
    assembled = assemble_call2(conn, slab_id)
    if assembled is None:
        return JSONResponse(status_code=404, content=_error("not_found", "Slab not found"))
    client = InferenceClient(base, key, model)
    text = assembled["text"]
    image_parts = assembled["image_parts"]
    photo_meta = assembled["photo_meta"]
    messages = [{"role": "user", "content": image_parts + [{"type": "text", "text": text}]}]
    t0 = time.time()
    try:
        raw = client.chat(messages)
    except InferenceConnectionError as e:
        return JSONResponse(status_code=503, content=_error("inference_unreachable", str(e), {"recoverable": True}))
    except InferenceError as e:
        return JSONResponse(status_code=502, content=_error("inference_failed", str(e), {"recoverable": True}))
    latency = int((time.time() - t0) * 1000)
    _log_call(
        conn, slab_id, 2, assembled["prompt_revision_id"], model, messages, raw, latency, photo_meta,
    )
    short, description = parse_call2_response(raw)
    if short:
        conn.execute(
            "UPDATE slabs SET short_description=?, description=?, content_source='llm', updated_at=? WHERE id=?",
            (short, description, _now(), slab_id),
        )
    else:
        conn.execute(
            "UPDATE slabs SET description=?, content_source='llm', updated_at=? WHERE id=?",
            (description, _now(), slab_id),
        )
    conn.commit()
    out: dict[str, Any] = {"ok": True, "description": description}
    if short:
        out["short_description"] = short
    return out


def server_description(slab: dict) -> str:
    """Inference-off long description (METADATA-70)."""
    return (
        f"A {slab.get('title', 'slab')} from the mill, measured at "
        f"{slab.get('length_in')}in long, {slab.get('width_avg_in')}in average width, and "
        f"{slab.get('thickness_in')}in thick ({slab.get('bdft')} bdft). "
        "Ready to ship."
    )


def _photo_meta_row(photo_id: str, seq: int, blob: bytes) -> dict[str, Any]:
    """Metadata for one photo slot in the inference log (#87)."""
    edge = None
    try:
        im = Image.open(io.BytesIO(blob))
        im.load()
        edge = max(im.size)
    except Exception:
        edge = None
    return {
        "photo_id": photo_id,
        "seq": seq,
        "bytes": len(blob) if blob is not None else 0,
        "infer_long_edge_px": edge,
        "label": f"photo seq={seq} id={photo_id}",
    }


def _redact_request_for_log(messages: list, photo_meta: list[dict] | None = None) -> list:
    """Replace image data-URLs with photo refs for inference_log (#87 / METADATA-66).

    Wire messages still carry base64 for the model; only the logged copy is redacted.
    Text parts are kept in full. Image parts become type=image_ref using photo_meta by
    image-part index (same order as built from slab_photos).
    """
    meta = list(photo_meta or [])
    out: list = []
    img_i = 0
    for msg in messages:
        if not isinstance(msg, dict):
            out.append(msg)
            continue
        content = msg.get("content")
        if not isinstance(content, list):
            out.append(msg)
            continue
        new_parts: list = []
        for part in content:
            if not isinstance(part, dict):
                new_parts.append(part)
                continue
            ptype = part.get("type")
            if ptype == "text":
                new_parts.append({"type": "text", "text": part.get("text", "")})
                continue
            if ptype == "image_url":
                url = ""
                iu = part.get("image_url")
                if isinstance(iu, dict):
                    url = str(iu.get("url") or "")
                if url.startswith("data:image") or "base64," in url:
                    ref = {
                        "type": "image_ref",
                        "photo_id": None,
                        "seq": None,
                        "bytes": None,
                        "infer_long_edge_px": None,
                        "label": f"photo slot={img_i}",
                    }
                    if img_i < len(meta):
                        m = meta[img_i]
                        ref.update({
                            "photo_id": m.get("photo_id"),
                            "seq": m.get("seq"),
                            "bytes": m.get("bytes"),
                            "infer_long_edge_px": m.get("infer_long_edge_px"),
                            "label": m.get("label") or ref["label"],
                        })
                    new_parts.append(ref)
                    img_i += 1
                else:
                    # Non-data URL (rare): keep URL string only, never invent bytes
                    new_parts.append({
                        "type": "image_ref",
                        "url": url[:200],
                        "label": f"external image slot={img_i}",
                    })
                    img_i += 1
                continue
            new_parts.append(part)
        out.append({**msg, "content": new_parts})
    return out


_LOG_PAYLOAD_CAP = 200_000


def _log_call(conn, slab_id: str, call: int, rev_id: int | None, model: str,
              request: list, response: str, latency_ms: int,
              photo_meta: list[dict] | None = None) -> None:
    """Append inference_log with redacted request (no image bytes) — #87 / METADATA-66."""
    redacted = _redact_request_for_log(request, photo_meta)
    payload = json.dumps(redacted, ensure_ascii=False)
    if len(payload) > _LOG_PAYLOAD_CAP:
        # Safety net only; after redaction normal calls are far under the cap.
        payload = payload[: _LOG_PAYLOAD_CAP - 20] + "\n/* truncated */"
    conn.execute(
        "INSERT INTO inference_log(slab_id, call, prompt_revision_id, model, request_payload, response_payload, latency_ms, created_at) "
        "VALUES (?,?,?,?,?,?,?,?)",
        (slab_id, call, rev_id, model, payload, response, latency_ms, _now()),
    )
    conn.commit()


@router.post("/admin/test-inference")
async def test_inference() -> dict:
    conn = get_connection()
    base, key, model = _inference_creds(conn)
    if not base or not key:
        return {"ok": False, "mode": "fail", "vision_capable": False, "detail": "Inference is not configured"}
    client = InferenceClient(base, key, model, timeout=20.0)
    t0 = time.time()
    try:
        client.chat([{"role": "user", "content": "Reply with the single word: ok"}])
    except InferenceConnectionError as e:
        return {"ok": False, "mode": "fail", "vision_capable": False, "detail": str(e)}
    except InferenceError as e:
        return {"ok": False, "mode": "fail", "vision_capable": False, "detail": str(e)}
    detail = f"reachable in {int((time.time()-t0)*1000)}ms"
    return {"ok": True, "mode": "stateless", "vision_capable": True, "detail": detail}