Spaces:
Running
Running
| """Proactive gates — port of ProactiveWorker.kt, blueprint 10.3. | |
| 2026-09-21 is a Monday and 2026-09-19 a Saturday. Times are IST. | |
| """ | |
| from datetime import datetime, timedelta | |
| from zoneinfo import ZoneInfo | |
| import pytest | |
| from brain.services.proactive import decide, in_quiet_hours | |
| from brain.store.memory import MemoryStore | |
| TZ = ZoneInfo("Asia/Kolkata") | |
| USER = "u1" | |
| def at(day: int, hour: int, minute: int = 0) -> datetime: | |
| return datetime(2026, 9, day, hour, minute, tzinfo=TZ) | |
| async def store_with_last( | |
| now: datetime, minutes_ago: int, role: str = "user", proactive: bool = False | |
| ): | |
| store = MemoryStore() | |
| store.messages.append( | |
| { | |
| "id": "m", | |
| "user_id": USER, | |
| "mode": "gf", | |
| "role": role, | |
| "content": "x", | |
| "is_proactive": proactive, | |
| "created_at": now - timedelta(minutes=minutes_ago), | |
| } | |
| ) | |
| return store | |
| def test_quiet_hours_including_windows_that_wrap_midnight(minute, start, end, quiet): | |
| assert in_quiet_hours(minute, start, end) is quiet | |
| async def test_everything_off_means_notifications_off(): | |
| now = at(21, 11) | |
| store = await store_with_last(now, 600) | |
| settings = {"proactive": {"gfEnabled": False, "normalEnabled": False}} | |
| assert (await decide(store, USER, settings, now)).gate == "notifications_off" | |
| async def test_weekday_quiet_hours_are_midnight_to_seven(): | |
| now = at(21, 6, 59) | |
| store = await store_with_last(now, 600) | |
| assert (await decide(store, USER, {}, now)).gate == "quiet_hours" | |
| async def test_weekend_quiet_hours_start_later(): | |
| # 00:30 on a Saturday is still awake time at the weekend; weekday rules would say quiet. | |
| now = at(19, 0, 30) | |
| store = await store_with_last(now, 600) | |
| decision = await decide(store, USER, {}, now) | |
| assert decision.gate != "quiet_hours" | |
| async def test_the_daily_cap(): | |
| now = at(21, 11) | |
| store = await store_with_last(now, 600) | |
| for _ in range(3): | |
| await store.log_proactive(USER, "checkin", None, ok=True) | |
| for row in store.proactive: | |
| row["sent_at"] = now - timedelta(hours=1) | |
| assert (await decide(store, USER, {}, now)).gate == "daily_cap" | |
| async def test_the_cap_is_clamped_to_one_through_five(): | |
| now = at(21, 11) | |
| store = await store_with_last(now, 600) | |
| for _ in range(5): | |
| await store.log_proactive(USER, "checkin", None, ok=True) | |
| for row in store.proactive: | |
| row["sent_at"] = now - timedelta(hours=1) | |
| decision = await decide(store, USER, {"proactive": {"maxPerDay": 50}}, now) | |
| assert decision.gate == "daily_cap" | |
| async def test_no_history_means_no_opening_line(): | |
| assert (await decide(MemoryStore(), USER, {}, at(21, 11))).gate == "no_history" | |
| async def test_a_weekday_needs_five_hours_of_silence(): | |
| now = at(21, 11) | |
| assert (await decide(await store_with_last(now, 299), USER, {}, now)).gate == "too_recent" | |
| assert (await decide(await store_with_last(now, 301), USER, {}, now)).send is True | |
| async def test_the_weekend_bar_is_an_hour(): | |
| now = at(19, 15) | |
| assert (await decide(await store_with_last(now, 61), USER, {}, now)).send is True | |
| async def test_a_routine_window_lowers_the_bar_to_45_minutes_and_is_stamped(): | |
| now = at(21, 14, 10) # ten minutes after lunch | |
| decision = await decide(await store_with_last(now, 50), USER, {}, now) | |
| assert decision.send is True | |
| assert decision.stamp == "2026-09-21:lunch" | |
| async def test_a_window_fires_once_per_day(): | |
| now = at(21, 14, 10) | |
| store = await store_with_last(now, 50) | |
| await store.log_proactive(USER, "checkin", "2026-09-21:lunch", ok=True) | |
| for row in store.proactive: | |
| row["sent_at"] = now - timedelta(days=1) # does not count toward today's cap | |
| decision = await decide(store, USER, {}, now) | |
| assert decision.gate == "too_recent" # back to the five-hour bar | |
| async def test_off_today_closes_the_routine_windows(): | |
| now = at(21, 14, 10) | |
| settings = {"offTodayDate": "2026-09-21"} | |
| decision = await decide(await store_with_last(now, 50), USER, settings, now) | |
| assert decision.gate == "too_recent" | |
| async def test_an_unanswered_check_in_blocks_the_next_one(): | |
| now = at(21, 17) | |
| store = await store_with_last(now, 330, role="assistant", proactive=True) | |
| assert (await decide(store, USER, {}, now)).gate == "unanswered_backoff" | |
| async def test_a_recent_failure_backs_off_for_six_hours(): | |
| now = at(21, 17) | |
| store = await store_with_last(now, 600) | |
| await store.log_proactive(USER, "failed", None, ok=False) | |
| store.proactive[-1]["sent_at"] = now - timedelta(hours=2) | |
| assert (await decide(store, USER, {}, now)).gate == "failure_backoff" | |
| async def test_test_mode_ignores_every_gate(): | |
| decision = await decide(MemoryStore(), USER, {}, at(21, 3), test_mode=True) | |
| assert decision.send is True | |
| assert decision.gate == "test_mode" | |
| async def test_gf_is_the_default_mode_and_normal_is_used_only_when_gf_is_off(): | |
| now = at(21, 11) | |
| store = await store_with_last(now, 600) | |
| assert (await decide(store, USER, {}, now)).mode.value == "gf" | |
| only_normal = {"proactive": {"gfEnabled": False, "normalEnabled": True}} | |
| assert (await decide(store, USER, only_normal, now)).mode.value == "normal" | |