"""Derive schedule risk signals from entries + daily check-ins. Shared by human plan routes and agent context — server-owned, not LLM. """ from __future__ import annotations from datetime import date, datetime, timedelta, timezone from typing import Any RISK_TAGS = frozenset( { "urge", "court", "comparison", "corn", "daydream", "bully", "family", "shame", "rerun", } ) LAST_HOUR_HIGH_TAGS = frozenset({"urge", "court", "corn", "rerun"}) def risk_from_stores( entry_store: Any, daily_store: Any, *, now: datetime | None = None, target_day: date | None = None, ) -> dict[str, Any]: """Return risk_1h, triggers_yesterday, and last_hour_high flags.""" now = now or datetime.now(timezone.utc) target = target_day or now.date() hour_ago = now - timedelta(hours=1) yesterday = (target - timedelta(days=1)).isoformat() tags_1h: list[str] = [] triggers_y: list[str] = [] for entry in entry_store._load(): ts = entry.ts if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) entry_tags = set(entry.tags) | set(entry.emotions) hit = sorted(entry_tags & RISK_TAGS) if ts >= hour_ago: tags_1h.extend(hit) if entry.ts.date().isoformat() == yesterday: triggers_y.extend(hit) y_daily = daily_store.get(date.fromisoformat(yesterday)) if y_daily: if y_daily.court == "court": triggers_y.append("court") if y_daily.corn_sessions >= 2: triggers_y.append("corn") if y_daily.daydream == "fc": triggers_y.append("daydream") tags_1h = sorted(set(tags_1h)) triggers_y = sorted(set(triggers_y)) last_hour_high = bool(LAST_HOUR_HIGH_TAGS & set(tags_1h)) return { "risk_1h_score": float(len(tags_1h)), "risk_1h_tags": tags_1h, "triggers_yesterday": triggers_y, "yesterday_trigger": len(triggers_y) > 0, "last_hour_high": last_hour_high, }