"""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 @router.post("/slabs/{slab_id}/inference/call2") async def call2(slab_id: str) -> dict: conn = get_connection() row = conn.execute("SELECT * FROM slabs WHERE id=? AND deleted_at IS NULL", (slab_id,)).fetchone() if row is None: return JSONResponse(status_code=404, content=_error("not_found", "Slab not found")) 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")) client = InferenceClient(base, key, model) act = _active_prompt(conn, "call2") prompt_text = act[1] if act else CALL2_SEED 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 = [] photo_meta: 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] image_parts.append(_image_part_from_processed(blob)) photo_meta.append(_photo_meta_row(pid, seq, blob)) brand = conn.execute("SELECT value FROM settings WHERE key='brand_voice'").fetchone() geo = conn.execute("SELECT value FROM settings WHERE key='geo_context'").fetchone() text = (f"{prompt_text}\n\nDIMENSION FACT BLOCK (exact, invent no numbers):\n" f"length={row['length_in']}in thickness={row['thickness_in']}in " f"avg_width={row['width_avg_in']}in width_min={row['width_min_in']}in width_max={row['width_max_in']}in " f"bdft={row['bdft']}\n\nBrand voice: {brand[0] if brand else ''}\nGeo: {geo[0] if geo else ''}") 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, act[0] if act else None, 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}