# -*- coding: utf-8 -*- # Copyright 2024-2026 Nexa Systems Inc. # License OPL-1 (Odoo Proprietary License v1.0) """Repair AI Service - single guardrailed entry point for client-portal AI. Per the design spec (Appendix A), this AbstractModel: 1) Builds a strict system prompt forbidding medical advice, diagnoses, stop-using recommendations, etc. 2) Calls fusion.api.service.call_openai() if available (try/fallback per fusion-api-integration rule - never installs as a hard dep) 3) JSON-schema validates the response and runs a forbidden-phrase regex 4) Always falls back to deterministic fusion.repair.self.check.rule records on any failure - intake must never be blocked by AI System prompt + JSON schema live in ir.config_parameter so the office can refine them without code changes. """ import hashlib import json import logging import re from odoo import api, fields, models _logger = logging.getLogger(__name__) # ----- Safety filters ----- FORBIDDEN_PATTERNS = [ re.compile(r'\b(diagnos(e|is|ed|ing))\b', re.I), re.compile(r'\byou have\b', re.I), re.compile(r'\bmedical condition\b', re.I), re.compile(r'\b(stop|should\s+stop)\s+using\b', re.I), re.compile(r'\bconsult\s+(your|a)\s+(doctor|physician|nurse)\b', re.I), re.compile(r'\b(blood\s+pressure|heart\s+rate|pulse|oxygen)\b', re.I), re.compile(r'(\$|CAD|USD)\s?\d+', re.I), # No price mentions ] # Universal hard-escalate: ANY equipment category - fire / smoke / sparks / # burning / injury / trapped is always an immediate escalation. Word # boundaries prevent "unhurt" matching "hurt" and "fireman" matching "fire". UNIVERSAL_ESCALATION_RE = re.compile( r'\b(fire|smoke|burning|spark|injur(y|ed)|hurt|bleeding|trapped)\b', re.I, ) # Category-specific safety symptoms - only fire if the category is flagged # safety_critical=True on fusion.repair.product.category (stairlifts, # porch lifts, power wheelchairs). "won.?t" handles both "won't" and "wont". SAFETY_SYMPTOMS_RE = re.compile( r"\b(stuck|motor|brake\s*fail|won.?t\s*stop|overshoot)\b", re.I, ) DEFAULT_SYSTEM_PROMPT = ( "You are a triage assistant for Fusion Repairs, a Canadian medical " "equipment service company. Your ONLY job is to suggest 1-3 safe, " "reversible self-check steps a client can try on their medical equipment " "before scheduling a technician visit.\n\n" "ABSOLUTE RULES:\n" "1. NEVER provide medical advice, diagnoses, or health recommendations.\n" "2. NEVER claim a definitive cause for the problem.\n" "3. NEVER recommend stopping use of medical equipment.\n" "4. NEVER use phrases like 'you have', 'I diagnose', 'you should stop', " "'medical condition', 'consult your doctor'.\n" "5. ONLY suggest steps that are: safe, reversible, require no tools, " "take under 2 minutes, and pose zero risk to the client or equipment.\n" "6. If symptoms involve smoke, sparks, burning smell, motors on " "stairlifts/porch lifts, OR if you are uncertain -> return " "escalate_immediately: true.\n" "7. Maximum 3 steps. Each step <= 1 sentence. Grade-6 reading level. " "No technical jargon.\n" "8. NEVER reference part numbers, prices, or other clients.\n" "9. If client reports injury, equipment fire, or person trapped -> " "escalate_immediately: true with escalation_reason: 'emergency'.\n" "10. You MUST output valid JSON matching the provided schema. No prose, " "no markdown, no commentary." ) class FusionRepairAIService(models.AbstractModel): _name = 'fusion.repair.ai.service' _description = 'Repair AI Service - guardrailed self-check engine' # ------------------------------------------------------------------ # PUBLIC API # ------------------------------------------------------------------ @api.model def suggest_self_check(self, product_category_id=None, symptoms=None, urgency=None): """Return a list of safe self-check steps for the client portal. Returns a dict with shape: { 'escalate_immediately': bool, 'escalation_reason': str | None, 'confidence': 'high' | 'medium' | 'low', 'steps': [{'instruction': str, 'expected_result': str, 'safety_note': str | None}, ...], 'source': 'ai' | 'fallback' | 'escalated', 'disclaimer': str, } """ symptoms = [s for s in (symptoms or []) if s] category = ( self.env['fusion.repair.product.category'].sudo().browse(product_category_id) if product_category_id else False ) # Pre-check: hard-escalate for safety-critical category + symptom combos # without consulting AI. This is BEFORE any AI call so even if AI is # down we still escalate the right way. if self._should_hard_escalate(category, symptoms, urgency): return self._escalated_response('safety') # Try the AI, fall back to deterministic rules on any failure. ai_result = self._try_ai(category, symptoms) if ai_result: ai_result['source'] = 'ai' ai_result['disclaimer'] = self._disclaimer() return ai_result return self._deterministic_fallback(category, symptoms) # ------------------------------------------------------------------ # HARD ESCALATION # ------------------------------------------------------------------ @api.model def _should_hard_escalate(self, category, symptoms, urgency): if urgency == 'safety': return True text = ' '.join(symptoms) # Universal: fire / smoke / spark / burning / injury / trapped escalate # regardless of equipment category. Electrical fire on a hospital bed # is exactly as urgent as on a stairlift. if UNIVERSAL_ESCALATION_RE.search(text): return True # Category-specific: 'stuck', 'motor', 'brake fail', etc. only escalate # on safety-critical categories (stairlifts, porch lifts, power chairs). if category and category.safety_critical and SAFETY_SYMPTOMS_RE.search(text): return True return False @api.model def _escalated_response(self, reason): return { 'escalate_immediately': True, 'escalation_reason': reason, 'confidence': 'high', 'steps': [], 'source': 'escalated', 'disclaimer': self._disclaimer(), } # ------------------------------------------------------------------ # AI CALL (try/fallback) # ------------------------------------------------------------------ @api.model def _try_ai(self, category, symptoms): try: ApiService = self.env.get('fusion.api.service') if not ApiService: return None messages = [ {'role': 'system', 'content': self._system_prompt()}, {'role': 'user', 'content': self._user_prompt(category, symptoms)}, ] cache_key = self._cache_key(category, symptoms) cached = self._cache_get(cache_key) if cached: return cached raw = ApiService.call_openai( consumer='fusion_repairs', feature='client_self_triage', messages=messages, max_tokens=400, ) if not raw: return None parsed = self._safe_parse(raw) if not parsed: self._log_incident('parse_failed', raw) return None self._cache_set(cache_key, parsed) return parsed except Exception as e: _logger.info('AI self-check skipped: %s', e) return None @api.model def _system_prompt(self): ICP = self.env['ir.config_parameter'].sudo() return ICP.get_param( 'fusion_repairs.ai_self_check_system_prompt', DEFAULT_SYSTEM_PROMPT, ) @api.model def _user_prompt(self, category, symptoms): cat_name = category.name if category else 'medical equipment' return ( f"Equipment category: {cat_name}\n" f"Reported symptoms: {'; '.join(symptoms) if symptoms else '(none provided)'}\n" "Output the JSON object only." ) # ------------------------------------------------------------------ # SAFE PARSE + VALIDATE # ------------------------------------------------------------------ @api.model def _safe_parse(self, raw): """Parse the AI response, validate against the JSON schema, and run forbidden-phrase regex filters. Returns None on any failure - caller falls back to deterministic rules.""" if not raw: return None text = raw.strip() # Strip code-fence wrapping if AI added it. if text.startswith('```'): text = re.sub(r'^```[a-zA-Z]*\n?', '', text) text = re.sub(r'\n?```$', '', text) try: data = json.loads(text) except (ValueError, TypeError): return None # Schema check (minimal - we don't pull in jsonschema as a dep) if not isinstance(data, dict): return None if not isinstance(data.get('escalate_immediately'), bool): return None confidence = data.get('confidence') if confidence not in ('high', 'medium', 'low'): return None steps = data.get('steps') if not isinstance(steps, list) or len(steps) > 3: return None # Coherence: not-escalated must have at least one step. if not data['escalate_immediately'] and not steps: return None # Per-step validation + forbidden-phrase scan. cleaned_steps = [] for step in steps: if not isinstance(step, dict): return None instr = step.get('instruction') expected = step.get('expected_result') if not isinstance(instr, str) or not instr.strip(): return None if not isinstance(expected, str) or not expected.strip(): return None if len(instr) > 200 or len(expected) > 200: return None if self._contains_forbidden(instr) or self._contains_forbidden(expected): return None note = step.get('safety_note') if note is not None and (not isinstance(note, str) or len(note) > 200): return None if note and self._contains_forbidden(note): return None cleaned_steps.append({ 'instruction': instr.strip(), 'expected_result': expected.strip(), 'safety_note': (note or '').strip() or None, }) return { 'escalate_immediately': data['escalate_immediately'], 'escalation_reason': data.get('escalation_reason') or None, 'confidence': confidence, 'steps': cleaned_steps, } @api.model def _contains_forbidden(self, text): if not text: return False return any(p.search(text) for p in FORBIDDEN_PATTERNS) # ------------------------------------------------------------------ # DETERMINISTIC FALLBACK # ------------------------------------------------------------------ @api.model def _normalise(self, text): """Strip punctuation + lowercase so 'wont move' matches 'won't move' and vice versa. IMPORTANT: apostrophes are REMOVED (not replaced with space), so "won't" -> "wont" matches user input "wont" (without apostrophe). Other punctuation collapses to a single space. """ s = (text or "").lower() # Remove ALL apostrophe variants (straight + curly) so contraction # forms collide with apostrophe-less forms. for apos in ("'", "\u2019", "\u2018", "\u02bc"): s = s.replace(apos, "") # Everything else non-alphanumeric -> single space. return re.sub(r"[^a-z0-9 ]+", " ", s) @api.model def _deterministic_fallback(self, category, symptoms): """Look up fusion.repair.self.check.rule records for the category and return the matching steps. Used when AI is unavailable or returns invalid / unsafe content.""" Rule = self.env['fusion.repair.self.check.rule'].sudo() steps = [] if category: haystack = self._normalise(' '.join(symptoms)) rules = Rule.search([ ('category_id', '=', category.id), ('active', '=', True), ], order='sequence') for r in rules: kws = [ self._normalise(k) for k in (r.symptom_keywords or '').split(',') if k.strip() ] if not kws or any(kw and kw in haystack for kw in kws): steps.append({ 'instruction': r.instruction or '', 'expected_result': r.expected_result or '', 'safety_note': r.safety_note or None, }) if len(steps) >= 3: break return { 'escalate_immediately': len(steps) == 0, 'escalation_reason': None if steps else 'no_match', 'confidence': 'medium' if steps else 'low', 'steps': steps, 'source': 'fallback', 'disclaimer': self._disclaimer(), } # ------------------------------------------------------------------ # CACHE (in-memory per worker, 24h) # ------------------------------------------------------------------ _CACHE = {} _CACHE_TTL = 24 * 3600 @api.model def _cache_key(self, category, symptoms): symptom_hash = hashlib.sha256( ('|'.join(sorted(s.lower() for s in symptoms))).encode() ).hexdigest()[:16] return f"{category.code if category else 'none'}:{symptom_hash}" @api.model def _cache_get(self, key): import time entry = self._CACHE.get(key) if not entry: return None ts, value = entry if time.time() - ts > self._CACHE_TTL: self._CACHE.pop(key, None) return None return value @api.model def _cache_set(self, key, value): import time # Bound cache size to ~512 entries. if len(self._CACHE) > 512: self._CACHE.clear() self._CACHE[key] = (time.time(), value) # ------------------------------------------------------------------ # MISC # ------------------------------------------------------------------ @api.model def _disclaimer(self): return ("This is not medical advice. If you're unsure, schedule a " "technician visit. In an emergency, call 9-1-1.") @api.model def _log_incident(self, kind, raw): _logger.warning('AI self-check incident (%s): %s', kind, (raw or '')[:300])