feat(review): View prompt + Call 2 declared attrs, Create-desc gate

Shared assemble_call2 for GET .../call2/prompt and POST call2 (seed +
DECLARED ATTRIBUTES + dimension fact + brand/geo; photo slots without
base64). Create desc disabled until content LLM + title/short + species +
wood cats/primary + edge + ≥1 figure + grade + moisture + price; show
missing. figure_term_ids required for ready. Moisture defaults to Kiln
dried (kd). Pin snl / liveedge in wood-cat UI. Ready-on-change via
MultiSelect onCommit.
This commit is contained in:
grokbot_ty
2026-09-26 10:11:50 -07:00
parent a801a757f8
commit 1ff082f77c
13 changed files with 790 additions and 44 deletions
+236 -27
View File
@@ -436,39 +436,244 @@ def _gate(parsed: dict) -> dict:
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
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 = []
photo_meta: list[dict] = []
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]
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 ''}")
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:
@@ -478,10 +683,14 @@ async def call2(slab_id: str) -> dict:
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)
_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.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()}
+1
View File
@@ -35,6 +35,7 @@ _REQUIRED = [
"sku", "length_in", "thickness_in", "sqft", "bdft",
"width_min_in", "width_max_in", "width_avg_in",
"species_id", "wood_category_ids", "edge_type_term_id",
"figure_term_ids", # UX requires ≥1 figure (Create-desc gate + ready)
"grade_term_id", "thickness_term_id", "moisture_term_id",
"price", "price_source", "title", "short_description", "description",
"content_source",
+113
View File
@@ -342,3 +342,116 @@ def test_call1_sends_long_edge_dormant_while_disabled(env):
r = client.post("/api/v1/slabs/missing/inference/call1")
assert r.status_code == 410
assert r.json()["error"]["code"] == "call1_disabled"
def test_assemble_call2_includes_declared_attrs_and_dims(env):
"""Call 2 text = seed + DECLARED ATTRIBUTES + dimension fact + brand/geo; shared assembler."""
client, conn = env
conn.execute(
"INSERT INTO woo_categories(id,name,slug,parent_id,woo_id) VALUES "
"(10,'Species','species',NULL,10),"
"(11,'Black Walnut','black-walnut',10,11),"
"(20,'Wcat','wcat',NULL,20),"
"(21,'Slabs & Lumber','snl',20,21),"
"(22,'Live edge wood','liveedge',21,22)"
)
conn.execute(
"INSERT INTO woo_attributes(id,name,slug,woo_id) VALUES "
"(6,'Edge type','pa_etype',6),(8,'Figure','pa_fig',8),"
"(9,'Grade','pa_gr',9),(7,'Moisture','pa_mst',7)"
)
conn.execute(
"INSERT INTO woo_attribute_terms(id,attribute_id,name,slug,woo_id) VALUES "
"(321,6,'Live Edge 1 side','le1',321),"
"(328,8,'Curl','curl',328),"
"(342,9,'Fancy','fnc',342),"
"(326,7,'Kiln dried','kd',326)"
)
conn.execute(
"INSERT INTO woo_tags(id,name,slug,woo_id) VALUES (1,'feat-knot','feat-knot',1)"
)
conn.execute(
"INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES "
"('brand_voice','Warm and precise.',NULL),"
"('geo_context','Pacific Northwest mill.',NULL)"
)
conn.execute(
"INSERT INTO slabs(id,sku,length_in,thickness_in,width_avg_in,width_min_in,width_max_in,bdft,"
"species_id,wood_category_ids,edge_type_term_id,figure_term_ids,grade_term_id,moisture_term_id,"
"feat_tag_ids,title,short_description,status,server_rev,client_rev,created_at,updated_at) "
"VALUES ('s1','SLAB-01',114.0,1.75,24.3,18.0,24.3,18.4,11,?,321,?,342,326,?,"
"'Figured Black Walnut Live Edge Slab','short line','ready',1,1,'t','t')",
(json.dumps([22]), json.dumps([328]), json.dumps([1])),
)
png = _rgba_png(200, 200)
conn.execute(
"INSERT INTO slab_photos(id,slab_id,seq,processed_blob,created_at) VALUES ('p9','s1',1,?, 't')",
(png,),
)
conn.commit()
from backend.inference import assemble_call2, assemble_call2_text, CALL2_SEED
assembled = assemble_call2(conn, "s1")
assert assembled is not None
text = assembled["text"]
assert "DECLARED ATTRIBUTES" in text
assert "species: Black Walnut" in text
assert "primary wood category: Live edge wood" in text
assert "edge: Live Edge 1 side" in text
assert "figure: Curl" in text
assert "grade: Fancy" in text
assert "moisture: Kiln dried" in text
assert "features: knot" in text
assert "DIMENSION FACT BLOCK" in text
assert "length=114.0in" in text
assert "Brand voice: Warm and precise." in text
assert "Geo: Pacific Northwest mill." in text
assert CALL2_SEED.split("\n")[0] in text
# No raw base64 in View-prompt shape
assert "photos" in assembled
assert assembled["photos"][0]["id"] == "p9"
assert assembled["photos"][0]["seq"] == 1
assert assembled["photos"][0]["bytes"] > 0
assert "base64" not in json.dumps(assembled["photos"])
# View prompt GET mirrors assembler (no model)
r = client.get("/api/v1/slabs/s1/inference/call2/prompt")
assert r.status_code == 200, r.text
body = r.json()
assert body["text"] == text
assert body["photos"][0]["id"] == "p9"
assert "data:image" not in json.dumps(body)
def test_call2_post_uses_same_assembler_text(env, monkeypatch):
"""POST call2 sends the same assembled text View prompt returns."""
client, conn = env
conn.execute("INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES ('content_llm_enabled','true',NULL)")
conn.execute(
"INSERT OR REPLACE INTO settings(key,value,value_enc) VALUES "
"('brand_voice','voice',NULL),('geo_context','geo',NULL)"
)
conn.execute(
"INSERT INTO slabs(id,sku,length_in,thickness_in,width_avg_in,width_min_in,width_max_in,bdft,"
"title,status,server_rev,client_rev,created_at,updated_at) "
"VALUES ('s1','SLAB-01',100,2,20,18,22,18,'Walnut Slab','ready',1,1,'t','t')"
)
conn.execute(
"INSERT INTO slab_photos(id,slab_id,seq,processed_blob,created_at) VALUES ('p1','s1',1,?, 't')",
(_rgba_png(64, 64),),
)
conn.commit()
from backend.inference import InferenceClient, assemble_call2
expected = assemble_call2(conn, "s1")["text"]
captured = {}
def fake_chat(self, messages):
captured["text"] = messages[0]["content"][-1]["text"]
return "<p>ok</p>"
monkeypatch.setattr(InferenceClient, "chat", fake_chat)
r = client.post("/api/v1/slabs/s1/inference/call2")
assert r.status_code == 200, r.text
assert captured["text"] == expected
assert "DECLARED ATTRIBUTES" in captured["text"]
assert "DIMENSION FACT BLOCK" in captured["text"]
+26
View File
@@ -194,3 +194,29 @@ def test_review_persists_owner_overrides(env):
assert got["sku"] == "SLAB-99"
# server_rev was incremented (the UPDATE fired).
assert got["server_rev"] == 2
def test_ready_requires_figure_term_ids(env):
"""UX requires ≥1 figure — figure_term_ids is in _REQUIRED."""
from backend.review import missing_required, compute_ready
slab = _slab(figure_term_ids=[])
miss = missing_required(slab)
assert "figure_term_ids" in miss
ready, missing = compute_ready({**slab, "price": 1.0, "price_source": "manual",
"title": "T", "short_description": "S",
"edge_type_term_id": 1, "grade_term_id": 1,
"thickness_term_id": 1, "moisture_term_id": 1})
assert ready is False
assert "figure_term_ids" in missing
ready2, missing2 = compute_ready({
**slab,
"figure_term_ids": [1],
"price": 1.0, "price_source": "manual",
"title": "T", "short_description": "S",
"edge_type_term_id": 1, "grade_term_id": 1,
"thickness_term_id": 1, "moisture_term_id": 1,
})
# May still miss other fields depending on fixture; figure must not be listed
assert "figure_term_ids" not in missing2