from __future__ import annotations import statistics from collections import Counter from typing import ClassVar from .base import BaseAgent, AgentInput, AgentOutput from .inference.history import UserHistory from .manifest import AgentManifest, InferredParam _DOW_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] # min_history required before quiet/peak inference is meaningful (issue #112) _MIN_HISTORY = 50 def _infer_preferred_hour(history: UserHistory) -> int: """Mode hour of day across all 'done' feedback events; falls back to 9.""" done_hours = [e.hour for e in history.events if e.action == "done"] if not done_hours: return 9 return Counter(done_hours).most_common(1)[0][0] def _quiet_window_hours(history: UserHistory) -> tuple[int, int]: """Return (start_hour, end_hour) of the longest below-baseline quiet window. Counts all engagement events by hour. Baseline = mean hourly count. Finds the longest contiguous run of below-baseline hours on the circular clock; that run defines the quiet window. """ by_hour: Counter[int] = Counter(e.hour for e in history.events) total = sum(by_hour.values()) baseline = total / 24 # Mark each of the 24 hours as below-baseline (True = quiet) quiet: list[bool] = [by_hour.get(h, 0) < baseline for h in range(24)] # Find longest contiguous run in circular array best_start, best_len = 0, 0 run_start, run_len = 0, 0 # Double the sequence to handle wrap-around for i in range(48): h = i % 24 if quiet[h]: if run_len == 0: run_start = i run_len += 1 if run_len > best_len: best_len = run_len best_start = run_start else: run_len = 0 if best_len == 0: return (22, 7) # fallback start = best_start % 24 end = (best_start + best_len) % 24 return (start, end) def _infer_quiet_start(history: UserHistory) -> str: start, _ = _quiet_window_hours(history) return f"{start:02d}:00" def _infer_quiet_end(history: UserHistory) -> str: _, end = _quiet_window_hours(history) return f"{end:02d}:00" def _infer_peak_hours(history: UserHistory) -> list[int]: """Top-quartile hours by done-event count. Computes done_count per hour, then returns hours above the 75th percentile of non-zero hourly counts, sorted ascending. """ done_by_hour: Counter[int] = Counter( e.hour for e in history.events if e.action == "done" ) if not done_by_hour: return [9, 14, 20] counts = list(done_by_hour.values()) threshold = statistics.quantiles(counts, n=4)[-1] # 75th percentile return sorted(h for h, c in done_by_hour.items() if c >= threshold) MANIFEST = AgentManifest( id="time-of-day", version="1.2.0", # #112: quiet_start/end + peak_hours + tz inference description="Frames the current moment relative to the user's productive peak and quiet hours.", pref_schema={ "type": "object", "additionalProperties": False, "properties": { "quiet_start": { "type": "string", "pattern": "^([01][0-9]|2[0-3]):[0-5][0-9]$", "description": "HH:MM start of quiet hours (24h, user's local TZ).", }, "quiet_end": { "type": "string", "pattern": "^([01][0-9]|2[0-3]):[0-5][0-9]$", "description": "HH:MM end of quiet hours.", }, "peak_hours": { "type": "array", "items": {"type": "integer", "minimum": 0, "maximum": 23}, "default": [9, 14, 20], "description": "Hours (0–23) with top-quartile completion density.", }, "tz": { "type": "string", "default": "UTC", "description": "IANA timezone; populated from auth provider, fallback UTC.", }, "preferred_hour": { "type": "integer", "minimum": 0, "maximum": 23, "description": "Mode done-hour (legacy; superseded by peak_hours).", }, }, }, context_schema=["profile.features"], required_consents=["data:core", "agent:time-of-day"], output_contract={"type": "snippet", "format": "free_text"}, ttl_sec=900, inferred_params=[ InferredParam( key="preferred_hour", ttl_sec=3_600, cold_start_default=None, min_history=10, infer=_infer_preferred_hour, ), InferredParam( key="quiet_start", ttl_sec=86_400, cold_start_default="22:00", min_history=_MIN_HISTORY, infer=_infer_quiet_start, ), InferredParam( key="quiet_end", ttl_sec=86_400, cold_start_default="07:00", min_history=_MIN_HISTORY, infer=_infer_quiet_end, ), InferredParam( key="peak_hours", ttl_sec=86_400, cold_start_default=[9, 14, 20], min_history=_MIN_HISTORY, infer=_infer_peak_hours, ), # tz is populated from the auth provider; no infer function. InferredParam( key="tz", ttl_sec=86_400, cold_start_default="UTC", min_history=999_999, # effectively never inferred — always cold_start infer=None, ), ], ) class TimeOfDayAgent(BaseAgent): """Frames the current moment relative to the user's productive peak.""" agent_id: ClassVar[str] = MANIFEST.id ttl_seconds: ClassVar[int] = MANIFEST.ttl_sec version: ClassVar[str] = MANIFEST.version def compute(self, inp: AgentInput) -> AgentOutput: hour = inp.now.hour dow = inp.now.weekday() is_weekend = dow >= 5 preferred_raw = inp.agent_prefs.get("preferred_hour", inp.profile.get("preferred_hour")) preferred = int(preferred_raw) if preferred_raw is not None else None quiet_start: str | None = inp.agent_prefs.get("quiet_start") quiet_end: str | None = inp.agent_prefs.get("quiet_end") peak_hours: list[int] = inp.agent_prefs.get("peak_hours", []) tz: str = inp.agent_prefs.get("tz", "UTC") in_quiet = self._in_quiet_window(hour, quiet_start, quiet_end) in_peak = hour in peak_hours parts = [f"It is {hour:02d}:00 on {_DOW_NAMES[dow]} ({self._label(hour)})."] if tz != "UTC": parts[0] = f"It is {hour:02d}:00 ({tz}) on {_DOW_NAMES[dow]} ({self._label(hour)})." if is_weekend: parts.append("Weekend context — prefer personal or reflective tips over work tasks.") if in_quiet: parts.append( f"User is in their quiet window ({quiet_start}–{quiet_end}) — " "avoid urgent or demanding tips." ) elif in_peak: parts.append( f"Hour {hour:02d}:00 is a peak productivity hour for this user — " "a high-impact or challenging tip is appropriate." ) elif peak_hours: # Report nearest peak so orchestrator can time advice accordingly. nearest = min(peak_hours, key=lambda p: min(abs(p - hour), 24 - abs(p - hour))) delta = min(abs(nearest - hour), 24 - abs(nearest - hour)) if delta <= 2: parts.append(f"Approaching peak productivity window ({nearest:02d}:00).") elif preferred is not None: delta = min(abs(hour - preferred), 24 - abs(hour - preferred)) if delta == 0: parts.append( f"This is the user's peak productivity hour ({preferred:02d}:00) — " "a high-impact tip is appropriate." ) elif delta <= 2: parts.append(f"Approaching the user's peak productivity window ({preferred:02d}:00).") else: parts.append("No preferred-hour data yet.") prompt = " ".join(parts) snapshot = { "hour": hour, "day_of_week": dow, "preferred_hour": preferred, "quiet_start": quiet_start, "quiet_end": quiet_end, "peak_hours": peak_hours, "in_quiet": in_quiet, "in_peak": in_peak, "tz": tz, } return self._make_output(inp, prompt, snapshot) @staticmethod def _in_quiet_window(hour: int, start: str | None, end: str | None) -> bool: if not start or not end: return False try: sh = int(start.split(":")[0]) eh = int(end.split(":")[0]) except (ValueError, IndexError): return False if sh <= eh: return sh <= hour < eh # wraps midnight e.g. 22:00–07:00 return hour >= sh or hour < eh @staticmethod def _label(hour: int) -> str: if 5 <= hour < 12: return "morning" if 12 <= hour < 17: return "afternoon" if 17 <= hour < 21: return "evening" return "night"