from __future__ import annotations from collections import Counter from datetime import datetime, timezone from typing import ClassVar from .base import BaseAgent, AgentInput, AgentOutput _SEVEN_DAYS_S = 7 * 86_400 class RecentPatternsAgent(BaseAgent): """Surfaces the user's reaction pattern from the last 7 days of feedback.""" agent_id: ClassVar[str] = "recent-patterns" ttl_seconds: ClassVar[int] = 86_400 # 24h version: ClassVar[str] = "1.0.0" def compute(self, inp: AgentInput) -> AgentOutput: now_ts = inp.now.timestamp() recent = [ f for f in inp.feedback_history if self._age_s(f.get("created_at", ""), now_ts) <= _SEVEN_DAYS_S ] counts: Counter[str] = Counter(f.get("action") for f in recent) total = len(recent) dwell_ms = inp.profile.get("mean_dwell_ms_30d") if total == 0: prompt = "No tip reactions recorded in the last 7 days." else: done = counts.get("done", 0) dismissed = counts.get("dismiss", 0) snoozed = counts.get("snooze", 0) parts = [ f"Last 7 days: {total} tip reaction{'s' if total != 1 else ''} — " f"{done} completed, {dismissed} dismissed, {snoozed} snoozed." ] if dwell_ms is not None: dwell_s = round(dwell_ms / 1000) if dwell_s < 15: parts.append( "Average dwell is very short — user may be acting on auto-pilot; vary tip content." ) elif dwell_s < 60: parts.append(f"Average dwell {dwell_s}s — tips are being read.") else: parts.append( f"Average dwell {dwell_s}s — user deliberates; prefer tips that reward reflection." ) prompt = " ".join(parts) snapshot = { "recent_total": total, "action_counts": dict(counts), "mean_dwell_ms_30d": dwell_ms, } return self._make_output(inp, prompt, snapshot) @staticmethod def _age_s(iso: str, now_ts: float) -> float: if not iso: return float("inf") try: dt = datetime.fromisoformat(iso.replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return now_ts - dt.timestamp() except Exception: return float("inf")