"""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 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": , "species_confidence": <0-1>, ' '"wood_category_ids": [,...], "wood_category_confidence": <0-1>, ' '"edge_type_term_id": , "edge_confidence": <0-1>, ' '"figure_term_ids": [,...], "figure_confidence": <0-1>, ' '"feature_tag_ids": [,...], "feature_tag_confidence": <0-1>}' ) CALL2_SEED = ( "Write the long product description for the slab from the supplied photos and listing " "facts. The name and short description are already built, so do not repeat them. Use the " "brand voice and geo context given. The dimensions in the fact block are exact: use those " "strings and invent no numbers. Write prose a premium lumber buyer would read." ) _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 '_confidence' keys; as a safety net it also understands the nested 'confidence: {: 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() short = (slab.get("short_description") or "").strip() if title: lines.append(f"title: {title}") if short: lines.append(f"short_description: {short}") 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"], } @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, ) # Store as description, mark content_source=llm. conn.execute( "UPDATE slabs SET description=?, content_source='llm', updated_at=? WHERE id=?", (raw.strip(), _now(), slab_id), ) conn.commit() return {"ok": True, "description": raw.strip()} 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}