from __future__ import annotations import json, re from typing import Any from agent_framework.guardrails.base import Guardrail, RailDecision from .tim_prompts.ausencia_oferta_proativa import build_aoferta_prompt from .tim_prompts.out_of_scope import build_oos_prompt from .tim_prompts.revprec import build_revprec_prompt from .tim_prompts.fraseologia import build_fraseologia_prompt def _llm(context: dict[str, Any]): return context.get('guardrail_llm') or context.get('llm') or context.get('model') def _context_text(context: dict[str, Any]) -> str: try: return json.dumps(context or {}, ensure_ascii=False, default=str)[:16000] except Exception: return str(context or {})[:16000] def _compact_aoferta_context(context: dict[str, Any]) -> str: """High-signal context for the proactive-offer LLM. The generic output context can contain very large MCP/workflow payloads. Sending the first N characters of that payload may truncate the customer request and the resolved transaction target, which makes the auditor treat legitimate fallback guidance as a new unsolicited action. Keep only authoritative facts needed by AOFERTA and cap nested operational results. """ ctx = context or {} history = [] for item in list(ctx.get('conversation_history') or [])[-8:]: if isinstance(item, dict): role = str(item.get('role') or item.get('type') or '') content = str(item.get('content') or '') if content: history.append({'role': role, 'content': content[:1200]}) pre = ctx.get('transaction_pre_validation') if isinstance(ctx.get('transaction_pre_validation'), dict) else {} requested = pre.get('requested_arguments') if isinstance(pre.get('requested_arguments'), dict) else {} resolved = pre.get('resolved_arguments') if isinstance(pre.get('resolved_arguments'), dict) else {} def compact_result(value: Any, depth: int = 0) -> Any: if depth > 6: return None if isinstance(value, list): return [compact_result(v, depth + 1) for v in value[:8]] if not isinstance(value, dict): return value if isinstance(value, (str, int, float, bool)) or value is None else str(value)[:500] keep = { 'tool_name', 'ok', 'error', 'status', 'transaction_status', 'workflow_status', 'subject', 'name', 'resolved_subject', 'resolved_subjects', 'success', 'reason', 'mensagem', 'message', 'terminal_status', 'effective_tool_name', 'requested_arguments', 'resolved_arguments', 'items', 'results', 'errors', 'cancelados', 'nao_cancelados', 'nao_encontrados', 'output', 'result', } out = {} for k, v in value.items(): if k in keep: cv = compact_result(v, depth + 1) if cv not in (None, {}, []): out[k] = cv return out payload = { 'current_user_message': str(ctx.get('current_user_message') or '')[:1200], 'current_intent': str(ctx.get('current_intent') or ctx.get('intent') or ''), 'current_route': str(ctx.get('current_route') or ctx.get('route') or ''), 'transaction_status': str(ctx.get('transaction_status') or ''), 'conversation_history': history, 'transaction_request': { 'requested_arguments': compact_result(requested), 'resolved_arguments': compact_result(resolved), 'effective_tool_name': pre.get('effective_tool_name'), 'confirmation_message': pre.get('confirmation_message'), }, 'current_execution': compact_result(ctx.get('mcp_results') or ctx.get('tool_result') or []), } return json.dumps(payload, ensure_ascii=False, default=str)[:16000] def _authorized_human_handoff(context: dict[str, Any]) -> bool: """Return True only for structurally authorized human handoff on this turn.""" ctx = context or {} route = str(ctx.get('current_route') or ctx.get('route') or '').strip().lower() intent = str(ctx.get('current_intent') or ctx.get('intent') or '').strip().lower() session_control = str(ctx.get('session_control') or '').strip().upper() requested = ctx.get('human_handoff_requested') is True handoff = ctx.get('handoff') is True route_decision = ctx.get('route_decision') if isinstance(ctx.get('route_decision'), dict) else {} route_meta = route_decision.get('metadata') if isinstance(route_decision.get('metadata'), dict) else {} rd_route = str(route_decision.get('route') or route_decision.get('agent') or '').strip().lower() rd_intent = str(route_decision.get('intent') or '').strip().lower() rd_handoff = route_decision.get('handoff') is True rd_session_control = str(route_meta.get('session_control') or '').strip().upper() control_evidence = ( session_control == 'HUMAN_HANDOFF' or requested or handoff or rd_handoff or rd_session_control == 'HUMAN_HANDOFF' ) route_evidence = ( route == 'human_handoff' or intent == 'human_handoff' or rd_route == 'human_handoff' or rd_intent == 'human_handoff' ) if control_evidence and route_evidence: return True # A resumed domain workflow may decide the handoff after the router has # already been bypassed for workflow continuation. In that case the # current route/intent legitimately remain the domain values, while the # *current-turn workflow result* is the authoritative orchestration # decision. Accept only a terminal, internally consistent handoff result; # an isolated `handoff=true` or transfer-like sentence is never enough. roots = [] for key in ('mcp_results', 'tool_result', 'workflow_result'): value = ctx.get(key) if value is not None: roots.append(value) def terminal_workflow_handoff(value: Any) -> bool: if isinstance(value, dict): workflow_control = str(value.get('session_control') or '').strip().upper() workflow_terminal = str(value.get('terminal_status') or '').strip().lower() workflow_requested = value.get('human_handoff_requested') is True workflow_handoff = value.get('handoff') is True workflow_session_ended = value.get('session_ended') is True has_control = workflow_control == 'HUMAN_HANDOFF' has_request = workflow_requested or workflow_handoff has_terminal = workflow_terminal == 'human_handoff' or workflow_session_ended if has_control and has_request and has_terminal: return True return any(terminal_workflow_handoff(item) for item in value.values()) if isinstance(value, (list, tuple)): return any(terminal_workflow_handoff(item) for item in value) return False return any(terminal_workflow_handoff(root) for root in roots) def _deterministic_tim_domain_response(text: str, context: dict[str, Any]) -> bool: """Conservative fast path for unmistakable TIM Contas domain output. This extension may know the TIM Contas vocabulary; the generic framework does not. Human-transfer language is intentionally excluded so handoff authorization remains exclusively structural. Ambiguous domain prose still goes to TIM_OOS LLM. """ normalized = re.sub(r'\s+', ' ', str(text or '').strip().lower()) if not normalized: return False handoff_markers = ( 'encaminhar', 'encaminhado', 'transferir', 'transferido', 'atendente', 'pessoa', 'humano', 'especialista', ) if any(m in normalized for m in handoff_markers): return False domain_markers = ( 'contestação', 'contestacao', 'fatura', 'cobrança', 'cobranca', 'plano', 'serviço', 'servico', 'vas', 'boleto', 'desconto', ) completion_markers = ( 'registrad', 'cancelad', 'concluíd', 'concluid', 'executad', 'processad', 'enviad', 'emitid', 'protocolo', ) # Keep the bypass intentionally narrow: an unmistakable domain noun plus an # operational/result marker. Ordinary explanations and ambiguous responses are # still semantically audited by TIM_OOS. return ( any(m in normalized for m in domain_markers) and any(m in normalized for m in completion_markers) ) def _parse_json(raw: Any) -> dict[str, Any]: text = str(getattr(raw, 'content', raw) or '').strip() m = re.search(r'\{[\s\S]*\}', text) if m: text=m.group(0) try: return json.loads(text) except Exception: return {'allowed': False, 'reason': f'Resposta inválida do guardrail TIM: {text[:300]}'} class _TimPromptRail(Guardrail): prompt_builder = None profile_name = 'guardrail' async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision: llm = _llm(context) if llm is None: return RailDecision(code=self.code, allowed=False, reason='LLM do framework indisponível para guardrail TIM', metadata={'external': True, 'fail_closed': True}) prompt = self.prompt_builder(text or '', _context_text(context)) raw = await llm.ainvoke( [{'role':'system','content':'Responda apenas JSON válido, sem markdown.'}, {'role':'user','content':prompt}], profile_name=self.profile_name, component_name=f'guardrail.external.{self.code.lower()}', generation_name=f'guardrail.external.{self.code.lower()}', ) out = _parse_json(raw) return RailDecision(code=self.code, allowed=bool(out.get('allowed', False)), reason=str(out.get('reason') or out.get('label') or ''), sanitized_text=text, metadata={'external': True, 'domain':'TIM_CONTAS', 'data':out}) class TimOutOfScopeRail(_TimPromptRail): code='TIM_OOS'; stage='output'; prompt_builder=staticmethod(build_oos_prompt) async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision: # A handoff structurally selected by the router/workflow is an authorized # orchestration response, not a domain answer to be judged as OOS by text. if _authorized_human_handoff(context): return RailDecision( code=self.code, allowed=True, reason='handoff_humano_autorizado', sanitized_text=text, metadata={ 'external': True, 'domain': 'TIM_CONTAS', 'mechanism': 'deterministic_handoff_bypass', 'data': { 'allowed': True, 'reason': 'handoff_humano_autorizado', }, }, ) if _deterministic_tim_domain_response(text, context): return RailDecision( code=self.code, allowed=True, reason='resposta suportada pelo contexto do domínio TIM Contas', sanitized_text=text, metadata={ 'external': True, 'domain': 'TIM_CONTAS', 'mechanism': 'deterministic_domain_bypass', 'data': { 'allowed': True, 'reason': 'resposta suportada pelo contexto do domínio TIM Contas', }, }, ) return await super().evaluate(text, context) class TimProactiveOfferRail(_TimPromptRail): code='TIM_AOFERTA'; stage='output'; prompt_builder=staticmethod(build_aoferta_prompt) # Mantém a mesma semântica do AOFERTA nativo do framework: mensagens de # continuidade de uma transação já aberta não constituem nova oferta # proativa. O bypass é estritamente dirigido pelo estado transacional, sem # inferência textual e sem desabilitar outros guardrails de saída. _TRANSACTION_CONTINUATION_STATUSES = { 'COLLECTING_PARAMETERS', 'AWAITING_CONFIRMATION', } @staticmethod def _authorized_human_handoff(context: dict[str, Any]) -> bool: return _authorized_human_handoff(context) @classmethod def _transaction_continuation_status(cls, context: dict[str, Any]) -> str | None: ctx = context or {} status = str(ctx.get('transaction_status') or '').strip().upper() if status in cls._TRANSACTION_CONTINUATION_STATUSES: return status # Compatibilidade com o contexto de output do Contas e com callers que # expõem o estado apenas no resultado da tool. É o mesmo fallback usado # pelo ProactiveOfferRail nativo: mcp_results -> tool_result. results = ctx.get('mcp_results') or ctx.get('tool_result') or [] if isinstance(results, dict): results = [results] for result in reversed(list(results)): if not isinstance(result, dict): continue result_status = str(result.get('transaction_status') or '').strip().upper() if result_status in cls._TRANSACTION_CONTINUATION_STATUSES: return result_status return None async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision: if self._authorized_human_handoff(context): return RailDecision( code=self.code, allowed=True, reason='handoff_humano_autorizado', sanitized_text=text, metadata={ 'external': True, 'domain': 'TIM_CONTAS', 'mechanism': 'deterministic_handoff_bypass', 'data': { 'allowed': True, 'reason': 'handoff_humano_autorizado', }, }, ) continuation_status = self._transaction_continuation_status(context) if continuation_status: return RailDecision( code=self.code, allowed=True, reason=f'continuidade_transacional:{continuation_status}', sanitized_text=text, metadata={ 'external': True, 'domain': 'TIM_CONTAS', 'mechanism': 'deterministic_transaction_bypass', 'transaction_status': continuation_status, 'data': { 'allowed': True, 'reason': f'continuidade_transacional:{continuation_status}', }, }, ) # Terminal/partial-success responses still need semantic auditing, but with # compact authoritative transaction evidence. Do not use a completed-state # bypass: the LLM must still block genuinely new targets/actions. llm = _llm(context) if llm is None: return RailDecision(code=self.code, allowed=False, reason='LLM do framework indisponível para guardrail TIM', metadata={'external': True, 'fail_closed': True}) compact_context = _compact_aoferta_context(context) prompt = self.prompt_builder(text or '', compact_context) raw = await llm.ainvoke( [{'role':'system','content':'Responda apenas JSON válido, sem markdown.'}, {'role':'user','content':prompt}], profile_name=self.profile_name, component_name=f'guardrail.external.{self.code.lower()}', generation_name=f'guardrail.external.{self.code.lower()}', ) out = _parse_json(raw) return RailDecision( code=self.code, allowed=bool(out.get('allowed', False)), reason=str(out.get('reason') or out.get('label') or ''), sanitized_text=text, metadata={ 'external': True, 'domain':'TIM_CONTAS', 'mechanism': 'llm_semantic_compact_transaction_context', 'data':out, }, ) class TimPrematureActionRail(_TimPromptRail): code='TIM_REVPREC'; stage='output'; profile_name='grl'; prompt_builder=staticmethod(build_revprec_prompt) _MESSAGE_KEYS = { 'mensagem', 'message', 'response', 'response_text', 'final_answer', 'customer_message', 'customer_response', 'answer', 'text', } @classmethod def _current_execution_evidence(cls, context: dict[str, Any]) -> list[dict[str, Any]]: """Return compact, current-turn execution evidence only. REVPREC must judge whether an operational claim is supported by what actually happened in this turn. Conversation history, LTM and previous workflow state are deliberately excluded from this evidence set. """ ctx = context or {} roots = ctx.get('mcp_results') if roots is None: roots = ctx.get('tool_result') if roots is None: return [] if isinstance(roots, dict): roots = [roots] if not isinstance(roots, (list, tuple)): return [] def compact(value: Any, depth: int = 0) -> Any: if depth > 7: return None if isinstance(value, (str, int, float, bool)) or value is None: if isinstance(value, str): return value[:2000] return value if isinstance(value, (list, tuple)): out = [compact(v, depth + 1) for v in value[:20]] return [v for v in out if v not in (None, {}, [])] if not isinstance(value, dict): return str(value)[:500] keep = { 'tool_name', 'server_name', 'ok', 'error', 'status', 'workflow_name', 'execution_id', 'success', 'allowed', 'reason', 'subject', 'name', 'resolved_subject', 'resolved_subjects', 'value', 'amount', 'protocol', 'protocolo', 'mensagem', 'message', 'response', 'response_text', 'final_answer', 'customer_message', 'customer_response', 'answer', 'text', 'epistemic_status', 'output', 'result', 'items', 'results', 'errors', 'cancelados', 'nao_cancelados', 'nao_encontrados', 'terminal_status', 'contestation_registered', 'contested_invoice_amount', 'contested_invoice_amount_open', 'sms_sent', 'barcode', 'itemName', 'itemsResponse', 'contested_items', 'sr', } out = {} for k, v in value.items(): if k in keep: cv = compact(v, depth + 1) if cv not in (None, {}, []): out[k] = cv return out evidence = [] for item in roots: if not isinstance(item, dict): continue # Only successful/current tool executions can prove that an action happened. # Failed results are still included so a contradictory success claim can be blocked. cv = compact(item) if isinstance(cv, dict) and cv: evidence.append(cv) return evidence @classmethod def _authoritative_output_messages(cls, context: dict[str, Any]) -> list[str]: """Extract canonical customer-facing messages from current successful tool outputs. This is generic: no tool/workflow names are known here. Only message-like fields underneath the current result/output tree are considered; state/input/history are intentionally ignored. """ ctx = context or {} roots = ctx.get('mcp_results') if roots is None: roots = ctx.get('tool_result') if isinstance(roots, dict): roots = [roots] if not isinstance(roots, (list, tuple)): return [] found: list[str] = [] def walk(value: Any, *, in_output: bool = False) -> None: if isinstance(value, dict): for k, v in value.items(): key = str(k).lower() child_output = in_output or key in {'output', 'result', 'results'} if child_output and key in cls._MESSAGE_KEYS and isinstance(v, str) and v.strip(): found.append(' '.join(v.split())) if key not in {'state', 'input', 'metadata', 'conversation_history', 'history'}: walk(v, in_output=child_output) elif isinstance(value, (list, tuple)): for v in value: walk(v, in_output=in_output) for item in roots: if not isinstance(item, dict): continue if item.get('ok') is False: continue walk(item.get('result', item), in_output=True) return list(dict.fromkeys(found)) @classmethod def _candidate_is_authoritative_output(cls, text: str, context: dict[str, Any]) -> bool: candidate = ' '.join(str(text or '').split()) if not candidate: return False return candidate in cls._authoritative_output_messages(context) @staticmethod def _normalized_text(value: Any) -> str: text = str(value or '').lower() text = text.replace('r$', ' ') text = re.sub(r'\s+', ' ', text) return text.strip() @classmethod def _candidate_has_operational_completion_claim(cls, text: str) -> bool: """Detect completion/effect claims, not ordinary explanations. REVPREC exists to stop claims that an operation already happened. A billing explanation containing values/dates is not such a claim and must not depend on a probabilistic binary LLM classification. """ normalized = cls._normalized_text(text) patterns = ( r'\b(cancelad[oa]s?|cancelei|cancelamos|cancelou|cancelamento .*conclu)', r'\b(contesta(?:ção|cao) .*?(?:registrad|abert|conclu|criad))', r'\b(registrad[oa]s?|executei|executamos|executou|executad[oa]s?|conclu[ií]d[oa]s?|processad[oa]s?)\b', r'\b(enviad[oa]s?|emitid[oa]s?|retirad[oa]s?|removid[oa]s?)\b', r'\b(protocol(?:o)? .*?(?:abert|gerad|registrad))', r'\b(?:foi|foram) (?:cancelad|contestad|registrad|enviad|emitid|retirad|removid)', r'\b(?:was|were|has been) (?:cancelled|canceled|registered|sent|issued|removed)', ) return any(re.search(pattern, normalized, re.I) for pattern in patterns) @classmethod def _current_execution_scalar_evidence(cls, context: dict[str, Any]) -> tuple[list[str], bool, bool]: """Flatten scalar facts from current successful result/output only. Returns (facts, completed, positive_effect). Inputs, conversation state, history and metadata are excluded so user claims can never prove an action. """ roots = (context or {}).get('mcp_results') if roots is None: roots = (context or {}).get('tool_result') if isinstance(roots, dict): roots = [roots] if not isinstance(roots, (list, tuple)): return [], False, False facts: list[str] = [] completed = False positive = False positive_status = {'completed', 'opened', 'open', 'iniciada', 'iniciado', 'success', 'succeeded', 'closed', 'fechado'} positive_keys = { 'success', 'executed', 'contestation_registered', 'sms_sent', 'cancelled', 'canceled', 'registered', 'created', 'updated', } skip = {'state', 'input', 'metadata', 'conversation_history', 'history', 'session', 'session_metadata'} def walk(value: Any, path: tuple[str, ...] = ()) -> None: nonlocal completed, positive if isinstance(value, dict): for k, v in value.items(): key = str(k).lower() if key in skip: continue if key == 'status' and str(v).strip().lower() == 'completed': completed = True if key in positive_keys and v is True: positive = True if key == 'status' and str(v).strip().lower() in positive_status: positive = True # Keys such as sms_sent/barcode are themselves meaningful # evidence and can support paraphrased customer-facing text. if key not in {'output', 'result', 'results'}: facts.append(key) walk(v, path + (key,)) elif isinstance(value, (list, tuple)): for item in value[:50]: walk(item, path) elif value not in (None, ''): facts.append(str(value)) for item in roots: if not isinstance(item, dict) or item.get('ok') is False: continue if str(((item.get('result') or {}) if isinstance(item.get('result'), dict) else {}).get('status') or '').upper() == 'COMPLETED': completed = True walk(item.get('result', item)) return facts, completed, positive @classmethod def _candidate_is_structurally_grounded_completion(cls, text: str, context: dict[str, Any]) -> bool: """Allow a paraphrased completion only when current execution proves it. This deliberately does not know tool names. It requires a completed current execution, a positive side-effect marker, grounding for every material number mentioned by the candidate, and at least one meaningful textual fact/entity shared with the execution result. """ if not cls._candidate_has_operational_completion_claim(text): return False facts, completed, positive = cls._current_execution_scalar_evidence(context) if not (completed and positive and facts): return False candidate = cls._normalized_text(text) evidence_text = cls._normalized_text(' '.join(facts)) def canon_number(raw: str) -> str: raw = raw.strip().replace(' ', '') if ',' in raw and '.' in raw: if raw.rfind(',') > raw.rfind('.'): raw = raw.replace('.', '').replace(',', '.') else: raw = raw.replace(',', '') else: raw = raw.replace(',', '.') try: num = float(raw) return (f'{num:.6f}').rstrip('0').rstrip('.') except ValueError: return re.sub(r'\D', '', raw) candidate_numbers = [canon_number(x) for x in re.findall(r'(?= 2] if not material_numbers: return False if any(n not in evidence_numbers for n in material_numbers): return False # Require an entity/message overlap beyond generic success vocabulary. candidate_words = set(re.findall(r'[a-zà-ÿ0-9+_-]{4,}', candidate, re.I)) stop = { 'sucesso', 'registrada', 'registrado', 'concluido', 'concluida', 'cliente', 'valor', 'protocolo', 'cobranca', 'contestacao', 'servico', 'mensal', 'novo', 'boleto', 'recebeu', 'detalhes', } evidence_words = set(re.findall(r'[a-zà-ÿ0-9+_-]{4,}', evidence_text, re.I)) meaningful_overlap = (candidate_words - stop) & (evidence_words - stop) return bool(meaningful_overlap) async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision: # Strongest possible proof: the exact customer-facing candidate was emitted by # the successful current-turn tool/workflow itself. REVPREC is about premature # operational claims, so re-asking an LLM whether this exact authoritative result # "really happened" only adds nondeterminism. Other safety rails still run. if self._candidate_is_authoritative_output(text, context): return RailDecision( code=self.code, allowed=True, reason='current_execution_authoritative_output', sanitized_text=text, metadata={ 'external': True, 'domain': 'TIM_CONTAS', 'mechanism': 'deterministic_current_execution_evidence', 'terminal_action': 'retry', }, ) # REVPREC is not a general factuality rail. Ordinary descriptions and # explanations contain no claim that an operation has already completed, # so they are deterministically outside this rail's blocking scope. if not self._candidate_has_operational_completion_claim(text): return RailDecision( code=self.code, allowed=True, reason='no_operational_completion_claim', sanitized_text=text, metadata={ 'external': True, 'domain': 'TIM_CONTAS', 'mechanism': 'deterministic_claim_scope', 'terminal_action': 'retry', }, ) # A composed/paraphrased response can still be proven by structured # current-turn execution evidence even when it is not byte-for-byte equal # to a workflow message. This removes nondeterministic false positives # while keeping unsupported or contradictory claims on the semantic path. if self._candidate_is_structurally_grounded_completion(text, context): return RailDecision( code=self.code, allowed=True, reason='current_execution_structurally_grounded', sanitized_text=text, metadata={ 'external': True, 'domain': 'TIM_CONTAS', 'mechanism': 'deterministic_structured_execution_evidence', 'terminal_action': 'retry', }, ) evidence = self._current_execution_evidence(context) llm = _llm(context) if llm is None: return RailDecision( code=self.code, allowed=False, reason='LLM do framework indisponível para guardrail TIM', metadata={'external': True, 'fail_closed': True}, ) # For composed/paraphrased answers, ask the semantic rail the correct question: # whether the operational claim is unsupported or contradicted by current-turn # evidence. This replaces the obsolete assumption that reaching REVPREC means # no tool ran. evidence_text = json.dumps(evidence, ensure_ascii=False, default=str)[:14000] prompt = self.prompt_builder(text or '', evidence_text) raw = await llm.ainvoke( [ {'role':'system','content':'Responda apenas com o dígito 1 ou 0, sem texto adicional.'}, {'role':'user','content':prompt}, ], profile_name=self.profile_name, component_name='guardrail.external.tim_revprec', generation_name='guardrail.external.tim_revprec', ) output = str(getattr(raw, 'content', raw) or '').strip() digits = [ch for ch in output if ch in '01'] # 1 = unsupported/contradicted operational completion claim; 0 = allowed. allowed = not digits or digits[-1] != '1' return RailDecision( code=self.code, allowed=allowed, reason='' if allowed else 'resultado operacional afirmado sem suporte na evidência atual', sanitized_text=text, metadata={ 'external': True, 'domain':'TIM_CONTAS', 'raw':output[:100], 'terminal_action':'retry', 'mechanism':'llm_current_execution_evidence', 'current_execution_evidence_count': len(evidence), }, ) class TimPhraseologyRail(_TimPromptRail): code='TIM_FRASEOLOGIA'; stage='output'; profile_name='grl'; prompt_builder=staticmethod(build_fraseologia_prompt) async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision: result = await super().evaluate(text, context) result.metadata = { **dict(result.metadata or {}), 'remediation': { 'type': 'rewrite', 'max_attempts': 1, 'prompt_id': 'FALLBACK', 'profile_name': 'grl', 'component_name': 'guardrail.wording.rewrite', 'generation_name': 'guardrail.wording.rewrite', }, } return result