Suriyakumarvijayanayagam
9cbb6f9704
With DISPATCH_AGENT autonomous=true in the agent registry, the assignment-failure
path sends a real customer message. It would have sent
"Delay alert: Order 36 may arrive later than expected. New ETA: {new_eta}"
with customer_id "unknown": the DELAYED template names {new_eta} but DispatchAgent
supplied only order_id, and it has no customer id to give.
- _send_via_channel refuses to send when any {placeholder} survives substitution
(or the template is missing), logs which variables the caller omitted, and
records a failed notification. Silence beats a broken message.
- customer_id is omitted from the payload when it is absent or "unknown" rather
than sent literally; the backend resolves the recipient from order_id.
- DispatchAgent supplies new_eta "being confirmed" — honest, since no ETA exists
at assignment-failure time.
- Tests: the old bug is refused, a complete payload renders clean, the id is
omitted/passed correctly, and DispatchAgent's vars satisfy the template.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012AJLYcbTHCe45fyFnMfEin
495 lines
21 KiB
Python
495 lines
21 KiB
Python
"""Customer Agent - Handles notifications, tracking, and customer communication."""
|
|
import re
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Any, Optional
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
|
|
import aiohttp
|
|
|
|
from core.agent import SpecializedAgent
|
|
from core.types import AgentTask, MessageType, OrderStatus
|
|
from core.logger import logger
|
|
from core.http_client import api_post, api_get
|
|
from core.registry import registry
|
|
from config.system_config import GO_API_BASE_URL, INTERNAL_API_KEY
|
|
|
|
# Any {placeholder} left after template substitution — a caller forgot a variable.
|
|
_UNFILLED_PLACEHOLDER = re.compile(r"\{([a-zA-Z_][a-zA-Z0-9_]*)\}")
|
|
|
|
|
|
class NotificationChannel(str, Enum):
|
|
SMS = "sms"
|
|
EMAIL = "email"
|
|
PUSH = "push"
|
|
WHATSAPP = "whatsapp"
|
|
|
|
|
|
class NotificationType(str, Enum):
|
|
ORDER_CONFIRMED = "order_confirmed"
|
|
PICKUP_SCHEDULED = "pickup_scheduled"
|
|
PICKED_UP = "picked_up"
|
|
IN_TRANSIT = "in_transit"
|
|
ARRIVED_AT_HUB = "arrived_at_hub"
|
|
OUT_FOR_DELIVERY = "out_for_delivery"
|
|
DELIVERED = "delivered"
|
|
DELAYED = "delayed"
|
|
RESCHEDULED = "rescheduled"
|
|
CANCELLED = "cancelled"
|
|
|
|
|
|
@dataclass
|
|
class Notification:
|
|
notification_id: str
|
|
order_id: str
|
|
customer_id: str
|
|
channel: NotificationChannel
|
|
notification_type: NotificationType
|
|
message: str
|
|
sent_at: datetime
|
|
delivered_at: Optional[datetime]
|
|
status: str # pending, sent, delivered, failed
|
|
|
|
|
|
class CustomerAgent(SpecializedAgent):
|
|
"""
|
|
Customer Agent - Manages customer communications and notifications.
|
|
Sends via Go backend (POST /api/v1/internal/notify → real FCM/SMS/WhatsApp).
|
|
Keeps local notification history for audit / query.
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
agent_id="CUSTOMER_AGENT",
|
|
domain="customer_communication",
|
|
description="Handles notifications, tracking updates, and customer communication"
|
|
)
|
|
self._customers: Dict[str, Dict] = {}
|
|
self._notifications: Dict[str, List[Notification]] = {}
|
|
self._templates = self._init_templates()
|
|
self._preferences: Dict[str, Dict] = {}
|
|
self._received_events: List[Dict[str, Any]] = []
|
|
|
|
# Event notifications other agents send without a task_type. They used to
|
|
# fall through to the base handle_message and be dropped at DEBUG level.
|
|
# They come from ExceptionAgent's generic exception tasks, which run over
|
|
# in-memory, simulated records (their order ids are not real bookings), so
|
|
# they are RECORDED here and logged — deliberately not turned into a real
|
|
# customer message through the backend's /internal/notify.
|
|
_RECORDED_EVENTS = {MessageType.ORDER_CANCELLED, MessageType.NOTIFICATION_SENT}
|
|
|
|
async def handle_message(self, message) -> None:
|
|
if message.message_type not in self._RECORDED_EVENTS:
|
|
await super().handle_message(message)
|
|
return
|
|
payload = message.payload if isinstance(message.payload, dict) else {}
|
|
event = {
|
|
"type": message.message_type.value,
|
|
"from": message.sender,
|
|
"order_id": payload.get("order_id"),
|
|
"received_at": datetime.now().isoformat(),
|
|
}
|
|
self._received_events = (self._received_events + [event])[-200:]
|
|
logger.info(
|
|
f"Customer Agent: recorded {event['type']} from {event['from']} for order {event['order_id']} "
|
|
"(simulated exception flow; no customer message sent)"
|
|
)
|
|
|
|
def _init_templates(self) -> Dict[str, Dict]:
|
|
return {
|
|
NotificationType.ORDER_CONFIRMED: {
|
|
"sms": "Your order {order_id} has been confirmed! Estimated delivery: {eta}",
|
|
"email": "Hi {customer_name}, your order {order_id} is confirmed. Track at: {tracking_url}",
|
|
"whatsapp": "Order {order_id} confirmed! Delivery by {eta}. Track: {tracking_url}"
|
|
},
|
|
NotificationType.PICKUP_SCHEDULED: {
|
|
"sms": "Pickup scheduled for {pickup_time}. Driver will arrive within 30 mins.",
|
|
"email": "Your pickup is scheduled for {pickup_time}",
|
|
"whatsapp": "Pickup scheduled for {pickup_time}. Driver arrives in 30 mins"
|
|
},
|
|
NotificationType.PICKED_UP: {
|
|
"sms": "Order {order_id} picked up and on its way!",
|
|
"email": "Your package has been picked up and is in transit.",
|
|
"whatsapp": "Order {order_id} picked up! Heading to {destination}"
|
|
},
|
|
NotificationType.IN_TRANSIT: {
|
|
"sms": "Your order {order_id} is now in transit.",
|
|
"email": "Update: Your order is on the move!",
|
|
"whatsapp": "Order {order_id} in transit. Current stop: {current_location}"
|
|
},
|
|
NotificationType.ARRIVED_AT_HUB: {
|
|
"sms": "Your order {order_id} has arrived at {hub_name}.",
|
|
"email": "Your package has arrived at the {hub_name} facility.",
|
|
"whatsapp": "Order {order_id} arrived at {hub_name}"
|
|
},
|
|
NotificationType.OUT_FOR_DELIVERY: {
|
|
"sms": "Out for delivery! Driver: {driver_name}, Contact: {driver_phone}",
|
|
"email": "Your order is out for delivery!",
|
|
"whatsapp": "Out for delivery! Driver: {driver_name} | {driver_phone}"
|
|
},
|
|
NotificationType.DELIVERED: {
|
|
"sms": "Order {order_id} delivered successfully! Thank you for choosing us.",
|
|
"email": "Your order has been delivered! We hope you enjoy your purchase.",
|
|
"whatsapp": "Delivered! Order {order_id}. Thank you for shopping with us!"
|
|
},
|
|
NotificationType.DELAYED: {
|
|
"sms": "Delay alert: Order {order_id} may arrive later than expected. New ETA: {new_eta}",
|
|
"email": "We apologize for the delay. Your order's new ETA is {new_eta}",
|
|
"whatsapp": "Delay update: Order {order_id}. New ETA: {new_eta}"
|
|
},
|
|
NotificationType.RESCHEDULED: {
|
|
"sms": "Delivery rescheduled. New date: {new_date}",
|
|
"email": "Your delivery has been rescheduled to {new_date}",
|
|
"whatsapp": "Rescheduled! New delivery: {new_date}"
|
|
},
|
|
NotificationType.CANCELLED: {
|
|
"sms": "Order {order_id} has been cancelled. Refund processing: {refund_info}",
|
|
"email": "Your order has been cancelled. {refund_info}",
|
|
"whatsapp": "Cancelled: Order {order_id}. {refund_info}"
|
|
}
|
|
}
|
|
|
|
async def handle_task(self, task: AgentTask) -> Dict[str, Any]:
|
|
handlers = {
|
|
"send_confirmation": self._send_confirmation,
|
|
"send_notification": self._send_notification,
|
|
"update_tracking": self._update_tracking,
|
|
"get_tracking": self._get_tracking,
|
|
"send_bulk_notification": self._send_bulk_notification,
|
|
"get_customer_history": self._get_customer_history,
|
|
"handle_query": self._handle_query,
|
|
}
|
|
handler = handlers.get(task.task_type, self._unknown_task)
|
|
return await handler(task)
|
|
|
|
async def _send_confirmation(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_data = task.data.get("order", {})
|
|
order_id = order_data.get("order_id")
|
|
customer_phone = order_data.get("customer_phone")
|
|
customer_email = order_data.get("customer_email")
|
|
customer_name = order_data.get("customer_name", "Customer")
|
|
eta = task.data.get("eta", "tomorrow")
|
|
|
|
logger.info(f"Customer Agent: Sending confirmation for order {order_id}")
|
|
|
|
if customer_phone and customer_phone not in self._customers:
|
|
self._customers[customer_phone] = {
|
|
"name": customer_name,
|
|
"email": customer_email,
|
|
"orders": []
|
|
}
|
|
|
|
if customer_phone in self._customers:
|
|
self._customers[customer_phone]["orders"].append(order_id)
|
|
|
|
template_vars = {
|
|
"order_id": order_id,
|
|
"customer_name": customer_name,
|
|
"eta": eta,
|
|
"tracking_url": f"https://track.doormile.com/{order_id}"
|
|
}
|
|
|
|
notifications_sent = []
|
|
|
|
if customer_phone:
|
|
notifications_sent.append(await self._send_via_channel(
|
|
order_id=order_id, customer_id=customer_phone,
|
|
channel=NotificationChannel.SMS,
|
|
notification_type=NotificationType.ORDER_CONFIRMED,
|
|
template_vars=template_vars,
|
|
))
|
|
notifications_sent.append(await self._send_via_channel(
|
|
order_id=order_id, customer_id=customer_phone,
|
|
channel=NotificationChannel.WHATSAPP,
|
|
notification_type=NotificationType.ORDER_CONFIRMED,
|
|
template_vars=template_vars,
|
|
))
|
|
|
|
if customer_email:
|
|
notifications_sent.append(await self._send_via_channel(
|
|
order_id=order_id, customer_id=customer_email,
|
|
channel=NotificationChannel.EMAIL,
|
|
notification_type=NotificationType.ORDER_CONFIRMED,
|
|
template_vars=template_vars,
|
|
))
|
|
|
|
return {"status": "sent", "order_id": order_id, "notifications": notifications_sent}
|
|
|
|
async def _send_notification(self, task: AgentTask) -> Dict[str, Any]:
|
|
if not registry.skill_enabled("customer_notifications"):
|
|
logger.info("Customer Agent: skill customer_notifications is disabled in the agent registry; not sending")
|
|
return {"status": "skipped", "reason": "customer_notifications disabled in the agent registry"}
|
|
order_id = task.data.get("order_id")
|
|
notification_type = task.data.get("notification_type")
|
|
template_vars = task.data.get("template_vars", {})
|
|
customer_id = task.data.get("customer_id", "unknown")
|
|
|
|
logger.info(f"Customer Agent: Sending {notification_type} for order {order_id}")
|
|
|
|
notification = await self._send_via_channel(
|
|
order_id=order_id,
|
|
customer_id=customer_id,
|
|
channel=NotificationChannel.SMS,
|
|
notification_type=NotificationType(notification_type),
|
|
template_vars=template_vars,
|
|
)
|
|
|
|
return {"status": "sent", "notification": notification}
|
|
|
|
async def _update_tracking(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_id = task.data.get("order_id")
|
|
status = task.data.get("status")
|
|
location = task.data.get("location")
|
|
eta = task.data.get("eta")
|
|
|
|
logger.info(f"Customer Agent: Updating tracking for order {order_id} | status={status} location={location}")
|
|
|
|
status_notification_map = {
|
|
"picked_up": NotificationType.PICKED_UP,
|
|
"in_transit": NotificationType.IN_TRANSIT,
|
|
"at_hub": NotificationType.ARRIVED_AT_HUB,
|
|
"out_for_delivery": NotificationType.OUT_FOR_DELIVERY,
|
|
"delivered": NotificationType.DELIVERED,
|
|
"delayed": NotificationType.DELAYED,
|
|
}
|
|
|
|
notification_type = status_notification_map.get(status)
|
|
if notification_type:
|
|
await self._send_via_channel(
|
|
order_id=order_id,
|
|
customer_id=task.data.get("customer_id", ""),
|
|
channel=NotificationChannel.SMS,
|
|
notification_type=notification_type,
|
|
template_vars={
|
|
"order_id": order_id,
|
|
"current_location": location,
|
|
"new_eta": eta,
|
|
"hub_name": location,
|
|
},
|
|
)
|
|
|
|
return {"status": "updated", "order_id": order_id, "new_status": status, "location": location}
|
|
|
|
async def _get_tracking(self, task: AgentTask) -> Dict[str, Any]:
|
|
booking_id = task.data.get("booking_id") or task.data.get("order_id")
|
|
|
|
result = await api_get(
|
|
f"{GO_API_BASE_URL}/api/v1/bookings/cache/{booking_id}",
|
|
headers={"X-Internal-Key": INTERNAL_API_KEY},
|
|
)
|
|
|
|
if not result:
|
|
logger.warning(f"Tracking lookup failed for booking {booking_id}")
|
|
return {"status": "not_found", "order_id": booking_id}
|
|
|
|
data = result.get("data") or result
|
|
return {
|
|
"order_id": booking_id,
|
|
"current_status": data.get("status", "unknown"),
|
|
"miler_id": data.get("assignedmileruserid"),
|
|
"miler_name": data.get("milername"),
|
|
"miler_phone": data.get("milerphone"),
|
|
"estimated_delivery": data.get("estimateddelivery"),
|
|
"pickup_address": data.get("pickupaddress"),
|
|
"delivery_address": data.get("deliveryaddress"),
|
|
"last_update": data.get("updatedat") or datetime.now().isoformat(),
|
|
}
|
|
|
|
async def _send_bulk_notification(self, task: AgentTask) -> Dict[str, Any]:
|
|
order_ids = task.data.get("order_ids", [])
|
|
notification_type = task.data.get("notification_type")
|
|
template_vars = task.data.get("template_vars", {})
|
|
|
|
logger.info(f"Customer Agent: Sending bulk notification to {len(order_ids)} customers")
|
|
|
|
results = []
|
|
for order_id in order_ids:
|
|
result = await self._send_notification(AgentTask(
|
|
task_id=f"bulk_{order_id}",
|
|
agent_type="customer",
|
|
task_type="send_notification",
|
|
data={
|
|
"order_id": order_id,
|
|
"notification_type": notification_type,
|
|
"template_vars": template_vars,
|
|
}
|
|
))
|
|
results.append(result)
|
|
|
|
return {
|
|
"status": "bulk_sent",
|
|
"total": len(order_ids),
|
|
"successful": sum(1 for r in results if r.get("status") == "sent"),
|
|
"failed": sum(1 for r in results if r.get("status") == "failed"),
|
|
}
|
|
|
|
async def _get_customer_history(self, task: AgentTask) -> Dict[str, Any]:
|
|
customer_id = task.data.get("customer_id")
|
|
notifications = self._notifications.get(customer_id, [])
|
|
return {
|
|
"customer_id": customer_id,
|
|
"total_notifications": len(notifications),
|
|
"notifications": [self._format_notification(n) for n in notifications[-20:]],
|
|
}
|
|
|
|
async def _handle_query(self, task: AgentTask) -> Dict[str, Any]:
|
|
query = task.data.get("query", "").lower()
|
|
order_id = task.data.get("order_id")
|
|
|
|
logger.info(f"Customer Agent: Handling query '{query}' for order {order_id}")
|
|
|
|
if "where" in query or "track" in query or "status" in query:
|
|
return await self._get_tracking(AgentTask(
|
|
task_id="query_tracking",
|
|
agent_type="customer",
|
|
task_type="get_tracking",
|
|
data={"order_id": order_id},
|
|
))
|
|
|
|
if "delay" in query or "late" in query:
|
|
return {
|
|
"response": "I understand your concern about the delay. Let me check the current status and get back to you with updated information.",
|
|
"action": "check_status",
|
|
}
|
|
|
|
if "cancel" in query:
|
|
return {
|
|
"response": f"I can help you with cancellation. Please confirm you want to cancel order {order_id}. Note: Cancellation is only possible before dispatch.",
|
|
"action": "await_confirmation",
|
|
}
|
|
|
|
return {
|
|
"response": "I'm here to help! You can ask about your order status, tracking, delays, or cancellation. How can I assist you?",
|
|
"action": "provide_options",
|
|
}
|
|
|
|
def _failed_notification(self, order_id, customer_id, channel, notification_type,
|
|
reason: str, message: str) -> Dict[str, Any]:
|
|
"""A not-sent result in the same shape as a sent one, recorded for audit."""
|
|
notification_id = f"NOTIF-{uuid.uuid4().hex[:8].upper()}"
|
|
key = customer_id or "unknown"
|
|
self._notifications.setdefault(key, []).append(Notification(
|
|
notification_id=notification_id,
|
|
order_id=order_id,
|
|
customer_id=key,
|
|
channel=channel,
|
|
notification_type=notification_type,
|
|
message=message,
|
|
sent_at=datetime.now(),
|
|
delivered_at=None,
|
|
status="failed",
|
|
))
|
|
return {
|
|
"notification_id": notification_id,
|
|
"channel": channel.value,
|
|
"status": "failed",
|
|
"reason": reason,
|
|
"message": message,
|
|
}
|
|
|
|
async def _send_via_channel(
|
|
self,
|
|
order_id: str,
|
|
customer_id: str,
|
|
channel: NotificationChannel,
|
|
notification_type: NotificationType,
|
|
template_vars: Dict[str, Any],
|
|
) -> Dict[str, Any]:
|
|
"""Render template and call Go backend (POST /api/v1/internal/notify).
|
|
|
|
Refuses to send a message that still contains an unsubstituted
|
|
{placeholder}: a caller that omits a template variable would otherwise
|
|
put the literal text in front of a customer. Silence is better than a
|
|
broken message, and the ERROR names the caller's missing variables."""
|
|
template = self._templates.get(notification_type, {}).get(channel.value, "")
|
|
if not template:
|
|
logger.error(
|
|
f"Customer Agent: no {channel.value} template for {notification_type.value}; not sending"
|
|
)
|
|
return self._failed_notification(order_id, customer_id, channel, notification_type,
|
|
"no template", "")
|
|
|
|
message = template
|
|
for key, value in template_vars.items():
|
|
message = message.replace(f"{{{key}}}", str(value))
|
|
|
|
missing = _UNFILLED_PLACEHOLDER.findall(message)
|
|
if missing:
|
|
logger.error(
|
|
f"Customer Agent: refusing to send {notification_type.value} for order {order_id} — "
|
|
f"template variables not supplied: {missing}"
|
|
)
|
|
return self._failed_notification(order_id, customer_id, channel, notification_type,
|
|
f"unfilled template variables: {missing}", message)
|
|
|
|
notification_id = f"NOTIF-{uuid.uuid4().hex[:8].upper()}"
|
|
|
|
payload = {
|
|
"notification_id": notification_id,
|
|
"order_id": order_id,
|
|
"channel": channel.value,
|
|
"notification_type": notification_type.value,
|
|
"message": message,
|
|
}
|
|
# Omit customer_id rather than sending a placeholder: callers that only
|
|
# know the booking (e.g. DispatchAgent on an assignment failure) leave the
|
|
# backend to resolve the recipient from order_id, which it can already do.
|
|
if customer_id and str(customer_id) != "unknown":
|
|
payload["customer_id"] = customer_id
|
|
|
|
result = await api_post(
|
|
f"{GO_API_BASE_URL}/api/v1/internal/notify",
|
|
json=payload,
|
|
headers={"X-Internal-Key": INTERNAL_API_KEY},
|
|
timeout=aiohttp.ClientTimeout(total=10),
|
|
)
|
|
|
|
if result is None:
|
|
status = "failed"
|
|
logger.warning(f"Notify API call failed for {notification_id}")
|
|
else:
|
|
status = "sent"
|
|
|
|
notification = Notification(
|
|
notification_id=notification_id,
|
|
order_id=order_id,
|
|
customer_id=customer_id,
|
|
channel=channel,
|
|
notification_type=notification_type,
|
|
message=message,
|
|
sent_at=datetime.now(),
|
|
delivered_at=None,
|
|
status=status,
|
|
)
|
|
history_key = customer_id or "unknown"
|
|
self._notifications.setdefault(history_key, []).append(notification)
|
|
|
|
logger.info(
|
|
f"[{channel.value.upper()}] {notification_type.value} -> "
|
|
f"{str(history_key)[:20]} [{status}]"
|
|
)
|
|
|
|
return {
|
|
"notification_id": notification_id,
|
|
"channel": channel.value,
|
|
"status": status,
|
|
"message": message,
|
|
}
|
|
|
|
def _format_notification(self, notification: Notification) -> Dict[str, Any]:
|
|
return {
|
|
"id": notification.notification_id,
|
|
"type": notification.notification_type.value,
|
|
"channel": notification.channel.value,
|
|
"message": notification.message,
|
|
"sent_at": notification.sent_at.isoformat(),
|
|
"status": notification.status,
|
|
}
|
|
|
|
async def _unknown_task(self, task: AgentTask) -> Dict[str, Any]:
|
|
return {"status": "error", "message": f"Unknown task: {task.task_type}"}
|
|
|
|
async def think(self, context: str, options: List[str] = None) -> str:
|
|
return f"[CUSTOMER_AGENT reasoning]: {context}"
|