from __future__ import annotations 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"] 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] MANIFEST = AgentManifest( id="time-of-day", version="1.1.0", # bumped: inferred_params added (ADR-0014 §3, #112) 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.", }, }, }, 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, # recompute hourly cold_start_default=None, min_history=10, # need at least 10 feedback events to be meaningful infer=_infer_preferred_hour, ), ], ) 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() # 0=Monday … 6=Sunday is_weekend = dow >= 5 # agent_prefs (inferred or user-set) take precedence over ML profile features. 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") in_quiet = self._in_quiet_window(hour, quiet_start, quiet_end) parts = [f"It is {hour:02d}:00 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." ) if 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, } 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"