sterilize bulk: scope walk + apply to user's own household

Mealie's group model spans multiple households (Hayes House group has
Redondo + Lake Elsinore — Cobb's family + Bay/mom share recipes
read-only across households). Members can list/read every recipe in
the group, but write access is per-household. Trying to sterilize a
foreign-household recipe returns 403 on the PUT; the food/unit creates
that ran first end up as orphan rows in the user's own household.

Walk path now resolves /api/users/self.householdId once (cached on the
Mealie client), and skips any recipe whose top-level householdId differs.
No proposal row is created; the recipe just counts as "skipped" along-
side already-clean ones.

Apply path adds the same defensive check (covers job 1's existing
proposals from before this fix landed) and translates any remaining
403 to a friendly "skipped: recipe belongs to a different household —
sterilize from that household's account" message instead of dumping
Mealie's raw permission-denied JSON into the UI.

Net: Cobb sterilizes Redondo recipes from his account; Bay or mom
each get their own walk-and-apply scoped to Lake Elsinore when they
sign in. Same architecture works for new households joining Sulkta —
each member's bulk job is automatically scoped to the household they
belong to. Zero cross-pollution.

cauldron/bulk_sterilize.py

@@ -50,6 +50,36 @@ def _recipe_needs_sterilizing(recipe: dict) -> bool:
     return any(_ingredient_needs_sterilizing(i) for i in ings)
 
 
+def _recipe_household_id(recipe: dict) -> str | None:
+    """Mealie 1.x returns householdId at the top level. Older shapes used
+    nested household.id — try both."""
+    hid = recipe.get("householdId") or recipe.get("household_id")
+    if hid:
+        return hid
+    h = recipe.get("household")
+    if isinstance(h, dict):
+        return h.get("id")
+    return None
+
+
+def _user_household_id(mealie) -> str | None:
+    """Resolve the authenticated user's householdId from /api/users/self.
+    Cached on the Mealie client instance to avoid hitting the endpoint
+    once per recipe."""
+    cache_attr = "_cached_household_id"
+    cached = getattr(mealie, cache_attr, None)
+    if cached is not None:
+        return cached
+    me = mealie.who_am_i()
+    hid = me.get("householdId") or me.get("household_id")
+    if not hid:
+        h = me.get("household")
+        if isinstance(h, dict):
+            hid = h.get("id")
+    setattr(mealie, cache_attr, hid)
+    return hid
+
+
 def run_bulk_preview(
     *,
     db: DB,
@@ -60,6 +90,13 @@ def run_bulk_preview(
     Skip already-clean recipes. Move job state on completion."""
     log.info("[bulk-sterilize:%s] starting walk", job_id)
     try:
+        # Resolve the user's household once. The walk skips any recipe in a
+        # different household — Mealie's group model lets all members read
+        # everything but write only within their own household, so trying
+        # to sterilize a foreign recipe would 403 at apply time. Skipping
+        # at preview means no orphan proposals.
+        user_household = _user_household_id(sterilizer.mealie)
+
         # Pull every recipe slug (paginated). Mealie's listing returns
         # items with slug + name; we resolve full recipes one at a time.
         slugs: list[tuple[str, str]] = []
@@ -107,6 +144,16 @@ def run_bulk_preview(
                 )
                 continue
 
+            if user_household:
+                rec_hh = _recipe_household_id(recipe)
+                if rec_hh and rec_hh != user_household:
+                    # Different household within the group — read-only for
+                    # this user. Skip silently; no proposal row created.
+                    db.update_sterilize_job_progress(
+                        job_id, skipped_delta=1, current_slug=slug
+                    )
+                    continue
+
             if not _recipe_needs_sterilizing(recipe):
                 db.update_sterilize_job_progress(
                     job_id, skipped_delta=1, current_slug=slug
@@ -162,15 +209,34 @@ def run_bulk_apply(
     any per-recipe failure is recorded but doesn't stop the loop."""
     log.info("[bulk-sterilize:%s] starting apply", job_id)
     try:
+        user_household = _user_household_id(sterilizer.mealie)
         approved = db.list_approved_unapplied_proposals(job_id)
         for row in approved:
             slug = row["recipe_slug"]
             try:
                 db.update_sterilize_job_progress(job_id, current_slug=slug)
+                # Pre-check household: if this proposal was created before
+                # the walk-side filter (e.g. legacy job 1), Mealie would
+                # 403 on the PUT and the food/unit creates would have
+                # already polluted our own household. Guard early.
+                if user_household:
+                    rec = sterilizer.mealie.get_recipe(slug)
+                    rec_hh = _recipe_household_id(rec)
+                    if rec_hh and rec_hh != user_household:
+                        msg = "skipped: recipe belongs to a different household — sterilize from that household's account"
+                        db.mark_proposal_applied(job_id, slug, error=msg)
+                        db.update_sterilize_job_progress(
+                            job_id, error_delta=1, current_slug=slug, last_error=msg
+                        )
+                        continue
                 sterilizer.apply_recipe(slug, create_missing=True)
                 db.mark_proposal_applied(job_id, slug)
             except (ForgeError, RuntimeError, MealieError) as e:
-                msg = str(e)[:500]
+                raw = str(e)
+                if "403" in raw:
+                    msg = "skipped: recipe belongs to a different household — sterilize from that household's account"
+                else:
+                    msg = raw[:500]
                 log.warning("[bulk-sterilize:%s] apply(%s): %s", job_id, slug, msg)
                 db.mark_proposal_applied(job_id, slug, error=msg)
                 db.update_sterilize_job_progress(