piotr/innercontext
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(api): align routine context windows with recent skin history

Browse source
This commit is contained in:
Piotr Oleszczyk 2026-03-08 11:53:59 +01:00
parent 1c457d62a3
commit fecfa0b9e4
3 changed files with 273 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

86
backend/innercontext/api/routines.py
View file

@ -49,6 +49,9 @@ from innercontext.validators.routine_validator import RoutineValidationContext
logger = logging.getLogger(__name__)
HISTORY_WINDOW_DAYS = 5
SNAPSHOT_FALLBACK_DAYS = 14
def _build_response_metadata(session: Session, log_id: Any) -> ResponseMetadata | None:
"""Build ResponseMetadata from AICallLog for Phase 3 observability."""
@ -284,12 +287,58 @@ def _ev(v: object) -> str:
return str(v)
def _build_skin_context(session: Session) -> str:
def _get_recent_skin_snapshot(
session: Session,
reference_date: date,
window_days: int = HISTORY_WINDOW_DAYS,
fallback_days: int = SNAPSHOT_FALLBACK_DAYS,
) -> SkinConditionSnapshot | None:
window_cutoff = reference_date - timedelta(days=window_days)
fallback_cutoff = reference_date - timedelta(days=fallback_days)
snapshot = session.exec(
select(SkinConditionSnapshot).order_by(
col(SkinConditionSnapshot.snapshot_date).desc()
)
select(SkinConditionSnapshot)
.where(SkinConditionSnapshot.snapshot_date <= reference_date)
.where(SkinConditionSnapshot.snapshot_date >= window_cutoff)
.order_by(col(SkinConditionSnapshot.snapshot_date).desc())
).first()
if snapshot is not None:
return snapshot
return session.exec(
select(SkinConditionSnapshot)
.where(SkinConditionSnapshot.snapshot_date <= reference_date)
.where(SkinConditionSnapshot.snapshot_date >= fallback_cutoff)
.order_by(col(SkinConditionSnapshot.snapshot_date).desc())
).first()
def _get_latest_skin_snapshot_within_days(
session: Session,
reference_date: date,
max_age_days: int = SNAPSHOT_FALLBACK_DAYS,
) -> SkinConditionSnapshot | None:
cutoff = reference_date - timedelta(days=max_age_days)
return session.exec(
select(SkinConditionSnapshot)
.where(SkinConditionSnapshot.snapshot_date <= reference_date)
.where(SkinConditionSnapshot.snapshot_date >= cutoff)
.order_by(col(SkinConditionSnapshot.snapshot_date).desc())
).first()
def _build_skin_context(
session: Session,
reference_date: date,
window_days: int = HISTORY_WINDOW_DAYS,
fallback_days: int = SNAPSHOT_FALLBACK_DAYS,
) -> str:
snapshot = _get_recent_skin_snapshot(
session,
reference_date=reference_date,
window_days=window_days,
fallback_days=fallback_days,
)
if snapshot is None:
return "SKIN CONDITION: no data\n"
ev = _ev
@ -369,10 +418,15 @@ def _build_upcoming_grooming_context(
return "\n".join(lines) + "\n"
def _build_recent_history(session: Session) -> str:
cutoff = date.today() - timedelta(days=7)
def _build_recent_history(
session: Session,
reference_date: date,
window_days: int = HISTORY_WINDOW_DAYS,
) -> str:
cutoff = reference_date - timedelta(days=window_days)
routines = session.exec(
select(Routine)
.where(Routine.routine_date <= reference_date)
.where(Routine.routine_date >= cutoff)
.order_by(col(Routine.routine_date).desc())
).all()
@ -672,14 +726,14 @@ def suggest_routine(
session: Session = Depends(get_session),
):
weekday = data.routine_date.weekday()
skin_ctx = _build_skin_context(session)
skin_ctx = _build_skin_context(session, reference_date=data.routine_date)
profile_ctx = build_user_profile_context(session, reference_date=data.routine_date)
upcoming_grooming_ctx = _build_upcoming_grooming_context(
session,
start_date=data.routine_date,
days=7,
)
history_ctx = _build_recent_history(session)
history_ctx = _build_recent_history(session, reference_date=data.routine_date)
day_ctx = _build_day_context(data.leaving_home)
available_products = _get_available_products(
session,
@ -848,10 +902,10 @@ def suggest_routine(
)
# Get skin snapshot for barrier state
stmt = select(SkinConditionSnapshot).order_by(
col(SkinConditionSnapshot.snapshot_date).desc()
skin_snapshot = _get_latest_skin_snapshot_within_days(
session,
reference_date=data.routine_date,
)
skin_snapshot = session.exec(stmt).first()
# Build validation context
products_by_id = {p.id: p for p in available_products}
@ -923,9 +977,9 @@ def suggest_batch(
{(data.from_date + timedelta(days=i)).weekday() for i in range(delta)}
)
profile_ctx = build_user_profile_context(session, reference_date=data.from_date)
skin_ctx = _build_skin_context(session)
skin_ctx = _build_skin_context(session, reference_date=data.from_date)
grooming_ctx = _build_grooming_context(session, weekdays=weekdays)
history_ctx = _build_recent_history(session)
history_ctx = _build_recent_history(session, reference_date=data.from_date)
batch_products = _get_available_products(
session,
include_minoxidil=data.include_minoxidil_beard,
@ -1030,10 +1084,10 @@ def suggest_batch(
)
# Get skin snapshot for barrier state
stmt = select(SkinConditionSnapshot).order_by(
col(SkinConditionSnapshot.snapshot_date).desc()
skin_snapshot = _get_latest_skin_snapshot_within_days(
session,
reference_date=data.from_date,
)
skin_snapshot = session.exec(stmt).first()
# Build validation context
products_by_id = {p.id: p for p in batch_products}

42
backend/tests/test_routines.py
View file

@ -1,6 +1,10 @@
import uuid
from datetime import date
from unittest.mock import patch
from innercontext.models import Routine, SkinConditionSnapshot
from innercontext.models.enums import BarrierState, OverallSkinState
# ---------------------------------------------------------------------------
# Routines
# ---------------------------------------------------------------------------
@ -223,6 +227,18 @@ def test_suggest_routine(client, session):
with patch(
"innercontext.api.routines.call_gemini_with_function_tools"
) as mock_gemini:
session.add(
SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=date(2026, 2, 22),
overall_state=OverallSkinState.GOOD,
hydration_level=4,
barrier_state=BarrierState.INTACT,
priorities=["hydration"],
)
)
session.commit()
# Mock the Gemini response
mock_response = type(
"Response",
@ -231,7 +247,7 @@ def test_suggest_routine(client, session):
"text": '{"steps": [{"product_id": null, "action_type": "shaving_razor"}], "reasoning": "because"}'
},
)
mock_gemini.return_value = mock_response
mock_gemini.return_value = (mock_response, None)
r = client.post(
"/routines/suggest",
@ -250,12 +266,32 @@ def test_suggest_routine(client, session):
kwargs = mock_gemini.call_args.kwargs
assert "USER PROFILE:" in kwargs["contents"]
assert "UPCOMING GROOMING (next 7 days):" in kwargs["contents"]
assert "snapshot from 2026-02-22" in kwargs["contents"]
assert "RECENT ROUTINES: none" in kwargs["contents"]
assert "function_handlers" in kwargs
assert "get_product_details" in kwargs["function_handlers"]
def test_suggest_batch(client, session):
with patch("innercontext.api.routines.call_gemini") as mock_gemini:
session.add(
Routine(
id=uuid.uuid4(),
routine_date=date(2026, 2, 27),
part_of_day="pm",
)
)
session.add(
SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=date(2026, 2, 20),
overall_state=OverallSkinState.GOOD,
hydration_level=4,
barrier_state=BarrierState.INTACT,
)
)
session.commit()
# Mock the Gemini response
mock_response = type(
"Response",
@ -264,7 +300,7 @@ def test_suggest_batch(client, session):
"text": '{"days": [{"date": "2026-03-03", "am_steps": [], "pm_steps": [], "reasoning": "none"}], "overall_reasoning": "batch test"}'
},
)
mock_gemini.return_value = mock_response
mock_gemini.return_value = (mock_response, None)
r = client.post(
"/routines/suggest-batch",
@ -281,6 +317,8 @@ def test_suggest_batch(client, session):
assert data["overall_reasoning"] == "batch test"
kwargs = mock_gemini.call_args.kwargs
assert "USER PROFILE:" in kwargs["contents"]
assert "2026-02-27 PM:" in kwargs["contents"]
assert "snapshot from 2026-02-20" in kwargs["contents"]
def test_suggest_batch_invalid_date_range(client):

216
backend/tests/test_routines_helpers.py
View file

@ -3,11 +3,11 @@ from datetime import date, timedelta
from sqlmodel import Session
from innercontext.api.llm_context import build_products_context_summary_list
from innercontext.api.routines import (
_build_day_context,
_build_grooming_context,
_build_objectives_context,
_build_products_context,
_build_recent_history,
_build_skin_context,
_build_upcoming_grooming_context,
@ -17,17 +17,19 @@ from innercontext.api.routines import (
_extract_requested_product_ids,
_filter_products_by_interval,
_get_available_products,
_get_latest_skin_snapshot_within_days,
_get_recent_skin_snapshot,
_is_minoxidil_product,
build_product_details_tool_handler,
)
from innercontext.models import (
GroomingSchedule,
Product,
ProductInventory,
Routine,
RoutineStep,
SkinConditionSnapshot,
)
from innercontext.models.enums import BarrierState, OverallSkinState, SkinConcern
def test_contains_minoxidil_text():
@ -78,32 +80,131 @@ def test_ev():
def test_build_skin_context(session: Session):
# Empty
assert _build_skin_context(session) == "SKIN CONDITION: no data\n"
reference_date = date(2026, 3, 10)
assert (
_build_skin_context(session, reference_date=reference_date)
== "SKIN CONDITION: no data\n"
)
# With data
snap = SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=date.today(),
overall_state="good",
snapshot_date=reference_date,
overall_state=OverallSkinState.GOOD,
hydration_level=4,
barrier_state="intact",
active_concerns=["acne", "dryness"],
barrier_state=BarrierState.INTACT,
active_concerns=[SkinConcern.ACNE, SkinConcern.DEHYDRATION],
priorities=["hydration"],
notes="Feeling good",
)
session.add(snap)
session.commit()
ctx = _build_skin_context(session)
ctx = _build_skin_context(session, reference_date=reference_date)
assert "SKIN CONDITION (snapshot from" in ctx
assert "Overall state: good" in ctx
assert "Hydration: 4/5" in ctx
assert "Barrier: intact" in ctx
assert "Active concerns: acne, dryness" in ctx
assert "Active concerns: acne, dehydration" in ctx
assert "Priorities: hydration" in ctx
assert "Notes: Feeling good" in ctx
def test_build_skin_context_falls_back_to_recent_snapshot_within_14_days(
session: Session,
):
reference_date = date(2026, 3, 20)
snap = SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=reference_date - timedelta(days=10),
overall_state=OverallSkinState.FAIR,
hydration_level=3,
barrier_state=BarrierState.COMPROMISED,
active_concerns=[SkinConcern.REDNESS],
priorities=["barrier"],
)
session.add(snap)
session.commit()
ctx = _build_skin_context(session, reference_date=reference_date)
assert f"snapshot from {reference_date - timedelta(days=10)}" in ctx
assert "Barrier: compromised" in ctx
def test_build_skin_context_ignores_snapshot_older_than_14_days(session: Session):
reference_date = date(2026, 3, 20)
snap = SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=reference_date - timedelta(days=15),
overall_state=OverallSkinState.FAIR,
hydration_level=3,
barrier_state=BarrierState.INTACT,
)
session.add(snap)
session.commit()
assert (
_build_skin_context(session, reference_date=reference_date)
== "SKIN CONDITION: no data\n"
)
def test_get_recent_skin_snapshot_prefers_window_match(session: Session):
reference_date = date(2026, 3, 20)
older = SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=reference_date - timedelta(days=10),
overall_state=OverallSkinState.POOR,
hydration_level=2,
barrier_state=BarrierState.COMPROMISED,
)
newer = SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=reference_date - timedelta(days=2),
overall_state=OverallSkinState.GOOD,
hydration_level=4,
barrier_state=BarrierState.INTACT,
)
session.add_all([older, newer])
session.commit()
snapshot = _get_recent_skin_snapshot(session, reference_date=reference_date)
assert snapshot is not None
assert snapshot.id == newer.id
def test_get_latest_skin_snapshot_within_days_uses_latest_within_14_days(
session: Session,
):
reference_date = date(2026, 3, 20)
older = SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=reference_date - timedelta(days=10),
overall_state=OverallSkinState.POOR,
hydration_level=2,
barrier_state=BarrierState.COMPROMISED,
)
newer = SkinConditionSnapshot(
id=uuid.uuid4(),
snapshot_date=reference_date - timedelta(days=2),
overall_state=OverallSkinState.GOOD,
hydration_level=4,
barrier_state=BarrierState.INTACT,
)
session.add_all([older, newer])
session.commit()
snapshot = _get_latest_skin_snapshot_within_days(
session,
reference_date=reference_date,
)
assert snapshot is not None
assert snapshot.id == newer.id
def test_build_grooming_context(session: Session):
assert _build_grooming_context(session) == "GROOMING SCHEDULE: none\n"
@ -146,12 +247,17 @@ def test_build_upcoming_grooming_context(session: Session):
def test_build_recent_history(session: Session):
assert _build_recent_history(session) == "RECENT ROUTINES: none\n"
reference_date = date(2026, 3, 10)
assert (
_build_recent_history(session, reference_date=reference_date)
== "RECENT ROUTINES: none\n"
)
r = Routine(id=uuid.uuid4(), routine_date=date.today(), part_of_day="am")
r = Routine(id=uuid.uuid4(), routine_date=reference_date, part_of_day="am")
session.add(r)
p = Product(
id=uuid.uuid4(),
short_id=str(uuid.uuid4())[:8],
name="Cleanser",
category="cleanser",
brand="Test",
@ -174,7 +280,7 @@ def test_build_recent_history(session: Session):
session.add_all([s1, s2, s3])
session.commit()
ctx = _build_recent_history(session)
ctx = _build_recent_history(session, reference_date=reference_date)
assert "RECENT ROUTINES:" in ctx
assert "AM:" in ctx
assert "cleanser [" in ctx
@ -182,10 +288,48 @@ def test_build_recent_history(session: Session):
assert "unknown [" in ctx
def test_build_products_context(session: Session):
def test_build_recent_history_uses_reference_window(session: Session):
reference_date = date(2026, 3, 10)
recent = Routine(
id=uuid.uuid4(),
routine_date=reference_date - timedelta(days=3),
part_of_day="pm",
)
old = Routine(
id=uuid.uuid4(),
routine_date=reference_date - timedelta(days=6),
part_of_day="am",
)
session.add_all([recent, old])
session.commit()
ctx = _build_recent_history(session, reference_date=reference_date)
assert str(recent.routine_date) in ctx
assert str(old.routine_date) not in ctx
def test_build_recent_history_excludes_future_routines(session: Session):
reference_date = date(2026, 3, 10)
future = Routine(
id=uuid.uuid4(),
routine_date=reference_date + timedelta(days=1),
part_of_day="am",
)
session.add(future)
session.commit()
assert (
_build_recent_history(session, reference_date=reference_date)
== "RECENT ROUTINES: none\n"
)
def test_build_products_context_summary_list(session: Session):
p1 = Product(
id=uuid.uuid4(),
name="Regaine",
short_id=str(uuid.uuid4())[:8],
name="Regaine Minoxidil",
category="serum",
is_medication=True,
brand="J&J",
@ -195,6 +339,7 @@ def test_build_products_context(session: Session):
)
p2 = Product(
id=uuid.uuid4(),
short_id=str(uuid.uuid4())[:8],
name="Sunscreen",
category="spf",
brand="Test",
@ -209,49 +354,14 @@ def test_build_products_context(session: Session):
session.add_all([p1, p2])
session.commit()
# Inventory
inv1 = ProductInventory(
id=uuid.uuid4(),
product_id=p2.id,
is_opened=True,
opened_at=date.today() - timedelta(days=10),
expiry_date=date.today() + timedelta(days=365),
)
inv2 = ProductInventory(id=uuid.uuid4(), product_id=p2.id, is_opened=False)
session.add_all([inv1, inv2])
session.commit()
# Usage
r = Routine(id=uuid.uuid4(), routine_date=date.today(), part_of_day="am")
session.add(r)
session.commit()
s = RoutineStep(id=uuid.uuid4(), routine_id=r.id, order_index=1, product_id=p2.id)
session.add(s)
session.commit()
products_am = _get_available_products(session, time_filter="am")
ctx = _build_products_context(session, products_am, reference_date=date.today())
# p1 is medication but not minoxidil (wait, Regaine name doesn't contain minoxidil!) -> skipped
assert "Regaine" not in ctx
ctx = build_products_context_summary_list(products_am, {p2.id})
# Let's fix p1 to be minoxidil
p1.name = "Regaine Minoxidil"
session.add(p1)
session.commit()
products_am = _get_available_products(session, time_filter="am")
ctx = _build_products_context(session, products_am, reference_date=date.today())
assert "Regaine Minoxidil" in ctx
assert "Sunscreen" in ctx
assert "inventory_status={active:2,opened:1,sealed:1}" in ctx
assert "nearest_open_expiry=" in ctx
assert "nearest_open_pao_deadline=" in ctx
assert "pao_months=6" in ctx
assert "effects={'hydration_immediate': 2}" in ctx
assert "context_rules={'safe_after_shaving': False}" in ctx
assert "min_interval_hours=12" in ctx
assert "max_frequency_per_week=7" in ctx
assert "used_in_last_7_days=1" in ctx
assert "[✓]" in ctx
assert "hydration=2" in ctx
assert "!post_shave" in ctx
def test_build_objectives_context():