new feature: transaction interruption for same intent replacement

This commit is contained in:
2026-09-18 11:47:25 -03:00
parent c318d9af27
commit 6eee4514e8
202 changed files with 6594 additions and 211 deletions

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,201 @@
"""Deterministic, framework-owned sanitization for DLEX_OUT.
The DLEX classifier decides whether a candidate response contains information
that must not be exposed. When the finding is recoverable, this module masks
only the sensitive values so the response can be revalidated and preserved.
Rules are intentionally generic and live in the framework, never in domain
agents. Deployments may extend the built-ins without changing agent code via:
* ``GUARDRAIL_DLEX_OUT_EXTRA_MASKS_JSON`` (JSON list of rule objects), or
* the DLEX_OUT rail policy key ``sanitize_patterns`` in guardrails.yaml.
Each extension rule accepts ``name``, ``pattern`` and ``replacement``. Patterns
are regular expressions. Raw matched values are never returned in findings.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
import os
import re
from typing import Any, Iterable
@dataclass(frozen=True, slots=True)
class DlexMaskRule:
name: str
pattern: str
replacement: str
flags: int = re.IGNORECASE
# Conservative defaults: values that are broadly unsafe to echo verbatim and
# can be redacted without changing the semantic meaning of an answer.
_BUILTIN_RULES: tuple[DlexMaskRule, ...] = (
DlexMaskRule(
"payment_barcode",
# Brazilian boleto/payment representations normally contain 44, 47 or
# 48 digits and may be formatted with dots/spaces. Keep the match on the
# numeric token only so surrounding prose is preserved.
r"(?<!\d)(?=[0-9.\-\s]{44,90}(?!\d))(?:\d[.\-\s]*){44,48}(?!\d)",
"[CODIGO_DE_PAGAMENTO_OCULTADO]",
flags=0,
),
DlexMaskRule(
"jwt",
r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b",
"[TOKEN_OCULTADO]",
flags=0,
),
DlexMaskRule(
"openai_style_secret",
r"\bsk-[A-Za-z0-9_-]{10,}\b",
"[CREDENCIAL_OCULTADA]",
flags=0,
),
DlexMaskRule(
"labelled_secret",
r"(?i)\b(api[_ -]?key|secret|segredo|token|senha|password|credential|credencial)\b(\s*[:=]\s*)([^\s,;]+)",
r"\1\2[DADO_PROTEGIDO]",
flags=0,
),
DlexMaskRule(
"cpf",
r"\b\d{3}\.\d{3}\.\d{3}-\d{2}\b",
"[CPF_OCULTADO]",
flags=0,
),
DlexMaskRule(
"cnpj",
r"\b\d{2}\.\d{3}\.\d{3}/\d{4}-\d{2}\b",
"[CNPJ_OCULTADO]",
flags=0,
),
DlexMaskRule(
"email",
r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b",
"[EMAIL_OCULTADO]",
),
DlexMaskRule(
"labelled_phone",
r"(?i)\b(telefone|celular|whatsapp|msisdn)\b(\s*[:=]?\s*)(?:\+?55\s*)?(?:\(?\d{2}\)?[\s.-]*)?\d{4,5}[\s.-]*\d{4}\b",
r"\1\2[TELEFONE_OCULTADO]",
flags=0,
),
DlexMaskRule(
"uuid",
r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}\b",
"[IDENTIFICADOR_OCULTADO]",
flags=0,
),
)
def _parse_rule(raw: Any) -> DlexMaskRule | None:
if not isinstance(raw, dict):
return None
name = str(raw.get("name") or "custom").strip()
pattern = str(raw.get("pattern") or "").strip()
replacement = str(raw.get("replacement") or "[DADO_PROTEGIDO]")
if not pattern:
return None
try:
re.compile(pattern)
except re.error:
return None
return DlexMaskRule(name=name, pattern=pattern, replacement=replacement, flags=0)
def _extra_rules_from_env() -> list[DlexMaskRule]:
raw = os.getenv("GUARDRAIL_DLEX_OUT_EXTRA_MASKS_JSON", "").strip()
if not raw:
return []
try:
data = json.loads(raw)
except Exception:
return []
if isinstance(data, dict):
data = [data]
return [rule for item in (data or []) if (rule := _parse_rule(item)) is not None]
def _extra_rules_from_policy(policy: dict[str, Any] | None) -> list[DlexMaskRule]:
data = dict(policy or {}).get("sanitize_patterns") or []
if isinstance(data, dict):
data = [data]
return [rule for item in data if (rule := _parse_rule(item)) is not None]
def _mask_payment_sequences(text: str) -> tuple[str, int]:
"""Mask 44/47/48-digit payment tokens without consuming surrounding prose.
Regex alone is awkward because formatted boletos have variable punctuation.
We scan numeric/punctuation runs, count digits, and only redact the run when
it has a canonical payment digit count.
"""
token_re = re.compile(r"(?<!\d)\d(?:[.\-\s]*\d){43,47}(?!\d)")
count = 0
def repl(match: re.Match[str]) -> str:
nonlocal count
token = match.group(0)
digits = re.sub(r"\D", "", token)
if len(digits) in {44, 47, 48}:
count += 1
return "[CODIGO_DE_PAGAMENTO_OCULTADO]"
return token
return token_re.sub(repl, text), count
def sanitize_dlex_output(
text: str,
*,
policy: dict[str, Any] | None = None,
exclude_values: Iterable[str] | None = None,
) -> tuple[str, list[dict[str, Any]]]:
"""Return ``(sanitized_text, findings)`` without exposing raw matches.
``exclude_values`` contains explicitly authorized values (for example a
workflow-generated protocol). They are temporarily protected so generic
masking never alters them.
"""
original = text or ""
sanitized = original
findings: list[dict[str, Any]] = []
protected: dict[str, str] = {}
for index, value in enumerate(exclude_values or []):
value = str(value or "").strip()
if not value or value not in sanitized:
continue
marker = f"__DLEX_AUTHORIZED_{index}__"
sanitized = sanitized.replace(value, marker)
protected[marker] = value
# Payment sequences use digit-count validation rather than a plain rule.
sanitized, payment_count = _mask_payment_sequences(sanitized)
if payment_count:
findings.append({"type": "payment_barcode", "action": "redact", "count": payment_count})
rules = [r for r in _BUILTIN_RULES if r.name != "payment_barcode"]
rules.extend(_extra_rules_from_env())
rules.extend(_extra_rules_from_policy(policy))
for rule in rules:
try:
regex = re.compile(rule.pattern, rule.flags)
except re.error:
continue
sanitized, count = regex.subn(rule.replacement, sanitized)
if count:
findings.append({"type": rule.name, "action": "redact", "count": count})
for marker, value in protected.items():
sanitized = sanitized.replace(marker, value)
return sanitized, findings
__all__ = ["DlexMaskRule", "sanitize_dlex_output"]

View File

@@ -25,6 +25,7 @@ from .calibrated.output_sanitization import mascarar_pii_output, sanitizar_toxic
from .calibrated.rules.pinj_patterns import _PINJ_PATTERNS, is_obvious_injection
from .calibrated.rules.tox_blocklist import _EXPLICIT_TERMS, _THREAT_PATTERNS, is_obvious_toxic
from .framework_llm_client import classify_with_framework_llm
from .dlex_sanitizer import sanitize_dlex_output
from agent_framework.workflows.input_contract import has_meaningful_unmatched_policy, has_semantic_classifier
@@ -323,10 +324,28 @@ class CoherenceRail(Guardrail):
"data": out,
},
)
out = await classify_with_framework_llm(
_llm(ctx), "COER", {"text": text or "", "context": ctx},
profile_name="guardrail", component_name="guardrail.coer", generation_name="guardrail.coer",
)
try:
out = await classify_with_framework_llm(
_llm(ctx), "COER", {"text": text or "", "context": ctx},
profile_name="guardrail", component_name="guardrail.coer", generation_name="guardrail.coer",
)
except Exception as exc:
# COER is a conversational-coherence rail, not a security boundary.
# If its semantic classifier/provider is unavailable, the graph must
# continue and let the normal router/workflow clarification logic own
# the turn instead of surfacing an infrastructure exception.
return RailDecision(
code=self.code,
allowed=True,
reason="Classificador de coerência indisponível; continuidade delegada ao runtime",
sanitized_text=text,
metadata={
"mechanism": "infrastructure_fail_open",
"calibrated": True,
"error_type": type(exc).__name__,
"error": str(exc)[:500],
},
)
return RailDecision(
code=self.code, allowed=bool(out.get("allowed", True)),
reason=str(out.get("reason") or out.get("label") or "COER avaliado"),
@@ -344,6 +363,9 @@ class LoopRail(Guardrail):
"COLLECTING_PARAMETERS", "AWAITING_CONFIRMATION", "EXECUTING",
"PAUSED", "WAITING_INPUT",
}
# A queued multi-intent transaction may be promoted immediately after
# the previous one completed. Prefer the newest live MCP evidence over
# a stale top-level COMPLETED status left by checkpoint merge order.
for result in reversed(list(ctx.get("mcp_results") or [])):
if isinstance(result, dict):
result_status = str(result.get("transaction_status") or "").strip().upper()
@@ -359,6 +381,11 @@ class LoopRail(Guardrail):
async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision:
ctx = _ctx(context)
transaction_status = self._transaction_status(ctx)
# Parameter values and short confirmations may legitimately repeat in
# the same session. VLOOP must not consume input owned by the active
# transaction contract; extraction/confirmation remains responsible
# for deciding whether the value is valid.
if transaction_status in {"COLLECTING_PARAMETERS", "AWAITING_CONFIRMATION"}:
return RailDecision(
code=self.code,
@@ -373,6 +400,7 @@ class LoopRail(Guardrail):
"calibrated": True,
},
)
normalized = _lower(text).strip()
history = [_lower(h).strip() for h in ctx.get("history_texts", [])[-6:]]
repeated = history.count(normalized) >= 2 if normalized else False
@@ -805,11 +833,64 @@ class DataLeakageOutputRail(Guardrail):
}
protocol_authorization_verified = True
# DLEX_OUT is permissive for recoverable leakage: when classification
# rejects the candidate, mask only generic sensitive values and re-run
# the same classifier. The agent/domain remains unaware of this detail.
# If sanitization cannot change the candidate, or the sanitized version
# is still unsafe, the original blocking behavior is preserved.
dlex_sanitized_text = None
dlex_findings: list[dict[str, Any]] = []
sanitization_recheck = None
original_dlex_result = dict(out)
if not bool(out.get("allowed", True)):
policy = dict(getattr(self, "_guardrail_policy", {}) or {})
candidate_sanitized, dlex_findings = sanitize_dlex_output(
original_text,
policy=policy,
exclude_values=matched_expected_protocols,
)
if candidate_sanitized != original_text:
sanitized_classifier_text = candidate_sanitized
sanitized_classifier_ctx: dict[str, Any] = ctx
if matched_expected_protocols:
sanitized_classifier_text = _mask_authorized_protocol_values(
sanitized_classifier_text, matched_expected_protocols
)
sanitized_classifier_ctx = _mask_authorized_protocol_values(
ctx, matched_expected_protocols
)
sanitization_recheck = await classify_with_framework_llm(
_llm(ctx),
"DLEX_OUT",
{"text": sanitized_classifier_text, "context": sanitized_classifier_ctx},
profile_name="grl",
component_name="guardrail.dlex_out.sanitization_recheck",
generation_name="guardrail.dlex_out.sanitization_recheck",
)
if bool(sanitization_recheck.get("allowed", True)):
dlex_sanitized_text = candidate_sanitized
out = {
"allowed": True,
"label": "SANITIZED",
"reason": "DLEX_OUT sanitizado e revalidado com segurança",
"original_reason": original_dlex_result.get("reason"),
}
metadata = {
"mechanism": "llm_rail",
"data": out,
"calibrated": True,
}
if dlex_findings:
metadata.update(
{
"action": "sanitize" if dlex_sanitized_text is not None else "block",
"sanitized": dlex_sanitized_text is not None,
"findings": dlex_findings,
"original_decision": original_dlex_result,
"sanitization_recheck": sanitization_recheck,
}
)
if matched_expected_protocols:
metadata.update(
{
@@ -823,7 +904,7 @@ class DataLeakageOutputRail(Guardrail):
code=self.code,
allowed=bool(out.get("allowed", True)),
reason=str(out.get("reason") or out.get("label") or "DLEX_OUT avaliado"),
sanitized_text=text,
sanitized_text=dlex_sanitized_text,
metadata=metadata,
)

View File

@@ -630,47 +630,144 @@ class EnterpriseRouter:
if state_decision:
tx_status = str(state.get("transaction_status") or "").strip().upper()
# Explicit abandonment must be evaluated before confirmation
# consumption. Otherwise an utterance such as
# "não quero cancelar; quero apenas consultar meus serviços" can be
# trapped by the confirmation state and never reach ABANDON. A plain
# yes/no still keeps deterministic confirmation precedence because it
# does not pass the abandonment gate.
if tx_status == "AWAITING_CONFIRMATION":
if self._looks_like_explicit_abandonment(str(text)):
active_tx = state.get("active_transaction") if isinstance(state.get("active_transaction"), dict) else {}
legacy_tx = state.get("pending_tool_call") or state.get("selected_tool_call") or {}
state_lock_name = str(
state_decision.next_state
or current_state
or state_decision.intent.removeprefix("state:")
or ""
).strip().upper()
transactional_state_lock = state_lock_name.startswith(
("WAITING_", "COLLECTING_", "AWAITING_")
)
# Rich transactional latches remain the primary source of truth.
# The domain state lock is only a recovery fallback for checkpoints
# where those latches were not restored. This prevents a generic
# COLLECTING_* state from stealing a legitimate parameter answer.
rich_transaction_context = bool(
active_tx.get("tool_name")
or (isinstance(legacy_tx, dict) and legacy_tx.get("tool_name"))
or tx_status in {"COLLECTING_PARAMETERS", "AWAITING_CONFIRMATION"}
)
fallback_state_lock_context = bool(
transactional_state_lock and not rich_transaction_context
)
has_transaction_context = bool(
rich_transaction_context or fallback_state_lock_context
)
# Some hosts restore the domain state but omit transaction_status.
# Infer only the generic phase needed by the precedence logic; do not
# replace richer transaction data when it exists.
effective_tx_status = tx_status
if not effective_tx_status:
if state_lock_name.startswith("COLLECTING_"):
effective_tx_status = "COLLECTING_PARAMETERS"
elif state_lock_name.startswith("AWAITING_") or "CONFIRMATION" in state_lock_name:
effective_tx_status = "AWAITING_CONFIRMATION"
precedence_state = state
if effective_tx_status and effective_tx_status != tx_status:
precedence_state = dict(state)
precedence_state["transaction_status"] = effective_tx_status
# Transactional interruption precedence is semantic, not lexical.
#
# COLLECTING_PARAMETERS is special: the pending parameter gets the
# first opportunity to consume the turn. If the input is pertinent
# to the parameter, extraction returns values and the transaction
# continues. Only when the input is not pertinent do we ask the
# semantic interruption classifier whether the user abandoned the
# active transaction. This preserves inputs such as
# ``motivo é que desisti`` as a legitimate ``reason`` value while
# still allowing ``não quero mais devolver; quero ver meus serviços``
# to abandon the current operation and start a new objective.
if effective_tx_status == "COLLECTING_PARAMETERS":
if rich_transaction_context:
relevance = await self._classify_transaction_parameter_relevance(
precedence_state, text=str(text)
)
if relevance is True:
consumed = await self._transaction_parameter_precedence(
precedence_state, text=str(text), state_decision=state_decision
)
if consumed is not None:
consumed.metadata = {
**(consumed.metadata or {}),
"transaction_parameter_relevant": True,
}
await self._emit(consumed, state)
return consumed
# Pertinent to the requested parameter, but extraction did
# not produce a structured value. Keep transaction
# ownership and let the agent reprompt/clarify; do not
# reinterpret the same input as ABANDON or SHIFT.
state_decision.metadata = {
**(state_decision.metadata or {}),
"transaction_parameter_relevant": True,
"transaction_parameter_extraction_empty": True,
}
await self._emit(state_decision, state)
return state_decision
if relevance is None:
# Compatibility/fail-safe when the relevance classifier is
# unavailable: retain the previous extractor-first behavior.
consumed = await self._transaction_parameter_precedence(
precedence_state, text=str(text), state_decision=state_decision
)
if consumed is not None:
consumed.metadata = {
**(consumed.metadata or {}),
"transaction_parameter_relevance": "unknown",
}
await self._emit(consumed, state)
return consumed
if has_transaction_context:
abandonment = await self._transaction_state_interruption_candidate(
state,
precedence_state,
text=str(text),
state_decision=state_decision,
abandon_only=True,
)
if abandonment is not None:
abandonment.metadata = {
**(abandonment.metadata or {}),
"transaction_parameter_relevant": False if rich_transaction_context else None,
}
await self._emit(abandonment, state)
return abandonment
consumed = await self._transaction_parameter_precedence(
state, text=str(text), state_decision=state_decision
# In confirmation/waiting phases there is no pending free-form
# parameter whose value can be confused with abandonment. Evaluate
# ABANDON semantically before confirmation/state-lock consumption.
# Plain confirmations (e.g. yes/no) are classified as CONTINUE by
# this probe and then follow the normal confirmation path below.
elif has_transaction_context:
abandonment = await self._transaction_state_interruption_candidate(
precedence_state,
text=str(text),
state_decision=state_decision,
abandon_only=True,
)
if consumed is not None:
await self._emit(consumed, state)
return consumed
if abandonment is not None:
await self._emit(abandonment, state)
return abandonment
# Transaction parameter precedence is absolute while collecting:
# first let the active transaction try to consume the current turn.
# Only when NO pending parameter can be extracted do we ask the
# semantic classifier whether the user changed goals. This prevents
# value/name/reference answers (for example "a de 14,99") from being
# stolen by a semantically plausible but incompatible intent.
if tx_status == "COLLECTING_PARAMETERS":
if effective_tx_status == "AWAITING_CONFIRMATION":
consumed = await self._transaction_parameter_precedence(
state, text=str(text), state_decision=state_decision
precedence_state, text=str(text), state_decision=state_decision
)
if consumed is not None:
await self._emit(consumed, state)
return consumed
interruption = await self._transaction_state_interruption_candidate(
state, text=str(text), state_decision=state_decision
precedence_state, text=str(text), state_decision=state_decision
)
if interruption is not None:
await self._emit(interruption, state)
@@ -701,34 +798,61 @@ class EnterpriseRouter:
method="state",
next_state=tx_status,
)
if tx_status == "AWAITING_CONFIRMATION":
if self._looks_like_explicit_abandonment(str(text)):
abandonment = await self._transaction_state_interruption_candidate(
state,
text=str(text),
state_decision=synthetic,
abandon_only=True,
)
if abandonment is not None:
abandonment.metadata = {
**(abandonment.metadata or {}),
"transaction_state_recovered": True,
}
await self._emit(abandonment, state)
return abandonment
consumed = await self._transaction_parameter_precedence(
state, text=str(text), state_decision=synthetic
)
if consumed is not None:
consumed.metadata = {
**(consumed.metadata or {}),
"transaction_state_recovered": True,
}
await self._emit(consumed, state)
return consumed
if tx_status == "COLLECTING_PARAMETERS":
relevance = await self._classify_transaction_parameter_relevance(
state, text=str(text)
)
if relevance is True:
consumed = await self._transaction_parameter_precedence(
state, text=str(text), state_decision=synthetic
)
if consumed is not None:
consumed.metadata = {
**(consumed.metadata or {}),
"transaction_state_recovered": True,
"transaction_parameter_relevant": True,
}
await self._emit(consumed, state)
return consumed
synthetic.metadata = {
**(synthetic.metadata or {}),
"transaction_state_recovered": True,
"transaction_parameter_relevant": True,
"transaction_parameter_extraction_empty": True,
}
await self._emit(synthetic, state)
return synthetic
if relevance is None:
consumed = await self._transaction_parameter_precedence(
state, text=str(text), state_decision=synthetic
)
if consumed is not None:
consumed.metadata = {
**(consumed.metadata or {}),
"transaction_state_recovered": True,
"transaction_parameter_relevance": "unknown",
}
await self._emit(consumed, state)
return consumed
# No parameter was consumed (or this is confirmation): now decide
# semantically whether the user abandoned the active transaction.
abandonment = await self._transaction_state_interruption_candidate(
state,
text=str(text),
state_decision=synthetic,
abandon_only=True,
)
if abandonment is not None:
abandonment.metadata = {
**(abandonment.metadata or {}),
"transaction_state_recovered": True,
}
await self._emit(abandonment, state)
return abandonment
if tx_status == "AWAITING_CONFIRMATION":
consumed = await self._transaction_parameter_precedence(
state, text=str(text), state_decision=synthetic
)
@@ -867,6 +991,82 @@ class EnterpriseRouter:
return decision
async def _classify_transaction_parameter_relevance(
self, state: dict[str, Any], *, text: str
) -> bool | None:
"""Semantically decide whether the turn answers a pending parameter.
This is intentionally separate from extraction. A parameter extractor
may be able to manufacture a syntactically valid value from unrelated
text; relevance answers the higher-level question first: does the user
appear to be responding to what the active transaction asked for?
Returns True/False when classification succeeds and None when the
classifier is unavailable or its output cannot be trusted.
"""
if not (self.enable_llm_router and self.llm is not None):
return None
missing = [str(name) for name in (state.get("missing_parameters") or []) if str(name).strip()]
if not missing:
return None
active = state.get("active_transaction") if isinstance(state.get("active_transaction"), dict) else {}
tool_name = str(active.get("tool_name") or "").strip()
if not tool_name:
return None
payload = {
"message": str(text or ""),
"transaction": {
"intent": active.get("started_from_intent") or state.get("intent"),
"tool_name": tool_name,
"tool_description": active.get("tool_description"),
"missing_parameters": missing,
"parameter_schema": active.get("parameter_schema") or {},
"known_arguments": active.get("arguments") or {},
"relevant_conversation_context": (
active.get("parameter_conversational_context")
or self._collect_transaction_parameter_context(state=state, current_text=str(text))
),
},
}
system = (
"Você decide somente se a mensagem do usuário é uma resposta pertinente ao(s) "
"parâmetro(s) que a transação ativa está pedindo. Avalie significado e contexto, "
"não palavras isoladas. Considere RELEVANT quando a mensagem fornece, esclarece, "
"corrige ou referencia plausivelmente o dado solicitado, mesmo em linguagem livre. "
"Considere NOT_RELEVANT quando a mensagem abandona a ação, inicia outro objetivo, "
"faz uma pergunta diferente ou não responde ao dado pedido. Não extraia valores e "
"não decida abandono/intent; classifique apenas pertinência. Retorne somente JSON "
"válido: {\"relevance\":\"RELEVANT|NOT_RELEVANT\",\"confidence\":0.0,\"reason\":\"...\"}."
)
try:
answer = await self.llm.ainvoke(
[
{"role": "system", "content": system},
{"role": "user", "content": json.dumps(payload, ensure_ascii=False)},
],
temperature=0.0,
max_tokens=256,
profile_name="router",
component_name="router.transaction_parameter_relevance",
generation_name="llm.transaction_parameter_relevance",
)
data = self._parse_json(answer)
except Exception as exc:
logger.warning("Falha ao classificar pertinência do parâmetro transacional: %s", exc)
return None
relevance = str(data.get("relevance") or "").strip().upper()
try:
confidence = float(data.get("confidence") or 0.0)
except (TypeError, ValueError):
return None
if confidence < self.intent_shift_threshold:
return None
if relevance == "RELEVANT":
return True
if relevance == "NOT_RELEVANT":
return False
return None
async def _transaction_parameter_precedence(
self,
state: dict[str, Any],
@@ -997,13 +1197,14 @@ class EnterpriseRouter:
*,
text: str,
state_decision: RouteDecision,
abandon_only: bool = False,
) -> RouteDecision | None:
"""Detecta semanticamente mudança de intenção durante uma transação.
Não existe lista de palavras para desistência ou mudança de assunto. Uma
interrupção nasce de uma intent diferente resolvida por uma keyword
configurada no ``routing.yaml`` ou, na ausência dela, por uma decisão
semântica do LLM com o contexto da transação pendente.
Não existe lista de palavras para desistência. ABANDON é sempre uma
decisão semântica do LLM baseada no contexto transacional. Uma possível
nova intenção pode aparecer como hint do ``routing.yaml``, mas esse hint
não decide abandono e não substitui a classificação semântica.
"""
active_tx = state.get("active_transaction") if isinstance(state.get("active_transaction"), dict) else {}
started_intent = str(active_tx.get("started_from_intent") or "").strip()
@@ -1017,7 +1218,17 @@ class EnterpriseRouter:
or (started_intent and configured_candidate.intent != started_intent)
or (previous_intent and not previous_intent.startswith("state:") and configured_candidate.intent != previous_intent)
)
if not different:
# A configured candidate that resolves to the same intent/agent is not
# sufficient to conclude that the active transaction continues. The
# user may be starting a *new instance of the same intent* with a
# different transactional target (for example cancelling TIM Fashion
# while Tamboro is awaiting confirmation). In LLM mode keep the
# candidate only as a semantic hint and let the transaction classifier
# decide CONTINUE/REPLACE/ABANDON. Without semantic classification we
# retain the legacy state lock because guessing replacement from a
# keyword alone would be unsafe.
if not different and not (self.enable_llm_router and self.llm is not None):
return None
# During parameter collection, a configured keyword may be present in
@@ -1058,6 +1269,11 @@ class EnterpriseRouter:
"current_intent": started_intent or previous_intent,
"transaction_status": state.get("transaction_status"),
"tool_name": active_tx.get("tool_name"),
# Structured arguments are part of the identity of the active
# transaction. They let the semantic classifier distinguish a new
# instance of the same intent from a continuation without relying on
# domain keywords or hard-coded product names.
"active_arguments": dict(active_tx.get("arguments") or {}),
"missing_parameters": list(state.get("missing_parameters") or []),
"configured_candidate": (
{
@@ -1069,18 +1285,36 @@ class EnterpriseRouter:
else None
),
}
system = (
"Você decide apenas se o turno atual continua a transação ativa ou muda de intenção. "
"Use o significado da mensagem e o contexto transacional; não use palavras isoladas como regra. "
"A extração dos parâmetros pendentes já foi tentada antes desta etapa e não consumiu o turno. "
"Se ainda assim a mensagem for apenas uma resposta referencial/valor/nome ao dado pendente, retorne CONTINUE. "
"Se o usuário passou claramente a perseguir outro objetivo sem desistir explicitamente da ação atual, "
"retorne SHIFT e a nova intent permitida. "
"Se o usuário desistiu explicitamente da ação/transação atual, retorne ABANDON. "
"Quando ABANDON vier acompanhado de um novo objetivo, também informe intent e agent desse novo objetivo. "
"Quando for apenas abandono sem novo objetivo, intent e agent podem ser nulos. "
"Retorne somente JSON válido com decision, intent, agent, confidence, reason."
)
if abandon_only:
system = (
"Você decide somente se o usuário desistiu explicitamente da ação/transação ativa. "
"Quando a transação estava coletando parâmetros, a pertinência ao parâmetro já foi avaliada antes desta etapa; "
"portanto não transforme uma resposta pertinente ao parâmetro em abandono. "
"Use o significado completo da mensagem e o contexto transacional; não use palavras isoladas como regra. "
"Se houver desistência explícita da ação atual, retorne ABANDON. "
"Quando o abandono vier acompanhado de um novo objetivo, também informe intent e agent desse novo objetivo. "
"Quando for apenas abandono sem novo objetivo, intent e agent podem ser nulos. "
"Se a mensagem apenas fornece parâmetro, confirma contexto, ou somente muda/adiciona outro objetivo sem "
"desistir explicitamente da ação atual, retorne CONTINUE. "
"Retorne somente JSON válido com decision, intent, agent, confidence, reason."
)
else:
system = (
"Você decide apenas se o turno atual continua a transação ativa, substitui a instância transacional ativa, "
"ou muda de intenção. Use o significado da mensagem e o contexto transacional; não use palavras isoladas como regra. "
"A extração dos parâmetros pendentes já foi tentada antes desta etapa e não consumiu o turno. "
"Se ainda assim a mensagem for apenas uma resposta referencial/valor/nome ao dado pendente, retorne CONTINUE. "
"Se o usuário continua com a MESMA intent, mas claramente inicia outra instância da operação com alvo/argumentos diferentes, "
"retorne REPLACE e informe a mesma intent e agent. REPLACE significa substituir a transação ativa; não significa estado novo. "
"Não use REPLACE quando o usuário apenas complementa/adiciona itens à mesma operação (por exemplo 'também', 'além desses'), "
"nem quando apenas reformula ou confirma o mesmo alvo: nesses casos retorne CONTINUE. "
"Se o usuário passou claramente a perseguir outro objetivo/intent sem desistir explicitamente da ação atual, "
"retorne SHIFT e a nova intent permitida. "
"Se o usuário desistiu explicitamente da ação/transação atual, retorne ABANDON. "
"Quando ABANDON vier acompanhado de um novo objetivo, também informe intent e agent desse novo objetivo. "
"Quando for apenas abandono sem novo objetivo, intent e agent podem ser nulos. "
"Retorne somente JSON válido com decision, intent, agent, confidence, reason."
)
user = {
"message": text,
"transaction": transaction_context,
@@ -1105,7 +1339,10 @@ class EnterpriseRouter:
return None
decision_kind = str(data.get("decision") or "").strip().upper()
if decision_kind not in {"SHIFT", "ABANDON"}:
if abandon_only:
if decision_kind != "ABANDON":
return None
elif decision_kind not in {"SHIFT", "REPLACE", "ABANDON"}:
return None
confidence = float(data.get("confidence") or 0.0)
if confidence < self.intent_shift_threshold:
@@ -1156,7 +1393,42 @@ class EnterpriseRouter:
mcp_tools=mcp_tools,
)
if not intent_name or intent_name == (started_intent or previous_intent):
current_intent = started_intent or previous_intent
if not intent_name:
return None
# Same-intent replacement is a transaction lifecycle event, not a new
# conversation state. It cancels only the active transaction instance
# and re-enters normal routing for the same configured intent.
if decision_kind == "REPLACE":
if not current_intent or intent_name != current_intent:
return None
agent = agent or str(self._agent_for_intent(intent_name) or "").strip()
if not agent:
return None
return RouteDecision(
route=agent,
agent=agent,
intent=intent_name,
confidence=confidence,
reason=str(data.get("reason") or "Nova instância transacional da mesma intenção."),
method="llm",
metadata={
"transaction_interruption": "same_intent_replacement",
"interrupted_state": state_decision.next_state,
"interrupted_agent": state_decision.agent,
"interrupted_intent": current_intent,
"interruption_source": "semantic_classifier",
"configured_routing_hint": (
configured_candidate.intent if configured_candidate is not None else None
),
"raw_llm_answer": answer[:1000],
},
domain=self._domain_for_intent(intent_name),
mcp_tools=self._tools_for_intent(intent_name),
)
if intent_name == current_intent:
return None
agent = agent or str(self._agent_for_intent(intent_name) or "").strip()
if not agent:

View File

@@ -3113,21 +3113,33 @@ class AgentRuntimeMixin:
self._normalize_transaction_lifecycle(state)
# Uma transação em coleta/confirmação não pode aprisionar a sessão. O
# EnterpriseRouter é a única fonte para interrupção semântica: SHIFT ou
# ABANDON. Não existe interpretação lexical de desistência no runtime; o
# runtime apenas aplica a decisão do router e limpa os latches da ação.
# EnterpriseRouter é a única fonte para interrupção semântica: SHIFT,
# ABANDON ou replacement da instância transacional na mesma intent. Não
# existe interpretação lexical de desistência no runtime; o runtime apenas
# aplica a decisão do router e limpa os latches da ação.
active_before_interruption = self._active_transaction(state)
interruption = str(route_meta.get("transaction_interruption") or "").strip().lower()
if active_before_interruption and interruption in {"intent_shift", "explicit_abandonment"}:
if active_before_interruption and interruption in {
"intent_shift",
"explicit_abandonment",
"same_intent_replacement",
}:
interrupted_tool = active_before_interruption.get("tool_name")
self._finish_active_transaction(state, "CANCELLED")
state["transaction_pre_validation"] = None
# Generic framework boundary: consumers do not need to know why the
# previous transactional context was invalidated. This deliberately
# avoids leaking same_intent_replacement (or any other interruption
# subtype) into domain agents/workflows.
state["operational_context_reset"] = True
if interruption == "explicit_abandonment":
action = "cancelled_by_explicit_abandonment"
elif interruption == "same_intent_replacement":
action = "cancelled_by_same_intent_replacement"
else:
action = "cancelled_by_intent_shift"
state["tool_policy_result"] = {
"action": (
"cancelled_by_explicit_abandonment"
if interruption == "explicit_abandonment"
else "cancelled_by_intent_shift"
),
"action": action,
"tool_name": interrupted_tool,
}

View File

@@ -53,6 +53,7 @@ src/agent_framework/guardrails/__init__.py
src/agent_framework/guardrails/base.py
src/agent_framework/guardrails/config_loader.py
src/agent_framework/guardrails/custom_rails.py
src/agent_framework/guardrails/dlex_sanitizer.py
src/agent_framework/guardrails/executor.py
src/agent_framework/guardrails/framework_llm_client.py
src/agent_framework/guardrails/langgraph_adapters.py

View File

@@ -0,0 +1,201 @@
"""Deterministic, framework-owned sanitization for DLEX_OUT.
The DLEX classifier decides whether a candidate response contains information
that must not be exposed. When the finding is recoverable, this module masks
only the sensitive values so the response can be revalidated and preserved.
Rules are intentionally generic and live in the framework, never in domain
agents. Deployments may extend the built-ins without changing agent code via:
* ``GUARDRAIL_DLEX_OUT_EXTRA_MASKS_JSON`` (JSON list of rule objects), or
* the DLEX_OUT rail policy key ``sanitize_patterns`` in guardrails.yaml.
Each extension rule accepts ``name``, ``pattern`` and ``replacement``. Patterns
are regular expressions. Raw matched values are never returned in findings.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
import os
import re
from typing import Any, Iterable
@dataclass(frozen=True, slots=True)
class DlexMaskRule:
name: str
pattern: str
replacement: str
flags: int = re.IGNORECASE
# Conservative defaults: values that are broadly unsafe to echo verbatim and
# can be redacted without changing the semantic meaning of an answer.
_BUILTIN_RULES: tuple[DlexMaskRule, ...] = (
DlexMaskRule(
"payment_barcode",
# Brazilian boleto/payment representations normally contain 44, 47 or
# 48 digits and may be formatted with dots/spaces. Keep the match on the
# numeric token only so surrounding prose is preserved.
r"(?<!\d)(?=[0-9.\-\s]{44,90}(?!\d))(?:\d[.\-\s]*){44,48}(?!\d)",
"[CODIGO_DE_PAGAMENTO_OCULTADO]",
flags=0,
),
DlexMaskRule(
"jwt",
r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b",
"[TOKEN_OCULTADO]",
flags=0,
),
DlexMaskRule(
"openai_style_secret",
r"\bsk-[A-Za-z0-9_-]{10,}\b",
"[CREDENCIAL_OCULTADA]",
flags=0,
),
DlexMaskRule(
"labelled_secret",
r"(?i)\b(api[_ -]?key|secret|segredo|token|senha|password|credential|credencial)\b(\s*[:=]\s*)([^\s,;]+)",
r"\1\2[DADO_PROTEGIDO]",
flags=0,
),
DlexMaskRule(
"cpf",
r"\b\d{3}\.\d{3}\.\d{3}-\d{2}\b",
"[CPF_OCULTADO]",
flags=0,
),
DlexMaskRule(
"cnpj",
r"\b\d{2}\.\d{3}\.\d{3}/\d{4}-\d{2}\b",
"[CNPJ_OCULTADO]",
flags=0,
),
DlexMaskRule(
"email",
r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b",
"[EMAIL_OCULTADO]",
),
DlexMaskRule(
"labelled_phone",
r"(?i)\b(telefone|celular|whatsapp|msisdn)\b(\s*[:=]?\s*)(?:\+?55\s*)?(?:\(?\d{2}\)?[\s.-]*)?\d{4,5}[\s.-]*\d{4}\b",
r"\1\2[TELEFONE_OCULTADO]",
flags=0,
),
DlexMaskRule(
"uuid",
r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}\b",
"[IDENTIFICADOR_OCULTADO]",
flags=0,
),
)
def _parse_rule(raw: Any) -> DlexMaskRule | None:
if not isinstance(raw, dict):
return None
name = str(raw.get("name") or "custom").strip()
pattern = str(raw.get("pattern") or "").strip()
replacement = str(raw.get("replacement") or "[DADO_PROTEGIDO]")
if not pattern:
return None
try:
re.compile(pattern)
except re.error:
return None
return DlexMaskRule(name=name, pattern=pattern, replacement=replacement, flags=0)
def _extra_rules_from_env() -> list[DlexMaskRule]:
raw = os.getenv("GUARDRAIL_DLEX_OUT_EXTRA_MASKS_JSON", "").strip()
if not raw:
return []
try:
data = json.loads(raw)
except Exception:
return []
if isinstance(data, dict):
data = [data]
return [rule for item in (data or []) if (rule := _parse_rule(item)) is not None]
def _extra_rules_from_policy(policy: dict[str, Any] | None) -> list[DlexMaskRule]:
data = dict(policy or {}).get("sanitize_patterns") or []
if isinstance(data, dict):
data = [data]
return [rule for item in data if (rule := _parse_rule(item)) is not None]
def _mask_payment_sequences(text: str) -> tuple[str, int]:
"""Mask 44/47/48-digit payment tokens without consuming surrounding prose.
Regex alone is awkward because formatted boletos have variable punctuation.
We scan numeric/punctuation runs, count digits, and only redact the run when
it has a canonical payment digit count.
"""
token_re = re.compile(r"(?<!\d)\d(?:[.\-\s]*\d){43,47}(?!\d)")
count = 0
def repl(match: re.Match[str]) -> str:
nonlocal count
token = match.group(0)
digits = re.sub(r"\D", "", token)
if len(digits) in {44, 47, 48}:
count += 1
return "[CODIGO_DE_PAGAMENTO_OCULTADO]"
return token
return token_re.sub(repl, text), count
def sanitize_dlex_output(
text: str,
*,
policy: dict[str, Any] | None = None,
exclude_values: Iterable[str] | None = None,
) -> tuple[str, list[dict[str, Any]]]:
"""Return ``(sanitized_text, findings)`` without exposing raw matches.
``exclude_values`` contains explicitly authorized values (for example a
workflow-generated protocol). They are temporarily protected so generic
masking never alters them.
"""
original = text or ""
sanitized = original
findings: list[dict[str, Any]] = []
protected: dict[str, str] = {}
for index, value in enumerate(exclude_values or []):
value = str(value or "").strip()
if not value or value not in sanitized:
continue
marker = f"__DLEX_AUTHORIZED_{index}__"
sanitized = sanitized.replace(value, marker)
protected[marker] = value
# Payment sequences use digit-count validation rather than a plain rule.
sanitized, payment_count = _mask_payment_sequences(sanitized)
if payment_count:
findings.append({"type": "payment_barcode", "action": "redact", "count": payment_count})
rules = [r for r in _BUILTIN_RULES if r.name != "payment_barcode"]
rules.extend(_extra_rules_from_env())
rules.extend(_extra_rules_from_policy(policy))
for rule in rules:
try:
regex = re.compile(rule.pattern, rule.flags)
except re.error:
continue
sanitized, count = regex.subn(rule.replacement, sanitized)
if count:
findings.append({"type": rule.name, "action": "redact", "count": count})
for marker, value in protected.items():
sanitized = sanitized.replace(marker, value)
return sanitized, findings
__all__ = ["DlexMaskRule", "sanitize_dlex_output"]

View File

@@ -25,6 +25,7 @@ from .calibrated.output_sanitization import mascarar_pii_output, sanitizar_toxic
from .calibrated.rules.pinj_patterns import _PINJ_PATTERNS, is_obvious_injection
from .calibrated.rules.tox_blocklist import _EXPLICIT_TERMS, _THREAT_PATTERNS, is_obvious_toxic
from .framework_llm_client import classify_with_framework_llm
from .dlex_sanitizer import sanitize_dlex_output
from agent_framework.workflows.input_contract import has_meaningful_unmatched_policy, has_semantic_classifier
@@ -832,11 +833,64 @@ class DataLeakageOutputRail(Guardrail):
}
protocol_authorization_verified = True
# DLEX_OUT is permissive for recoverable leakage: when classification
# rejects the candidate, mask only generic sensitive values and re-run
# the same classifier. The agent/domain remains unaware of this detail.
# If sanitization cannot change the candidate, or the sanitized version
# is still unsafe, the original blocking behavior is preserved.
dlex_sanitized_text = None
dlex_findings: list[dict[str, Any]] = []
sanitization_recheck = None
original_dlex_result = dict(out)
if not bool(out.get("allowed", True)):
policy = dict(getattr(self, "_guardrail_policy", {}) or {})
candidate_sanitized, dlex_findings = sanitize_dlex_output(
original_text,
policy=policy,
exclude_values=matched_expected_protocols,
)
if candidate_sanitized != original_text:
sanitized_classifier_text = candidate_sanitized
sanitized_classifier_ctx: dict[str, Any] = ctx
if matched_expected_protocols:
sanitized_classifier_text = _mask_authorized_protocol_values(
sanitized_classifier_text, matched_expected_protocols
)
sanitized_classifier_ctx = _mask_authorized_protocol_values(
ctx, matched_expected_protocols
)
sanitization_recheck = await classify_with_framework_llm(
_llm(ctx),
"DLEX_OUT",
{"text": sanitized_classifier_text, "context": sanitized_classifier_ctx},
profile_name="grl",
component_name="guardrail.dlex_out.sanitization_recheck",
generation_name="guardrail.dlex_out.sanitization_recheck",
)
if bool(sanitization_recheck.get("allowed", True)):
dlex_sanitized_text = candidate_sanitized
out = {
"allowed": True,
"label": "SANITIZED",
"reason": "DLEX_OUT sanitizado e revalidado com segurança",
"original_reason": original_dlex_result.get("reason"),
}
metadata = {
"mechanism": "llm_rail",
"data": out,
"calibrated": True,
}
if dlex_findings:
metadata.update(
{
"action": "sanitize" if dlex_sanitized_text is not None else "block",
"sanitized": dlex_sanitized_text is not None,
"findings": dlex_findings,
"original_decision": original_dlex_result,
"sanitization_recheck": sanitization_recheck,
}
)
if matched_expected_protocols:
metadata.update(
{
@@ -850,7 +904,7 @@ class DataLeakageOutputRail(Guardrail):
code=self.code,
allowed=bool(out.get("allowed", True)),
reason=str(out.get("reason") or out.get("label") or "DLEX_OUT avaliado"),
sanitized_text=text,
sanitized_text=dlex_sanitized_text,
metadata=metadata,
)

Some files were not shown because too many files have changed in this diff Show More