fix: prevent email loop by detecting bounces and capping conversation length
- email_channel.py: add detect_automated_message() that inspects RFC 3834 Auto-Submitted, mailer-daemon/postmaster sender patterns, X-Loop, multipart/report Content-Type, Precedence, and subject heuristics. fetch_unread_messages() now includes is_automated / automated_reason in every message dict. - main.py: if is_automated is set, call agent.handle_automated_message() instead of process_message() — no reply is ever sent to a bounce source. - agent/core.py: add MAX_USER_MESSAGES = 20 cap; process_message() returns "" without replying once a conversation exceeds the limit and calls notifier.notify_loop_escalation() on first breach. New public method handle_automated_message() records the event and triggers the same one-shot admin alert. - models/conversation.py: add loop_escalated: bool field (persisted) so the admin alert fires at most once per conversation. - notifications/notifier.py: add notify_loop_escalation() which sends a plain-text warning to the admin CC list (Markus Graf / spielgruppen@). https://claude.ai/code/session_01KwvR5hDPjSuJg4kvw5b5e5
This commit is contained in:
@@ -98,6 +98,27 @@ def run_poll_loop(agent: EmailAgent, channel: EmailChannel, poll_interval: int)
|
|||||||
for msg in messages:
|
for msg in messages:
|
||||||
logger.info("Processing message from %s", msg["from"])
|
logger.info("Processing message from %s", msg["from"])
|
||||||
try:
|
try:
|
||||||
|
# ----------------------------------------------------------
|
||||||
|
# Bounce / automated-sender guard
|
||||||
|
# If the channel layer flagged this as an automated message
|
||||||
|
# (bounce, out-of-office, delivery failure, …) we must NOT
|
||||||
|
# reply — that would create or worsen an email loop.
|
||||||
|
# Instead, alert the admin once and drop the message.
|
||||||
|
# ----------------------------------------------------------
|
||||||
|
if msg.get("is_automated"):
|
||||||
|
logger.warning(
|
||||||
|
"Automated/bounce message from %s — reason: %s — not replying",
|
||||||
|
msg["from"],
|
||||||
|
msg.get("automated_reason", "unknown"),
|
||||||
|
)
|
||||||
|
agent.handle_automated_message(
|
||||||
|
sender_email=msg["from"],
|
||||||
|
subject=msg["subject"],
|
||||||
|
reason=msg.get("automated_reason", "automated sender detected"),
|
||||||
|
inbound_message_id=msg["message_id"],
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
|
||||||
# Prepend email headers so the LLM can extract the
|
# Prepend email headers so the LLM can extract the
|
||||||
# sender's address and subject (e.g. to fill in
|
# sender's address and subject (e.g. to fill in
|
||||||
# parentGuardian.email automatically).
|
# parentGuardian.email automatically).
|
||||||
|
|||||||
@@ -14,6 +14,11 @@ from .response_parser import apply_updates, fallback_message, parse_llm_response
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# Maximum number of inbound user messages before the conversation is stopped and
|
||||||
|
# escalated to the admin. This prevents runaway loops that slip through automated
|
||||||
|
# sender detection (e.g. a forwarding alias that bounces the agent's own replies).
|
||||||
|
MAX_USER_MESSAGES = 20
|
||||||
|
|
||||||
|
|
||||||
class EmailAgent:
|
class EmailAgent:
|
||||||
"""Processes one inbound email and returns the agent's reply text.
|
"""Processes one inbound email and returns the agent's reply text.
|
||||||
@@ -78,6 +83,43 @@ class EmailAgent:
|
|||||||
# Append the user's message to history
|
# Append the user's message to history
|
||||||
state.messages.append(ChatMessage(role="user", content=message_text))
|
state.messages.append(ChatMessage(role="user", content=message_text))
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# Hard message-count cap — stop conversations that have gone on too
|
||||||
|
# long without completing (covers loops that bypass automated-sender
|
||||||
|
# detection, e.g. a broken forwarding alias).
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
user_msg_count = sum(1 for m in state.messages if m.role == "user")
|
||||||
|
if user_msg_count > MAX_USER_MESSAGES:
|
||||||
|
if not state.loop_escalated:
|
||||||
|
state.loop_escalated = True
|
||||||
|
state.updated_at = now
|
||||||
|
self._store.save(state)
|
||||||
|
reason = (
|
||||||
|
f"conversation exceeded {MAX_USER_MESSAGES} inbound messages "
|
||||||
|
f"without completing"
|
||||||
|
)
|
||||||
|
logger.warning(
|
||||||
|
"Conversation %s exceeded message limit (%d user messages) — escalating",
|
||||||
|
email_key,
|
||||||
|
user_msg_count,
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
self._notifier.notify_loop_escalation(
|
||||||
|
sender_email=parent_email,
|
||||||
|
conversation_id=email_key,
|
||||||
|
reason=reason,
|
||||||
|
message_count=user_msg_count,
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
logger.exception("Failed to send loop escalation notification for %s", email_key)
|
||||||
|
else:
|
||||||
|
logger.warning(
|
||||||
|
"Conversation %s still exceeding message limit — already escalated, ignoring",
|
||||||
|
email_key,
|
||||||
|
)
|
||||||
|
self._store.save(state)
|
||||||
|
return ""
|
||||||
|
|
||||||
# Route to the appropriate handler
|
# Route to the appropriate handler
|
||||||
if state.completed:
|
if state.completed:
|
||||||
reply_text = self._handle_post_completion(state)
|
reply_text = self._handle_post_completion(state)
|
||||||
@@ -91,6 +133,60 @@ class EmailAgent:
|
|||||||
|
|
||||||
return reply_text
|
return reply_text
|
||||||
|
|
||||||
|
def handle_automated_message(
|
||||||
|
self,
|
||||||
|
sender_email: str,
|
||||||
|
subject: str,
|
||||||
|
reason: str,
|
||||||
|
inbound_message_id: str = "",
|
||||||
|
) -> None:
|
||||||
|
"""Handle an inbound message detected as automated/bounce.
|
||||||
|
|
||||||
|
Does NOT send any reply (to avoid looping). Alerts the admin once per
|
||||||
|
conversation — subsequent automated messages from the same sender are
|
||||||
|
silently dropped after the first alert.
|
||||||
|
"""
|
||||||
|
email_key = normalize_email(sender_email)
|
||||||
|
state = self._store.load(email_key)
|
||||||
|
if state is None:
|
||||||
|
state = ConversationState(
|
||||||
|
conversation_id=email_key,
|
||||||
|
parent_email=email_key,
|
||||||
|
)
|
||||||
|
|
||||||
|
now = datetime.now(timezone.utc).isoformat()
|
||||||
|
state.last_activity = now
|
||||||
|
if inbound_message_id:
|
||||||
|
state.last_inbound_message_id = inbound_message_id
|
||||||
|
|
||||||
|
message_count = sum(1 for m in state.messages if m.role == "user")
|
||||||
|
|
||||||
|
if state.loop_escalated:
|
||||||
|
logger.info(
|
||||||
|
"Automated message from %s (already escalated) — dropping silently", sender_email
|
||||||
|
)
|
||||||
|
self._store.save(state)
|
||||||
|
return
|
||||||
|
|
||||||
|
state.loop_escalated = True
|
||||||
|
state.updated_at = now
|
||||||
|
self._store.save(state)
|
||||||
|
|
||||||
|
logger.warning(
|
||||||
|
"Automated/bounce message from %s — reason: %s — alerting admin", sender_email, reason
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
self._notifier.notify_loop_escalation(
|
||||||
|
sender_email=sender_email,
|
||||||
|
conversation_id=email_key,
|
||||||
|
reason=reason,
|
||||||
|
message_count=message_count,
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
logger.exception(
|
||||||
|
"Failed to send loop escalation notification for automated sender %s", email_key
|
||||||
|
)
|
||||||
|
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
# Registration flow
|
# Registration flow
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
|
|||||||
@@ -89,6 +89,75 @@ def _strip_quoted_text(text: str) -> str:
|
|||||||
return "\n".join(result).strip()
|
return "\n".join(result).strip()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Automated / bounce message detection
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
# Local parts of sender addresses that are never humans (RFC 5321 §4.5.4, common practice).
|
||||||
|
_AUTOMATED_SENDER_RE = re.compile(
|
||||||
|
r"^(mailer-daemon|postmaster|noreply|no-reply|no\.reply|do-not-reply|"
|
||||||
|
r"donotreply|bounce|bounce\+.*|delivery|mail-delivery|mail\.delivery)$",
|
||||||
|
re.IGNORECASE,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Subject lines that indicate delivery failure or automated responses.
|
||||||
|
_AUTOMATED_SUBJECT_RE = re.compile(
|
||||||
|
r"(undelivered mail|undeliverable|delivery (failed|status|notification)|"
|
||||||
|
r"mail delivery (failed|error)|returned to sender|mailer-daemon|"
|
||||||
|
r"auto.?reply|out of office|außer haus|abwesenheitsnotiz|automatische antwort)",
|
||||||
|
re.IGNORECASE,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def detect_automated_message(raw_msg: email.message.Message, from_addr: str) -> tuple[bool, str]:
|
||||||
|
"""Detect whether an email was generated by an automated system, not a human.
|
||||||
|
|
||||||
|
Checks (in order of reliability):
|
||||||
|
1. Sender local-part (mailer-daemon, postmaster, noreply, …)
|
||||||
|
2. Auto-Submitted header (RFC 3834)
|
||||||
|
3. X-Auto-Response-Suppress header (Microsoft Exchange)
|
||||||
|
4. Content-Type: multipart/report (RFC 3462 — Delivery Status Notifications)
|
||||||
|
5. X-Loop header
|
||||||
|
6. Precedence: bulk/junk
|
||||||
|
7. Subject-line heuristics
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
(True, reason_string) if automated, (False, "") otherwise.
|
||||||
|
"""
|
||||||
|
local = from_addr.split("@")[0] if "@" in from_addr else from_addr
|
||||||
|
if _AUTOMATED_SENDER_RE.match(local):
|
||||||
|
return True, f"sender matches automated address pattern: {from_addr}"
|
||||||
|
|
||||||
|
# RFC 3834 — Auto-Submitted header
|
||||||
|
auto_submitted = raw_msg.get("Auto-Submitted", "").strip().lower()
|
||||||
|
if auto_submitted and auto_submitted != "no":
|
||||||
|
return True, f"Auto-Submitted: {auto_submitted}"
|
||||||
|
|
||||||
|
# Microsoft Exchange — suppresses auto-replies
|
||||||
|
if raw_msg.get("X-Auto-Response-Suppress"):
|
||||||
|
return True, "X-Auto-Response-Suppress header present"
|
||||||
|
|
||||||
|
# RFC 3462 — multipart/report is used for DSNs and MDNs
|
||||||
|
if raw_msg.get_content_type() == "multipart/report":
|
||||||
|
return True, "Content-Type: multipart/report (delivery status notification)"
|
||||||
|
|
||||||
|
# X-Loop — set by mailing-list managers and some MTAs to break loops
|
||||||
|
if raw_msg.get("X-Loop"):
|
||||||
|
return True, f"X-Loop: {raw_msg.get('X-Loop')}"
|
||||||
|
|
||||||
|
# Precedence header
|
||||||
|
precedence = raw_msg.get("Precedence", "").strip().lower()
|
||||||
|
if precedence in ("bulk", "junk", "auto_reply"):
|
||||||
|
return True, f"Precedence: {precedence}"
|
||||||
|
|
||||||
|
# Subject heuristic (weakest — only flag when highly specific)
|
||||||
|
subject = _decode_header(raw_msg.get("Subject", ""))
|
||||||
|
if _AUTOMATED_SUBJECT_RE.search(subject):
|
||||||
|
return True, f"subject matches automated pattern: {subject!r}"
|
||||||
|
|
||||||
|
return False, ""
|
||||||
|
|
||||||
|
|
||||||
def _generate_message_id(from_addr: str) -> str:
|
def _generate_message_id(from_addr: str) -> str:
|
||||||
domain = from_addr.split("@")[-1] if "@" in from_addr else "meister-eder.local"
|
domain = from_addr.split("@")[-1] if "@" in from_addr else "meister-eder.local"
|
||||||
return f"<{time.time():.6f}.{id(from_addr)}@{domain}>"
|
return f"<{time.time():.6f}.{id(from_addr)}@{domain}>"
|
||||||
@@ -185,6 +254,8 @@ class EmailChannel:
|
|||||||
imap.store(num, "+FLAGS", "\\Seen")
|
imap.store(num, "+FLAGS", "\\Seen")
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
is_automated, automated_reason = detect_automated_message(msg, from_addr)
|
||||||
|
|
||||||
messages.append(
|
messages.append(
|
||||||
{
|
{
|
||||||
"from": from_addr,
|
"from": from_addr,
|
||||||
@@ -194,6 +265,8 @@ class EmailChannel:
|
|||||||
"references": references,
|
"references": references,
|
||||||
"body": body,
|
"body": body,
|
||||||
"raw_body": raw_body,
|
"raw_body": raw_body,
|
||||||
|
"is_automated": is_automated,
|
||||||
|
"automated_reason": automated_reason,
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
imap.store(num, "+FLAGS", "\\Seen")
|
imap.store(num, "+FLAGS", "\\Seen")
|
||||||
|
|||||||
@@ -35,6 +35,9 @@ class ConversationState:
|
|||||||
# Most recent inbound Message-ID — used for reply threading headers only,
|
# Most recent inbound Message-ID — used for reply threading headers only,
|
||||||
# NOT for conversation matching (which is always by email address).
|
# NOT for conversation matching (which is always by email address).
|
||||||
last_inbound_message_id: str = ""
|
last_inbound_message_id: str = ""
|
||||||
|
# Loop / automated-sender prevention.
|
||||||
|
# Set to True once the admin has been notified; prevents repeated alerts.
|
||||||
|
loop_escalated: bool = False
|
||||||
|
|
||||||
def to_dict(self) -> dict:
|
def to_dict(self) -> dict:
|
||||||
return {
|
return {
|
||||||
@@ -54,6 +57,7 @@ class ConversationState:
|
|||||||
"completed": self.completed,
|
"completed": self.completed,
|
||||||
"reminder_count": self.reminder_count,
|
"reminder_count": self.reminder_count,
|
||||||
"last_inbound_message_id": self.last_inbound_message_id,
|
"last_inbound_message_id": self.last_inbound_message_id,
|
||||||
|
"loop_escalated": self.loop_escalated,
|
||||||
}
|
}
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@@ -78,4 +82,5 @@ class ConversationState:
|
|||||||
state.completed = data.get("completed", False)
|
state.completed = data.get("completed", False)
|
||||||
state.reminder_count = data.get("reminder_count", 0)
|
state.reminder_count = data.get("reminder_count", 0)
|
||||||
state.last_inbound_message_id = data.get("last_inbound_message_id", "")
|
state.last_inbound_message_id = data.get("last_inbound_message_id", "")
|
||||||
|
state.loop_escalated = data.get("loop_escalated", False)
|
||||||
return state
|
return state
|
||||||
|
|||||||
@@ -131,6 +131,46 @@ class AdminNotifier:
|
|||||||
reply_to=registration.parent_guardian.email or "",
|
reply_to=registration.parent_guardian.email or "",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def notify_loop_escalation(
|
||||||
|
self,
|
||||||
|
sender_email: str,
|
||||||
|
conversation_id: str,
|
||||||
|
reason: str,
|
||||||
|
message_count: int,
|
||||||
|
) -> None:
|
||||||
|
"""Alert the admin that a conversation was stopped due to a loop or automated sender.
|
||||||
|
|
||||||
|
Sent to the CC list (Markus Graf / admin) only — no playgroup leader routing needed.
|
||||||
|
"""
|
||||||
|
if not self._cc_emails:
|
||||||
|
logger.warning(
|
||||||
|
"No admin CC email configured — loop escalation NOT sent for %s", conversation_id
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
|
subject = f"[WARNUNG] Automatische E-Mail / Endlosschleife erkannt: {sender_email}"
|
||||||
|
body = (
|
||||||
|
f"Das Anmeldungssystem hat eine Konversation automatisch gestoppt.\n\n"
|
||||||
|
f"Absender: {sender_email}\n"
|
||||||
|
f"Konversations-ID: {conversation_id}\n"
|
||||||
|
f"Nachrichten: {message_count}\n"
|
||||||
|
f"Grund: {reason}\n\n"
|
||||||
|
f"Es wurde keine weitere Antwort gesendet. Bitte prüfen Sie den Sachverhalt "
|
||||||
|
f"manuell und leiten Sie die Konversation bei Bedarf weiter.\n\n"
|
||||||
|
f"---\nMeister-Eder Anmeldungssystem"
|
||||||
|
)
|
||||||
|
self._send(
|
||||||
|
to=self._cc_emails,
|
||||||
|
cc=[],
|
||||||
|
subject=subject,
|
||||||
|
body=body,
|
||||||
|
)
|
||||||
|
logger.info(
|
||||||
|
"Loop escalation notification sent to admin for conversation %s (reason: %s)",
|
||||||
|
conversation_id,
|
||||||
|
reason,
|
||||||
|
)
|
||||||
|
|
||||||
def notify_parent(
|
def notify_parent(
|
||||||
self,
|
self,
|
||||||
registration: RegistrationData,
|
registration: RegistrationData,
|
||||||
|
|||||||
Reference in New Issue
Block a user