cauldron/cauldron/db.py

"""DB access + migrations against sulkta-mariadb.

Uses PyMySQL with a tiny per-request connection (no pool) — Cauldron is
LAN-only family-internal, traffic is single-digit qps. If load ever grows
swap in DBUtils.PooledDB or SQLAlchemy.
"""
from contextlib import contextmanager
from pathlib import Path

import pymysql
import pymysql.cursors


MIGRATIONS = [
    # 001 — bookkeeping
    """
    CREATE TABLE IF NOT EXISTS schema_migrations (
        version VARCHAR(16) PRIMARY KEY,
        applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 002 — users (Authentik subject is the PK)
    """
    CREATE TABLE IF NOT EXISTS cauldron_users (
        authentik_sub VARCHAR(190) PRIMARY KEY,
        email VARCHAR(255) NOT NULL,
        display_name VARCHAR(255),
        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        last_seen DATETIME,
        INDEX idx_email (email)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 003 — per-user encrypted Mealie tokens
    """
    CREATE TABLE IF NOT EXISTS cauldron_user_mealie_tokens (
        authentik_sub VARCHAR(190) PRIMARY KEY,
        encrypted_token BLOB NOT NULL,
        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        last_validated DATETIME,
        last_failure_at DATETIME,
        last_failure_reason VARCHAR(500),
        FOREIGN KEY (authentik_sub) REFERENCES cauldron_users(authentik_sub)
            ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 004 — chat / AI run log (joins to clawdforge runs server-side)
    """
    CREATE TABLE IF NOT EXISTS cauldron_chat_log (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        authentik_sub VARCHAR(190) NOT NULL,
        ts DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        intent VARCHAR(64),
        forge_duration_ms INT,
        forge_model VARCHAR(64),
        prompt_chars INT,
        result_chars INT,
        ok BOOLEAN NOT NULL DEFAULT TRUE,
        error VARCHAR(500),
        INDEX idx_user_ts (authentik_sub, ts)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 005 — meal picks: per-user list of recipes the user wants in the next
    # AI meal plan run. Pre-populated wishlist that the planner respects.
    """
    CREATE TABLE IF NOT EXISTS cauldron_meal_picks (
        authentik_sub VARCHAR(190) NOT NULL,
        recipe_slug   VARCHAR(255) NOT NULL,
        recipe_name   VARCHAR(500) NOT NULL,
        added_at      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        PRIMARY KEY (authentik_sub, recipe_slug),
        INDEX idx_user_added (authentik_sub, added_at)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 006 — households (cached mirror of Mealie's household) + membership.
    # Keyed by Mealie's UUID. Multiple cauldron users join via the same
    # Mealie household to share picks/plans.
    """
    CREATE TABLE IF NOT EXISTS cauldron_households (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        mealie_household_id VARCHAR(64) UNIQUE NOT NULL,
        name VARCHAR(255) NOT NULL,
        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    """
    CREATE TABLE IF NOT EXISTS cauldron_household_members (
        household_id BIGINT NOT NULL,
        authentik_sub VARCHAR(190) NOT NULL,
        role VARCHAR(32) NOT NULL DEFAULT 'member',
        joined_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        PRIMARY KEY (household_id, authentik_sub),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
        FOREIGN KEY (authentik_sub) REFERENCES cauldron_users(authentik_sub) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 007 — meal plans (per household per week). Lock state + race metadata.
    # week_start = Monday (date) of the week.
    """
    CREATE TABLE IF NOT EXISTS cauldron_meal_plans (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        household_id BIGINT NOT NULL,
        week_start DATE NOT NULL,
        generated_by_sub VARCHAR(190),
        generated_at DATETIME,
        locked_by_sub VARCHAR(190),
        locked_at DATETIME,
        locked_reason ENUM('user','auto') DEFAULT NULL,
        UNIQUE KEY uk_household_week (household_id, week_start),
        INDEX idx_locked_by (locked_by_sub),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 008 — local recipe index for fast in-process search. Mirrors enough
    # of Mealie's recipe shape to fuzzy-rank without round-tripping to
    # Mealie on every keystroke. Refreshed on demand (on first /recipes
    # load, after pin/unpin, every 5min, or on /me 'refresh' button).
    """
    CREATE TABLE IF NOT EXISTS cauldron_recipe_index (
        household_id BIGINT NOT NULL,
        slug         VARCHAR(255) NOT NULL,
        name         VARCHAR(500) NOT NULL,
        description  TEXT,
        tags_text    TEXT,
        cats_text    TEXT,
        foods_text   TEXT,
        ings_text    TEXT,
        date_updated DATETIME,
        date_added   DATETIME,
        last_made    DATETIME,
        total_time   VARCHAR(64),
        recipe_yield VARCHAR(255),
        raw_json     JSON,
        indexed_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        PRIMARY KEY (household_id, slug),
        FULLTEXT KEY ft_text (name, description, tags_text, cats_text, foods_text),
        INDEX idx_household (household_id),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 009 — refresh state per household
    """
    CREATE TABLE IF NOT EXISTS cauldron_recipe_index_state (
        household_id BIGINT PRIMARY KEY,
        last_refreshed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        recipe_count INT NOT NULL DEFAULT 0,
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 010 — canonical foods table for the unit-aware aggregator. Each row is
    # ONE food (e.g. "rice", "butter", "onion") with density + unit class.
    # Seeded from USDA SR Legacy via scripts/build_foods_seed.py; will be
    # extended with claude-curated entries in v0.3 step 2.
    """
    CREATE TABLE IF NOT EXISTS cauldron_foods (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        canonical_name VARCHAR(255) NOT NULL,
        plural_name VARCHAR(255),
        category VARCHAR(64),
        density_g_per_ml DECIMAL(6,3),
        common_size_g DECIMAL(8,2),
        default_unit_class ENUM('mass','volume','count','mixed') NOT NULL DEFAULT 'mass',
        usda_fdc_id INT,
        usda_description VARCHAR(500),
        notes JSON,
        source ENUM('usda','claude','manual') NOT NULL DEFAULT 'usda',
        last_updated DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
        UNIQUE KEY uk_canonical (canonical_name),
        INDEX idx_category (category),
        INDEX idx_usda (usda_fdc_id)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 011 — Mealie food_id → cauldron food_id mapping per household. The
    # foods dedupe step (v0.3 A2) populates this. Aggregator joins through
    # this to group ingredients across recipes by canonical food.
    """
    CREATE TABLE IF NOT EXISTS cauldron_food_mapping (
        household_id BIGINT NOT NULL,
        mealie_food_id VARCHAR(64) NOT NULL,
        cauldron_food_id BIGINT NOT NULL,
        confidence DECIMAL(4,2) NOT NULL DEFAULT 1.00,
        mapped_by ENUM('exact','fuzzy','claude','manual') NOT NULL DEFAULT 'fuzzy',
        mapped_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        PRIMARY KEY (household_id, mealie_food_id),
        INDEX idx_canonical (cauldron_food_id),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
        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
    """,
    # 015 — bulk sterilizer jobs. One row per bulk-parse run kicked off by a
    # household member. State machine:
    #   running  → walking recipes, calling Sterilizer.preview_recipe
    #   review   → walk done, awaiting human approve/reject in /sterilize UI
    #   applying → applying approved proposals back to Mealie
    #   done     → complete
    #   failed   → unrecoverable error or stuck-job recovery marked it
    """
    CREATE TABLE IF NOT EXISTS cauldron_sterilize_jobs (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        household_id BIGINT NOT NULL,
        started_by_sub VARCHAR(190) NOT NULL,
        started_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        last_progress_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        finished_at DATETIME,
        total_recipes INT NOT NULL DEFAULT 0,
        processed_count INT NOT NULL DEFAULT 0,
        skipped_count INT NOT NULL DEFAULT 0,
        error_count INT NOT NULL DEFAULT 0,
        current_slug VARCHAR(255),
        last_error VARCHAR(500),
        state ENUM('running','review','applying','done','failed','cancelled')
              NOT NULL DEFAULT 'running',
        INDEX idx_household_state (household_id, state),
        INDEX idx_started_by (started_by_sub),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
        FOREIGN KEY (started_by_sub) REFERENCES cauldron_users(authentik_sub) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 016 — per-recipe proposals for a sterilize job. proposal_json is the
    # full Sterilizer.preview_recipe() output. approved=NULL means awaiting
    # review; TRUE = will apply; FALSE = skipped on apply. apply_error
    # captures any failure when applying that specific recipe.
    """
    CREATE TABLE IF NOT EXISTS cauldron_sterilize_proposals (
        job_id BIGINT NOT NULL,
        recipe_slug VARCHAR(255) NOT NULL,
        recipe_name VARCHAR(500),
        ingredient_count INT NOT NULL DEFAULT 0,
        proposal_json JSON,
        preview_error VARCHAR(500),
        approved BOOLEAN,
        applied_at DATETIME,
        apply_error VARCHAR(500),
        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        PRIMARY KEY (job_id, recipe_slug),
        INDEX idx_approved (job_id, approved),
        FOREIGN KEY (job_id) REFERENCES cauldron_sterilize_jobs(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 017 — Per-Mealie-food cooking metadata, keyed by Mealie's UUID food.id.
    # Replaces the old parallel name catalog (cauldron_foods) — Mealie owns
    # canonical food names, cauldron only owns density + unit class + common
    # size for shopping list aggregation. Populated lazily by the /list
    # foods_lookup path on the first sighting of each food (claude fallback).
    # Carried-over USDA densities will get re-keyed to mealie_food_id during
    # the Step 2 backfill, then cauldron_foods can be dropped (Step 4).
    """
    CREATE TABLE IF NOT EXISTS cauldron_food_metadata (
        mealie_food_id VARCHAR(64) PRIMARY KEY,
        food_name VARCHAR(255),
        density_g_per_ml DECIMAL(6,3),
        common_size_g DECIMAL(8,2),
        default_unit_class ENUM('mass','volume','count','mixed') NOT NULL DEFAULT 'mass',
        category VARCHAR(64),
        source ENUM('seed','claude','manual') NOT NULL DEFAULT 'claude',
        notes JSON,
        last_updated DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
        INDEX idx_food_name (food_name),
        INDEX idx_category (category)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 018 — Foods consolidator jobs. Mirrors the sterilize job pattern but
    # for the foods table itself: scan Mealie household foods, cluster by
    # similarity, ask Sonnet which clusters are dupes, apply via Mealie's
    # PUT /api/foods/merge endpoint. One job at a time per household.
    """
    CREATE TABLE IF NOT EXISTS cauldron_consolidate_jobs (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        household_id BIGINT NOT NULL,
        started_by_sub VARCHAR(190) NOT NULL,
        started_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        last_progress_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        finished_at DATETIME,
        total_clusters INT NOT NULL DEFAULT 0,
        processed_count INT NOT NULL DEFAULT 0,
        merged_count INT NOT NULL DEFAULT 0,
        error_count INT NOT NULL DEFAULT 0,
        current_cluster VARCHAR(255),
        last_error VARCHAR(500),
        state ENUM('running','review','applying','done','failed','cancelled')
              NOT NULL DEFAULT 'running',
        INDEX idx_household_state (household_id, state),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
        FOREIGN KEY (started_by_sub) REFERENCES cauldron_users(authentik_sub) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 019 — Per-cluster proposal. cluster_json holds the list of Mealie
    # food IDs+names in the cluster; sonnet_decision stores
    # {merge: bool, canonical_name: <picked>, canonical_food_id: <id>,
    # discard_food_ids: [...]}.
    """
    CREATE TABLE IF NOT EXISTS cauldron_consolidate_proposals (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        job_id BIGINT NOT NULL,
        cluster_key VARCHAR(255) NOT NULL,
        cluster_size INT NOT NULL DEFAULT 0,
        cluster_json JSON,
        sonnet_decision JSON,
        approved BOOLEAN,
        applied_at DATETIME,
        apply_error VARCHAR(500),
        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        UNIQUE KEY uk_job_cluster (job_id, cluster_key),
        INDEX idx_approved (job_id, approved),
        FOREIGN KEY (job_id) REFERENCES cauldron_consolidate_jobs(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 020 — Drop the game-system pick-points table. Stripped from /plan
    # 2026-04-30; tables and methods kept around as dead code for a few
    # days to let it bake. Now that the simpler base flow is proven,
    # the ledger comes out. Recovery path: revert this migration's number
    # and re-add the table from db.py history; the points data was never
    # widely populated anyway.
    """
    DROP TABLE IF EXISTS cauldron_pick_points
    """,
    # 021 — Recipe-dedupe bulk job state (mirrors the consolidate pattern
    # but for recipes themselves — name + ingredient similarity → Sonnet
    # decides → user-confirms → DELETE via Mealie API).
    """
    CREATE TABLE IF NOT EXISTS cauldron_recipe_dedupe_jobs (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        household_id BIGINT NOT NULL,
        started_by_sub VARCHAR(190) NOT NULL,
        started_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        last_progress_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        finished_at DATETIME,
        total_clusters INT NOT NULL DEFAULT 0,
        processed_count INT NOT NULL DEFAULT 0,
        deleted_count INT NOT NULL DEFAULT 0,
        error_count INT NOT NULL DEFAULT 0,
        current_cluster VARCHAR(255),
        last_error VARCHAR(500),
        state ENUM('running','review','applying','done','failed','cancelled')
              NOT NULL DEFAULT 'running',
        INDEX idx_household_state (household_id, state),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
        FOREIGN KEY (started_by_sub) REFERENCES cauldron_users(authentik_sub) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 022 — Per-cluster recipe-dedupe proposal. cluster_json holds the
    # full recipe summaries (slug+name+ingredients+source); sonnet_decision
    # = {duplicates: bool, canonical_slug, delete_slugs: [...], reason}.
    """
    CREATE TABLE IF NOT EXISTS cauldron_recipe_dedupe_proposals (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        job_id BIGINT NOT NULL,
        cluster_key VARCHAR(255) NOT NULL,
        cluster_size INT NOT NULL DEFAULT 0,
        cluster_json JSON,
        sonnet_decision JSON,
        approved BOOLEAN,
        applied_at DATETIME,
        apply_error VARCHAR(500),
        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        UNIQUE KEY uk_job_cluster (job_id, cluster_key),
        INDEX idx_approved (job_id, approved),
        FOREIGN KEY (job_id) REFERENCES cauldron_recipe_dedupe_jobs(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 023 — Per-week diet/vibe preference on the plan generator. Free-form
    # text the user types before generating ("high protein low carb",
    # "carb load this week", "light recovery, no fish") that biases
    # Sonnet's slot picks. Picks still take precedence — preference only
    # influences AI-chosen slots when the pool is bigger than picks.
    """
    ALTER TABLE cauldron_meal_plans
      ADD COLUMN IF NOT EXISTS preference_prompt VARCHAR(1000)
    """,
    # 024 — Per-recipe AI-generated metadata. Sonnet looks at the full
    # recipe (name, description, ingredients, steps, yields) and returns
    # a structured blob: tags, cuisine, complexity, estimated_minutes,
    # meal_type, primary_protein, primary_carb, veg_forward, comfort_tier,
    # season_fit, summary, best_for. Plan generator uses this so "high
    # protein week" becomes a real query, not just a vibe-prompt.
    # enrich_version lets us bump the prompt and re-enrich without
    # losing the prior data.
    """
    CREATE TABLE IF NOT EXISTS cauldron_recipe_meta (
        household_id BIGINT NOT NULL,
        recipe_slug VARCHAR(255) NOT NULL,
        meta_json JSON,
        enrich_version INT NOT NULL DEFAULT 1,
        last_enriched_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
                                   ON UPDATE CURRENT_TIMESTAMP,
        PRIMARY KEY (household_id, recipe_slug),
        INDEX idx_household (household_id),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    # 025 — Recipe-enrichment bulk job state. Runs through every household
    # recipe, calls Sonnet, persists meta. No apply/review step — meta is
    # purely additive so we just write it. Same daemon-thread runner +
    # cancel + stuck-recovery pattern.
    """
    CREATE TABLE IF NOT EXISTS cauldron_enrich_jobs (
        id BIGINT PRIMARY KEY AUTO_INCREMENT,
        household_id BIGINT NOT NULL,
        started_by_sub VARCHAR(190) NOT NULL,
        started_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        last_progress_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        finished_at DATETIME,
        total_recipes INT NOT NULL DEFAULT 0,
        enriched_count INT NOT NULL DEFAULT 0,
        skipped_count INT NOT NULL DEFAULT 0,
        error_count INT NOT NULL DEFAULT 0,
        current_slug VARCHAR(255),
        last_error VARCHAR(500),
        state ENUM('running','done','failed','cancelled')
              NOT NULL DEFAULT 'running',
        INDEX idx_household_state (household_id, state),
        FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
        FOREIGN KEY (started_by_sub) REFERENCES cauldron_users(authentik_sub) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
]


class DB:
    def __init__(self, *, host: str, port: int, name: str, user: str, password: str):
        self.kwargs = dict(
            host=host,
            port=port,
            user=user,
            password=password,
            database=name,
            charset="utf8mb4",
            cursorclass=pymysql.cursors.DictCursor,
            autocommit=False,
        )

    @contextmanager
    def conn(self):
        c = pymysql.connect(**self.kwargs)
        try:
            yield c
            c.commit()
        except Exception:
            c.rollback()
            raise
        finally:
            c.close()

    def migrate(self) -> list[str]:
        """Apply pending migrations. Returns list of versions applied."""
        applied: list[str] = []
        with self.conn() as c:
            with c.cursor() as cur:
                cur.execute(MIGRATIONS[0])  # bootstrap migrations table
                cur.execute("SELECT version FROM schema_migrations")
                done = {r["version"] for r in cur.fetchall()}
                for i, sql in enumerate(MIGRATIONS, start=1):
                    ver = f"{i:03d}"
                    if ver in done:
                        continue
                    cur.execute(sql)
                    # IGNORE to tolerate the multi-worker boot race where two
                    # gunicorn workers both bootstrap an empty migrations table
                    cur.execute(
                        "INSERT IGNORE INTO schema_migrations (version) VALUES (%s)", (ver,)
                    )
                    applied.append(ver)
        return applied

    # --- user ops -----------------------------------------------------------

    def upsert_user(self, *, sub: str, email: str, display_name: str | None) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT INTO cauldron_users (authentik_sub, email, display_name, last_seen)
                VALUES (%s, %s, %s, NOW())
                ON DUPLICATE KEY UPDATE
                    email = VALUES(email),
                    display_name = COALESCE(VALUES(display_name), display_name),
                    last_seen = NOW()
                """,
                (sub, email, display_name),
            )

    def get_user(self, sub: str) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT authentik_sub, email, display_name, last_seen FROM cauldron_users WHERE authentik_sub=%s",
                (sub,),
            )
            return cur.fetchone()

    # --- mealie token ops ---------------------------------------------------

    def first_usable_mealie_token(self) -> dict | None:
        """Return the freshest (last_validated DESC) per-user encrypted Mealie
        token row. Used at boot as a fallback when the system MEALIE_API_TOKEN
        env is dead — group admins can list foods, so any user's token works
        for the backfill scan."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT authentik_sub, encrypted_token
                   FROM cauldron_user_mealie_tokens
                   WHERE last_failure_at IS NULL OR
                         last_validated > COALESCE(last_failure_at, '1970-01-01')
                   ORDER BY last_validated DESC, created_at DESC
                   LIMIT 1"""
            )
            row = cur.fetchone()
            return dict(row) if row else None

    def get_user_mealie_token_blob(self, sub: str) -> bytes | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT encrypted_token FROM cauldron_user_mealie_tokens WHERE authentik_sub=%s",
                (sub,),
            )
            row = cur.fetchone()
            return row["encrypted_token"] if row else None

    def set_user_mealie_token_blob(self, sub: str, blob: bytes) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT INTO cauldron_user_mealie_tokens (authentik_sub, encrypted_token, last_validated)
                VALUES (%s, %s, NOW())
                ON DUPLICATE KEY UPDATE
                    encrypted_token = VALUES(encrypted_token),
                    last_validated = NOW(),
                    last_failure_at = NULL,
                    last_failure_reason = NULL
                """,
                (sub, blob),
            )

    def delete_user_mealie_token(self, sub: str) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "DELETE FROM cauldron_user_mealie_tokens WHERE authentik_sub=%s",
                (sub,),
            )

    def mark_user_mealie_token_failure(self, sub: str, reason: str) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                UPDATE cauldron_user_mealie_tokens
                   SET last_failure_at = NOW(), last_failure_reason = %s
                 WHERE authentik_sub = %s
                """,
                (reason[:500], sub),
            )

    # --- households ---------------------------------------------------------

    def upsert_household(self, *, mealie_household_id: str, name: str) -> int:
        """Create or update a household record. Returns local PK (id)."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT INTO cauldron_households (mealie_household_id, name)
                VALUES (%s, %s)
                ON DUPLICATE KEY UPDATE name = VALUES(name), id = LAST_INSERT_ID(id)
                """,
                (mealie_household_id, name),
            )
            return cur.lastrowid

    def add_household_member(self, household_id: int, sub: str, role: str = "member") -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT IGNORE INTO cauldron_household_members
                  (household_id, authentik_sub, role)
                VALUES (%s, %s, %s)
                """,
                (household_id, sub, role),
            )

    def get_user_household_id(self, sub: str) -> int | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT household_id FROM cauldron_household_members WHERE authentik_sub=%s LIMIT 1",
                (sub,),
            )
            row = cur.fetchone()
            return row["household_id"] if row else None

    def list_household_member_subs(self, household_id: int) -> list[str]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT authentik_sub FROM cauldron_household_members WHERE household_id=%s",
                (household_id,),
            )
            return [r["authentik_sub"] for r in cur.fetchall()]

    # --- meal plans (per household per week) -------------------------------

    def get_or_create_plan(self, household_id: int, week_start) -> dict:
        """Get the plan record for a (household, week_start), creating an
        empty one if it doesn't exist. week_start is a date (Monday)."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT IGNORE INTO cauldron_meal_plans (household_id, week_start)
                VALUES (%s, %s)
                """,
                (household_id, week_start),
            )
            cur.execute(
                "SELECT * FROM cauldron_meal_plans WHERE household_id=%s AND week_start=%s",
                (household_id, week_start),
            )
            return dict(cur.fetchone())

    def set_plan_preference(self, plan_id: int, preference: str | None) -> None:
        """Persist the per-week diet/vibe preference. Empty/whitespace-only
        strings are stored as NULL so the prompt only includes the section
        when the user actually filled it in."""
        clean = (preference or "").strip()[:1000] or None
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "UPDATE cauldron_meal_plans SET preference_prompt=%s WHERE id=%s",
                (clean, plan_id),
            )

    def lock_plan(self, plan_id: int, *, sub: str, reason: str = "user") -> dict:
        """Lock a plan if not already locked. Returns updated plan dict."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                UPDATE cauldron_meal_plans
                   SET locked_by_sub = %s, locked_at = NOW(), locked_reason = %s
                 WHERE id = %s AND locked_at IS NULL
                """,
                (sub, reason, plan_id),
            )
            cur.execute("SELECT * FROM cauldron_meal_plans WHERE id=%s", (plan_id,))
            return dict(cur.fetchone())

    def auto_lock_past_unlocked_plans(self, household_id: int, before_date) -> int:
        """Mark any past unlocked plans as auto-locked. Returns count."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                UPDATE cauldron_meal_plans
                   SET locked_at = NOW(), locked_reason = 'auto'
                 WHERE household_id = %s AND week_start < %s AND locked_at IS NULL
                """,
                (household_id, before_date),
            )
            return cur.rowcount

    # --- plan slots (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)."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute("DELETE FROM cauldron_meal_plan_slots WHERE plan_id=%s", (plan_id,))
            return cur.rowcount

    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 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:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT IGNORE INTO cauldron_meal_picks (authentik_sub, recipe_slug, recipe_name)
                VALUES (%s, %s, %s)
                """,
                (sub, slug, name[:500]),
            )
            return cur.rowcount > 0

    def remove_meal_pick(self, sub: str, slug: str) -> bool:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "DELETE FROM cauldron_meal_picks WHERE authentik_sub=%s AND recipe_slug=%s",
                (sub, slug),
            )
            return cur.rowcount > 0

    def list_meal_picks(self, sub: str) -> list[dict]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT recipe_slug, recipe_name, added_at FROM cauldron_meal_picks "
                "WHERE authentik_sub=%s ORDER BY added_at DESC",
                (sub,),
            )
            return [dict(r) for r in cur.fetchall()]

    def list_meal_pick_slugs(self, sub: str) -> set[str]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT recipe_slug FROM cauldron_meal_picks WHERE authentik_sub=%s",
                (sub,),
            )
            return {r["recipe_slug"] for r in cur.fetchall()}

    def list_household_pick_slugs(self, household_id: int) -> set[str]:
        """Union of picks across all members of the household."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                SELECT DISTINCT p.recipe_slug
                FROM cauldron_meal_picks p
                JOIN cauldron_household_members m ON m.authentik_sub = p.authentik_sub
                WHERE m.household_id = %s
                """,
                (household_id,),
            )
            return {r["recipe_slug"] for r in cur.fetchall()}

    def list_household_picks_with_pickers(self, household_id: int) -> list[dict]:
        """All picks across the household, grouped by slug, with the list of
        members who picked each (so the UI can show 'pinned by Cobb · Abby').
        Latest pick added_at per slug for ordering."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                SELECT
                    p.recipe_slug AS slug,
                    MIN(p.recipe_name) AS name,
                    GROUP_CONCAT(
                        DISTINCT COALESCE(NULLIF(u.display_name, ''),
                                          SUBSTRING_INDEX(u.email, '@', 1))
                        ORDER BY p.added_at ASC
                        SEPARATOR '|'
                    ) AS pickers,
                    GROUP_CONCAT(
                        DISTINCT u.authentik_sub
                        ORDER BY p.added_at ASC
                        SEPARATOR '|'
                    ) AS picker_subs,
                    MAX(p.added_at) AS last_pick_at,
                    COUNT(*)        AS pick_count
                FROM cauldron_meal_picks p
                JOIN cauldron_household_members m ON m.authentik_sub = p.authentik_sub
                LEFT JOIN cauldron_users u ON u.authentik_sub = p.authentik_sub
                WHERE m.household_id = %s
                GROUP BY p.recipe_slug
                ORDER BY last_pick_at DESC
                """,
                (household_id,),
            )
            out = []
            for r in cur.fetchall():
                d = dict(r)
                d["pickers"] = (d["pickers"] or "").split("|") if d["pickers"] else []
                d["picker_subs"] = (d["picker_subs"] or "").split("|") if d["picker_subs"] else []
                out.append(d)
            return out

    # --- recipe index -------------------------------------------------------

    def get_index_state(self, household_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT last_refreshed_at, recipe_count FROM cauldron_recipe_index_state WHERE household_id=%s",
                (household_id,),
            )
            return cur.fetchone()

    def replace_recipe_index(self, household_id: int, rows: list[dict]) -> int:
        """Atomic-ish replace of the index for one household. Drops + reinserts."""
        import json as _json
        with self.conn() as c, c.cursor() as cur:
            cur.execute("DELETE FROM cauldron_recipe_index WHERE household_id=%s", (household_id,))
            for r in rows:
                cur.execute(
                    """
                    INSERT INTO cauldron_recipe_index
                      (household_id, slug, name, description, tags_text, cats_text,
                       foods_text, ings_text, date_updated, date_added, last_made,
                       total_time, recipe_yield, raw_json)
                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                    """,
                    (
                        household_id,
                        r["slug"],
                        r["name"][:500],
                        (r.get("description") or "")[:65000],
                        (r.get("tags_text") or "")[:65000],
                        (r.get("cats_text") or "")[:65000],
                        (r.get("foods_text") or "")[:65000],
                        (r.get("ings_text") or "")[:65000],
                        r.get("date_updated"),
                        r.get("date_added"),
                        r.get("last_made"),
                        (r.get("total_time") or "")[:64],
                        (r.get("recipe_yield") or "")[:255],
                        _json.dumps(r.get("raw") or {}, default=str),
                    ),
                )
            cur.execute(
                """
                INSERT INTO cauldron_recipe_index_state (household_id, last_refreshed_at, recipe_count)
                VALUES (%s, NOW(), %s)
                ON DUPLICATE KEY UPDATE last_refreshed_at=NOW(), recipe_count=VALUES(recipe_count)
                """,
                (household_id, len(rows)),
            )
            return len(rows)

    def list_indexed_recipes(self, household_id: int, *, category: str | None = None,
                             order_by: str = "date_added", order_dir: str = "desc",
                             limit: int = 1000, offset: int = 0) -> list[dict]:
        """Pull the indexed recipe rows. Used both for non-search browse + as
        the candidate set for in-process fuzzy ranking on search."""
        order_col = {
            "date_added": "date_added",
            "date_updated": "date_updated",
            "last_made": "last_made",
            "name": "name",
        }.get(order_by, "date_added")
        order_dir_sql = "DESC" if order_dir.lower() != "asc" else "ASC"
        sql = f"""
            SELECT slug, name, description, tags_text, cats_text, foods_text,
                   date_updated, date_added, last_made, total_time, recipe_yield, raw_json
            FROM cauldron_recipe_index
            WHERE household_id = %s
        """
        params: list = [household_id]
        if category:
            sql += " AND cats_text LIKE %s"
            params.append(f"%{category}%")
        sql += f" ORDER BY {order_col} {order_dir_sql} LIMIT %s OFFSET %s"
        params += [limit, offset]
        with self.conn() as c, c.cursor() as cur:
            cur.execute(sql, params)
            return [dict(r) for r in cur.fetchall()]

    # --- chat log -----------------------------------------------------------

    def log_chat(
        self,
        *,
        sub: str,
        intent: str,
        duration_ms: int,
        model: str,
        prompt_chars: int,
        result_chars: int,
        ok: bool,
        error: str | None = None,
    ) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT INTO cauldron_chat_log
                  (authentik_sub, intent, forge_duration_ms, forge_model,
                   prompt_chars, result_chars, ok, error)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
                """,
                (sub, intent, duration_ms, model, prompt_chars, result_chars, ok, (error or "")[:500] or None),
            )

    # --- bulk sterilizer ----------------------------------------------------

    def create_sterilize_job(
        self, *, household_id: int, started_by_sub: str, total: int
    ) -> int:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT INTO cauldron_sterilize_jobs
                  (household_id, started_by_sub, total_recipes, state)
                VALUES (%s, %s, %s, 'running')
                """,
                (household_id, started_by_sub, total),
            )
            return cur.lastrowid

    def get_sterilize_job(self, job_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT * FROM cauldron_sterilize_jobs WHERE id=%s", (job_id,)
            )
            return cur.fetchone()

    def get_sterilize_job_state(self, job_id: int) -> str | None:
        """Cheap state-only fetch used by the runner to detect external
        cancels mid-loop without dragging the whole row over the wire."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "SELECT state FROM cauldron_sterilize_jobs WHERE id=%s", (job_id,)
            )
            row = cur.fetchone()
            return row["state"] if row else None

    def latest_sterilize_job_for_household(self, household_id: int) -> dict | None:
        """Most recent job (by started_at) for the household — used by the
        UI to figure out what to render on /sterilize."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT * FROM cauldron_sterilize_jobs
                   WHERE household_id=%s
                   ORDER BY started_at DESC
                   LIMIT 1""",
                (household_id,),
            )
            return cur.fetchone()

    def running_sterilize_job_for_household(self, household_id: int) -> dict | None:
        """Active (running or applying) job; used to gate concurrent starts."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT * FROM cauldron_sterilize_jobs
                   WHERE household_id=%s AND state IN ('running','applying')
                   ORDER BY started_at DESC
                   LIMIT 1""",
                (household_id,),
            )
            return cur.fetchone()

    def update_sterilize_job_progress(
        self,
        job_id: int,
        *,
        processed_delta: int = 0,
        skipped_delta: int = 0,
        error_delta: int = 0,
        current_slug: str | None = None,
        last_error: str | None = None,
    ) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                UPDATE cauldron_sterilize_jobs
                SET processed_count = processed_count + %s,
                    skipped_count   = skipped_count   + %s,
                    error_count     = error_count     + %s,
                    current_slug    = %s,
                    last_error      = COALESCE(%s, last_error),
                    last_progress_at = NOW()
                WHERE id=%s
                """,
                (processed_delta, skipped_delta, error_delta,
                 current_slug, last_error, job_id),
            )

    def finalize_sterilize_job(self, job_id: int, *, state: str) -> None:
        """Move job to a new state. Will NOT overwrite a terminal state
        (done / failed / cancelled) — that's the anti-zombie guard that
        keeps user cancels from being silently replaced when the daemon
        thread limps to the finish line.

        Allowed source states: running, applying, review. The review state
        is part of the normal flow (walk done → review → user approves
        → applying), so transitions out of review must work."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                UPDATE cauldron_sterilize_jobs
                SET state=%s,
                    finished_at = CASE WHEN %s IN ('done','failed','cancelled')
                                       THEN NOW() ELSE finished_at END,
                    last_progress_at = NOW(),
                    current_slug = NULL
                WHERE id=%s
                  AND state NOT IN ('done','failed','cancelled')
                """,
                (state, state, job_id),
            )

    def insert_sterilize_proposal(
        self,
        *,
        job_id: int,
        recipe_slug: str,
        recipe_name: str | None,
        ingredient_count: int,
        proposal_json: str | None,
        preview_error: str | None,
    ) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """
                INSERT INTO cauldron_sterilize_proposals
                  (job_id, recipe_slug, recipe_name, ingredient_count,
                   proposal_json, preview_error)
                VALUES (%s, %s, %s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                  recipe_name=VALUES(recipe_name),
                  ingredient_count=VALUES(ingredient_count),
                  proposal_json=VALUES(proposal_json),
                  preview_error=VALUES(preview_error)
                """,
                (job_id, recipe_slug, (recipe_name or "")[:500],
                 ingredient_count, proposal_json, (preview_error or "")[:500] or None),
            )

    def list_sterilize_proposals(self, job_id: int) -> list[dict]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT recipe_slug, recipe_name, ingredient_count,
                          proposal_json, preview_error,
                          approved, applied_at, apply_error
                   FROM cauldron_sterilize_proposals
                   WHERE job_id=%s
                   ORDER BY recipe_name ASC""",
                (job_id,),
            )
            return [dict(r) for r in cur.fetchall()]

    def set_proposal_approval(
        self, job_id: int, recipe_slug: str, approved: bool
    ) -> bool:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_sterilize_proposals
                   SET approved=%s
                   WHERE job_id=%s AND recipe_slug=%s""",
                (1 if approved else 0, job_id, recipe_slug),
            )
            return cur.rowcount > 0

    def bulk_set_proposal_approvals(
        self, job_id: int, approved_slugs: list[str]
    ) -> None:
        """Set approved=TRUE for the listed slugs, FALSE for everything else
        on the job. Idempotent — safe to call before /bulk-apply."""
        with self.conn() as c, c.cursor() as cur:
            # Reset everything to FALSE first (NULL stays NULL only on rows
            # that aren't part of this job, which is none)
            cur.execute(
                "UPDATE cauldron_sterilize_proposals SET approved=0 WHERE job_id=%s",
                (job_id,),
            )
            if approved_slugs:
                placeholders = ",".join(["%s"] * len(approved_slugs))
                cur.execute(
                    f"""UPDATE cauldron_sterilize_proposals SET approved=1
                        WHERE job_id=%s AND recipe_slug IN ({placeholders})""",
                    (job_id, *approved_slugs),
                )

    def mark_proposal_applied(
        self, job_id: int, recipe_slug: str, *, error: str | None = None
    ) -> None:
        """On success: applied_at=NOW(), apply_error=NULL. On error: leave
        applied_at NULL so a rerun can retry, but record the error for
        review. The list_approved_unapplied_proposals query keys off
        applied_at IS NULL, so this directly drives retryability."""
        if error:
            with self.conn() as c, c.cursor() as cur:
                cur.execute(
                    """UPDATE cauldron_sterilize_proposals
                       SET apply_error=%s
                       WHERE job_id=%s AND recipe_slug=%s""",
                    (error[:500], job_id, recipe_slug),
                )
        else:
            with self.conn() as c, c.cursor() as cur:
                cur.execute(
                    """UPDATE cauldron_sterilize_proposals
                       SET applied_at=NOW(), apply_error=NULL
                       WHERE job_id=%s AND recipe_slug=%s""",
                    (job_id, recipe_slug),
                )

    def list_approved_unapplied_proposals(self, job_id: int) -> list[dict]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT recipe_slug, recipe_name
                   FROM cauldron_sterilize_proposals
                   WHERE job_id=%s AND approved=1 AND applied_at IS NULL""",
                (job_id,),
            )
            return [dict(r) for r in cur.fetchall()]

    # --- consolidate (foods merge) jobs -------------------------------------

    def create_consolidate_job(
        self, *, household_id: int, started_by_sub: str
    ) -> int:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """INSERT INTO cauldron_consolidate_jobs
                     (household_id, started_by_sub, state)
                   VALUES (%s, %s, 'running')""",
                (household_id, started_by_sub),
            )
            return cur.lastrowid

    def get_consolidate_job(self, job_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute("SELECT * FROM cauldron_consolidate_jobs WHERE id=%s", (job_id,))
            return cur.fetchone()

    def get_consolidate_job_state(self, job_id: int) -> str | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute("SELECT state FROM cauldron_consolidate_jobs WHERE id=%s", (job_id,))
            row = cur.fetchone()
            return row["state"] if row else None

    def latest_consolidate_job_for_household(self, household_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT * FROM cauldron_consolidate_jobs
                   WHERE household_id=%s ORDER BY started_at DESC LIMIT 1""",
                (household_id,),
            )
            return cur.fetchone()

    def running_consolidate_job_for_household(self, household_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT * FROM cauldron_consolidate_jobs
                   WHERE household_id=%s AND state IN ('running','applying')
                   ORDER BY started_at DESC LIMIT 1""",
                (household_id,),
            )
            return cur.fetchone()

    def update_consolidate_job_progress(
        self,
        job_id: int,
        *,
        processed_delta: int = 0,
        merged_delta: int = 0,
        error_delta: int = 0,
        current_cluster: str | None = None,
        last_error: str | None = None,
    ) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_consolidate_jobs
                   SET processed_count = processed_count + %s,
                       merged_count    = merged_count    + %s,
                       error_count     = error_count     + %s,
                       current_cluster = COALESCE(%s, current_cluster),
                       last_error      = COALESCE(%s, last_error),
                       last_progress_at = NOW()
                   WHERE id=%s""",
                (processed_delta, merged_delta, error_delta,
                 current_cluster, last_error, job_id),
            )

    def finalize_consolidate_job(self, job_id: int, *, state: str) -> None:
        """Same anti-zombie guard as finalize_sterilize_job — won't overwrite
        a terminal state (done/failed/cancelled), but allows the normal
        running→review→applying→done flow."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_consolidate_jobs
                   SET state=%s,
                       finished_at = CASE WHEN %s IN ('done','failed','cancelled')
                                          THEN NOW() ELSE finished_at END,
                       last_progress_at = NOW(),
                       current_cluster = NULL
                   WHERE id=%s
                     AND state NOT IN ('done','failed','cancelled')""",
                (state, state, job_id),
            )

    def insert_consolidate_proposal(
        self,
        *,
        job_id: int,
        cluster_key: str,
        cluster: list[dict],
        decision: dict | None,
        error: str | None,
    ) -> None:
        import json as _j
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """INSERT INTO cauldron_consolidate_proposals
                     (job_id, cluster_key, cluster_size, cluster_json,
                      sonnet_decision, apply_error)
                   VALUES (%s, %s, %s, %s, %s, %s)
                   ON DUPLICATE KEY UPDATE
                     cluster_size=VALUES(cluster_size),
                     cluster_json=VALUES(cluster_json),
                     sonnet_decision=VALUES(sonnet_decision),
                     apply_error=VALUES(apply_error)""",
                (
                    job_id, cluster_key[:255], len(cluster),
                    _j.dumps(cluster, ensure_ascii=False, default=str),
                    _j.dumps(decision, ensure_ascii=False) if decision else None,
                    (error or "")[:500] or None,
                ),
            )

    def list_consolidate_proposals(self, job_id: int) -> list[dict]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT id, cluster_key, cluster_size, cluster_json,
                          sonnet_decision, approved, applied_at, apply_error
                   FROM cauldron_consolidate_proposals
                   WHERE job_id=%s
                   ORDER BY cluster_size DESC, cluster_key""",
                (job_id,),
            )
            return [dict(r) for r in cur.fetchall()]

    def list_approved_unapplied_consolidate(self, job_id: int) -> list[dict]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT id, cluster_key, cluster_size, sonnet_decision
                   FROM cauldron_consolidate_proposals
                   WHERE job_id=%s AND approved=1 AND applied_at IS NULL""",
                (job_id,),
            )
            return [dict(r) for r in cur.fetchall()]

    def bulk_set_consolidate_approvals(
        self, job_id: int, approved_ids: list[int]
    ) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "UPDATE cauldron_consolidate_proposals SET approved=0 WHERE job_id=%s",
                (job_id,),
            )
            if approved_ids:
                placeholders = ",".join(["%s"] * len(approved_ids))
                cur.execute(
                    f"""UPDATE cauldron_consolidate_proposals SET approved=1
                        WHERE job_id=%s AND id IN ({placeholders})""",
                    (job_id, *approved_ids),
                )

    def mark_consolidate_proposal_applied(
        self, proposal_id: int, *, error: str | None = None
    ) -> None:
        if error:
            with self.conn() as c, c.cursor() as cur:
                cur.execute(
                    """UPDATE cauldron_consolidate_proposals
                       SET apply_error=%s
                       WHERE id=%s""",
                    (error[:500], proposal_id),
                )
        else:
            with self.conn() as c, c.cursor() as cur:
                cur.execute(
                    """UPDATE cauldron_consolidate_proposals
                       SET applied_at=NOW(), apply_error=NULL
                       WHERE id=%s""",
                    (proposal_id,),
                )

    # --- recipe-dedupe jobs ------------------------------------------------

    def create_recipe_dedupe_job(self, *, household_id: int, started_by_sub: str) -> int:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """INSERT INTO cauldron_recipe_dedupe_jobs
                     (household_id, started_by_sub, state)
                   VALUES (%s, %s, 'running')""",
                (household_id, started_by_sub),
            )
            return cur.lastrowid

    def get_recipe_dedupe_job(self, job_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute("SELECT * FROM cauldron_recipe_dedupe_jobs WHERE id=%s", (job_id,))
            return cur.fetchone()

    def get_recipe_dedupe_job_state(self, job_id: int) -> str | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute("SELECT state FROM cauldron_recipe_dedupe_jobs WHERE id=%s", (job_id,))
            row = cur.fetchone()
            return row["state"] if row else None

    def latest_recipe_dedupe_job_for_household(self, household_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT * FROM cauldron_recipe_dedupe_jobs
                   WHERE household_id=%s ORDER BY started_at DESC LIMIT 1""",
                (household_id,),
            )
            return cur.fetchone()

    def running_recipe_dedupe_job_for_household(self, household_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT * FROM cauldron_recipe_dedupe_jobs
                   WHERE household_id=%s AND state IN ('running','applying')
                   ORDER BY started_at DESC LIMIT 1""",
                (household_id,),
            )
            return cur.fetchone()

    def update_recipe_dedupe_job_progress(
        self,
        job_id: int,
        *,
        processed_delta: int = 0,
        deleted_delta: int = 0,
        error_delta: int = 0,
        current_cluster: str | None = None,
        last_error: str | None = None,
    ) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_recipe_dedupe_jobs
                   SET processed_count = processed_count + %s,
                       deleted_count   = deleted_count   + %s,
                       error_count     = error_count     + %s,
                       current_cluster = COALESCE(%s, current_cluster),
                       last_error      = COALESCE(%s, last_error),
                       last_progress_at = NOW()
                   WHERE id=%s""",
                (processed_delta, deleted_delta, error_delta,
                 current_cluster, last_error, job_id),
            )

    def finalize_recipe_dedupe_job(self, job_id: int, *, state: str) -> None:
        """Same anti-zombie guard as the others — won't overwrite terminal."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_recipe_dedupe_jobs
                   SET state=%s,
                       finished_at = CASE WHEN %s IN ('done','failed','cancelled')
                                          THEN NOW() ELSE finished_at END,
                       last_progress_at = NOW(),
                       current_cluster = NULL
                   WHERE id=%s
                     AND state NOT IN ('done','failed','cancelled')""",
                (state, state, job_id),
            )

    def insert_recipe_dedupe_proposal(
        self,
        *,
        job_id: int,
        cluster_key: str,
        cluster: list[dict],
        decision: dict | None,
        error: str | None,
    ) -> None:
        import json as _j
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """INSERT INTO cauldron_recipe_dedupe_proposals
                     (job_id, cluster_key, cluster_size, cluster_json,
                      sonnet_decision, apply_error)
                   VALUES (%s, %s, %s, %s, %s, %s)
                   ON DUPLICATE KEY UPDATE
                     cluster_size=VALUES(cluster_size),
                     cluster_json=VALUES(cluster_json),
                     sonnet_decision=VALUES(sonnet_decision),
                     apply_error=VALUES(apply_error)""",
                (
                    job_id, cluster_key[:255], len(cluster),
                    _j.dumps(cluster, ensure_ascii=False, default=str),
                    _j.dumps(decision, ensure_ascii=False) if decision else None,
                    (error or "")[:500] or None,
                ),
            )

    def list_recipe_dedupe_proposals(self, job_id: int) -> list[dict]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT id, cluster_key, cluster_size, cluster_json,
                          sonnet_decision, approved, applied_at, apply_error
                   FROM cauldron_recipe_dedupe_proposals
                   WHERE job_id=%s
                   ORDER BY cluster_size DESC, cluster_key""",
                (job_id,),
            )
            return [dict(r) for r in cur.fetchall()]

    def list_approved_unapplied_recipe_dedupe(self, job_id: int) -> list[dict]:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT id, cluster_key, cluster_size, sonnet_decision
                   FROM cauldron_recipe_dedupe_proposals
                   WHERE job_id=%s AND approved=1 AND applied_at IS NULL""",
                (job_id,),
            )
            return [dict(r) for r in cur.fetchall()]

    def bulk_set_recipe_dedupe_approvals(self, job_id: int, approved_ids: list[int]) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                "UPDATE cauldron_recipe_dedupe_proposals SET approved=0 WHERE job_id=%s",
                (job_id,),
            )
            if approved_ids:
                placeholders = ",".join(["%s"] * len(approved_ids))
                cur.execute(
                    f"""UPDATE cauldron_recipe_dedupe_proposals SET approved=1
                        WHERE job_id=%s AND id IN ({placeholders})""",
                    (job_id, *approved_ids),
                )

    def mark_recipe_dedupe_proposal_applied(
        self, proposal_id: int, *, error: str | None = None
    ) -> None:
        if error:
            with self.conn() as c, c.cursor() as cur:
                cur.execute(
                    "UPDATE cauldron_recipe_dedupe_proposals SET apply_error=%s WHERE id=%s",
                    (error[:500], proposal_id),
                )
        else:
            with self.conn() as c, c.cursor() as cur:
                cur.execute(
                    "UPDATE cauldron_recipe_dedupe_proposals SET applied_at=NOW(), apply_error=NULL WHERE id=%s",
                    (proposal_id,),
                )

    # --- recipe enrichment ------------------------------------------------

    ENRICH_VERSION = 1

    def get_recipe_meta(self, household_id: int, recipe_slug: str) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT recipe_slug, meta_json, enrich_version, last_enriched_at
                   FROM cauldron_recipe_meta
                   WHERE household_id=%s AND recipe_slug=%s""",
                (household_id, recipe_slug),
            )
            row = cur.fetchone()
            return dict(row) if row else None

    def list_recipe_meta_for_household(self, household_id: int) -> list[dict]:
        """Used by the plan generator to splice meta into the recipe pool prompt."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT recipe_slug, meta_json, enrich_version
                   FROM cauldron_recipe_meta
                   WHERE household_id=%s""",
                (household_id,),
            )
            return [dict(r) for r in cur.fetchall()]

    def upsert_recipe_meta(
        self,
        *,
        household_id: int,
        recipe_slug: str,
        meta_json: str,
        version: int,
    ) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """INSERT INTO cauldron_recipe_meta
                     (household_id, recipe_slug, meta_json, enrich_version)
                   VALUES (%s, %s, %s, %s)
                   ON DUPLICATE KEY UPDATE
                     meta_json = VALUES(meta_json),
                     enrich_version = VALUES(enrich_version)""",
                (household_id, recipe_slug, meta_json, version),
            )

    def create_enrich_job(self, *, household_id: int, started_by_sub: str) -> int:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """INSERT INTO cauldron_enrich_jobs
                     (household_id, started_by_sub, state)
                   VALUES (%s, %s, 'running')""",
                (household_id, started_by_sub),
            )
            return cur.lastrowid

    def get_enrich_job(self, job_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute("SELECT * FROM cauldron_enrich_jobs WHERE id=%s", (job_id,))
            return cur.fetchone()

    def get_enrich_job_state(self, job_id: int) -> str | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute("SELECT state FROM cauldron_enrich_jobs WHERE id=%s", (job_id,))
            row = cur.fetchone()
            return row["state"] if row else None

    def latest_enrich_job_for_household(self, household_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT * FROM cauldron_enrich_jobs
                   WHERE household_id=%s ORDER BY started_at DESC LIMIT 1""",
                (household_id,),
            )
            return cur.fetchone()

    def running_enrich_job_for_household(self, household_id: int) -> dict | None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """SELECT * FROM cauldron_enrich_jobs
                   WHERE household_id=%s AND state='running'
                   ORDER BY started_at DESC LIMIT 1""",
                (household_id,),
            )
            return cur.fetchone()

    def update_enrich_job_progress(
        self,
        job_id: int,
        *,
        enriched_delta: int = 0,
        skipped_delta: int = 0,
        error_delta: int = 0,
        current_slug: str | None = None,
        last_error: str | None = None,
    ) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_enrich_jobs
                   SET enriched_count = enriched_count + %s,
                       skipped_count  = skipped_count  + %s,
                       error_count    = error_count    + %s,
                       current_slug   = COALESCE(%s, current_slug),
                       last_error     = COALESCE(%s, last_error),
                       last_progress_at = NOW()
                   WHERE id=%s""",
                (enriched_delta, skipped_delta, error_delta,
                 current_slug, last_error, job_id),
            )

    def finalize_enrich_job(self, job_id: int, *, state: str) -> None:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_enrich_jobs
                   SET state=%s,
                       finished_at = CASE WHEN %s IN ('done','failed','cancelled')
                                          THEN NOW() ELSE finished_at END,
                       last_progress_at = NOW(),
                       current_slug = NULL
                   WHERE id=%s
                     AND state NOT IN ('done','failed','cancelled')""",
                (state, state, job_id),
            )

    def fail_stuck_enrich_jobs(self, *, stale_minutes: int = 15) -> int:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_enrich_jobs
                   SET state='failed',
                       finished_at=NOW(),
                       last_error=COALESCE(last_error, 'recovery: worker exited mid-run')
                   WHERE state='running'
                     AND last_progress_at < NOW() - INTERVAL %s MINUTE""",
                (stale_minutes,),
            )
            return cur.rowcount

    def fail_stuck_recipe_dedupe_jobs(self, *, stale_minutes: int = 15) -> int:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_recipe_dedupe_jobs
                   SET state='failed',
                       finished_at=NOW(),
                       last_error=COALESCE(last_error,
                                  'recovery: worker exited mid-run')
                   WHERE state IN ('running','applying')
                     AND last_progress_at < NOW() - INTERVAL %s MINUTE""",
                (stale_minutes,),
            )
            return cur.rowcount

    def fail_stuck_consolidate_jobs(self, *, stale_minutes: int = 15) -> int:
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_consolidate_jobs
                   SET state='failed',
                       finished_at=NOW(),
                       last_error=COALESCE(last_error,
                                  'recovery: worker exited mid-run')
                   WHERE state IN ('running','applying')
                     AND last_progress_at < NOW() - INTERVAL %s MINUTE""",
                (stale_minutes,),
            )
            return cur.rowcount

    # --- sterilize stuck-job recovery (already used at boot) ---------------

    def fail_stuck_sterilize_jobs(self, *, stale_minutes: int = 10) -> int:
        """Recover jobs stuck in 'running'/'applying' with no progress for
        N minutes. Called at app startup. Returns count of jobs failed."""
        with self.conn() as c, c.cursor() as cur:
            cur.execute(
                """UPDATE cauldron_sterilize_jobs
                   SET state='failed',
                       finished_at=NOW(),
                       last_error=COALESCE(last_error,
                                  'recovery: worker exited mid-run')
                   WHERE state IN ('running','applying')
                     AND last_progress_at < NOW() - INTERVAL %s MINUTE""",
                (stale_minutes,),
            )
            return cur.rowcount