Rhinox13/chatapi
0
1from __future__ import annotations
2
3import json
4import regex
5import threading
6import time
7import uuid
8from dataclasses import dataclass
9from typing import Any
10
11from .output_controller import TurnOutputController
12from .pending import PendingTurn
13
14LEGACY_HEARTBEAT_RULE_ID = "legacy-heartbeat"
15REGEX_MATCH_TIMEOUT_SECONDS = 0.1
16REGEX_MAX_PATTERN_LENGTH = 512
17
18
19
20def _as_string_list(value: Any) -> list[str]:
21 if isinstance(value, str):
22 text = value.strip()
23 return [text] if text else []
24 if not isinstance(value, list):
25 return []
26 result: list[str] = []
27 for item in value:
28 if not isinstance(item, str):
29 continue
30 text = item.strip()
31 if text:
32 result.append(text)
33 return result
34
35
36def _as_non_negative_float(value: Any) -> float:
37 try:
38 return float(value or 0.0)
39 except (TypeError, ValueError):
40 raise ValueError("timing values must be numbers")
41
42
43def _normalize_match_entries(value: Any) -> list[dict[str, str]]:
44 if not isinstance(value, list):
45 return []
46 result: list[dict[str, str]] = []
47 for item in value:
48 if isinstance(item, str):
49 text = item.strip()
50 if text:
51 result.append({"match_type": "substring", "pattern": text})
52 continue
53 if not isinstance(item, dict):
54 continue
55 pattern = str(item.get("pattern") or "").strip()
56 if not pattern:
57 continue
58 match_type = str(item.get("match_type") or "substring").strip() or "substring"
59 result.append(
60 {
61 "match_type": match_type,
62 "pattern": pattern,
63 }
64 )
65 return result
66
67
68@dataclass(frozen=True)
69class AutomationRule:
70 id: str
71 enabled: bool
72 contains: list[dict[str, str]]
73 excludes: list[dict[str, str]]
74 delay_seconds: float
75 repeat_interval_seconds: float
76 max_output_count: int
77 action_type: str
78 action_text: str
79 error_message: str
80 tool_name: str = ""
81 tool_arguments: str = ""
82 tool_call_id: str = ""
83
84 def matches(self, input_text: str) -> bool:
85 if self.contains and not all(_match_pattern(item, input_text) for item in self.contains):
86 return False
87 if any(_match_pattern(item, input_text) for item in self.excludes):
88 return False
89 return True
90
91
92def _match_pattern(item: dict[str, str], input_text: str) -> bool:
93 match_type = str(item.get("match_type") or "substring").strip() or "substring"
94 pattern = str(item.get("pattern") or "")
95 if not pattern:
96 return False
97 if match_type == "regex":
98 if len(pattern) > REGEX_MAX_PATTERN_LENGTH:
99 return False
100 try:
101 return regex.search(pattern, input_text, timeout=REGEX_MATCH_TIMEOUT_SECONDS) is not None
102 except (regex.error, TimeoutError):
103 return False
104 return pattern in input_text
105
106
107def _validate_tool_arguments(arguments_json: str, schema: dict[str, Any]) -> bool:
108 try:
109 args = json.loads(arguments_json) if arguments_json.strip() else {}
110 except (json.JSONDecodeError, ValueError):
111 return False
112 if not isinstance(args, dict):
113 return False
114 properties = schema.get("properties")
115 if isinstance(properties, dict) and properties:
116 for key in args:
117 if key not in properties:
118 return False
119 required = schema.get("required")
120 if isinstance(required, list):
121 for field in required:
122 if field not in args:
123 return False
124 return True
125
126
127def normalize_rule_payload(raw_rule: dict[str, Any]) -> dict[str, Any]:
128 conditions = raw_rule.get("conditions")
129 timing = raw_rule.get("timing")
130 action = raw_rule.get("action")
131 if not isinstance(conditions, dict):
132 conditions = {}
133 if not isinstance(timing, dict):
134 timing = {}
135 if not isinstance(action, dict):
136 action = {}
137 delay_seconds = _as_non_negative_float(timing.get("delay_seconds"))
138 repeat_interval_seconds = _as_non_negative_float(timing.get("repeat_interval_seconds"))
139 try:
140 max_output_count = int(timing.get("max_output_count") or 120)
141 except (TypeError, ValueError):
142 max_output_count = 120
143 max_output_count = max(1, max_output_count)
144 return {
145 "id": str(raw_rule.get("id") or f"rule_{uuid.uuid4().hex[:8]}"),
146 "enabled": bool(raw_rule.get("enabled", True)),
147 "conditions": {
148 "contains": _normalize_match_entries(conditions.get("contains")),
149 "excludes": _normalize_match_entries(conditions.get("excludes")),
150 },
151 "timing": {
152 "delay_seconds": delay_seconds,
153 "repeat_interval_seconds": repeat_interval_seconds,
154 "max_output_count": max_output_count,
155 },
156 "action": {
157 "type": str(action.get("type") or "").strip(),
158 "text": str(action.get("text") or ""),
159 "error_message": str(action.get("error_message") or ""),
160 "tool_name": str(action.get("tool_name") or ""),
161 "tool_arguments": str(action.get("tool_arguments") or ""),
162 "tool_call_id": str(action.get("tool_call_id") or ""),
163 },
164 }
165
166
167def validate_rule_payload(raw_rule: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None]:
168 if not isinstance(raw_rule, dict):
169 return None, "rule must be an object"
170 try:
171 normalized = normalize_rule_payload(raw_rule)
172 except ValueError as error:
173 return None, str(error)
174 if normalized["timing"]["delay_seconds"] < 0:
175 return None, "delay_seconds must be greater than or equal to 0"
176 if normalized["timing"]["repeat_interval_seconds"] < 0:
177 return None, "repeat_interval_seconds must be greater than or equal to 0"
178 if normalized["timing"].get("max_output_count", 120) < 1:
179 return None, "max_output_count must be greater than 0"
180 for group_name in ("contains", "excludes"):
181 for item in normalized["conditions"][group_name]:
182 match_type = str(item.get("match_type") or "")
183 pattern = str(item.get("pattern") or "")
184 if match_type not in {"substring", "regex"}:
185 return None, "condition match_type must be substring or regex"
186 if not pattern:
187 return None, "condition pattern is required"
188 if match_type == "regex":
189 if len(pattern) > REGEX_MAX_PATTERN_LENGTH:
190 return None, f"condition regex pattern is too long, max {REGEX_MAX_PATTERN_LENGTH} characters"
191 try:
192 regex.compile(pattern)
193 except regex.error as error:
194 return None, f"invalid regex: {error}"
195 action_type = normalized["action"]["type"]
196 if action_type not in {"output_text", "complete", "error", "tool_call"}:
197 return None, "action.type must be one of output_text, complete, error, tool_call"
198 if action_type == "output_text" and not normalized["action"]["text"]:
199 return None, "output_text rule requires action.text"
200 if action_type == "error" and not normalized["action"]["error_message"]:
201 return None, "error rule requires action.error_message"
202 if action_type == "tool_call" and not normalized["action"].get("tool_name"):
203 return None, "tool_call rule requires action.tool_name"
204 return normalized, None
205
206
207def materialize_rule(payload: dict[str, Any]) -> AutomationRule:
208 action = payload.get("action", {})
209 return AutomationRule(
210 id=str(payload["id"]),
211 enabled=bool(payload["enabled"]),
212 contains=list(payload["conditions"]["contains"]),
213 excludes=list(payload["conditions"]["excludes"]),
214 delay_seconds=float(payload["timing"]["delay_seconds"]),
215 repeat_interval_seconds=float(payload["timing"]["repeat_interval_seconds"]),
216 max_output_count=max(1, int(payload["timing"].get("max_output_count", 120) or 120)),
217 action_type=str(action.get("type") or ""),
218 action_text=str(action.get("text") or ""),
219 error_message=str(action.get("error_message") or ""),
220 tool_name=str(action.get("tool_name") or ""),
221 tool_arguments=str(action.get("tool_arguments") or ""),
222 tool_call_id=str(action.get("tool_call_id") or ""),
223 )
224
225
226class AutomationRuleEngine:
227 def __init__(self, *, user_store: Any, output_controller: TurnOutputController):
228 self._user_store = user_store
229 self._output_controller = output_controller
230
231 def load_rule_payloads(self, owner_id: str) -> list[dict[str, Any]]:
232 raw = self._user_store.get_automation_rules(owner_id)
233 try:
234 data = json.loads(raw)
235 except json.JSONDecodeError:
236 data = []
237 if not isinstance(data, list):
238 return []
239 result: list[dict[str, Any]] = []
240 for item in data:
241 normalized, error = validate_rule_payload(item)
242 if normalized is None or error is not None:
243 continue
244 result.append(normalized)
245 return result
246
247 def save_rule_payloads(self, owner_id: str, rules: list[dict[str, Any]]) -> list[dict[str, Any]]:
248 validated: list[dict[str, Any]] = []
249 for item in rules:
250 normalized, error = validate_rule_payload(item)
251 if normalized is None:
252 raise ValueError(error or "invalid rule")
253 validated.append(normalized)
254 self._user_store.set_automation_rules(owner_id, json.dumps(validated, ensure_ascii=False))
255 return validated
256
257 def get_heartbeat_rule_settings(self, owner_id: str) -> dict[str, Any]:
258 for rule in self.load_rule_payloads(owner_id):
259 if str(rule.get("id")) != LEGACY_HEARTBEAT_RULE_ID:
260 continue
261 text = str(rule["action"]["text"])
262 interval = float(rule["timing"]["repeat_interval_seconds"] or 0.0)
263 return {
264 "heartbeat_text": text,
265 "heartbeat_interval_seconds": interval,
266 }
267 return {
268 "heartbeat_text": "",
269 "heartbeat_interval_seconds": 0.0,
270 }
271
272 def update_heartbeat_rule_settings(self, owner_id: str, *, heartbeat_text: str, interval_seconds: float) -> dict[str, Any]:
273 rules = [rule for rule in self.load_rule_payloads(owner_id) if str(rule.get("id")) != LEGACY_HEARTBEAT_RULE_ID]
274 if heartbeat_text and interval_seconds > 0:
275 rules.append(
276 {
277 "id": LEGACY_HEARTBEAT_RULE_ID,
278 "enabled": True,
279 "conditions": {"contains": [], "excludes": []},
280 "timing": {
281 "delay_seconds": float(interval_seconds),
282 "repeat_interval_seconds": float(interval_seconds),
283 "max_output_count": 120,
284 },
285 "action": {
286 "type": "output_text",
287 "text": heartbeat_text,
288 "error_message": "",
289 },
290 }
291 )
292 self.save_rule_payloads(owner_id, rules)
293 return {
294 "heartbeat_text": heartbeat_text,
295 "heartbeat_interval_seconds": float(interval_seconds),
296 }
297
298 def start_for_pending(self, pending: PendingTurn) -> None:
299 input_text = pending.input_text
300 owner_id = pending.owner_id
301 rules = [
302 materialize_rule(payload)
303 for payload in self.load_rule_payloads(owner_id)
304 if bool(payload.get("enabled", True))
305 ]
306 for rule in rules:
307 if not rule.matches(input_text):
308 continue
309 worker = threading.Thread(
310 target=self._run_rule,
311 args=(pending, rule),
312 daemon=True,
313 name=f"automation-rule-{rule.id}",
314 )
315 worker.start()
316
317 def _run_rule(self, pending: PendingTurn, rule: AutomationRule) -> None:
318 owner_id = pending.owner_id
319 conversation_id = pending.conversation_id
320 if self._sleep_until_ready(pending, rule.delay_seconds):
321 return
322 output_count = 0
323 while True:
324 if pending.event.is_set():
325 return
326 try:
327 if rule.action_type == "output_text":
328 self._output_controller.add_text_delta(
329 conversation_id=conversation_id,
330 owner_id=owner_id,
331 text=rule.action_text,
332 )
333 output_count += 1
334 if output_count >= rule.max_output_count:
335 return
336 elif rule.action_type == "complete":
337 self._output_controller.complete_assistant_message(
338 conversation_id=conversation_id,
339 owner_id=owner_id,
340 provider="rule",
341 )
342 return
343 elif rule.action_type == "error":
344 self._output_controller.abort(
345 conversation_id=conversation_id,
346 owner_id=owner_id,
347 error_message=rule.error_message,
348 )
349 return
350 elif rule.action_type == "tool_call":
351 tool_name = str(rule.tool_name or "").strip()
352 tool_arguments = str(rule.tool_arguments or "")
353 tool_call_id = str(rule.tool_call_id or "") or None
354 if not tool_name:
355 return
356 if pending.available_tool_names and tool_name not in pending.available_tool_names:
357 return
358 tool_schema = pending.available_tool_schemas.get(tool_name)
359 if tool_schema is not None:
360 if not _validate_tool_arguments(tool_arguments, tool_schema):
361 return
362 self._output_controller.complete_tool_call(
363 conversation_id=conversation_id,
364 owner_id=owner_id,
365 tool_name=tool_name,
366 arguments=tool_arguments,
367 provider="rule",
368 tool_call_id=tool_call_id,
369 )
370 return
371 except ValueError:
372 return
373
374 if rule.repeat_interval_seconds <= 0:
375 return
376 if self._sleep_until_ready(pending, rule.repeat_interval_seconds):
377 return
378
379 @staticmethod
380 def _sleep_until_ready(pending: PendingTurn, duration_seconds: float) -> bool:
381 remaining = max(0.0, float(duration_seconds))
382 while remaining > 0:
383 if pending.event.wait(min(0.25, remaining)):
384 return True
385 remaining -= min(0.25, remaining)
386 return pending.event.is_set()
387