refactor: value object pattern for Role, FinishReason, SourceType, and time triggers

ntrp/automation/triggers.py

@@ -34,42 +34,88 @@
 _TIME_RE = re.compile(r"^(\d{1,2}):(\d{2})$")
 
 
-def parse_days(raw: str) -> frozenset[int]:
-    days: set[int] = set()
-
-    for part in raw.split(","):
-        token = part.strip().lower()
-        if not token:
-            continue
-
-        if token in DAY_KEYWORDS:
-            days.update(DAY_KEYWORDS[token])
-        elif token in DAY_NAMES:
-            days.add(DAY_NAMES[token])
-        else:
-            raise ValueError(f"Invalid day: '{token}'. Use: {', '.join(DAY_NAMES)} / daily / weekdays")
+@dataclass(frozen=True)
+class TimeOfDay:
+    hour: int
+    minute: int
+
+    def __post_init__(self) -> None:
+        if not (0 <= self.hour <= 23 and 0 <= self.minute <= 59):
+            raise ValueError(f"Invalid time: {self.hour:02d}:{self.minute:02d}")
 
-    return frozenset(days) if days else ALL_DAYS
+    @classmethod
+    def parse(cls, raw: str) -> "TimeOfDay":
+        match = _TIME_RE.match(raw.strip())
+        if not match:
+            raise ValueError(f"Invalid time format '{raw}'. Use HH:MM (24h)")
+        return cls(hour=int(match.group(1)), minute=int(match.group(2)))
 
+    def to_time(self) -> time:
+        return time(self.hour, self.minute)
 
-def resolve_days(days: str) -> frozenset[int]:
-    return parse_days(days)
+    def __str__(self) -> str:
+        return f"{self.hour:02d}:{self.minute:02d}"
 
 
-def validate_days(days: str) -> str:
-    parse_days(days)
-    return days.strip().lower()
+@dataclass(frozen=True)
+class Interval:
+    delta: timedelta
+
+    def __post_init__(self) -> None:
+        if self.delta < timedelta(minutes=1):
+            raise ValueError("Interval must be at least 1 minute")
+
+    @classmethod
+    def parse(cls, raw: str) -> "Interval":
+        if not (match := _INTERVAL_RE.match(raw.strip().lower())) or not (match.group(1) or match.group(2)):
+            raise ValueError(f"Invalid interval: '{raw}'. Use e.g. '30m', '2h', '1h30m'")
+        hours = int(match.group(1) or 0)
+        minutes = int(match.group(2) or 0)
+        return cls(delta=timedelta(hours=hours, minutes=minutes))
+
+    def __str__(self) -> str:
+        total_minutes = int(self.delta.total_seconds() / 60)
+        h, m = divmod(total_minutes, 60)
+        if h and m:
+            return f"{h}h{m}m"
+        if h:
+            return f"{h}h"
+        return f"{m}m"
+
+
+@dataclass(frozen=True)
+class DaySpec:
+    days: frozenset[int]
+
+    @classmethod
+    def parse(cls, raw: str) -> "DaySpec":
+        days: set[int] = set()
+        for part in raw.split(","):
+            token = part.strip().lower()
+            if not token:
+                continue
+            if token in DAY_KEYWORDS:
+                days.update(DAY_KEYWORDS[token])
+            elif token in DAY_NAMES:
+                days.add(DAY_NAMES[token])
+            else:
+                raise ValueError(f"Invalid day: '{token}'. Use: {', '.join(DAY_NAMES)} / daily / weekdays")
+        return cls(days=frozenset(days) if days else ALL_DAYS)
+
+    def __str__(self) -> str:
+        if self.days == ALL_DAYS:
+            return "daily"
+        if self.days == WEEKDAY_SET:
+            return "weekdays"
+        reverse = {v: k for k, v in DAY_NAMES.items()}
+        return ",".join(reverse[d] for d in sorted(self.days))
+
+    def __contains__(self, weekday: int) -> bool:
+        return weekday in self.days
 
 
 def parse_interval(raw: str) -> timedelta:
-    if not (match := _INTERVAL_RE.match(raw.strip().lower())) or not (match.group(1) or match.group(2)):
-        raise ValueError(f"Invalid interval: '{raw}'. Use e.g. '30m', '2h', '1h30m'")
-    hours = int(match.group(1) or 0)
-    minutes = int(match.group(2) or 0)
-    total = timedelta(hours=hours, minutes=minutes)
-    if total < timedelta(minutes=1):
-        raise ValueError("Interval must be at least 1 minute")
-    return total
+    return Interval.parse(raw).delta
 
 
 def normalize_lead_minutes(raw: int | str | None) -> int | None:
@@ -81,111 +127,105 @@ def normalize_lead_minutes(raw: int | str | None) -> int | None:
     return int(raw)
 
 
-def _advance_to_days(candidate: datetime, target_days: frozenset[int]) -> datetime:
+def _advance_to_days(candidate: datetime, day_spec: DaySpec) -> datetime:
     for _ in range(DAYS_IN_WEEK):
-        if candidate.weekday() in target_days:
+        if candidate.weekday() in day_spec:
             return candidate
         candidate += timedelta(days=1)
     return candidate
 
 
-def compute_next_schedule(at: str, days: str, after: datetime) -> datetime:
+def compute_next_schedule(at: TimeOfDay | str, days: DaySpec | str | None, after: datetime) -> datetime:
+    if isinstance(at, str):
+        at = TimeOfDay.parse(at)
+    day_spec = DaySpec.parse(days) if isinstance(days, str) else (days or DaySpec(ALL_DAYS))
     local_now = after.astimezone()
-    hour, minute = (int(x) for x in at.split(":"))
-    target_days = resolve_days(days)
 
-    candidate = local_now.replace(hour=hour, minute=minute, second=0, microsecond=0)
+    candidate = local_now.replace(hour=at.hour, minute=at.minute, second=0, microsecond=0)
     if candidate <= local_now:
         candidate += timedelta(days=1)
 
-    candidate = _advance_to_days(candidate, target_days)
+    candidate = _advance_to_days(candidate, day_spec)
     return candidate.astimezone(UTC)
 
 
 def compute_next_interval(
-    every: str,
-    days: str | None,
+    every: Interval | str,
+    days: DaySpec | str | None,
     after: datetime,
-    start: str | None = None,
-    end: str | None = None,
+    start: TimeOfDay | str | None = None,
+    end: TimeOfDay | str | None = None,
 ) -> datetime:
-    delta = parse_interval(every)
-    local_now = after.astimezone()
-    candidate = local_now + delta
+    if isinstance(every, str):
+        every = Interval.parse(every)
+    if isinstance(start, str):
+        start = TimeOfDay.parse(start)
+    if isinstance(end, str):
+        end = TimeOfDay.parse(end)
 
-    if start:
-        h, m = (int(x) for x in start.split(":"))
-        window_start = time(h, m)
-    else:
-        window_start = None
-
-    if end:
-        h, m = (int(x) for x in end.split(":"))
-        window_end = time(h, m)
-    else:
-        window_end = None
+    local_now = after.astimezone()
+    candidate = local_now + every.delta
 
-    if window_start and candidate.time() < window_start:
-        candidate = candidate.replace(hour=window_start.hour, minute=window_start.minute, second=0, microsecond=0)
+    if start and candidate.time() < start.to_time():
+        candidate = candidate.replace(hour=start.hour, minute=start.minute, second=0, microsecond=0)
 
-    if window_end and candidate.time() >= window_end:
+    if end and candidate.time() >= end.to_time():
         candidate = (candidate + timedelta(days=1)).replace(
-            hour=window_start.hour if window_start else 0,
-            minute=window_start.minute if window_start else 0,
+            hour=start.hour if start else 0,
+            minute=start.minute if start else 0,
             second=0,
             microsecond=0,
         )
 
     if days:
-        candidate = _advance_to_days(candidate, resolve_days(days))
+        day_spec = DaySpec.parse(days) if isinstance(days, str) else days
+        candidate = _advance_to_days(candidate, day_spec)
 
     return candidate.astimezone(UTC)
 
 
 @dataclass
 class TimeTrigger:
     type: Literal["time"] = "time"
-    at: str | None = None
-    days: str | None = None
-    every: str | None = None
-    start: str | None = None
-    end: str | None = None
+    at: TimeOfDay | None = None
+    days: DaySpec | None = None
+    every: Interval | None = None
+    start: TimeOfDay | None = None
+    end: TimeOfDay | None = None
 
     def __post_init__(self) -> None:
-        if self.at:
-            self.at = _validate_time(self.at, "at")
-        if self.days:
-            self.days = validate_days(self.days)
-        if self.start:
-            self.start = _validate_time(self.start, "start")
-        if self.end:
-            self.end = _validate_time(self.end, "end")
-        if self.every:
-            parse_interval(self.every)
+        if isinstance(self.at, str):
+            self.at = TimeOfDay.parse(self.at)
+        if isinstance(self.days, str):
+            self.days = DaySpec.parse(self.days)
+        if isinstance(self.every, str):
+            self.every = Interval.parse(self.every)
+        if isinstance(self.start, str):
+            self.start = TimeOfDay.parse(self.start)
+        if isinstance(self.end, str):
+            self.end = TimeOfDay.parse(self.end)
 
         if not self.at and not self.every:
             raise ValueError("Either 'at' (specific time) or 'every' (interval) is required")
         if self.at and self.every:
             raise ValueError("'at' and 'every' are mutually exclusive")
-        if self.every and not self.days and not self.start and not self.end:
-            pass  # Pure interval, runs continuously
         if (self.start or self.end) and not self.every:
             raise ValueError("'start'/'end' time windows require 'every' (interval mode)")
-        if self.start and self.end and self.start >= self.end:
+        if self.start and self.end and self.start.to_time() >= self.end.to_time():
             raise ValueError(f"'start' ({self.start}) must be before 'end' ({self.end})")
 
     def params(self) -> dict:
         d: dict = {}
         if self.at:
-            d["at"] = self.at
+            d["at"] = str(self.at)
         if self.days:
-            d["days"] = self.days
+            d["days"] = str(self.days)
         if self.every:
-            d["every"] = self.every
+            d["every"] = str(self.every)
         if self.start:
-            d["start"] = self.start
+            d["start"] = str(self.start)
         if self.end:
-            d["end"] = self.end
+            d["end"] = str(self.end)
         return d
 
     @property
@@ -199,7 +239,7 @@ def label(self) -> str:
             if self.start and self.end:
                 base += f" ({self.start}\u2013{self.end})"
         else:
-            base = self.at or ""
+            base = str(self.at) if self.at else ""
         if self.days:
             base += f"  {self.days}"
         return base
@@ -293,19 +333,10 @@ def next_run(self, after: datetime) -> datetime | None:
 Trigger = TimeTrigger | EventTrigger | IdleTrigger | CountTrigger
 
 
-def _validate_time(value: str, label: str) -> str:
-    match = _TIME_RE.match(value.strip())
-    if not match or not (0 <= int(match.group(1)) <= 23 and 0 <= int(match.group(2)) <= 59):
-        raise ValueError(f"Invalid {label} format '{value}'. Use HH:MM (24h)")
-    return f"{int(match.group(1)):02d}:{int(match.group(2)):02d}"
-
-
 def _next_run_for_time(trigger: TimeTrigger, now: datetime) -> datetime:
     if trigger.every:
         return compute_next_interval(trigger.every, trigger.days, now, trigger.start, trigger.end)
-    if trigger.days:
-        return compute_next_schedule(trigger.at, trigger.days, now)
-    return compute_next_schedule(trigger.at, "daily", now)
+    return compute_next_schedule(trigger.at, trigger.days, now)
 
 
 BuildHandler = Callable[..., tuple[Trigger, datetime | None]]

ntrp/core/agent.py

@@ -10,7 +10,7 @@
 from ntrp.events.sse import SSEEvent, TextDeltaEvent, TextEvent, ToolResultEvent
 from ntrp.llm.models import get_model
 from ntrp.llm.router import get_completion_client
-from ntrp.llm.types import CompletionResponse, ToolCall
+from ntrp.llm.types import CompletionResponse, Role, ToolCall
 from ntrp.logging import get_logger
 from ntrp.tools.core.context import ToolContext
 from ntrp.tools.executor import ToolExecutor
@@ -64,12 +64,12 @@ def _init_messages(self, task: str, history: list[dict] | None) -> None:
         self.messages.clear()
         if history:
             self.messages.extend(history)
-            self.messages.append({"role": "user", "content": task})
+            self.messages.append({"role": Role.USER, "content": task})
         else:
             self.messages.extend(
                 [
-                    {"role": "system", "content": self.system_prompt},
-                    {"role": "user", "content": task},
+                    {"role": Role.SYSTEM, "content": self.system_prompt},
+                    {"role": Role.USER, "content": task},
                 ]
             )
 
@@ -127,7 +127,7 @@ def _append_tool_results(self, tool_calls: list[ToolCall], results: dict[str, st
                 result = "Error: tool execution failed"
             self.messages.append(
                 {
-                    "role": "tool",
+                    "role": Role.TOOL,
                     "tool_call_id": tc.id,
                     "content": result,
                 }

ntrp/core/compactor.py

@@ -11,6 +11,7 @@
 from ntrp.context.prompts import SUMMARIZE_PROMPT_TEMPLATE
 from ntrp.llm.models import get_model
 from ntrp.llm.router import get_completion_client
+from ntrp.llm.types import Role
 from ntrp.llm.utils import blocks_to_text
 
 
@@ -58,7 +59,7 @@ def compactable_range(
     keep_count = max(4, int(compressible * keep_ratio))
     tail_start = n - keep_count
 
-    while tail_start < n and messages[tail_start]["role"] == "tool":
+    while tail_start < n and messages[tail_start]["role"] == Role.TOOL:
         tail_start += 1
 
     if tail_start <= 1 or tail_start >= n:
@@ -70,7 +71,7 @@ def compactable_range(
 def _build_conversation_text(messages: list, start: int, end: int) -> str:
     text_parts = []
     for msg in messages[start:end]:
-        if (role := msg["role"]) == "tool":
+        if (role := msg["role"]) == Role.TOOL:
             continue
         content = blocks_to_text(msg["content"])
         if not content:
@@ -88,8 +89,8 @@ def _build_summarize_request(conversation_text: str, model: str, summary_max_tok
     return {
         "model": model,
         "messages": [
-            {"role": "system", "content": prompt},
-            {"role": "user", "content": conversation_text},
+            {"role": Role.SYSTEM, "content": prompt},
+            {"role": Role.USER, "content": conversation_text},
         ],
         "temperature": 0.3,
         "max_tokens": summary_max_tokens,
@@ -118,7 +119,7 @@ async def compact_summarize(
 def _build_compacted_messages(messages: list[dict], end: int, summary: str) -> list[dict]:
     return [
         messages[0],
-        {"role": "assistant", "content": f"[Session State Handoff]\n{summary}"},
+        {"role": Role.ASSISTANT, "content": f"[Session State Handoff]\n{summary}"},
         *messages[end:],
     ]