From 36aba73f662af1cc39da66657c8b684eb2dc1e97 Mon Sep 17 00:00:00 2001 From: Kayos Date: Wed, 29 Apr 2026 06:26:54 -0700 Subject: [PATCH] v0.3 step 3+4: AI plan generator + /list shopping aggregation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - migrations 012 + 013: cauldron_meal_plan_slots + cauldron_pick_points - db: list_plan_slots, save_plan_slots, delete_plan_slots, mark_plan_generated, clear_plan_generated, award_pick_points, enrich_plan_with_slots; scoreboard extended with points (sum from pick_points) and weeks_locked alias - forge.generate_plan: sonnet prompt builds 7-day plan respecting picks, validates slot count + day uniqueness + slug-in-pool, fills picker_subs from ground-truth picks (model output is advisory) - POST /api/plan/generate: race-safe (existing slots → 409 with plan), lock-aware (locked → 409), idempotent - POST /api/plan/regenerate: re-roll for the original generator, gated by ownership + lock; wipes slots + pick_points then re-runs generate - plan.html: generate CTA + 7 day cards with picker chips + AI reason + re-roll button (generator-only, pre-lock); scoreboard now shows points + wins - /list: pulls plan slots, queries Mealie for ingredients, runs aggregator, renders 48px-tall checkbox shopping list with localStorage state per plan_id - tests: 13 new tests across forge.generate_plan + /api/plan/generate routes + /list view + scoreboard SQL inspection. conftest+_testenv stub pymysql/oidc/foods at import time so tests run against module-level app without a live DB. Both pytest and `unittest discover` paths green (27/27). Defers: bulk sterilizer admin (A1), foods dedupe (A2), Mealie shopping-list- export (button rendered but disabled). 7-slot count is fixed at the endpoint (no UI for slot-count selection yet). Spec: memory/spec-cauldron-v0.3.md --- cauldron/db.py | 235 +++++++++++++++++++-- cauldron/forge.py | 178 ++++++++++++++++ cauldron/server.py | 296 +++++++++++++++++++++++++-- cauldron/templates/list.html | 229 +++++++++++++++++++++ cauldron/templates/plan.html | 173 +++++++++++++++- tests/_testenv.py | 85 ++++++++ tests/conftest.py | 5 + tests/test_list_view.py | 175 ++++++++++++++++ tests/test_plan_generator.py | 381 +++++++++++++++++++++++++++++++++++ 9 files changed, 1724 insertions(+), 33 deletions(-) create mode 100644 cauldron/templates/list.html create mode 100644 tests/_testenv.py create mode 100644 tests/conftest.py create mode 100644 tests/test_list_view.py create mode 100644 tests/test_plan_generator.py diff --git a/cauldron/db.py b/cauldron/db.py index 9242461..96f8dc5 100644 --- a/cauldron/db.py +++ b/cauldron/db.py @@ -186,6 +186,48 @@ MIGRATIONS = [ FOREIGN KEY (cauldron_food_id) REFERENCES cauldron_foods(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, + # 012 — AI-generated meal plan slots. One row per (plan, day). Created + # when a household member triggers /api/plan/generate. picker_subs JSON + # holds the authentik_subs of household members who pinned this slot's + # recipe (empty list if AI-chosen). reason is the AI's user-facing + # rationale. notes is reserved for future swap/edit history. + """ + CREATE TABLE IF NOT EXISTS cauldron_meal_plan_slots ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + plan_id BIGINT NOT NULL, + day VARCHAR(10) NOT NULL, + recipe_slug VARCHAR(255) NOT NULL, + recipe_name VARCHAR(500) NOT NULL, + source ENUM('mealie','pick') NOT NULL DEFAULT 'mealie', + picker_subs JSON, + reason VARCHAR(500), + notes JSON, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE KEY uk_plan_day (plan_id, day), + INDEX idx_plan (plan_id), + FOREIGN KEY (plan_id) REFERENCES cauldron_meal_plans(id) ON DELETE CASCADE + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + """, + # 013 — pick-points ledger. 1pt awarded when a member's pick lands in + # a generated plan ('pick_used'). Reserved reasons for v0.4: first-to- + # lock + streak bonuses. Joins to households + plans + users so a row + # disappears cleanly if any of them are removed. + """ + CREATE TABLE IF NOT EXISTS cauldron_pick_points ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + household_id BIGINT NOT NULL, + plan_id BIGINT NOT NULL, + authentik_sub VARCHAR(190) NOT NULL, + points INT NOT NULL, + reason ENUM('pick_used','first_to_lock','streak_bonus') NOT NULL, + awarded_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + INDEX idx_household_user (household_id, authentik_sub), + INDEX idx_plan (plan_id), + FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE, + FOREIGN KEY (plan_id) REFERENCES cauldron_meal_plans(id) ON DELETE CASCADE, + FOREIGN KEY (authentik_sub) REFERENCES cauldron_users(authentik_sub) ON DELETE CASCADE + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + """, ] @@ -393,31 +435,59 @@ class DB: return cur.rowcount def household_scoreboard(self, household_id: int) -> list[dict]: - """Per-user lock counts + most recent lock time. Joins to users for - display name. Excludes auto-locks (those are no-one's win).""" + """Per-user lock counts + pick-points + most recent lock time. + + Three numbers per row: + wins — user-locked weeks (excludes auto-locks) + weeks_locked — alias for wins, preserved for older callers + points — sum of cauldron_pick_points for this user/household + + Sort: points desc, then wins desc, then last_win desc — points are + the headline metric in v0.3 (every pick lands → matters daily). + + We compute lock counts and points as separate scalar subqueries so + the JOIN doesn't blow up on the cartesian (members × plans × points). + """ with self.conn() as c, c.cursor() as cur: cur.execute( """ SELECT - u.authentik_sub AS sub, - u.email AS email, - u.display_name AS display_name, - COUNT(p.id) AS wins, - MAX(p.locked_at) AS last_win + u.authentik_sub AS sub, + u.email AS email, + u.display_name AS display_name, + COALESCE(( + SELECT COUNT(*) FROM cauldron_meal_plans p + WHERE p.locked_by_sub = m.authentik_sub + AND p.household_id = m.household_id + AND p.locked_reason = 'user' + ), 0) AS wins, + ( + SELECT MAX(p.locked_at) FROM cauldron_meal_plans p + WHERE p.locked_by_sub = m.authentik_sub + AND p.household_id = m.household_id + AND p.locked_reason = 'user' + ) AS last_win, + COALESCE(( + SELECT SUM(pp.points) FROM cauldron_pick_points pp + WHERE pp.household_id = m.household_id + AND pp.authentik_sub = m.authentik_sub + ), 0) AS points FROM cauldron_household_members m LEFT JOIN cauldron_users u ON u.authentik_sub = m.authentik_sub - LEFT JOIN cauldron_meal_plans p - ON p.locked_by_sub = m.authentik_sub - AND p.household_id = m.household_id - AND p.locked_reason = 'user' WHERE m.household_id = %s - GROUP BY u.authentik_sub, u.email, u.display_name - ORDER BY wins DESC, last_win DESC + ORDER BY points DESC, wins DESC, last_win DESC """, (household_id,), ) - return [dict(r) for r in cur.fetchall()] + out = [] + for r in cur.fetchall(): + d = dict(r) + d["points"] = int(d.get("points") or 0) + d["wins"] = int(d.get("wins") or 0) + d["weeks_locked"] = d["wins"] + out.append(d) + return out def household_streak(self, household_id: int) -> dict | None: """Compute current win streak: walk back from most recent locked week, @@ -451,6 +521,143 @@ class DB: "count": count, } + # --- plan slots + pick points (v0.3 A4) -------------------------------- + + # Day order is stable Mon..Sun. Used everywhere we need to render slots + # in calendar order. + PLAN_DAYS = ("monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday") + + def list_plan_slots(self, plan_id: int) -> list[dict]: + """All slots for a plan, ordered Mon..Sun. picker_subs is decoded + from JSON to a list (or [] if null).""" + import json as _json + with self.conn() as c, c.cursor() as cur: + cur.execute( + """ + SELECT id, plan_id, day, recipe_slug, recipe_name, source, + picker_subs, reason, notes, created_at + FROM cauldron_meal_plan_slots + WHERE plan_id = %s + """, + (plan_id,), + ) + rows = [dict(r) for r in cur.fetchall()] + for r in rows: + ps = r.get("picker_subs") + if isinstance(ps, str): + try: + r["picker_subs"] = _json.loads(ps) + except Exception: + r["picker_subs"] = [] + elif ps is None: + r["picker_subs"] = [] + n = r.get("notes") + if isinstance(n, str): + try: + r["notes"] = _json.loads(n) + except Exception: + r["notes"] = None + order = {d: i for i, d in enumerate(self.PLAN_DAYS)} + rows.sort(key=lambda r: order.get((r.get("day") or "").lower(), 99)) + return rows + + def save_plan_slots(self, plan_id: int, slots: list[dict]) -> int: + """INSERT IGNORE every slot. Returns count actually inserted — + callers can use this to detect race contention (zero rows = someone + else already saved this plan).""" + import json as _json + if not slots: + return 0 + inserted = 0 + with self.conn() as c, c.cursor() as cur: + for s in slots: + cur.execute( + """ + INSERT IGNORE INTO cauldron_meal_plan_slots + (plan_id, day, recipe_slug, recipe_name, source, + picker_subs, reason, notes) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + plan_id, + (s.get("day") or "").lower()[:10], + s["recipe_slug"][:255], + (s.get("recipe_name") or s["recipe_slug"])[:500], + s.get("source") or ("pick" if s.get("picker_subs") else "mealie"), + _json.dumps(s.get("picker_subs") or []), + (s.get("reason") or "")[:500] or None, + _json.dumps(s["notes"]) if s.get("notes") is not None else None, + ), + ) + inserted += cur.rowcount + return inserted + + def delete_plan_slots(self, plan_id: int) -> int: + """Wipe slots for a plan (used by re-roll). Also nukes the matching + pick_points so the ledger doesn't double-count on regenerate.""" + with self.conn() as c, c.cursor() as cur: + cur.execute("DELETE FROM cauldron_meal_plan_slots WHERE plan_id=%s", (plan_id,)) + slots_removed = cur.rowcount + cur.execute("DELETE FROM cauldron_pick_points WHERE plan_id=%s", (plan_id,)) + return slots_removed + + def mark_plan_generated(self, plan_id: int, sub: str) -> dict: + """Set generated_by_sub + generated_at IF not already set. Returns + the post-update plan row. Idempotent for the same generator.""" + with self.conn() as c, c.cursor() as cur: + cur.execute( + """ + UPDATE cauldron_meal_plans + SET generated_by_sub = %s, + generated_at = NOW() + WHERE id = %s AND generated_at IS NULL + """, + (sub, plan_id), + ) + cur.execute("SELECT * FROM cauldron_meal_plans WHERE id=%s", (plan_id,)) + return dict(cur.fetchone()) + + def clear_plan_generated(self, plan_id: int) -> None: + """Re-roll path: clear the generated_by/at marker so the next + generate writes fresh metadata.""" + with self.conn() as c, c.cursor() as cur: + cur.execute( + """ + UPDATE cauldron_meal_plans + SET generated_by_sub = NULL, + generated_at = NULL + WHERE id = %s + """, + (plan_id,), + ) + + def award_pick_points( + self, + household_id: int, + plan_id: int, + sub: str, + points: int, + reason: str = "pick_used", + ) -> int: + """Insert one ledger row. Returns the new row id. Reason must be one + of the ENUM values; we don't validate here — DB will reject bad ones.""" + with self.conn() as c, c.cursor() as cur: + cur.execute( + """ + INSERT INTO cauldron_pick_points + (household_id, plan_id, authentik_sub, points, reason) + VALUES (%s, %s, %s, %s, %s) + """, + (household_id, plan_id, sub, int(points), reason), + ) + return cur.lastrowid + + def enrich_plan_with_slots(self, plan: dict) -> dict: + """In-place: add `slots` key to a plan dict. Returns the same dict + for chaining. Empty list if there are no slots yet.""" + plan["slots"] = self.list_plan_slots(plan["id"]) if plan.get("id") else [] + return plan + # --- meal picks --------------------------------------------------------- def add_meal_pick(self, sub: str, slug: str, name: str) -> bool: diff --git a/cauldron/forge.py b/cauldron/forge.py index 7e2c9e9..f6fed2e 100644 --- a/cauldron/forge.py +++ b/cauldron/forge.py @@ -1,4 +1,7 @@ """Thin HTTP client for clawdforge — we're a consumer.""" +import json +import re + import requests @@ -6,6 +9,9 @@ class ForgeError(RuntimeError): pass +_DAYS = ("monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday") + + class Forge: def __init__(self, *, base_url: str, token: str, default_model: str, default_timeout: int): self.base_url = base_url.rstrip("/") @@ -62,3 +68,175 @@ class Forge: raise ForgeError(f"upstream {r.status_code}: {r.text[:500]}") return r.json() + + def generate_plan( + self, + *, + picks: list[dict], + recipes: list[dict], + slots: int = 7, + week_start: str, + model: str | None = None, + ) -> list[dict]: + """Ask Sonnet for a {slots}-day plan. Returns a list of slot dicts + shaped like: + {"day": "monday", "recipe_slug": "...", "recipe_name": "...", + "picker_subs": [...], "reason": "...", "source": "pick"|"mealie"} + + Validates structure aggressively — wrong shape / wrong slot count / + slug-not-in-pool → ForgeError. Caller surfaces a 502 to the user. + + recipes: [{slug, name, tags?}], picks: [{slug, name, picker_subs}]. + Picks are the family's pinned recipes; the prompt mandates each one + appears exactly once when the pool is large enough. + """ + if slots < 1 or slots > 14: + raise ForgeError(f"bad slot count: {slots}") + if not recipes: + raise ForgeError("recipe pool empty — cannot generate") + + # Build a slug → name map for validation. Use the recipe pool plus + # picks (picks should already be in the pool, but be defensive). + valid_by_slug: dict[str, str] = {} + for r in recipes: + slug = r.get("slug") + if slug: + valid_by_slug[slug] = r.get("name") or slug + for p in picks: + slug = p.get("slug") + if slug: + valid_by_slug.setdefault(slug, p.get("name") or slug) + + prompt = self._build_plan_prompt( + picks=picks, recipes=recipes, slots=slots, week_start=week_start, + ) + result = self.run(prompt, model=model or "sonnet") + parsed = _extract_plan_slots(result) + if not isinstance(parsed, list): + raise ForgeError("model output: 'slots' must be a list") + if len(parsed) != slots: + raise ForgeError(f"model output: got {len(parsed)} slots, expected {slots}") + + # Pick attribution lookup keyed by slug → list[sub] + pick_subs_by_slug: dict[str, list[str]] = {} + for p in picks: + slug = p.get("slug") + if slug: + pick_subs_by_slug[slug] = list(p.get("picker_subs") or []) + + out = [] + seen_days: set[str] = set() + for raw in parsed: + if not isinstance(raw, dict): + raise ForgeError("model output: each slot must be an object") + day = (raw.get("day") or "").strip().lower() + slug = (raw.get("recipe_slug") or "").strip() + if day not in _DAYS: + raise ForgeError(f"model output: bad day '{day}'") + if day in seen_days: + raise ForgeError(f"model output: duplicate day '{day}'") + seen_days.add(day) + if not slug or slug not in valid_by_slug: + raise ForgeError(f"model output: unknown recipe_slug '{slug}'") + + # Trust the model's picker_subs only if they intersect the real + # set. We have ground truth in pick_subs_by_slug — prefer it. + real_pickers = pick_subs_by_slug.get(slug, []) + model_pickers = raw.get("picker_subs") or [] + if not isinstance(model_pickers, list): + model_pickers = [] + picker_subs = real_pickers if real_pickers else [ + s for s in model_pickers if isinstance(s, str) + ] + source = "pick" if real_pickers else "mealie" + + out.append({ + "day": day, + "recipe_slug": slug, + "recipe_name": valid_by_slug[slug], + "picker_subs": picker_subs, + "reason": (raw.get("reason") or "")[:500], + "source": source, + }) + return out + + @staticmethod + def _build_plan_prompt(*, picks, recipes, slots, week_start) -> str: + pool_lines = [] + for r in recipes: + slug = r.get("slug") or "" + name = r.get("name") or slug + tags = r.get("tags") or [] + tag_str = "" + if tags: + # First 3 tags only — keeps prompt token count under control + cleaned = [] + for t in tags[:3]: + if isinstance(t, dict): + cleaned.append(t.get("name") or "") + elif isinstance(t, str): + cleaned.append(t) + cleaned = [c for c in cleaned if c] + if cleaned: + tag_str = f" [{', '.join(cleaned)}]" + pool_lines.append(f"- {slug}: {name}{tag_str}") + + pick_lines = [] + for p in picks: + slug = p.get("slug") or "" + name = p.get("name") or slug + pickers = p.get("pickers") or [] + picker_subs = p.get("picker_subs") or [] + who = ", ".join(pickers) if pickers else "household" + subs_repr = json.dumps(picker_subs) + pick_lines.append(f"- {slug}: {name} (picked by [{who}], picker_subs={subs_repr})") + + picks_block = "\n".join(pick_lines) if pick_lines else "(none)" + pool_block = "\n".join(pool_lines) + + return ( + f"You are a family meal planner. Build a {slots}-day dinner plan " + f"for the week of {week_start}.\n\n" + f"POOL (all available recipes):\n{pool_block}\n\n" + f"PICKS (recipes the family pre-selected — every pick MUST appear " + f"if pool size >= {slots}; no repeats unless pool < {slots}):\n" + f"{picks_block}\n\n" + "Output JSON ONLY, no prose: " + '{"slots": [{"day": "monday", "recipe_slug": "...", ' + '"picker_subs": [...] or [], "reason": "..."}, ...]}\n\n' + "Rules:\n" + f"- Use exactly {slots} recipes\n" + "- Distribute picks evenly across the week — don't bunch them\n" + "- \"reason\" is a one-line user-facing rationale " + "(e.g., \"balances heavy and light meals\", \"honors abby's pick\")\n" + "- \"picker_subs\" is the array of authentik_sub strings of family " + "members who picked this recipe (empty list if AI-chosen)\n" + "- Day order: monday..sunday\n" + ) + + +def _extract_plan_slots(forge_result: dict): + """clawdforge wraps its return; the JSON we asked for can sit in a few + different shapes. Normalize aggressively.""" + if not isinstance(forge_result, dict): + raise ForgeError("forge result not a dict") + inner = forge_result.get("result", forge_result) + # `result` may be a string when claude returned non-JSON — try to scrape + if isinstance(inner, str): + inner = _parse_json_blob(inner) + if isinstance(inner, dict) and "slots" in inner: + return inner["slots"] + if isinstance(inner, list): + return inner + raise ForgeError(f"forge result missing 'slots' key: {str(inner)[:200]}") + + +def _parse_json_blob(s: str): + s = s.strip() + # Strip code fences if Sonnet wrapped its output + s = re.sub(r"^```(?:json)?\s*", "", s) + s = re.sub(r"\s*```$", "", s) + try: + return json.loads(s) + except Exception as e: + raise ForgeError(f"could not parse model JSON: {e}; head={s[:200]!r}") from e diff --git a/cauldron/server.py b/cauldron/server.py index 7b75536..39728c7 100644 --- a/cauldron/server.py +++ b/cauldron/server.py @@ -29,8 +29,8 @@ from flask import Flask, jsonify, redirect, render_template, request, session, u from .config import load from .crypto import TokenCrypto from .db import DB -from .forge import Forge -from . import foods +from .forge import Forge, ForgeError +from . import aggregator, foods from .mealie import Mealie, MealieError from .oidc import init_oauth from .recipe_index import flatten_recipe, refresh_household_index, search_index @@ -500,27 +500,29 @@ def create_app() -> Flask: db.auto_lock_past_unlocked_plans(hid, this_monday) plan = db.get_or_create_plan(hid, this_monday) + db.enrich_plan_with_slots(plan) scoreboard = db.household_scoreboard(hid) streak = db.household_streak(hid) pick_count = len(db.list_household_pick_slugs(hid)) - # Look up display name for locked_by + # Resolve display names for any subs we render — locked_by + every + # picker referenced by any slot. One round-trip; small set. + sub_display: dict[str, str] = _resolve_sub_displays(db, plan) locked_by_display = None if plan.get("locked_by_sub"): - with db.conn() as c, c.cursor() as cur: - cur.execute( - "SELECT display_name, email FROM cauldron_users WHERE authentik_sub=%s", - (plan["locked_by_sub"],), - ) - r = cur.fetchone() - if r: - locked_by_display = r["display_name"] or (r["email"] or "").split("@")[0] + locked_by_display = sub_display.get(plan["locked_by_sub"]) + + generated_by_display = None + if plan.get("generated_by_sub"): + generated_by_display = sub_display.get(plan["generated_by_sub"]) return render_template( "plan.html", week_start=plan["week_start"], plan=plan, locked_by_display=locked_by_display, + generated_by_display=generated_by_display, + sub_display=sub_display, scoreboard=scoreboard, streak=streak, current_user_sub=u["sub"], @@ -543,10 +545,211 @@ def create_app() -> Flask: updated = db.lock_plan(plan["id"], sub=u["sub"], reason="user") return jsonify({"ok": True, "locked_at": updated["locked_at"].isoformat() if updated["locked_at"] else None}) + @app.post("/api/plan/generate") + @require_session + def plan_generate(): + u = session["user"] + hid = current_household_id() + if not hid: + return jsonify({"error": "no household"}), 409 + + today = date.today() + this_monday = monday_of(today) + plan = db.get_or_create_plan(hid, this_monday) + + if plan.get("locked_at"): + return jsonify({ + "error": "plan_locked", + "locked_by_sub": plan.get("locked_by_sub"), + "locked_at": plan["locked_at"].isoformat() if plan.get("locked_at") else None, + }), 409 + + # Idempotency / race: if slots already exist, return them. Two + # concurrent generators end up with the first writer's slots; the + # second sees them and returns 409 with the existing plan. + existing = db.list_plan_slots(plan["id"]) + if existing: + db.enrich_plan_with_slots(plan) + return jsonify({ + "error": "plan_already_generated", + "plan": _plan_payload(plan), + }), 409 + + # Pull picks (with picker_subs) + recipe pool (slug+name+tags only) + picks = db.list_household_picks_with_pickers(hid) + rows = db.list_indexed_recipes(hid, limit=2000, offset=0) + recipes = [] + for r in rows: + tags = [] + raw = r.get("raw_json") + if isinstance(raw, str): + try: + raw = _json_loads(raw) + except Exception: + raw = None + if isinstance(raw, dict): + tags = raw.get("tags") or [] + recipes.append({"slug": r["slug"], "name": r["name"], "tags": tags}) + + if not recipes: + return jsonify({"error": "no_recipes_indexed"}), 409 + + try: + slots = forge.generate_plan( + picks=picks, recipes=recipes, + slots=7, week_start=this_monday.isoformat(), + ) + except ForgeError as e: + return jsonify({"error": "forge_failed", "detail": str(e)}), 502 + + inserted = db.save_plan_slots(plan["id"], slots) + if inserted == 0: + # Race lost — re-read and return the winner's plan + db.enrich_plan_with_slots(plan) + return jsonify({ + "error": "plan_already_generated", + "plan": _plan_payload(plan), + }), 409 + + # Award 1pt per pick that landed (one row per (sub, pick_used) per + # plan, deduped client-side: a pick with two pickers gives both 1pt). + for s in slots: + for sub in s.get("picker_subs") or []: + try: + db.award_pick_points(hid, plan["id"], sub, 1, "pick_used") + except Exception as exc: + app.logger.warning("award_pick_points failed for %s: %s", sub, exc) + + plan = db.mark_plan_generated(plan["id"], u["sub"]) + db.enrich_plan_with_slots(plan) + return jsonify({"ok": True, "plan": _plan_payload(plan)}) + + @app.post("/api/plan/regenerate") + @require_session + def plan_regenerate(): + """Re-roll: only the original generator can do this, only before + lock. Wipes slots + pick_points for this plan, then reuses the + generate path. Defensive — returns 409 on lock or wrong owner.""" + u = session["user"] + hid = current_household_id() + if not hid: + return jsonify({"error": "no household"}), 409 + + today = date.today() + this_monday = monday_of(today) + plan = db.get_or_create_plan(hid, this_monday) + + if plan.get("locked_at"): + return jsonify({"error": "plan_locked"}), 409 + if not plan.get("generated_at"): + # Nothing to re-roll — caller probably wanted plain /generate + return jsonify({"error": "plan_not_generated"}), 409 + if plan.get("generated_by_sub") != u["sub"]: + return jsonify({"error": "not_generator"}), 403 + + db.delete_plan_slots(plan["id"]) + db.clear_plan_generated(plan["id"]) + + # Now fall through to the same logic as generate + picks = db.list_household_picks_with_pickers(hid) + rows = db.list_indexed_recipes(hid, limit=2000, offset=0) + recipes = [] + for r in rows: + tags = [] + raw = r.get("raw_json") + if isinstance(raw, str): + try: + raw = _json_loads(raw) + except Exception: + raw = None + if isinstance(raw, dict): + tags = raw.get("tags") or [] + recipes.append({"slug": r["slug"], "name": r["name"], "tags": tags}) + + if not recipes: + return jsonify({"error": "no_recipes_indexed"}), 409 + + try: + slots = forge.generate_plan( + picks=picks, recipes=recipes, + slots=7, week_start=this_monday.isoformat(), + ) + except ForgeError as e: + return jsonify({"error": "forge_failed", "detail": str(e)}), 502 + + db.save_plan_slots(plan["id"], slots) + for s in slots: + for sub in s.get("picker_subs") or []: + try: + db.award_pick_points(hid, plan["id"], sub, 1, "pick_used") + except Exception as exc: + app.logger.warning("award_pick_points failed for %s: %s", sub, exc) + plan = db.mark_plan_generated(plan["id"], u["sub"]) + db.enrich_plan_with_slots(plan) + return jsonify({"ok": True, "plan": _plan_payload(plan)}) + @app.get("/list") @require_session def list_view(): - return render_template("stub.html", title="list", coming="aggregated shopping list from this week's plan", active="list") + u = session["user"] + hid = current_household_id() + if not hid: + return redirect(url_for("connect_mealie_get")) + + today = date.today() + this_monday = monday_of(today) + plan = db.get_or_create_plan(hid, this_monday) + db.enrich_plan_with_slots(plan) + + if not plan.get("slots"): + return render_template( + "list.html", + plan=plan, lines=[], active="list", + empty_reason="no_plan", + missing_recipes=[], + ) + + client = current_user_mealie() + if not client: + return redirect(url_for("connect_mealie_get")) + + raw_ings: list[aggregator.Ingredient] = [] + missing: list[str] = [] + for s in plan["slots"]: + try: + recipe = client.get_recipe(s["recipe_slug"]) + except Exception: + missing.append(s["recipe_slug"]) + continue + for ri in recipe.get("recipeIngredient", []) or []: + qty = ri.get("quantity") + unit_obj = ri.get("unit") or {} + unit = (unit_obj.get("name") if isinstance(unit_obj, dict) else "") or "" + food_obj = ri.get("food") or {} + food_name = (food_obj.get("name") if isinstance(food_obj, dict) else "") or "" + note = ri.get("note") or "" + if not food_name and not note: + continue + raw_ings.append(aggregator.Ingredient( + qty=float(qty) if qty not in (None, "") else None, + unit=unit, + food_name=food_name or note, + note=note if food_name else None, + source_recipe_slug=s["recipe_slug"], + original_text=ri.get("display") or _ing_render(qty, unit, food_name, note), + )) + + def foods_lookup(name: str): + hits = foods.search_food(db, name, limit=1) + return hits[0] if hits else None + + lines = aggregator.aggregate(raw_ings, foods_lookup) + return render_template( + "list.html", + plan=plan, lines=lines, active="list", + empty_reason=None, + missing_recipes=missing, + ) @app.get("/api/recipes/.json") @require_session @@ -714,5 +917,74 @@ def _const_eq(a: str, b: str) -> bool: return diff == 0 +def _json_loads(s): + import json as _j + return _j.loads(s) + + +def _resolve_sub_displays(db, plan: dict) -> dict[str, str]: + """Return {sub: display_name} for every sub referenced by this plan's + slots + locked_by + generated_by. One DB round-trip; small set.""" + subs: set[str] = set() + if plan.get("locked_by_sub"): + subs.add(plan["locked_by_sub"]) + if plan.get("generated_by_sub"): + subs.add(plan["generated_by_sub"]) + for s in plan.get("slots") or []: + for sub in s.get("picker_subs") or []: + if sub: + subs.add(sub) + if not subs: + return {} + placeholders = ", ".join(["%s"] * len(subs)) + out: dict[str, str] = {} + with db.conn() as c, c.cursor() as cur: + cur.execute( + f"SELECT authentik_sub, display_name, email FROM cauldron_users " + f"WHERE authentik_sub IN ({placeholders})", + tuple(subs), + ) + for r in cur.fetchall(): + sub = r["authentik_sub"] + disp = r.get("display_name") + if not disp: + disp = (r.get("email") or "").split("@")[0] + out[sub] = disp or sub + return out + + +def _plan_payload(plan: dict) -> dict: + """JSON-serializable view of a plan dict (datetimes → iso strings).""" + p = dict(plan) + for k in ("week_start", "generated_at", "locked_at"): + v = p.get(k) + if v is not None and hasattr(v, "isoformat"): + p[k] = v.isoformat() + slots = p.get("slots") or [] + out_slots = [] + for s in slots: + s2 = dict(s) + ca = s2.get("created_at") + if ca is not None and hasattr(ca, "isoformat"): + s2["created_at"] = ca.isoformat() + out_slots.append(s2) + p["slots"] = out_slots + return p + + +def _ing_render(qty, unit, food_name, note) -> str: + """Tiny fallback for `display` when Mealie didn't render the line.""" + parts: list[str] = [] + if qty not in (None, ""): + parts.append(str(qty)) + if unit: + parts.append(unit) + if food_name: + parts.append(food_name) + if note and not food_name: + parts.append(note) + return " ".join(parts).strip() + + # gunicorn entrypoint app = create_app() diff --git a/cauldron/templates/list.html b/cauldron/templates/list.html new file mode 100644 index 0000000..1cfb392 --- /dev/null +++ b/cauldron/templates/list.html @@ -0,0 +1,229 @@ +{% extends "_base.html" %} +{% block title %}Shopping List · Cauldron{% endblock %} +{% block content %} + + + +
+
// list · week of {{ plan.week_start.strftime('%b %-d') }}
+

shopping list

+
+ {% if empty_reason == 'no_plan' %} + no plan yet — head over to /plan and summon one first. + {% elif lines %} + {{ lines|length }} line{{ '' if lines|length == 1 else 's' }} · {{ plan.slots|length }} recipe{{ '' if plan.slots|length == 1 else 's' }} · check off as you shop, state survives refresh. + {% else %} + plan is set but no aggregatable ingredients came back. + {% endif %} +
+
+ +{% if empty_reason == 'no_plan' %} +
+
+

no plan yet

+
+
+
🪄
+

summon this week's plan first — every recipe's ingredients flow here automatically, density-aware aggregated.

+ go to /plan → +
+
+ +{% else %} + +{% if missing_recipes %} +
+ ⚠ {{ missing_recipes|length }} recipe{{ '' if missing_recipes|length == 1 else 's' }} couldn't be loaded from mealie: + {% for slug in missing_recipes %}{{ slug }}{% if not loop.last %}, {% endif %}{% endfor %}. + the list below is partial — check /plan and re-roll if needed. +
+{% endif %} + +
+
+

the list

+ {{ lines|length }} item{{ '' if lines|length == 1 else 's' }} +
+ + {% if lines %} + {# Flat list for now — aggregator output doesn't carry category. v0.4 may + re-introduce per-category sections via foods.category once dedupe lands. #} +
    + {% for ln in lines %} +
  • + +
    +
    + {% if ln.qty is not none %}{{ ln.qty }} {{ ln.unit }}{% endif %} + {{ ln.food }} + {% if ln.is_split %}(split){% endif %} +
    + {% if ln.contributors %} +
    from {{ ln.contributors|length }} recipe{{ '' if ln.contributors|length == 1 else 's' }}
    + {% endif %} + {% if ln.notes %} +
    {{ ln.notes|join(', ') }}
    + {% endif %} +
    +
    • + {% endfor %} +
+ {% else %} +

no aggregatable ingredients. check /plan.

+ {% endif %} +
+ +
+ + + +
+ + + +{% endif %} + +{% endblock %} diff --git a/cauldron/templates/plan.html b/cauldron/templates/plan.html index 815870a..b1ef3f4 100644 --- a/cauldron/templates/plan.html +++ b/cauldron/templates/plan.html @@ -2,6 +2,83 @@ {% block title %}This Week · Cauldron{% endblock %} {% block content %} + +
// plan · week of {{ week_start.strftime('%b %-d') }}

this week

@@ -14,8 +91,11 @@ automatically {% endif %} · {{ plan.locked_at.strftime('%a %-I:%M %p') }} + {% elif plan.slots %} + generated{% if generated_by_display %} by {{ generated_by_display }}{% endif %}. + lock when ready — first to lock takes the week. {% else %} - pool has {{ pick_count }} pinned. ai generation lands in v0.3 — for now, lock when ready to mark this week done. + {{ pick_count }} pinned in the pool. summon the planner to build a 7-day plan, then race to lock. {% endif %}
@@ -25,6 +105,8 @@

state

{% if plan.locked_at %} {{ 'auto-locked' if plan.locked_reason == 'auto' else 'locked' }} + {% elif plan.slots %} + generated {% else %} open {% endif %} @@ -36,19 +118,51 @@ {% if plan.locked_reason == 'user' and plan.locked_by_sub == current_user_sub %}

🏆 you locked this one in.

{% endif %} + {% elif not plan.slots %} +

no plan yet. summon claude to build one from the {{ pick_count }} pinned pick{{ '' if pick_count == 1 else 's' }} + the rest of the grimoire.

+ +
sonnet · ~30s
{% else %} -

nothing's stopping you from locking right now. or wait — generator's coming and the race is part of the deal.

+

plan's set. lock it in to claim the week — first to lock wins.

- view pool ({{ pick_count }}) + {% if plan.generated_by_sub == current_user_sub %} + + {% endif %} + view list →
{% endif %} +{% if plan.slots %} +
+
+

the week

+ {{ plan.slots|length }} day{{ '' if plan.slots|length == 1 else 's' }} +
+
+ {% for s in plan.slots %} + +
{{ s.day }}
+
{{ s.recipe_name }}
+ {% if s.picker_subs %} +
+ {% for sub in s.picker_subs %} + 🍄 {{ sub_display.get(sub, 'family') }} + {% endfor %} +
+ {% endif %} + {% if s.reason %}
{{ s.reason }}
{% endif %} + + {% endfor %} +
+
+{% endif %} +

scoreboard

- user-locked weeks only + picks landed · weeks won
{% if streak and streak.count >= 2 %} @@ -58,11 +172,12 @@

{% endif %} - {% if scoreboard and scoreboard|selectattr("wins")|list %} + {% if scoreboard and scoreboard|selectattr("points")|list or scoreboard and scoreboard|selectattr("wins")|list %} + @@ -72,11 +187,12 @@ +
memberpts wins last
{{ loop.index }}. - + {{ s.display_name or s.email.split('@')[0] }} {% if s.sub == current_user_sub %}· you{% endif %} {{ s.points or 0 }} {{ s.wins or 0 }} {{ s.last_win.strftime('%b %-d') if s.last_win else '—' }} @@ -86,7 +202,7 @@
{% else %} -

nobody's locked a week yet. be the first.

+

nobody's locked or generated yet. be the first.

{% endif %}
@@ -101,6 +217,49 @@ async function lockPlan(btn) { btn.disabled = false; btn.innerHTML = '🔒 lock this week'; } } + +async function generatePlan(btn) { + btn.disabled = true; + btn.innerHTML = '🪄 summoning…'; + const meta = document.getElementById('gen-meta'); + if (meta) meta.textContent = 'sonnet building plan — hold tight'; + try { + const r = await fetch('/api/plan/generate', { method: 'POST' }); + if (r.status === 409) { + // Someone beat us to it — just reload to see their plan + location.reload(); + return; + } + if (!r.ok) { + const data = await r.json().catch(() => ({})); + throw new Error(data.detail || data.error || r.status); + } + location.reload(); + } catch (e) { + btn.disabled = false; + btn.innerHTML = '🪄 generate this week\'s plan'; + if (meta) meta.textContent = 'failed: ' + e.message; + } +} + +async function rerollPlan(btn) { + if (!confirm('re-roll the week? slots + points reset.')) return; + btn.disabled = true; btn.textContent = '…'; + try { + const r = await fetch('/api/plan/regenerate', { method: 'POST' }); + if (!r.ok) { + const data = await r.json().catch(() => ({})); + throw new Error(data.detail || data.error || r.status); + } + location.reload(); + } catch (e) { + btn.disabled = false; btn.textContent = '↻ re-roll'; + alert('re-roll failed: ' + e.message); + } +} + +// Day cards have class .recipe-card → the base modal handler picks them up +// automatically (single-click → modal, ctrl/cmd-click → new tab). {% endblock %} diff --git a/tests/_testenv.py b/tests/_testenv.py new file mode 100644 index 0000000..5ca283b --- /dev/null +++ b/tests/_testenv.py @@ -0,0 +1,85 @@ +"""Test environment bootstrap. + +Cauldron's server module instantiates a DB + OAuth client + Mealie client at +import time, so tests must stub those before importing cauldron.server. + +This module performs that stubbing exactly once (idempotent — safe to import +multiple times). Both pytest's conftest.py and direct `unittest discover` +runs route through here. +""" +import os +from unittest.mock import MagicMock, patch + +from cryptography.fernet import Fernet + + +_BOOTSTRAPPED = False + + +def _stub_env() -> None: + os.environ.setdefault("SECRET_KEY", "test-secret") + os.environ.setdefault("MEALIE_BASE_URL", "http://mealie.test") + os.environ.setdefault("MEALIE_API_TOKEN", "mealie-test-token") + os.environ.setdefault("CLAWDFORGE_URL", "http://forge.test") + os.environ.setdefault("CLAWDFORGE_TOKEN", "forge-test-token") + os.environ.setdefault("ADMIN_BEARER", "admin-test-bearer") + os.environ.setdefault("OIDC_ISSUER", "http://authentik.test/application/o/cauldron") + os.environ.setdefault("OIDC_CLIENT_ID", "test-client") + os.environ.setdefault("OIDC_CLIENT_SECRET", "test-secret-client") + os.environ.setdefault("OIDC_REDIRECT_URI", "http://localhost/auth/callback") + os.environ.setdefault("DB_HOST", "localhost") + os.environ.setdefault("DB_NAME", "cauldron_test") + os.environ.setdefault("DB_USER", "cauldron") + os.environ.setdefault("DB_PASSWORD", "test") + os.environ.setdefault("CAULDRON_FERNET_KEY", Fernet.generate_key().decode()) + + +class _FakeCursor: + def __init__(self): + self.lastrowid = 0 + self.rowcount = 0 + def execute(self, *a, **kw): pass + def fetchone(self): return None + def fetchall(self): return [] + def __enter__(self): return self + def __exit__(self, *a): return False + + +class _FakeConn: + def __init__(self, *a, **kw): pass + def cursor(self): return _FakeCursor() + def commit(self): pass + def rollback(self): pass + def close(self): pass + + +def bootstrap() -> None: + """Apply env + import-time monkey patches. Idempotent.""" + global _BOOTSTRAPPED + if _BOOTSTRAPPED: + return + _stub_env() + + # pymysql.connect → no-op fake + patch("pymysql.connect", _FakeConn).start() + + # Pre-import dotted modules so the next patches resolve + import cauldron.db # noqa: F401 + import cauldron.oidc # noqa: F401 + import cauldron.foods # noqa: F401 + + # Skip migrations + patch("cauldron.db.DB.migrate", lambda self: []).start() + + # Skip OIDC metadata fetch + oauth_obj = MagicMock() + oauth_obj.cauldron = MagicMock() + patch("cauldron.oidc.init_oauth", lambda *a, **kw: oauth_obj).start() + + # Skip foods seed loader + patch("cauldron.foods.load_seed_if_empty", lambda db: 0).start() + + _BOOTSTRAPPED = True + + +bootstrap() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..4194f61 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,5 @@ +"""Pytest auto-loaded conftest. Routes through _testenv so both pytest and +`unittest discover` see the same stubs.""" +import os, sys +sys.path.insert(0, os.path.dirname(__file__)) +import _testenv # noqa: E402, F401 diff --git a/tests/test_list_view.py b/tests/test_list_view.py new file mode 100644 index 0000000..4cad272 --- /dev/null +++ b/tests/test_list_view.py @@ -0,0 +1,175 @@ +"""Tests for the /list shopping-list view. + +Two paths exercised: + 1. plan with no slots → renders the "go to /plan" CTA + 2. plan with slots + a (mocked) Mealie client returning canned recipes → + aggregator runs and the rendered HTML contains the expected lines +""" +import unittest +from datetime import date +from unittest.mock import MagicMock, patch + +# Trigger import-time stubs (env, pymysql, oidc) before pulling cauldron.server. +# Absolute import so unittest discover finds it. +import os, sys +sys.path.insert(0, os.path.dirname(__file__)) +import _testenv # noqa: E402, F401 + +from cauldron import server as srv + + +class _ListTestBase(unittest.TestCase): + def setUp(self): + self.client = srv.app.test_client() + with self.client.session_transaction() as s: + s["user"] = {"sub": "sub-cobb", "email": "cobb@sulkta.com", "name": "Cobb"} + + +def _fake_db(*, plan, slots=None): + fake = MagicMock() + fake.get_user_household_id.return_value = 1 + fake.get_or_create_plan.return_value = dict(plan) + fake.list_plan_slots.return_value = slots or [] + fake.list_household_member_subs.return_value = ["sub-1"] + + def _enrich(p): + p["slots"] = slots or [] + return p + fake.enrich_plan_with_slots.side_effect = _enrich + + # connection ctx for any auxiliary lookups (none needed here, but defensive) + from contextlib import contextmanager + @contextmanager + def _conn(): + yield FakeConn() + fake.conn.side_effect = _conn + + return fake + + +class FakeCursor: + def execute(self, *a, **kw): pass + def fetchone(self): return None + def fetchall(self): return [] + def __enter__(self): return self + def __exit__(self, *a): return False + + +class FakeConn: + def cursor(self): return FakeCursor() + def commit(self): pass + def rollback(self): pass + def close(self): pass + + +class TestListNoPlan(_ListTestBase): + def test_list_renders_no_plan(self): + plan = { + "id": 99, "household_id": 1, + "week_start": date(2026, 4, 27), + "generated_by_sub": None, "generated_at": None, + "locked_by_sub": None, "locked_at": None, "locked_reason": None, + } + fake_db = _fake_db(plan=plan, slots=[]) + with patch.object(srv, "db", fake_db): + r = self.client.get("/list") + self.assertEqual(r.status_code, 200) + body = r.get_data(as_text=True).lower() + # CTA text + self.assertIn("no plan yet", body) + self.assertIn("/plan", body) + + +class TestListAggregated(_ListTestBase): + def test_list_renders_aggregated(self): + plan = { + "id": 100, "household_id": 1, + "week_start": date(2026, 4, 27), + "generated_by_sub": "sub-cobb", + "generated_at": None, + "locked_by_sub": None, "locked_at": None, "locked_reason": None, + } + slots = [ + {"id": 1, "plan_id": 100, "day": "monday", + "recipe_slug": "stew", "recipe_name": "Stew", + "source": "mealie", "picker_subs": [], "reason": "", + "notes": None, "created_at": None}, + {"id": 2, "plan_id": 100, "day": "tuesday", + "recipe_slug": "rice-bowl", "recipe_name": "Rice Bowl", + "source": "mealie", "picker_subs": [], "reason": "", + "notes": None, "created_at": None}, + ] + + # Canned Mealie recipes: stew has 1 lb rice, rice-bowl has 2 cup rice + # → density-mixed agg should produce a single rice line + fake_mealie = MagicMock() + def _get_recipe(slug): + if slug == "stew": + return { + "name": "Stew", + "recipeIngredient": [ + {"quantity": 1, "unit": {"name": "lb"}, + "food": {"name": "rice"}, "note": "", + "display": "1 lb rice"}, + {"quantity": 2, "unit": {"name": "tbsp"}, + "food": {"name": "olive oil"}, "note": "", + "display": "2 tbsp olive oil"}, + ], + } + if slug == "rice-bowl": + return { + "name": "Rice Bowl", + "recipeIngredient": [ + {"quantity": 2, "unit": {"name": "cup"}, + "food": {"name": "rice"}, "note": "", + "display": "2 cups rice"}, + {"quantity": 1, "unit": {"name": "cup"}, + "food": {"name": "olive oil"}, "note": "", + "display": "1 cup olive oil"}, + ], + } + raise KeyError(slug) + fake_mealie.get_recipe.side_effect = _get_recipe + + fake_db = _fake_db(plan=plan, slots=slots) + + # foods.search_food returns canonical names + density for rice + olive oil + def _search_food(db, name, *, limit=1): + n = (name or "").strip().lower() + if "rice" in n: + return [{"id": 1, "canonical_name": "rice", + "density_g_per_ml": 0.85, + "default_unit_class": "mass", + "common_size_g": None, "category": "grain"}] + if "olive" in n or "oil" in n: + return [{"id": 2, "canonical_name": "olive oil", + "density_g_per_ml": 0.92, + "default_unit_class": "volume", + "common_size_g": None, "category": "oils"}] + return [] + + with patch.object(srv, "db", fake_db), \ + patch.object(srv, "current_user_mealie", create=True), \ + patch("cauldron.foods.search_food", side_effect=_search_food): + # patch the closure-bound current_user_mealie inside server.py + # via patching the helper Mealie constructor route + with patch("cauldron.server.Mealie", return_value=fake_mealie): + # Make the user-token blob path return a fake encrypted blob + fake_db.get_user_mealie_token_blob.return_value = b"fake-blob" + with patch.object(srv.crypto, "decrypt", return_value="t"): + r = self.client.get("/list") + + self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) + body = r.get_data(as_text=True).lower() + # Rice should appear exactly once as an aggregated line; olive oil too + self.assertIn("rice", body) + self.assertIn("olive oil", body) + # Aggregated rice: 1 lb + 2 cup × 0.85 g/ml × 236.588 ml/cup + # = 453.6g + 402g = ~855g → display_mass picks lb (rounded .25) + self.assertIn("lb", body) + # localStorage data attribute should be present + self.assertIn('data-plan-id="100"', body) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_plan_generator.py b/tests/test_plan_generator.py new file mode 100644 index 0000000..194ba08 --- /dev/null +++ b/tests/test_plan_generator.py @@ -0,0 +1,381 @@ +"""Tests for the AI plan generator (forge.generate_plan + the +/api/plan/generate endpoint). + +The endpoint tests use Flask's test client. db methods on the module-level +cauldron.server.db object are swapped out with MagicMocks per-test — this +avoids needing a real MariaDB to test routing + the orchestration logic. +""" +import json +import unittest +from unittest.mock import MagicMock, patch + +# Run conftest's import-time patches BEFORE pulling in cauldron.server. +# pytest auto-loads conftest, but unittest doesn't, so do it explicitly. +# Import path is absolute so `unittest discover` (which doesn't treat tests/ +# as a package) and pytest both resolve it. +import os, sys +sys.path.insert(0, os.path.dirname(__file__)) +import _testenv # noqa: E402, F401 + +from cauldron import server as srv +from cauldron.forge import Forge, ForgeError + + +# ---------- forge.generate_plan unit tests -------------------------------- + + +class TestForgeGeneratePlan(unittest.TestCase): + def setUp(self): + self.forge = Forge( + base_url="http://forge.test", token="t", + default_model="sonnet", default_timeout=60, + ) + + def _ok_run(self, slots_payload): + """Patch self.forge.run to return a dict shaped like clawdforge's.""" + return patch.object( + self.forge, "run", + return_value={"result": {"slots": slots_payload}}, + ) + + def test_validates_slot_count_matches(self): + recipes = [ + {"slug": "r1", "name": "Stew"}, + {"slug": "r2", "name": "Tacos"}, + {"slug": "r3", "name": "Pasta"}, + {"slug": "r4", "name": "Pie"}, + {"slug": "r5", "name": "Curry"}, + {"slug": "r6", "name": "Bowl"}, + {"slug": "r7", "name": "Soup"}, + ] + # Model returns only 5 slots — must raise + bad = [{"day": d, "recipe_slug": "r1", "picker_subs": [], "reason": ""} + for d in ("monday", "tuesday", "wednesday", "thursday", "friday")] + with self._ok_run(bad): + with self.assertRaises(ForgeError) as cm: + self.forge.generate_plan(picks=[], recipes=recipes, slots=7, week_start="2026-04-27") + self.assertIn("expected 7", str(cm.exception)) + + def test_rejects_unknown_slug(self): + recipes = [{"slug": "r1", "name": "A"}] + bad = [{"day": "monday", "recipe_slug": "r-not-real", "picker_subs": [], "reason": ""}] + with self._ok_run(bad): + with self.assertRaises(ForgeError) as cm: + self.forge.generate_plan(picks=[], recipes=recipes, slots=1, week_start="2026-04-27") + self.assertIn("unknown recipe_slug", str(cm.exception)) + + def test_rejects_duplicate_day(self): + recipes = [{"slug": "r1", "name": "A"}, {"slug": "r2", "name": "B"}] + bad = [ + {"day": "monday", "recipe_slug": "r1", "picker_subs": [], "reason": ""}, + {"day": "monday", "recipe_slug": "r2", "picker_subs": [], "reason": ""}, + ] + with self._ok_run(bad): + with self.assertRaises(ForgeError) as cm: + self.forge.generate_plan(picks=[], recipes=recipes, slots=2, week_start="2026-04-27") + self.assertIn("duplicate day", str(cm.exception)) + + def test_picker_attribution_uses_real_subs(self): + """Even if the model omits picker_subs, our ground-truth pick map + is what ends up on the slot.""" + recipes = [{"slug": "r1", "name": "Stew"}] + picks = [{"slug": "r1", "name": "Stew", "picker_subs": ["sub-abby", "sub-cobb"]}] + # Model returns empty picker_subs — we should fill from the picks + slots_in = [{"day": "monday", "recipe_slug": "r1", "picker_subs": [], "reason": "honors picks"}] + with self._ok_run(slots_in): + out = self.forge.generate_plan( + picks=picks, recipes=recipes, slots=1, week_start="2026-04-27", + ) + self.assertEqual(len(out), 1) + self.assertEqual(out[0]["picker_subs"], ["sub-abby", "sub-cobb"]) + self.assertEqual(out[0]["source"], "pick") + self.assertEqual(out[0]["recipe_name"], "Stew") + self.assertEqual(out[0]["reason"], "honors picks") + + def test_string_response_is_parsed(self): + """clawdforge sometimes returns the JSON as a string in `result`.""" + recipes = [{"slug": "r1", "name": "A"}] + payload = {"slots": [{"day": "monday", "recipe_slug": "r1", + "picker_subs": [], "reason": "ai"}]} + with patch.object(self.forge, "run", + return_value={"result": json.dumps(payload)}): + out = self.forge.generate_plan( + picks=[], recipes=recipes, slots=1, week_start="2026-04-27", + ) + self.assertEqual(len(out), 1) + self.assertEqual(out[0]["recipe_slug"], "r1") + self.assertEqual(out[0]["source"], "mealie") # no picks → mealie source + + def test_code_fenced_response_is_parsed(self): + recipes = [{"slug": "r1", "name": "A"}] + payload = {"slots": [{"day": "monday", "recipe_slug": "r1", + "picker_subs": [], "reason": ""}]} + fenced = "```json\n" + json.dumps(payload) + "\n```" + with patch.object(self.forge, "run", return_value={"result": fenced}): + out = self.forge.generate_plan( + picks=[], recipes=recipes, slots=1, week_start="2026-04-27", + ) + self.assertEqual(out[0]["recipe_slug"], "r1") + + +# ---------- /api/plan/generate route tests -------------------------------- + + +def _make_db_stub(*, plan, picks=None, recipe_rows=None, + existing_slots=None, save_inserted=None): + """Build a fake db with the methods the route uses.""" + fake = MagicMock() + fake.list_household_picks_with_pickers.return_value = picks or [] + fake.list_indexed_recipes.return_value = recipe_rows or [] + fake.list_plan_slots.return_value = existing_slots or [] + fake.get_or_create_plan.return_value = dict(plan) + fake.auto_lock_past_unlocked_plans.return_value = 0 + fake.list_household_member_subs.return_value = ["sub-1"] + fake.get_user_household_id.return_value = 1 + fake.list_household_pick_slugs.return_value = set() + fake.household_scoreboard.return_value = [] + fake.household_streak.return_value = None + fake.upsert_user.return_value = None + + # save_plan_slots returns inserted count (1+ default, or override for race tests) + if save_inserted is None: + save_inserted = lambda plan_id, slots: len(slots) + fake.save_plan_slots.side_effect = save_inserted + + # mark_plan_generated returns updated plan dict + def _mark(plan_id, sub): + p = dict(plan) + p["generated_by_sub"] = sub + from datetime import datetime + p["generated_at"] = datetime(2026, 4, 27, 12, 0, 0) + return p + fake.mark_plan_generated.side_effect = _mark + + # enrich_plan_with_slots adds slots to the plan dict in-place + def _enrich(p): + p["slots"] = fake.list_plan_slots.return_value + return p + fake.enrich_plan_with_slots.side_effect = _enrich + + # conn() context manager stub for the display-name resolution + from contextlib import contextmanager + @contextmanager + def _conn(): + yield FakeConn() + fake.conn.side_effect = _conn + + return fake + + +class FakeCursor: + def __init__(self): + self._rows = [] + def execute(self, *a, **kw): pass + def fetchone(self): return None + def fetchall(self): return [] + def __enter__(self): return self + def __exit__(self, *a): return False + + +class FakeConn: + def cursor(self): return FakeCursor() + def commit(self): pass + def rollback(self): pass + def close(self): pass + + +class _RouteTestBase(unittest.TestCase): + def setUp(self): + self.client = srv.app.test_client() + # Inject a session via a context override + with self.client.session_transaction() as s: + s["user"] = {"sub": "sub-cobb", "email": "cobb@sulkta.com", "name": "Cobb"} + + def _patch_db(self, fake_db): + return patch.object(srv, "db", fake_db) + + +class TestGenerateRoute(_RouteTestBase): + def test_generate_creates_slots(self): + from datetime import date + plan = { + "id": 42, "household_id": 1, + "week_start": date(2026, 4, 27), + "generated_by_sub": None, "generated_at": None, + "locked_by_sub": None, "locked_at": None, "locked_reason": None, + } + recipe_rows = [ + {"slug": f"r{i}", "name": f"Recipe {i}", "raw_json": "{}"} + for i in range(1, 11) + ] + slots_returned = [ + {"day": d, "recipe_slug": "r1", "recipe_name": "Recipe 1", + "picker_subs": [], "reason": "ai", "source": "mealie"} + for d in ("monday", "tuesday", "wednesday", "thursday", + "friday", "saturday", "sunday") + ] + # Make slots use unique recipes per day for realism + for i, s in enumerate(slots_returned): + s["recipe_slug"] = f"r{i+1}" + s["recipe_name"] = f"Recipe {i+1}" + + fake_db = _make_db_stub(plan=plan, recipe_rows=recipe_rows) + with self._patch_db(fake_db), \ + patch.object(srv.forge, "generate_plan", return_value=slots_returned): + r = self.client.post("/api/plan/generate") + self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) + body = r.get_json() + self.assertTrue(body["ok"]) + # save_plan_slots called with the plan id and the slots list + fake_db.save_plan_slots.assert_called_once() + args, _ = fake_db.save_plan_slots.call_args + self.assertEqual(args[0], 42) + self.assertEqual(len(args[1]), 7) + self.assertEqual(args[1][0]["day"], "monday") + # mark_plan_generated called with cobb's sub + fake_db.mark_plan_generated.assert_called_once_with(42, "sub-cobb") + + def test_generate_when_locked_409(self): + from datetime import date, datetime + plan = { + "id": 7, "household_id": 1, "week_start": date(2026, 4, 27), + "generated_by_sub": None, "generated_at": None, + "locked_by_sub": "sub-abby", + "locked_at": datetime(2026, 4, 27, 18, 0), + "locked_reason": "user", + } + fake_db = _make_db_stub(plan=plan) + with self._patch_db(fake_db), \ + patch.object(srv.forge, "generate_plan") as gp: + r = self.client.post("/api/plan/generate") + self.assertEqual(r.status_code, 409) + self.assertEqual(r.get_json()["error"], "plan_locked") + gp.assert_not_called() + + def test_generate_when_already_generated_409(self): + from datetime import date + plan = { + "id": 9, "household_id": 1, "week_start": date(2026, 4, 27), + "generated_by_sub": "sub-abby", "generated_at": None, + "locked_by_sub": None, "locked_at": None, "locked_reason": None, + } + existing = [{ + "id": 1, "plan_id": 9, "day": "monday", + "recipe_slug": "r1", "recipe_name": "Stew", + "source": "mealie", "picker_subs": [], "reason": "", "notes": None, + "created_at": None, + }] + fake_db = _make_db_stub(plan=plan, existing_slots=existing) + with self._patch_db(fake_db), \ + patch.object(srv.forge, "generate_plan") as gp: + r = self.client.post("/api/plan/generate") + self.assertEqual(r.status_code, 409) + body = r.get_json() + self.assertEqual(body["error"], "plan_already_generated") + self.assertIn("plan", body) + self.assertEqual(len(body["plan"]["slots"]), 1) + gp.assert_not_called() + + def test_pick_points_awarded_on_pick_use(self): + from datetime import date + plan = { + "id": 11, "household_id": 1, "week_start": date(2026, 4, 27), + "generated_by_sub": None, "generated_at": None, + "locked_by_sub": None, "locked_at": None, "locked_reason": None, + } + recipe_rows = [ + {"slug": "stew", "name": "Stew", "raw_json": "{}"}, + {"slug": "tacos", "name": "Tacos", "raw_json": "{}"}, + ] + picks = [ + {"slug": "stew", "name": "Stew", + "pickers": ["abby"], "picker_subs": ["sub-abby"]}, + {"slug": "tacos", "name": "Tacos", + "pickers": ["cobb", "abby"], "picker_subs": ["sub-cobb", "sub-abby"]}, + ] + # Slot fixture: monday = stew (abby picks), tuesday = tacos (cobb + + # abby picks), wed-sun = stew (ai-chosen, no pickers). + slots_full = [] + days = ("monday", "tuesday", "wednesday", "thursday", + "friday", "saturday", "sunday") + for i, d in enumerate(days): + if i == 0: + slots_full.append({ + "day": d, "recipe_slug": "stew", "recipe_name": "Stew", + "picker_subs": ["sub-abby"], "reason": "abby's pick", + "source": "pick", + }) + elif i == 1: + slots_full.append({ + "day": d, "recipe_slug": "tacos", "recipe_name": "Tacos", + "picker_subs": ["sub-cobb", "sub-abby"], "reason": "co", + "source": "pick", + }) + else: + slots_full.append({ + "day": d, "recipe_slug": "stew", "recipe_name": "Stew", + "picker_subs": [], "reason": "ai", "source": "mealie", + }) + + fake_db = _make_db_stub(plan=plan, picks=picks, recipe_rows=recipe_rows) + with self._patch_db(fake_db), \ + patch.object(srv.forge, "generate_plan", return_value=slots_full): + r = self.client.post("/api/plan/generate") + self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) + + # 1pt for sub-abby on monday + 1pt sub-cobb + 1pt sub-abby on tuesday + # = 3 award_pick_points calls total + self.assertEqual(fake_db.award_pick_points.call_count, 3) + # All calls should be (1, 11, , 1, "pick_used") + called_subs = [c.args[2] for c in fake_db.award_pick_points.call_args_list] + self.assertEqual(sorted(called_subs), ["sub-abby", "sub-abby", "sub-cobb"]) + for call in fake_db.award_pick_points.call_args_list: + self.assertEqual(call.args[3], 1) # points + self.assertEqual(call.args[4], "pick_used") + + +# ---------- household_scoreboard SQL test -------------------------------- + + +class TestScoreboardSchema(unittest.TestCase): + """The scoreboard SELECT must reference cauldron_pick_points and return + a `points` field. Verified by inspecting the generated SQL via a + capturing fake cursor.""" + + def test_scoreboard_query_includes_points(self): + from cauldron.db import DB + captured = {"sql": None} + + class CapCursor: + def execute(self, sql, params=None): + captured["sql"] = sql + def fetchall(self): + return [ + { + "sub": "sub-cobb", "email": "cobb@x.com", + "display_name": "Cobb", + "wins": 2, "last_win": None, "points": 5, + }, + ] + def __enter__(self): return self + def __exit__(self, *a): return False + + class CapConn: + def cursor(self): return CapCursor() + def commit(self): pass + def rollback(self): pass + def close(self): pass + + db = DB(host="x", port=3306, name="x", user="x", password="x") + with patch("pymysql.connect", lambda **kw: CapConn()): + rows = db.household_scoreboard(1) + + self.assertIn("cauldron_pick_points", captured["sql"]) + self.assertIn("points", captured["sql"]) + # And the row decoder coerces points to int + adds weeks_locked alias + self.assertEqual(rows[0]["points"], 5) + self.assertEqual(rows[0]["weeks_locked"], 2) + + +if __name__ == "__main__": + unittest.main()