68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
"""Decide whether external messages should be mirrored into a Human Handoff thread."""
|
|
|
|
MESSAGE_SCOPE_EXTERNAL = "EXTERNAL"
|
|
|
|
# Message provenance — who authored / generated the content
|
|
MESSAGE_ORIGIN_EXTERNAL_CUSTOMER = "EXTERNAL_CUSTOMER"
|
|
MESSAGE_ORIGIN_HUMAN_OWNER = "HUMAN_OWNER"
|
|
MESSAGE_ORIGIN_AGENT = "AGENT"
|
|
MESSAGE_ORIGIN_SYSTEM = "SYSTEM"
|
|
MESSAGE_ORIGIN_HANDOFF_MIRROR = "HANDOFF_MIRROR"
|
|
|
|
GENERATED_BY_CUSTOMER = "CUSTOMER"
|
|
GENERATED_BY_HUMAN = "HUMAN"
|
|
GENERATED_BY_AGENT = "AGENT"
|
|
GENERATED_BY_HANDOFF_BRIDGE = "HANDOFF_BRIDGE"
|
|
GENERATED_BY_SYSTEM = "SYSTEM"
|
|
GENERATED_BY_RELAY_ACK = "RELAY_ACK"
|
|
|
|
RELAY_STATUS_SENT = "SENT"
|
|
RELAY_STATUS_FAILED = "FAILED"
|
|
RELAY_STATUS_PENDING = "PENDING"
|
|
RELAY_STATUS_BLOCKED = "BLOCKED"
|
|
|
|
|
|
def should_mirror_to_human(message: dict) -> bool:
|
|
"""Return True only for new external customer content worth mirroring."""
|
|
scope = message.get("message_scope", MESSAGE_SCOPE_EXTERNAL)
|
|
if scope != MESSAGE_SCOPE_EXTERNAL:
|
|
return False
|
|
|
|
if message.get("direction") not in ("inbound", "outbound"):
|
|
return False
|
|
|
|
generated_by = message.get("generated_by")
|
|
if generated_by in (
|
|
GENERATED_BY_HANDOFF_BRIDGE,
|
|
GENERATED_BY_RELAY_ACK,
|
|
GENERATED_BY_SYSTEM,
|
|
):
|
|
return False
|
|
|
|
origin = message.get("message_origin")
|
|
if origin in (
|
|
MESSAGE_ORIGIN_HUMAN_OWNER,
|
|
MESSAGE_ORIGIN_SYSTEM,
|
|
MESSAGE_ORIGIN_HANDOFF_MIRROR,
|
|
):
|
|
return False
|
|
|
|
if message.get("generated_from_handoff_message_id"):
|
|
return False
|
|
|
|
if message.get("direction") == "outbound":
|
|
return (
|
|
generated_by == GENERATED_BY_AGENT
|
|
or origin == MESSAGE_ORIGIN_AGENT
|
|
)
|
|
return True
|
|
|
|
|
|
def is_relay_acknowledgment_message(message: dict) -> bool:
|
|
"""Detect legacy or internal delivery-receipt messages in handoff threads."""
|
|
if message.get("generated_by") == GENERATED_BY_RELAY_ACK:
|
|
return True
|
|
if message.get("message_origin") == MESSAGE_ORIGIN_SYSTEM:
|
|
return True
|
|
body = (message.get("body_text") or message.get("snippet") or "").strip()
|
|
return message.get("direction") == "outbound" and body.startswith("Sent to customer")
|