Public Access

Files

T

admin c96b41ec26 feat(api): R3-A recipe engine — search, tags, family scope, recommendations

- GET /api/recipes: added cuisine, protein, dietary, ingredient, max_time,
  spice_max, calorie_max query params
- GET /api/recipes?family_profile_id=… hides never-suggest blocklist recipes
- GET /api/recipes/recommended: returns feedback-driven recipe suggestions
- Update docs: remove completed open items from planner-algorithm.md

2026-05-24 19:45:46 -07:00

61 KiB

Raw Blame History

Phase 9 — Meal-Planner Algorithm Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Generate a 3-dinner weekly meal plan via POST /api/admin/meal-plans/generate. Inputs: this week's grocery_item sale data, the family's recipes, blocklists, calorie target, and pantry. Output: a persisted MealPlan with 3 MealPlanItem rows ready for the existing email-approval flow.

Architecture: Two-stage filter-then-rank.

Filter removes recipes violating any of the 6 hard constraints (NeverSuggest, recency, calories, time, cost).
Score assigns each surviving recipe an individual score from 5 weighted signals (savings $, sale coverage %, pantry hit %, time bonus, recency bonus).
Select ranks the feasible set by individual score, takes top K=20, enumerates all C(20, 3) = 1,140 three-recipe combinations, and picks the combination with the highest combined score after a diversity penalty (protein/cuisine repeats).

Each stage is its own module with its own unit tests. The orchestrator wires them together and persists the result.

Tech Stack: FastAPI, SQLAlchemy 2.0, Pydantic v2, PostgreSQL 15, pytest. Pure-Python algorithm — no new dependencies.

Spec reference: docs/specs/2026-05-05-meal-planner-algorithm-design.md §2, §5.

Depends on: docs/superpowers/plans/2026-05-05-thin-phase-4-recipe-engine.md (must be complete first — Phase 9 reads ingredient_grocery_match and the seeded recipes).

File Structure

New files:

backend/app/services/planner/__init__.py — empty marker
backend/app/services/planner/config.py — weights, thresholds, K
backend/app/services/planner/types.py — dataclasses for ScoredRecipe, FilterResult, GenerationResult
backend/app/services/planner/filter.py — hard-constraint filter
backend/app/services/planner/score.py — individual recipe scoring
backend/app/services/planner/select.py — top-K set enumeration with diversity penalty
backend/app/services/planner/cost.py — recipe cost + savings estimation against ingredient_grocery_match
backend/app/services/planner/generate.py — orchestrator
backend/app/schemas/meal_plan_generation.py — Pydantic GenerateRequest/Response, RegenerateRequest, debug payload
backend/app/api/meal_plans.py — generate / regenerate / get endpoints (replaces or augments existing meals.py read paths)
backend/tests/test_planner_cost.py
backend/tests/test_planner_filter.py
backend/tests/test_planner_score.py
backend/tests/test_planner_select.py
backend/tests/test_planner_generate.py
backend/tests/test_meal_plan_generate_api.py

Modified files:

backend/app/main.py — wire new router
docs/ORIENTATION.md — phase 9 status
docs/HANDOFF.md — phase 9 status

Task 1: Planner config + types

Files:

Create: backend/app/services/planner/__init__.py
Create: backend/app/services/planner/config.py
Create: backend/app/services/planner/types.py
Step 1: Create the package marker

Create backend/app/services/planner/__init__.py with a single newline (empty).

Step 2: Write the config module

Create backend/app/services/planner/config.py:

"""Planner constants. Tune here without touching algorithm code."""
from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class PlannerConfig:
    # Hard constraints
    recency_weeks: int = 4              # constraint #3: no repeat within N weeks
    calorie_tolerance_pct: int = 20     # constraint #4: ±X% of family.calorie_target
    max_total_minutes: int = 45         # constraint #5: prep + cook
    max_meal_cost: float = 30.00        # constraint #6: dollars per meal

    # Scoring weights (must sum to 1.0)
    w_savings: float = 0.30
    w_coverage: float = 0.25
    w_pantry: float = 0.10
    w_time: float = 0.15
    w_recency: float = 0.20

    # Time bonus boundaries
    time_ideal_minutes: int = 25        # full bonus at <= this
    time_full_minutes: int = 45         # zero bonus at this; matches max_total_minutes

    # Recency bonus boundary
    recency_full_weeks: int = 12        # full bonus when last cooked >= this many weeks ago

    # Set selection
    top_k: int = 20                     # how many feasible recipes to enumerate over
    set_size: int = 3                   # 3 dinners/week
    p_protein: float = 0.15             # diversity penalty per shared-protein pair
    p_cuisine: float = 0.10             # diversity penalty per shared-cuisine pair

    def validate(self) -> None:
        total = self.w_savings + self.w_coverage + self.w_pantry + self.w_time + self.w_recency
        if abs(total - 1.0) > 1e-6:
            raise ValueError(f"weights must sum to 1.0, got {total}")


DEFAULT = PlannerConfig()
DEFAULT.validate()

Step 3: Write the types module

Create backend/app/services/planner/types.py:

"""Shared planner data structures."""
from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal
from typing import List, Optional
from uuid import UUID


@dataclass
class IngredientCost:
    ingredient_id: UUID
    qty: float
    unit: Optional[str]
    grocery_item_id: Optional[UUID]      # None when no match found
    grocery_item_name: Optional[str]
    current_price: Optional[Decimal]
    regular_price: Optional[Decimal]
    is_on_sale: bool
    estimated_cost: Decimal               # current_price scaled to recipe qty (best-effort)
    estimated_savings: Decimal            # max(regular - current, 0) scaled to qty
    matched: bool                         # False if no ingredient_grocery_match row found


@dataclass
class RecipeCost:
    recipe_id: UUID
    total_cost: Decimal
    total_savings: Decimal
    sale_ingredient_count: int
    matched_ingredient_count: int
    total_ingredient_count: int
    pantry_hit_count: int
    line_items: List[IngredientCost] = field(default_factory=list)

    @property
    def sale_coverage_pct(self) -> float:
        if self.total_ingredient_count == 0:
            return 0.0
        return self.sale_ingredient_count / self.total_ingredient_count

    @property
    def pantry_hit_pct(self) -> float:
        if self.total_ingredient_count == 0:
            return 0.0
        return self.pantry_hit_count / self.total_ingredient_count


@dataclass
class FilterResult:
    feasible_recipe_ids: List[UUID]
    rejected: dict                # recipe_id -> reason string


@dataclass
class ScoredRecipe:
    recipe_id: UUID
    score: float
    components: dict              # signal name -> raw value (for debug payload)
    cost: RecipeCost
    protein: Optional[str]
    cuisine: Optional[str]


@dataclass
class GenerationResult:
    meal_plan_id: UUID
    selected: List[ScoredRecipe]
    feasible_count: int
    rejected_summary: dict        # reason -> count
    set_score: float

Step 4: Smoke test imports

docker compose --env-file .env.test exec backend python -c "
from app.services.planner.config import DEFAULT
from app.services.planner.types import RecipeCost, ScoredRecipe
print('weights ok')
"

Expected: weights ok.

Step 5: Commit

git add backend/app/services/planner/__init__.py backend/app/services/planner/config.py backend/app/services/planner/types.py
git commit -m "feat: planner config (weights, thresholds, K) and shared types"

Task 2: Cost + savings estimator

Files:

Create: backend/app/services/planner/cost.py
Test: backend/tests/test_planner_cost.py
Step 1: Write failing tests

Create backend/tests/test_planner_cost.py:

from decimal import Decimal
from uuid import uuid4

import pytest

from app.services.planner.cost import compute_recipe_cost
from app.services.planner.types import RecipeCost


def _ingredient(ing_id, qty=1.0, unit=None):
    return {"ingredient_id": ing_id, "qty": qty, "unit": unit}


def _match(grocery_id, name, current, regular, is_on_sale, confidence=0.9):
    return {
        "grocery_item_id": grocery_id,
        "grocery_item_name": name,
        "current_price": Decimal(str(current)),
        "regular_price": Decimal(str(regular)),
        "is_on_sale": is_on_sale,
        "confidence": Decimal(str(confidence)),
    }


def test_compute_recipe_cost_sums_ingredient_costs():
    recipe_id = uuid4()
    i1, i2 = uuid4(), uuid4()
    g1, g2 = uuid4(), uuid4()
    matches = {
        i1: [_match(g1, "Chicken", current=4.00, regular=5.00, is_on_sale=True)],
        i2: [_match(g2, "Olive Oil", current=8.00, regular=8.00, is_on_sale=False)],
    }
    pantry_ingredient_ids = set()

    result = compute_recipe_cost(
        recipe_id=recipe_id,
        ingredients=[_ingredient(i1, qty=2.0), _ingredient(i2, qty=1.0)],
        match_index=matches,
        pantry_ingredient_ids=pantry_ingredient_ids,
    )

    assert isinstance(result, RecipeCost)
    assert result.recipe_id == recipe_id
    assert result.total_ingredient_count == 2
    assert result.matched_ingredient_count == 2
    assert result.sale_ingredient_count == 1
    assert result.total_cost == Decimal("16.00")  # 2*4 + 1*8
    assert result.total_savings == Decimal("2.00")  # 2*(5-4)


def test_unmatched_ingredient_zero_cost_and_savings():
    recipe_id = uuid4()
    i1 = uuid4()
    result = compute_recipe_cost(
        recipe_id=recipe_id,
        ingredients=[_ingredient(i1, qty=1.0)],
        match_index={},
        pantry_ingredient_ids=set(),
    )
    assert result.matched_ingredient_count == 0
    assert result.total_cost == Decimal("0.00")
    assert result.total_savings == Decimal("0.00")


def test_pantry_hits_counted():
    recipe_id = uuid4()
    i1, i2 = uuid4(), uuid4()
    g1, g2 = uuid4(), uuid4()
    matches = {
        i1: [_match(g1, "X", current=2.00, regular=2.00, is_on_sale=False)],
        i2: [_match(g2, "Y", current=3.00, regular=3.00, is_on_sale=False)],
    }
    result = compute_recipe_cost(
        recipe_id=recipe_id,
        ingredients=[_ingredient(i1), _ingredient(i2)],
        match_index=matches,
        pantry_ingredient_ids={i1},
    )
    assert result.pantry_hit_count == 1
    assert result.pantry_hit_pct == 0.5

Step 2: Run test, expect failure

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_cost.py -v

Expected: ImportError on app.services.planner.cost.

Step 3: Implement the cost estimator

Create backend/app/services/planner/cost.py:

"""Compute per-recipe cost and savings against ingredient_grocery_match.

Inputs:
    ingredients: list[dict] from recipe.ingredients JSONB
    match_index: dict[ingredient_id, list[match_dict]] — pre-fetched, sorted by confidence DESC
    pantry_ingredient_ids: set of ingredient_ids in home_pantry

Strategy:
    For each ingredient in the recipe, take the top-confidence match
    (or skip if none). Cost = current_price * qty (best-effort scaling
    that ignores unit conversion — see Limitations below).
    Savings = max(regular - current, 0) * qty.

Limitations:
    Unit conversion (lb vs oz, cup vs ml) is NOT implemented in this
    pass. The qty multiplier is treated as dimensionless. This produces
    a biased-but-monotonic ranking signal: recipes that use more of an
    expensive ingredient still rank as more expensive, which is what
    the planner needs. Real dollar accuracy can come later.
"""
from __future__ import annotations

from decimal import Decimal
from typing import Dict, Iterable, List, Set
from uuid import UUID

from app.services.planner.types import IngredientCost, RecipeCost


def _decimal(v) -> Decimal:
    if v is None:
        return Decimal("0")
    return v if isinstance(v, Decimal) else Decimal(str(v))


def _scale(price: Decimal, qty: float) -> Decimal:
    return (price * Decimal(str(qty))).quantize(Decimal("0.01"))


def compute_recipe_cost(
    *,
    recipe_id: UUID,
    ingredients: Iterable[dict],
    match_index: Dict[UUID, List[dict]],
    pantry_ingredient_ids: Set[UUID],
) -> RecipeCost:
    line_items: List[IngredientCost] = []
    total_cost = Decimal("0.00")
    total_savings = Decimal("0.00")
    sale_count = 0
    matched_count = 0
    pantry_hits = 0
    total = 0

    for raw in ingredients:
        total += 1
        ing_id = raw["ingredient_id"]
        if isinstance(ing_id, str):
            ing_id = UUID(ing_id)
        qty = float(raw.get("qty") or 1.0)
        unit = raw.get("unit")

        if ing_id in pantry_ingredient_ids:
            pantry_hits += 1

        candidates = match_index.get(ing_id) or []
        if not candidates:
            line_items.append(
                IngredientCost(
                    ingredient_id=ing_id,
                    qty=qty,
                    unit=unit,
                    grocery_item_id=None,
                    grocery_item_name=None,
                    current_price=None,
                    regular_price=None,
                    is_on_sale=False,
                    estimated_cost=Decimal("0.00"),
                    estimated_savings=Decimal("0.00"),
                    matched=False,
                )
            )
            continue

        best = candidates[0]
        current = _decimal(best.get("current_price"))
        regular = _decimal(best.get("regular_price"))
        is_on_sale = bool(best.get("is_on_sale"))
        line_cost = _scale(current, qty)
        line_savings = _scale(max(regular - current, Decimal("0")), qty)

        matched_count += 1
        if is_on_sale:
            sale_count += 1
        total_cost += line_cost
        total_savings += line_savings

        line_items.append(
            IngredientCost(
                ingredient_id=ing_id,
                qty=qty,
                unit=unit,
                grocery_item_id=best.get("grocery_item_id"),
                grocery_item_name=best.get("grocery_item_name"),
                current_price=current,
                regular_price=regular,
                is_on_sale=is_on_sale,
                estimated_cost=line_cost,
                estimated_savings=line_savings,
                matched=True,
            )
        )

    return RecipeCost(
        recipe_id=recipe_id,
        total_cost=total_cost.quantize(Decimal("0.01")),
        total_savings=total_savings.quantize(Decimal("0.01")),
        sale_ingredient_count=sale_count,
        matched_ingredient_count=matched_count,
        total_ingredient_count=total,
        pantry_hit_count=pantry_hits,
        line_items=line_items,
    )

Step 4: Run tests, expect pass

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_cost.py -v

Expected: 3 passed.

Step 5: Commit

git add backend/app/services/planner/cost.py backend/tests/test_planner_cost.py
git commit -m "feat: planner cost+savings estimator against ingredient_grocery_match"

Task 3: Hard-constraint filter

Files:

Create: backend/app/services/planner/filter.py
Test: backend/tests/test_planner_filter.py
Step 1: Write failing tests

Create backend/tests/test_planner_filter.py:

from datetime import date, datetime, timezone
from decimal import Decimal
from uuid import uuid4

import pytest

from app.services.planner.config import PlannerConfig
from app.services.planner.filter import filter_recipes
from app.services.planner.types import RecipeCost


_CFG = PlannerConfig()


def _recipe(
    recipe_id=None,
    name="r",
    prep=10,
    cook=20,
    calories=2000,
    protein="chicken",
    cuisine_tags=("american",),
):
    return {
        "id": recipe_id or uuid4(),
        "name": name,
        "prep_time_minutes": prep,
        "cook_time_minutes": cook,
        "calories_per_serving": calories,
        "protein_type": protein,
        "cuisine_tags": list(cuisine_tags),
    }


def _cost(recipe_id, total_cost):
    return RecipeCost(
        recipe_id=recipe_id,
        total_cost=Decimal(str(total_cost)),
        total_savings=Decimal("0"),
        sale_ingredient_count=0,
        matched_ingredient_count=1,
        total_ingredient_count=1,
        pantry_hit_count=0,
        line_items=[],
    )


def test_filter_blocks_by_never_suggest_ingredient():
    r1 = _recipe(name="has-mushrooms")
    r2 = _recipe(name="clean")
    blocked_ingredient = uuid4()
    recipe_ingredient_ids = {r1["id"]: {blocked_ingredient}, r2["id"]: {uuid4()}}

    result = filter_recipes(
        recipes=[r1, r2],
        recipe_ingredient_ids=recipe_ingredient_ids,
        recipe_costs={r1["id"]: _cost(r1["id"], 10), r2["id"]: _cost(r2["id"], 10)},
        blocked_ingredient_ids={blocked_ingredient},
        blocked_recipe_ids=set(),
        last_cooked_at={},
        family_calorie_target=2000,
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert r2["id"] in result.feasible_recipe_ids
    assert r1["id"] not in result.feasible_recipe_ids
    assert "blocked_ingredient" in result.rejected[r1["id"]]


def test_filter_blocks_by_never_suggest_recipe():
    r1 = _recipe()
    result = filter_recipes(
        recipes=[r1],
        recipe_ingredient_ids={r1["id"]: set()},
        recipe_costs={r1["id"]: _cost(r1["id"], 10)},
        blocked_ingredient_ids=set(),
        blocked_recipe_ids={r1["id"]},
        last_cooked_at={},
        family_calorie_target=2000,
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert r1["id"] not in result.feasible_recipe_ids
    assert "blocked_recipe" in result.rejected[r1["id"]]


def test_filter_blocks_by_recency():
    r1 = _recipe()
    last_week = date(2026, 5, 5) - __import__("datetime").timedelta(days=7)
    result = filter_recipes(
        recipes=[r1],
        recipe_ingredient_ids={r1["id"]: set()},
        recipe_costs={r1["id"]: _cost(r1["id"], 10)},
        blocked_ingredient_ids=set(),
        blocked_recipe_ids=set(),
        last_cooked_at={r1["id"]: last_week},
        family_calorie_target=2000,
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert "recency" in result.rejected[r1["id"]]


def test_filter_blocks_by_calories():
    over = _recipe(calories=3500)  # 75% over a 2000 target
    under = _recipe(calories=2200)
    result = filter_recipes(
        recipes=[over, under],
        recipe_ingredient_ids={over["id"]: set(), under["id"]: set()},
        recipe_costs={over["id"]: _cost(over["id"], 10), under["id"]: _cost(under["id"], 10)},
        blocked_ingredient_ids=set(),
        blocked_recipe_ids=set(),
        last_cooked_at={},
        family_calorie_target=2000,
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert under["id"] in result.feasible_recipe_ids
    assert "calories" in result.rejected[over["id"]]


def test_filter_blocks_by_time():
    slow = _recipe(prep=30, cook=20)  # 50 min total
    fast = _recipe(prep=10, cook=20)
    result = filter_recipes(
        recipes=[slow, fast],
        recipe_ingredient_ids={slow["id"]: set(), fast["id"]: set()},
        recipe_costs={slow["id"]: _cost(slow["id"], 10), fast["id"]: _cost(fast["id"], 10)},
        blocked_ingredient_ids=set(),
        blocked_recipe_ids=set(),
        last_cooked_at={},
        family_calorie_target=2000,
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert "time" in result.rejected[slow["id"]]
    assert fast["id"] in result.feasible_recipe_ids


def test_filter_blocks_by_cost():
    pricey = _recipe()
    cheap = _recipe()
    result = filter_recipes(
        recipes=[pricey, cheap],
        recipe_ingredient_ids={pricey["id"]: set(), cheap["id"]: set()},
        recipe_costs={pricey["id"]: _cost(pricey["id"], 50), cheap["id"]: _cost(cheap["id"], 25)},
        blocked_ingredient_ids=set(),
        blocked_recipe_ids=set(),
        last_cooked_at={},
        family_calorie_target=2000,
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert "cost" in result.rejected[pricey["id"]]
    assert cheap["id"] in result.feasible_recipe_ids


def test_filter_passes_when_calorie_target_is_none():
    r = _recipe(calories=9999)  # absurd, but no target → can't enforce
    result = filter_recipes(
        recipes=[r],
        recipe_ingredient_ids={r["id"]: set()},
        recipe_costs={r["id"]: _cost(r["id"], 10)},
        blocked_ingredient_ids=set(),
        blocked_recipe_ids=set(),
        last_cooked_at={},
        family_calorie_target=None,
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert r["id"] in result.feasible_recipe_ids


def test_filter_blocks_recipe_with_null_calories_when_target_set():
    r = _recipe(calories=None)
    result = filter_recipes(
        recipes=[r],
        recipe_ingredient_ids={r["id"]: set()},
        recipe_costs={r["id"]: _cost(r["id"], 10)},
        blocked_ingredient_ids=set(),
        blocked_recipe_ids=set(),
        last_cooked_at={},
        family_calorie_target=2000,
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert "calories" in result.rejected[r["id"]]

Step 2: Run test, expect failure

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_filter.py -v

Expected: ImportError on app.services.planner.filter.

Step 3: Implement filter

Create backend/app/services/planner/filter.py:

"""Hard-constraint filter (constraints #1–#6 from the spec)."""
from __future__ import annotations

from datetime import date, timedelta
from typing import Dict, Iterable, List, Optional, Set
from uuid import UUID

from app.services.planner.config import PlannerConfig
from app.services.planner.types import FilterResult, RecipeCost


def _calorie_band(target: int, pct: int) -> tuple[int, int]:
    delta = target * pct / 100.0
    return (int(target - delta), int(target + delta))


def filter_recipes(
    *,
    recipes: Iterable[dict],
    recipe_ingredient_ids: Dict[UUID, Set[UUID]],
    recipe_costs: Dict[UUID, RecipeCost],
    blocked_ingredient_ids: Set[UUID],
    blocked_recipe_ids: Set[UUID],
    last_cooked_at: Dict[UUID, date],
    family_calorie_target: Optional[int],
    config: PlannerConfig,
    today: date,
) -> FilterResult:
    feasible: List[UUID] = []
    rejected: Dict[UUID, str] = {}

    cal_band: Optional[tuple[int, int]] = None
    if family_calorie_target is not None:
        cal_band = _calorie_band(family_calorie_target, config.calorie_tolerance_pct)

    recency_cutoff = today - timedelta(weeks=config.recency_weeks)

    for r in recipes:
        rid = r["id"]
        if isinstance(rid, str):
            rid = UUID(rid)

        # #2: per-recipe blocklist
        if rid in blocked_recipe_ids:
            rejected[rid] = "blocked_recipe"
            continue

        # #1: ingredient blocklist intersection
        if blocked_ingredient_ids & recipe_ingredient_ids.get(rid, set()):
            rejected[rid] = "blocked_ingredient"
            continue

        # #3: recency
        last = last_cooked_at.get(rid)
        if last and last >= recency_cutoff:
            rejected[rid] = "recency"
            continue

        # #4: calories
        if cal_band is not None:
            cps = r.get("calories_per_serving")
            if cps is None:
                rejected[rid] = "calories_missing"
                continue
            lo, hi = cal_band
            if cps < lo or cps > hi:
                rejected[rid] = "calories"
                continue

        # #5: time
        prep = int(r.get("prep_time_minutes") or 0)
        cook = int(r.get("cook_time_minutes") or 0)
        if prep + cook > config.max_total_minutes:
            rejected[rid] = "time"
            continue

        # #6: cost
        cost = recipe_costs.get(rid)
        if cost is None:
            rejected[rid] = "no_cost"
            continue
        if float(cost.total_cost) > config.max_meal_cost:
            rejected[rid] = "cost"
            continue

        feasible.append(rid)

    return FilterResult(feasible_recipe_ids=feasible, rejected=rejected)

(Note: the test's __import__("datetime") workaround is unnecessary cleanup. The test should import datetime at the top instead. Implementer may clean up.)

Step 4: Run tests, expect pass

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_filter.py -v

Expected: 8 passed.

Step 5: Commit

git add backend/app/services/planner/filter.py backend/tests/test_planner_filter.py
git commit -m "feat: planner hard-constraint filter for the 6 spec constraints"

Task 4: Individual scoring

Files:

Create: backend/app/services/planner/score.py
Test: backend/tests/test_planner_score.py
Step 1: Write failing tests

Create backend/tests/test_planner_score.py:

from datetime import date, timedelta
from decimal import Decimal
from uuid import uuid4

from app.services.planner.config import PlannerConfig
from app.services.planner.score import (
    score_recipes,
    time_bonus,
    recency_bonus,
    normalize_savings,
)
from app.services.planner.types import RecipeCost


_CFG = PlannerConfig()


def test_time_bonus_capped_at_ideal():
    assert time_bonus(20, _CFG) == 1.0
    assert time_bonus(25, _CFG) == 1.0


def test_time_bonus_decays_to_zero_at_full():
    assert time_bonus(45, _CFG) == 0.0


def test_time_bonus_linear_midpoint():
    # 35 min is halfway between 25 and 45
    assert abs(time_bonus(35, _CFG) - 0.5) < 1e-6


def test_recency_bonus_full_when_long_ago():
    today = date(2026, 5, 5)
    long_ago = today - timedelta(weeks=20)
    assert recency_bonus(long_ago, today, _CFG) == 1.0


def test_recency_bonus_zero_when_just_eligible():
    # right at recency_weeks boundary → 0
    today = date(2026, 5, 5)
    cutoff = today - timedelta(weeks=_CFG.recency_weeks)
    assert recency_bonus(cutoff, today, _CFG) == 0.0


def test_recency_bonus_full_when_never_cooked():
    assert recency_bonus(None, date(2026, 5, 5), _CFG) == 1.0


def test_normalize_savings_minmax():
    out = normalize_savings([Decimal("0"), Decimal("5"), Decimal("10")])
    assert out == [0.0, 0.5, 1.0]


def test_normalize_savings_uniform_returns_zeros():
    out = normalize_savings([Decimal("3"), Decimal("3"), Decimal("3")])
    assert out == [0.0, 0.0, 0.0]


def test_score_recipes_orders_by_combined_score():
    recipes = [
        {
            "id": uuid4(),
            "name": "low_savings",
            "prep_time_minutes": 10,
            "cook_time_minutes": 30,
            "protein_type": "chicken",
            "cuisine_tags": ["american"],
        },
        {
            "id": uuid4(),
            "name": "high_savings",
            "prep_time_minutes": 10,
            "cook_time_minutes": 15,  # also lower time
            "protein_type": "beef",
            "cuisine_tags": ["mexican"],
        },
    ]
    costs = {
        recipes[0]["id"]: RecipeCost(
            recipe_id=recipes[0]["id"],
            total_cost=Decimal("10"),
            total_savings=Decimal("1"),
            sale_ingredient_count=1,
            matched_ingredient_count=4,
            total_ingredient_count=4,
            pantry_hit_count=0,
            line_items=[],
        ),
        recipes[1]["id"]: RecipeCost(
            recipe_id=recipes[1]["id"],
            total_cost=Decimal("12"),
            total_savings=Decimal("8"),
            sale_ingredient_count=3,
            matched_ingredient_count=4,
            total_ingredient_count=4,
            pantry_hit_count=2,
            line_items=[],
        ),
    }
    scored = score_recipes(
        recipes=recipes,
        recipe_costs=costs,
        last_cooked_at={},
        config=_CFG,
        today=date(2026, 5, 5),
    )
    assert scored[0].recipe_id == recipes[1]["id"]  # high_savings first

Step 2: Run test, expect failure

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_score.py -v

Expected: ImportError.

Step 3: Implement scoring

Create backend/app/services/planner/score.py:

"""Per-recipe scoring with the 5 weighted signals from the spec."""
from __future__ import annotations

from datetime import date
from decimal import Decimal
from typing import Dict, Iterable, List, Optional
from uuid import UUID

from app.services.planner.config import PlannerConfig
from app.services.planner.types import RecipeCost, ScoredRecipe


def time_bonus(total_minutes: int, config: PlannerConfig) -> float:
    if total_minutes <= config.time_ideal_minutes:
        return 1.0
    if total_minutes >= config.time_full_minutes:
        return 0.0
    span = config.time_full_minutes - config.time_ideal_minutes
    over = total_minutes - config.time_ideal_minutes
    return max(0.0, min(1.0, 1.0 - over / span))


def recency_bonus(
    last_cooked: Optional[date],
    today: date,
    config: PlannerConfig,
) -> float:
    if last_cooked is None:
        return 1.0
    weeks_ago = (today - last_cooked).days / 7
    if weeks_ago >= config.recency_full_weeks:
        return 1.0
    # Below recency_weeks the recipe wouldn't be in the feasible set, so we treat
    # exactly recency_weeks as score 0 and recency_full_weeks as score 1.
    if weeks_ago <= config.recency_weeks:
        return 0.0
    span = config.recency_full_weeks - config.recency_weeks
    over = weeks_ago - config.recency_weeks
    return max(0.0, min(1.0, over / span))


def normalize_savings(values: List[Decimal]) -> List[float]:
    if not values:
        return []
    floats = [float(v) for v in values]
    lo, hi = min(floats), max(floats)
    if hi == lo:
        return [0.0] * len(floats)
    return [(v - lo) / (hi - lo) for v in floats]


def score_recipes(
    *,
    recipes: Iterable[dict],
    recipe_costs: Dict[UUID, RecipeCost],
    last_cooked_at: Dict[UUID, date],
    config: PlannerConfig,
    today: date,
) -> List[ScoredRecipe]:
    """Returns recipes scored DESC. Caller passes only the feasible set."""
    materialized = list(recipes)
    if not materialized:
        return []

    savings = [recipe_costs[r["id"]].total_savings for r in materialized]
    norm_savings = normalize_savings(savings)

    out: List[ScoredRecipe] = []
    for r, ns in zip(materialized, norm_savings):
        rid = r["id"]
        if isinstance(rid, str):
            rid = UUID(rid)
        cost = recipe_costs[rid]
        total_min = int(r.get("prep_time_minutes") or 0) + int(r.get("cook_time_minutes") or 0)
        tb = time_bonus(total_min, config)
        rb = recency_bonus(last_cooked_at.get(rid), today, config)

        components = {
            "savings_normalized": ns,
            "sale_coverage_pct": cost.sale_coverage_pct,
            "pantry_hit_pct": cost.pantry_hit_pct,
            "time_bonus": tb,
            "recency_bonus": rb,
            "savings_dollars": float(cost.total_savings),
        }

        score = (
            config.w_savings * ns
            + config.w_coverage * cost.sale_coverage_pct
            + config.w_pantry * cost.pantry_hit_pct
            + config.w_time * tb
            + config.w_recency * rb
        )

        cuisine_tags = r.get("cuisine_tags") or []
        primary_cuisine = cuisine_tags[0] if cuisine_tags else None

        out.append(
            ScoredRecipe(
                recipe_id=rid,
                score=score,
                components=components,
                cost=cost,
                protein=r.get("protein_type"),
                cuisine=primary_cuisine,
            )
        )

    out.sort(key=lambda s: s.score, reverse=True)
    return out

Step 4: Run tests, expect pass

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_score.py -v

Expected: 9 passed.

Step 5: Commit

git add backend/app/services/planner/score.py backend/tests/test_planner_score.py
git commit -m "feat: planner per-recipe scoring with 5 weighted signals"

Task 5: Top-K set selection with diversity

Files:

Create: backend/app/services/planner/select.py
Test: backend/tests/test_planner_select.py
Step 1: Write failing tests

Create backend/tests/test_planner_select.py:

from decimal import Decimal
from uuid import uuid4

from app.services.planner.config import PlannerConfig
from app.services.planner.select import select_set, set_diversity_penalty
from app.services.planner.types import RecipeCost, ScoredRecipe


_CFG = PlannerConfig()


def _mk(score, protein, cuisine):
    rid = uuid4()
    return ScoredRecipe(
        recipe_id=rid,
        score=score,
        components={},
        cost=RecipeCost(
            recipe_id=rid,
            total_cost=Decimal("10"),
            total_savings=Decimal("1"),
            sale_ingredient_count=1,
            matched_ingredient_count=1,
            total_ingredient_count=1,
            pantry_hit_count=0,
            line_items=[],
        ),
        protein=protein,
        cuisine=cuisine,
    )


def test_diversity_penalty_zero_when_all_unique():
    a = _mk(0.5, "chicken", "american")
    b = _mk(0.5, "beef", "mexican")
    c = _mk(0.5, "fish", "italian")
    assert set_diversity_penalty([a, b, c], _CFG) == 0.0


def test_diversity_penalty_three_chickens():
    a = _mk(0.5, "chicken", "american")
    b = _mk(0.5, "chicken", "italian")
    c = _mk(0.5, "chicken", "mexican")
    # 3 protein pairs * 0.15 = 0.45, no cuisine pairs
    p = set_diversity_penalty([a, b, c], _CFG)
    assert abs(p - 0.45) < 1e-6


def test_select_set_picks_diverse_over_homogeneous():
    """Three chicken/american (each 0.95) lose to mixed (each 0.90)
    once protein and cuisine penalties apply."""
    high1 = _mk(0.95, "chicken", "american")
    high2 = _mk(0.95, "chicken", "american")
    high3 = _mk(0.95, "chicken", "american")
    mix1 = _mk(0.90, "chicken", "american")
    mix2 = _mk(0.90, "beef", "mexican")
    mix3 = _mk(0.90, "fish", "italian")

    chosen, set_score = select_set([high1, high2, high3, mix1, mix2, mix3], _CFG)
    chosen_ids = {s.recipe_id for s in chosen}
    assert chosen_ids == {mix1.recipe_id, mix2.recipe_id, mix3.recipe_id}


def test_select_set_handles_too_few():
    a = _mk(0.5, "x", "y")
    b = _mk(0.5, "x", "y")
    chosen, _ = select_set([a, b], _CFG)
    assert len(chosen) == 2  # less than set_size returns what we have


def test_select_set_empty_input():
    chosen, score = select_set([], _CFG)
    assert chosen == []
    assert score == 0.0

Step 2: Run test, expect failure

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_select.py -v

Expected: ImportError.

Step 3: Implement set selection

Create backend/app/services/planner/select.py:

"""Top-K set enumeration with diversity penalty.

Takes the top K (=20) scored recipes, enumerates all C(K, set_size)
combinations, applies a pairwise diversity penalty for shared
protein and cuisine, and returns the highest-scoring combination.
"""
from __future__ import annotations

from itertools import combinations
from typing import Iterable, List, Tuple

from app.services.planner.config import PlannerConfig
from app.services.planner.types import ScoredRecipe


def set_diversity_penalty(
    chosen: List[ScoredRecipe],
    config: PlannerConfig,
) -> float:
    penalty = 0.0
    for a, b in combinations(chosen, 2):
        if a.protein and b.protein and a.protein == b.protein:
            penalty += config.p_protein
        if a.cuisine and b.cuisine and a.cuisine == b.cuisine:
            penalty += config.p_cuisine
    return penalty


def _set_score(chosen: List[ScoredRecipe], config: PlannerConfig) -> float:
    return sum(s.score for s in chosen) - set_diversity_penalty(chosen, config)


def select_set(
    scored: Iterable[ScoredRecipe],
    config: PlannerConfig,
) -> Tuple[List[ScoredRecipe], float]:
    pool = list(scored)
    if not pool:
        return [], 0.0
    if len(pool) <= config.set_size:
        return pool, _set_score(pool, config)

    pool.sort(key=lambda s: s.score, reverse=True)
    candidate_pool = pool[: config.top_k]

    best: List[ScoredRecipe] = []
    best_score = float("-inf")
    for combo in combinations(candidate_pool, config.set_size):
        s = _set_score(list(combo), config)
        if s > best_score:
            best_score = s
            best = list(combo)
    return best, best_score

Step 4: Run tests, expect pass

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_select.py -v

Expected: 5 passed.

Step 5: Commit

git add backend/app/services/planner/select.py backend/tests/test_planner_select.py
git commit -m "feat: planner top-K set enumeration with protein/cuisine diversity penalty"

Task 6: Generation orchestrator

Files:

Create: backend/app/services/planner/generate.py
Test: backend/tests/test_planner_generate.py
Step 1: Write failing test

Create backend/tests/test_planner_generate.py:

from datetime import date
from decimal import Decimal
from uuid import UUID, uuid4

import pytest

pytestmark = pytest.mark.requires_postgres


def test_generate_meal_plan_against_seeded_data():
    """Smoke test: with the seeded 30 recipes and a seeded grocery scrape,
    generate produces a MealPlan with 3 items.
    """
    from datetime import datetime, timezone

    from app.database import SessionLocal
    from app.models import (
        FamilyProfile,
        GroceryItem,
        Ingredient,
        IngredientGroceryMatch,
        IngredientMatchSource,
        MealPlan,
        MealPlanItem,
    )
    from app.services.matcher import run_match_job
    from app.services.planner.generate import generate_meal_plan

    setup = SessionLocal()
    try:
        family = FamilyProfile(
            id=uuid4(),
            name="Generate Smoke Family",
            household_size=4,
            adult_count=2,
            child_count=2,
            calorie_target=2400,
        )
        setup.add(family)
        setup.commit()
        family_id = family.id

        # Seed at least one grocery_item per ingredient that the recipes use
        # so cost matching can find prices. Pick a handful to keep runtime low.
        for ing in setup.query(Ingredient).limit(20).all():
            setup.add(
                GroceryItem(
                    id=uuid4(),
                    name=ing.name,
                    source="lucky_california",
                    external_id=f"ext-test-{ing.id}",
                    current_price=Decimal("3.99"),
                    regular_price=Decimal("4.99"),
                    is_on_sale=True,
                    scraped_at=datetime.now(timezone.utc),
                )
            )
        setup.commit()
    finally:
        setup.close()

    work = SessionLocal()
    try:
        run_match_job(work)
        result = generate_meal_plan(work, family_id=family_id, week_start_date=date(2026, 5, 11))
        assert result.meal_plan_id is not None
        assert 1 <= len(result.selected) <= 3  # at least 1, at most 3
        plan = work.query(MealPlan).filter(MealPlan.id == result.meal_plan_id).first()
        assert plan is not None
        items = work.query(MealPlanItem).filter(MealPlanItem.meal_plan_id == plan.id).all()
        assert len(items) == len(result.selected)
    finally:
        cleanup = SessionLocal()
        try:
            cleanup.query(MealPlanItem).filter(
                MealPlanItem.meal_plan_id.in_(
                    [r[0] for r in cleanup.query(MealPlan.id).filter(MealPlan.family_profile_id == family_id).all()]
                )
            ).delete(synchronize_session=False)
            cleanup.query(MealPlan).filter(MealPlan.family_profile_id == family_id).delete(synchronize_session=False)
            cleanup.query(FamilyProfile).filter(FamilyProfile.id == family_id).delete(synchronize_session=False)
            cleanup.query(GroceryItem).filter(GroceryItem.external_id.like("ext-test-%")).delete(synchronize_session=False)
            cleanup.query(IngredientGroceryMatch).filter(
                IngredientGroceryMatch.source == IngredientMatchSource.AUTO
            ).delete(synchronize_session=False)
            cleanup.commit()
        finally:
            cleanup.close()
        work.close()

Step 2: Run test, expect failure

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_generate.py -v

Expected: ImportError on app.services.planner.generate.

Step 3: Implement orchestrator

Create backend/app/services/planner/generate.py:

"""End-to-end planner orchestration: load → filter → score → select → persist."""
from __future__ import annotations

from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Dict, List, Optional, Set
from uuid import UUID

from sqlalchemy.orm import Session

from app.models import (
    FamilyProfile,
    GroceryItem,
    HomePantry,
    Ingredient,
    IngredientGroceryMatch,
    IngredientMatchSource,
    MealPlan,
    MealPlanItem,
    MealPlanItemStatus,
    MealPlanStatus,
    MealType,
    NeverSuggest,
    Recipe,
)
from app.services.planner.config import DEFAULT, PlannerConfig
from app.services.planner.cost import compute_recipe_cost
from app.services.planner.filter import filter_recipes
from app.services.planner.score import score_recipes
from app.services.planner.select import select_set
from app.services.planner.types import GenerationResult


def _load_match_index(db: Session) -> Dict[UUID, List[dict]]:
    """ingredient_id → list of match dicts ordered by confidence DESC."""
    rows = (
        db.query(IngredientGroceryMatch, GroceryItem)
        .join(GroceryItem, GroceryItem.id == IngredientGroceryMatch.grocery_item_id)
        .order_by(IngredientGroceryMatch.confidence.desc())
        .all()
    )
    index: Dict[UUID, List[dict]] = {}
    for match, grocery in rows:
        index.setdefault(match.ingredient_id, []).append(
            {
                "grocery_item_id": grocery.id,
                "grocery_item_name": grocery.name,
                "current_price": grocery.current_price,
                "regular_price": grocery.regular_price,
                "is_on_sale": bool(grocery.is_on_sale),
                "confidence": match.confidence,
            }
        )
    return index


def _load_blocklists(
    db: Session, family_id: UUID
) -> tuple[Set[UUID], Set[UUID]]:
    blocked_ingredients: Set[UUID] = set()
    blocked_recipes: Set[UUID] = set()
    for row in db.query(NeverSuggest).filter(NeverSuggest.family_profile_id == family_id).all():
        if row.ingredient_id is not None:
            blocked_ingredients.add(row.ingredient_id)
        if row.recipe_id is not None:
            blocked_recipes.add(row.recipe_id)
    return blocked_ingredients, blocked_recipes


def _load_pantry(db: Session, family_id: UUID) -> Set[UUID]:
    return {
        row.ingredient_id
        for row in db.query(HomePantry).filter(HomePantry.family_profile_id == family_id).all()
        if row.ingredient_id is not None
    }


def _load_last_cooked(db: Session, family_id: UUID) -> Dict[UUID, date]:
    rows = (
        db.query(MealPlanItem, MealPlan)
        .join(MealPlan, MealPlan.id == MealPlanItem.meal_plan_id)
        .filter(MealPlan.family_profile_id == family_id)
        .all()
    )
    last: Dict[UUID, date] = {}
    for item, plan in rows:
        if item.recipe_id is None:
            continue
        if item.recipe_id not in last or plan.week_start_date > last[item.recipe_id]:
            last[item.recipe_id] = plan.week_start_date
    return last


def generate_meal_plan(
    db: Session,
    *,
    family_id: UUID,
    week_start_date: date,
    config: PlannerConfig = DEFAULT,
    today: Optional[date] = None,
) -> GenerationResult:
    today = today or date.today()
    family = db.query(FamilyProfile).filter(FamilyProfile.id == family_id).first()
    if family is None:
        raise ValueError(f"family_profile {family_id} not found")

    recipes = db.query(Recipe).all()
    recipe_dicts = [
        {
            "id": r.id,
            "name": r.name,
            "prep_time_minutes": r.prep_time_minutes,
            "cook_time_minutes": r.cook_time_minutes,
            "calories_per_serving": r.calories_per_serving,
            "protein_type": r.protein_type,
            "cuisine_tags": list(r.cuisine_tags or []),
            "ingredients": list(r.ingredients or []),
        }
        for r in recipes
    ]

    recipe_ingredient_ids: Dict[UUID, Set[UUID]] = {}
    for r in recipe_dicts:
        ids: Set[UUID] = set()
        for line in r["ingredients"]:
            ing_id = line.get("ingredient_id")
            if isinstance(ing_id, str):
                ing_id = UUID(ing_id)
            if ing_id is not None:
                ids.add(ing_id)
        recipe_ingredient_ids[r["id"]] = ids

    match_index = _load_match_index(db)
    pantry_ids = _load_pantry(db, family_id)
    blocked_ings, blocked_recipes = _load_blocklists(db, family_id)
    last_cooked = _load_last_cooked(db, family_id)

    recipe_costs = {
        r["id"]: compute_recipe_cost(
            recipe_id=r["id"],
            ingredients=r["ingredients"],
            match_index=match_index,
            pantry_ingredient_ids=pantry_ids,
        )
        for r in recipe_dicts
    }

    filtered = filter_recipes(
        recipes=recipe_dicts,
        recipe_ingredient_ids=recipe_ingredient_ids,
        recipe_costs=recipe_costs,
        blocked_ingredient_ids=blocked_ings,
        blocked_recipe_ids=blocked_recipes,
        last_cooked_at=last_cooked,
        family_calorie_target=family.calorie_target,
        config=config,
        today=today,
    )

    feasible_recipes = [r for r in recipe_dicts if r["id"] in filtered.feasible_recipe_ids]
    scored = score_recipes(
        recipes=feasible_recipes,
        recipe_costs=recipe_costs,
        last_cooked_at=last_cooked,
        config=config,
        today=today,
    )
    chosen, set_score = select_set(scored, config)

    plan = MealPlan(
        family_profile_id=family_id,
        week_start_date=week_start_date,
        status=MealPlanStatus.DRAFT,
        total_estimated_cost=sum(
            (s.cost.total_cost for s in chosen), Decimal("0.00")
        ),
    )
    db.add(plan)
    db.flush()

    for index, scored_recipe in enumerate(chosen):
        item = MealPlanItem(
            meal_plan_id=plan.id,
            recipe_id=scored_recipe.recipe_id,
            day_of_week=index + 1,         # Mon=1, Tue=2, Wed=3 by default
            meal_type=MealType.DINNER,
            approval_status=MealPlanItemStatus.PENDING,
            estimated_cost=scored_recipe.cost.total_cost,
        )
        db.add(item)

    db.commit()
    db.refresh(plan)

    rejected_summary: Dict[str, int] = {}
    for reason in filtered.rejected.values():
        rejected_summary[reason] = rejected_summary.get(reason, 0) + 1

    return GenerationResult(
        meal_plan_id=plan.id,
        selected=chosen,
        feasible_count=len(filtered.feasible_recipe_ids),
        rejected_summary=rejected_summary,
        set_score=set_score,
    )

Step 4: Run integration test, expect pass

docker compose --env-file .env.test exec backend pytest -q tests/test_planner_generate.py -v

Expected: 1 passed.

Step 5: Run all tests for regressions

docker compose --env-file .env.test exec backend pytest -q tests/ -v

Expected: all green.

Step 6: Commit

git add backend/app/services/planner/generate.py backend/tests/test_planner_generate.py
git commit -m "feat: planner orchestrator - load, filter, score, select, persist"

Task 7: Generate API endpoint

Files:

Create: backend/app/schemas/meal_plan_generation.py
Create: backend/app/api/meal_plans.py
Modify: backend/app/main.py
Test: backend/tests/test_meal_plan_generate_api.py
Step 1: Write the schemas

Create backend/app/schemas/meal_plan_generation.py:

from __future__ import annotations

from datetime import date
from decimal import Decimal
from typing import Dict, List, Optional
from uuid import UUID

from pydantic import BaseModel, Field


class GenerateRequest(BaseModel):
    family_profile_id: UUID
    week_start_date: date


class RegenerateRequest(BaseModel):
    family_profile_id: UUID
    week_start_date: date
    exclude_recipe_ids: List[UUID] = Field(default_factory=list)
    relax_time_max_minutes: Optional[int] = Field(default=None, ge=0)
    relax_calorie_pct: Optional[int] = Field(default=None, ge=0, le=100)
    relax_max_meal_cost: Optional[float] = Field(default=None, ge=0)


class GenerationItem(BaseModel):
    recipe_id: UUID
    day_of_week: int
    estimated_cost: Decimal
    score: float
    components: Dict[str, float]


class GenerationDebug(BaseModel):
    feasible_count: int
    rejected_summary: Dict[str, int]
    set_score: float


class GenerationResponse(BaseModel):
    meal_plan_id: UUID
    week_start_date: date
    items: List[GenerationItem]
    debug: GenerationDebug

Step 2: Write failing API tests

Create backend/tests/test_meal_plan_generate_api.py:

from datetime import datetime, timezone
from decimal import Decimal
from uuid import uuid4

import pytest

pytestmark = pytest.mark.requires_postgres


def _admin() -> dict:
    return {"Authorization": "Bearer test-admin-token"}


@pytest.fixture
def family_with_groceries(db_session):
    from app.models import FamilyProfile, GroceryItem, Ingredient

    family = FamilyProfile(
        id=uuid4(),
        name="API Smoke Family",
        household_size=4,
        adult_count=2,
        child_count=2,
        calorie_target=2400,
    )
    db_session.add(family)
    db_session.commit()

    for ing in db_session.query(Ingredient).limit(20).all():
        db_session.add(
            GroceryItem(
                id=uuid4(),
                name=ing.name,
                source="lucky_california",
                external_id=f"ext-api-{ing.id}",
                current_price=Decimal("3.99"),
                regular_price=Decimal("4.99"),
                is_on_sale=True,
                scraped_at=datetime.now(timezone.utc),
            )
        )
    db_session.commit()

    from app.services.matcher import run_match_job

    run_match_job(db_session)
    return family


def test_generate_endpoint_returns_meal_plan(client, family_with_groceries):
    body = {
        "family_profile_id": str(family_with_groceries.id),
        "week_start_date": "2026-05-11",
    }
    r = client.post("/api/admin/meal-plans/generate", json=body, headers=_admin())
    assert r.status_code == 201, r.text
    data = r.json()
    assert "meal_plan_id" in data
    assert 1 <= len(data["items"]) <= 3
    assert "debug" in data
    assert data["debug"]["feasible_count"] >= 1


def test_generate_endpoint_requires_admin_token(client, family_with_groceries):
    body = {
        "family_profile_id": str(family_with_groceries.id),
        "week_start_date": "2026-05-18",
    }
    r = client.post("/api/admin/meal-plans/generate", json=body)
    assert r.status_code == 401


def test_generate_endpoint_returns_404_for_unknown_family(client):
    body = {
        "family_profile_id": str(uuid4()),
        "week_start_date": "2026-05-11",
    }
    r = client.post("/api/admin/meal-plans/generate", json=body, headers=_admin())
    assert r.status_code == 404

Step 3: Run tests, expect failure

docker compose --env-file .env.test exec backend pytest -q tests/test_meal_plan_generate_api.py -v

Expected: 404 — endpoint missing.

Step 4: Implement the router

Create backend/app/api/meal_plans.py:

from __future__ import annotations

from dataclasses import replace
from typing import List
from uuid import UUID

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.database import get_db
from app.models import FamilyProfile, MealPlan, MealPlanItem
from app.schemas.meal_plan_generation import (
    GenerateRequest,
    GenerationDebug,
    GenerationItem,
    GenerationResponse,
    RegenerateRequest,
)
from app.security import require_admin
from app.services.planner.config import DEFAULT
from app.services.planner.generate import generate_meal_plan
from app.services.planner.types import GenerationResult


admin_router = APIRouter(
    prefix="/api/admin/meal-plans",
    tags=["meal-plans-admin"],
    dependencies=[Depends(require_admin)],
)
public_router = APIRouter(prefix="/api/meal-plans", tags=["meal-plans"])


def _to_response(week_start, plan_id, items: List[MealPlanItem], result: GenerationResult) -> GenerationResponse:
    item_payloads: List[GenerationItem] = []
    score_by_recipe = {s.recipe_id: s for s in result.selected}
    for it in items:
        scored = score_by_recipe.get(it.recipe_id)
        item_payloads.append(
            GenerationItem(
                recipe_id=it.recipe_id,
                day_of_week=it.day_of_week,
                estimated_cost=it.estimated_cost or 0,
                score=scored.score if scored else 0.0,
                components={k: float(v) for k, v in (scored.components.items() if scored else [])},
            )
        )
    return GenerationResponse(
        meal_plan_id=plan_id,
        week_start_date=week_start,
        items=item_payloads,
        debug=GenerationDebug(
            feasible_count=result.feasible_count,
            rejected_summary=result.rejected_summary,
            set_score=result.set_score,
        ),
    )


@admin_router.post("/generate", response_model=GenerationResponse, status_code=status.HTTP_201_CREATED)
def generate(payload: GenerateRequest, db: Session = Depends(get_db)) -> GenerationResponse:
    family = db.query(FamilyProfile).filter(FamilyProfile.id == payload.family_profile_id).first()
    if family is None:
        raise HTTPException(status_code=404, detail="family_profile not found")
    try:
        result = generate_meal_plan(
            db,
            family_id=payload.family_profile_id,
            week_start_date=payload.week_start_date,
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    items = (
        db.query(MealPlanItem)
        .filter(MealPlanItem.meal_plan_id == result.meal_plan_id)
        .order_by(MealPlanItem.day_of_week)
        .all()
    )
    return _to_response(payload.week_start_date, result.meal_plan_id, items, result)


@admin_router.post("/regenerate", response_model=GenerationResponse, status_code=status.HTTP_201_CREATED)
def regenerate(payload: RegenerateRequest, db: Session = Depends(get_db)) -> GenerationResponse:
    family = db.query(FamilyProfile).filter(FamilyProfile.id == payload.family_profile_id).first()
    if family is None:
        raise HTTPException(status_code=404, detail="family_profile not found")

    config = DEFAULT
    if payload.relax_time_max_minutes is not None:
        config = replace(config, max_total_minutes=payload.relax_time_max_minutes)
    if payload.relax_calorie_pct is not None:
        config = replace(config, calorie_tolerance_pct=payload.relax_calorie_pct)
    if payload.relax_max_meal_cost is not None:
        config = replace(config, max_meal_cost=payload.relax_max_meal_cost)

    if payload.exclude_recipe_ids:
        # Add the excluded recipes to the family's NeverSuggest for THIS run only?
        # Simpler: pass them through by extending blocked_recipe_ids inside generate.
        # The current orchestrator doesn't accept exclude_recipe_ids; for v1 we
        # delete any prior plan for the same week and rerun without exclusion
        # support. The TODO is captured in §Open items; the endpoint accepts the
        # field for forward compatibility.
        pass

    db.query(MealPlan).filter(
        MealPlan.family_profile_id == payload.family_profile_id,
        MealPlan.week_start_date == payload.week_start_date,
    ).delete(synchronize_session=False)
    db.commit()

    result = generate_meal_plan(
        db,
        family_id=payload.family_profile_id,
        week_start_date=payload.week_start_date,
        config=config,
    )
    items = (
        db.query(MealPlanItem)
        .filter(MealPlanItem.meal_plan_id == result.meal_plan_id)
        .order_by(MealPlanItem.day_of_week)
        .all()
    )
    return _to_response(payload.week_start_date, result.meal_plan_id, items, result)


@public_router.get("/{plan_id}", response_model=GenerationResponse)
def get_plan(plan_id: UUID, db: Session = Depends(get_db)) -> GenerationResponse:
    plan = db.query(MealPlan).filter(MealPlan.id == plan_id).first()
    if plan is None:
        raise HTTPException(status_code=404, detail="meal_plan not found")
    items = (
        db.query(MealPlanItem)
        .filter(MealPlanItem.meal_plan_id == plan.id)
        .order_by(MealPlanItem.day_of_week)
        .all()
    )
    # When read after the fact we don't have the GenerationResult, so the
    # debug payload returns zeros and items have score=0. The generate
    # response is still authoritative immediately after creation.
    return GenerationResponse(
        meal_plan_id=plan.id,
        week_start_date=plan.week_start_date,
        items=[
            GenerationItem(
                recipe_id=it.recipe_id,
                day_of_week=it.day_of_week,
                estimated_cost=it.estimated_cost or 0,
                score=0.0,
                components={},
            )
            for it in items
        ],
        debug=GenerationDebug(feasible_count=0, rejected_summary={}, set_score=0.0),
    )

Step 5: Wire the routers

In backend/app/main.py:

from app.api import meal_plans as meal_plans_api

app.include_router(meal_plans_api.admin_router)
app.include_router(meal_plans_api.public_router)

Step 6: Run tests, expect pass

docker compose --env-file .env.test exec backend pytest -q tests/test_meal_plan_generate_api.py -v

Expected: 3 passed.

Step 7: Commit

git add backend/app/api/meal_plans.py backend/app/schemas/meal_plan_generation.py backend/app/main.py backend/tests/test_meal_plan_generate_api.py
git commit -m "feat: POST /api/admin/meal-plans/generate + regenerate + get endpoints"

Task 8: Final regression sweep + docs refresh

Files:

Modify: docs/ORIENTATION.md
Modify: docs/HANDOFF.md
Step 1: Run full pytest + alembic round-trip

docker compose --env-file .env.test exec backend pytest -q tests/ -v
docker compose --env-file .env.test exec backend alembic downgrade base
docker compose --env-file .env.test exec backend alembic upgrade head
docker compose --env-file .env.test exec backend pytest -q tests/ -v

Expected: all green both runs; alembic round-trip clean.

Step 2: Update ORIENTATION.md

Open docs/ORIENTATION.md. In the phase status table:

| 9 | Meal-planner generation algorithm | **Complete** — POST /api/admin/meal-plans/generate produces 3-dinner plans against seeded recipes + matched grocery prices. Filter, score (5 signals), top-K=20 set enumeration with diversity penalty. |

Step 3: Update HANDOFF.md

In the "What is real (verified)" section, append:

- Phase 9: meal-plan generation. POST /api/admin/meal-plans/generate runs the full filter→score→set-select pipeline against seeded recipes and produces a persisted MealPlan with 3 MealPlanItem dinners. Regenerate endpoint accepts relaxed constraint overrides for the same week (deletes prior plan first). Per-meal cost matched against ingredient_grocery_match using the top-confidence grocery row.

In the "What is stubbed or missing" section, remove the Phase 9 bullet. Note the carry-over:

- regenerate endpoint accepts `exclude_recipe_ids` for forward compatibility but does not yet apply them; the orchestrator only honors the family's NeverSuggest blocklist. Tracked as a follow-up.
- GET /api/meal-plans/{id} returns persisted items but with empty score/components/debug — those are only available in the immediate `generate` response. Acceptable for the email-approval flow which uses the generate response directly.

Step 4: Commit

git add docs/ORIENTATION.md docs/HANDOFF.md
git commit -m "docs: phase 9 complete - planner algorithm shipped"

Verification gate

pytest -q tests/ green (60+ tests, no failures)
Alembic round-trip clean (downgrade base → upgrade head)
POST /api/admin/meal-plans/generate returns 201 with 3 items against seeded data
POST /api/admin/meal-plans/regenerate for the same (family, week_start_date) deletes prior plan and produces a new one
GET /api/meal-plans/{id} returns 200 with items
Hard-constraint enforcement verified by unit tests on every constraint
Diversity penalty verified by unit test (mixed 0.90 wins over homogeneous 0.95)
No regression in prior 50+ tests

Open items (deferred, tracked here)

GET /api/meal-plans/{id} returns score/components as zeros. If the UI needs them after the generate response is gone, persist set_score, per-item score, and per-item components at MealPlanItem-create time. New columns; not in this plan.
Per-member ingredient blocklists. Schema currently uses household-level NeverSuggest only.
Tunable weights via admin UI. Today they live in planner/config.py.

Out of scope (other phases)

Phase 5 orchestration (chaining scrape → generate → email → vote → finalize)
Phase 6 SendGrid implementation
Phase 8 feedback UI
Phase 10 image strategy
Phase 11 polish (APScheduler, variety analytics, budget tracking)
Frontend UI for /api/admin/meal-plans/* (separate spec; this plan ships API only)

61 KiB Raw Blame History Unescape Escape

Phase 9 — Meal-Planner Algorithm Implementation Plan

File Structure

Task 1: Planner config + types

Task 2: Cost + savings estimator

Task 3: Hard-constraint filter

Task 4: Individual scoring

Task 5: Top-K set selection with diversity

Task 6: Generation orchestrator

Task 7: Generate API endpoint

Task 8: Final regression sweep + docs refresh

Verification gate

Open items (deferred, tracked here)

Out of scope (other phases)

61 KiB

Raw Blame History