Spaces:
Running
Running
File size: 5,628 Bytes
210ef30 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | """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
@pytest.mark.parametrize(
("minute", "start", "end", "quiet"),
[
(60, 0, 420, True),
(420, 0, 420, False), # end is exclusive
(1400, 1380, 420, True), # wraps midnight: 23:00-07:00
(200, 1380, 420, True),
(720, 1380, 420, False),
(500, 600, 600, False), # an empty window is never quiet
],
)
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"
|