tools.py

"""
tools.py — Tool definitions (passed to Groq/LLM) + tool executor (runs locally).

Groq uses the same OpenAI-compatible tool-calling format.
"""

from datetime import date


# ── Tool schemas for Groq (OpenAI format) ────────────────────────────────────
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_customer",
            "description": "Retrieve a customer record by email address. Returns customer details including tier, notes, and membership history.",
            "parameters": {
                "type": "object",
                "properties": {
                    "email": {
                        "type": "string",
                        "description": "The customer's registered email address."
                    }
                },
                "required": ["email"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_order",
            "description": "Retrieve a specific order by order ID. Returns order status, delivery date, return deadline, refund status, and notes.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "The order ID (e.g. ORD-1001)."
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_orders_by_email",
            "description": "Retrieve all orders for a customer using their email address. Useful when no order ID is provided.",
            "parameters": {
                "type": "object",
                "properties": {
                    "email": {
                        "type": "string",
                        "description": "The customer's registered email address."
                    }
                },
                "required": ["email"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_product",
            "description": "Retrieve product details by product ID. Returns category, return window days, warranty period, and any special return notes.",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_id": {
                        "type": "string",
                        "description": "The product ID (e.g. P001)."
                    }
                },
                "required": ["product_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "check_refund_eligibility",
            "description": "Check whether a refund is eligible for a given order. Evaluates return window, product rules, customer tier, and current refund status.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "The order ID to evaluate."
                    },
                    "reason": {
                        "type": "string",
                        "description": "The reason for the refund request (e.g. 'defective', 'wrong item', 'changed mind', 'damaged on arrival')."
                    }
                },
                "required": ["order_id", "reason"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "cancel_order",
            "description": "Cancel an order. Only works if the order is still in 'processing' status. Returns success or failure with reason.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "The order ID to cancel."
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "Search the ShopWave support knowledge base for policy information. Use for questions about returns, refunds, warranties, exchanges, or cancellations.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The policy question or topic to search for."
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "issue_refund",
            "description": "Issue a refund for an order. IRREVERSIBLE — must call check_refund_eligibility first and confirm eligible=true before calling this. Never call this without verifying eligibility.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "The order ID to refund."
                    },
                    "amount": {
                        "type": "number",
                        "description": "The refund amount in USD."
                    },
                    "reason": {
                        "type": "string",
                        "description": "The reason for the refund."
                    }
                },
                "required": ["order_id", "amount", "reason"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "send_reply",
            "description": "Send a response message to the customer for a given ticket. Use this to formally deliver the final resolution message to the customer.",
            "parameters": {
                "type": "object",
                "properties": {
                    "ticket_id": {
                        "type": "string",
                        "description": "The ticket ID to reply to."
                    },
                    "message": {
                        "type": "string",
                        "description": "The message to send to the customer."
                    }
                },
                "required": ["ticket_id", "message"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "escalate_ticket",
            "description": "Escalate this ticket to a human support agent. Use when: warranty claims, replacement requests (not refunds), refund > $200, fraud suspected, or you are not confident.",
            "parameters": {
                "type": "object",
                "properties": {
                    "ticket_id": {
                        "type": "string",
                        "description": "The ticket ID being escalated."
                    },
                    "reason": {
                        "type": "string",
                        "description": "Clear reason for escalation."
                    },
                    "summary": {
                        "type": "string",
                        "description": "Brief summary of what the agent verified and attempted."
                    },
                    "recommended_action": {
                        "type": "string",
                        "description": "What the agent recommends the human agent should do."
                    },
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high", "urgent"],
                        "description": "Priority level for the escalation."
                    }
                },
                "required": ["ticket_id", "reason", "summary", "recommended_action", "priority"]
            }
        }
    }
]


# ── Tool router ───────────────────────────────────────────────────────────────

def execute_tool(name: str, inputs: dict, data: dict) -> dict:
    """Route a tool call by name to its implementation."""
    handlers = {
        "get_customer"            : _get_customer,
        "get_order"               : _get_order,
        "get_orders_by_email"     : _get_orders_by_email,
        "get_product"             : _get_product,
        "check_refund_eligibility": _check_refund_eligibility,
        "cancel_order"            : _cancel_order,
        "issue_refund"            : _issue_refund,
        "send_reply"              : _send_reply,
        "search_knowledge_base"   : _search_knowledge_base,
        "escalate_ticket"         : _escalate_ticket,
    }
    handler = handlers.get(name)
    if not handler:
        return {"error": f"Unknown tool: {name}"}
    return handler(inputs, data)


# ── Tool implementations ──────────────────────────────────────────────────────

def _get_customer(inputs: dict, data: dict) -> dict:
    email = inputs["email"].strip().lower()
    for customer in data["customers"]:
        if customer["email"].lower() == email:
            return {"found": True, "customer": customer}
    return {"found": False, "error": f"No customer found with email: {email}"}


def _get_order(inputs: dict, data: dict) -> dict:
    order_id = inputs["order_id"].strip().upper()
    for order in data["orders"]:
        if order["order_id"].upper() == order_id:
            product  = next((p for p in data["products"]  if p["product_id"]  == order["product_id"]),  None)
            customer = next((c for c in data["customers"] if c["customer_id"] == order["customer_id"]), None)
            return {"found": True, "order": order, "product": product, "customer": customer}
    return {"found": False, "error": f"No order found with ID: {order_id}"}


def _get_orders_by_email(inputs: dict, data: dict) -> dict:
    email    = inputs["email"].strip().lower()
    customer = next((c for c in data["customers"] if c["email"].lower() == email), None)
    if not customer:
        return {"found": False, "error": f"No customer found with email: {email}"}
    cid    = customer["customer_id"]
    orders = [o for o in data["orders"] if o["customer_id"] == cid]
    enriched = []
    for o in orders:
        product = next((p for p in data["products"] if p["product_id"] == o["product_id"]), None)
        enriched.append({**o, "product_name": product["name"] if product else "Unknown"})
    return {"found": True, "customer": customer, "orders": enriched}


def _get_product(inputs: dict, data: dict) -> dict:
    pid = inputs["product_id"].strip().upper()
    for product in data["products"]:
        if product["product_id"].upper() == pid:
            return {"found": True, "product": product}
    return {"found": False, "error": f"No product found with ID: {pid}"}


def _check_refund_eligibility(inputs: dict, data: dict) -> dict:
    order_id = inputs["order_id"].strip().upper()
    reason   = inputs.get("reason", "").lower()

    order = next((o for o in data["orders"] if o["order_id"].upper() == order_id), None)
    if not order:
        return {"eligible": False, "reason": f"Order {order_id} not found."}

    product  = next((p for p in data["products"]  if p["product_id"]  == order["product_id"]),  None)
    customer = next((c for c in data["customers"] if c["customer_id"] == order["customer_id"]), None)

    # Already refunded?
    if order.get("refund_status") == "refunded":
        return {"eligible": False, "reason": "A refund has already been processed for this order."}

    # Damaged / defective — always eligible regardless of window
    if any(kw in reason for kw in ["damage", "defect", "broken", "cracked", "defective", "manufacturing"]):
        return {
            "eligible"      : True,
            "reason"        : "Item arrived damaged or defective — eligible for full refund regardless of return window.",
            "refund_amount" : order["amount"],
            "requires_photo": True,
        }

    # Wrong item delivered
    if any(kw in reason for kw in ["wrong item", "wrong colour", "wrong color", "wrong size", "incorrect"]):
        return {
            "eligible"    : True,
            "reason"      : "Wrong item delivered — eligible for refund or exchange.",
            "refund_amount": order["amount"],
        }

    # Cannot refund undelivered orders
    if order["status"] in ("processing", "shipped"):
        return {"eligible": False, "reason": f"Order is '{order['status']}'. Cannot process refund before delivery."}

    # Return window check
    today = date.today()
    if order.get("return_deadline"):
        deadline  = date.fromisoformat(order["return_deadline"])
        days_past = (today - deadline).days

        if days_past > 0:
            tier = customer["tier"].lower() if customer else "standard"

            # VIP with pre-approved exception
            if tier == "vip" and "extended return" in (customer.get("notes") or "").lower():
                return {
                    "eligible"    : True,
                    "reason"      : "Return window expired but VIP exception pre-approved by management.",
                    "refund_amount": order["amount"],
                }

            # Premium borderline (1–3 days late)
            if tier == "premium" and days_past <= 3:
                return {
                    "eligible"               : True,
                    "reason"                 : f"Return window expired by {days_past} day(s). Premium member — agent discretion approved.",
                    "refund_amount"          : order["amount"],
                    "requires_supervisor_note": True,
                }

            return {
                "eligible"     : False,
                "reason"       : f"Return window expired {days_past} day(s) ago. Deadline was {order['return_deadline']}.",
                "customer_tier": tier,
            }

    # Product registered online — non-returnable
    if product:
        if "registered online" in (product.get("notes") or "").lower() and \
           "registered online" in (order.get("notes") or "").lower():
            return {"eligible": False, "reason": "Product was registered online — non-returnable per policy."}
        if not product.get("returnable", True):
            return {"eligible": False, "reason": f"{product['name']} is marked as non-returnable."}

    return {
        "eligible"    : True,
        "reason"      : "Order is within the return window and meets all refund criteria.",
        "refund_amount": order["amount"],
    }


def _cancel_order(inputs: dict, data: dict) -> dict:
    order_id = inputs["order_id"].strip().upper()
    order    = next((o for o in data["orders"] if o["order_id"].upper() == order_id), None)
    if not order:
        return {"success": False, "reason": f"Order {order_id} not found."}
    if order["status"] == "processing":
        return {
            "success" : True,
            "order_id": order_id,
            "message" : f"Order {order_id} has been successfully cancelled. A confirmation email will be sent within 1 hour.",
        }
    return {
        "success": False,
        "reason" : f"Order {order_id} is in '{order['status']}' status and cannot be cancelled. "
                   "The customer must wait for delivery and then initiate a return if needed.",
    }


def _search_knowledge_base(inputs: dict, data: dict) -> dict:
    """Keyword search on the knowledge base markdown."""
    query   = inputs["query"].lower()
    kb_text = data["knowledge_base"]

    keyword_map = {
        "return"  : ["return policy", "return window"],
        "refund"  : ["refund policy", "refund eligibility"],
        "warrant" : ["warranty policy", "warranty period"],
        "cancel"  : ["order cancellation"],
        "exchange": ["exchange policy"],
        "tier"    : ["customer tiers"],
        "escalat" : ["escalation guidelines"],
    }

    relevant_kws = [kw for kw in keyword_map if kw in query] or [query]
    lines        = kb_text.split("\n")
    matched      = []

    for i, line in enumerate(lines):
        if any(kw in line.lower() for kw in relevant_kws):
            chunk = "\n".join(lines[max(0, i-1): min(len(lines), i+6)])
            matched.append(chunk)

    return {
        "found"  : True,
        "results": matched[:3] if matched else [kb_text[:1500]],
    }


def _issue_refund(inputs: dict, data: dict) -> dict:
    """Simulate issuing a refund. IRREVERSIBLE — agent must check eligibility first."""
    import random, time
    order_id = inputs["order_id"].strip().upper()
    amount   = inputs.get("amount", 0)
    reason   = inputs.get("reason", "")

    order = next((o for o in data["orders"] if o["order_id"].upper() == order_id), None)
    if not order:
        return {"success": False, "reason": f"Order {order_id} not found. Refund aborted."}

    if order.get("refund_status") == "refunded":
        return {"success": False, "reason": f"Refund already processed for {order_id}. Cannot issue duplicate refund."}

    # Simulate occasional timeout (realistic mock)
    if random.random() < 0.05:  # 5% chance of timeout
        time.sleep(0.1)
        return {"success": False, "reason": "Refund service timeout. Please retry.", "error_code": "TIMEOUT"}

    # Mark as refunded in memory (simulated)
    order["refund_status"] = "refunded"

    print(f"\n  💰 REFUND ISSUED — {order_id} | ${amount:.2f} | Reason: {reason}")
    return {
        "success"          : True,
        "order_id"         : order_id,
        "amount_refunded"  : amount,
        "refund_id"        : f"REF-{order_id}-{date.today().strftime('%Y%m%d')}",
        "message"          : f"Refund of ${amount:.2f} successfully issued for {order_id}. Will appear in 5-7 business days.",
        "processing_time"  : "5–7 business days",
    }


def _send_reply(inputs: dict, data: dict) -> dict:
    """Simulate sending a reply to the customer."""
    ticket_id = inputs.get("ticket_id", "UNKNOWN")
    message   = inputs.get("message", "")

    print(f"\n  📧 REPLY SENT — Ticket: {ticket_id}")
    print(f"     Message: {message[:100]}...")

    return {
        "success"   : True,
        "ticket_id" : ticket_id,
        "message"   : "Reply successfully sent to customer.",
        "channel"   : "email",
        "timestamp" : date.today().isoformat(),
    }


def _escalate_ticket(inputs: dict, data: dict) -> dict:
    """Simulate ticket escalation to a human agent."""
    print(f"\n  🚨 ESCALATION — Priority: {inputs.get('priority','medium').upper()}")
    print(f"     Reason: {inputs.get('reason','')}")
    return {
        "success"           : True,
        "escalation_id"     : f"ESC-{inputs.get('ticket_id','UNKNOWN')}-{date.today().strftime('%Y%m%d')}",
        "assigned_to"       : "specialist-team@shopwave.com",
        "priority"          : inputs.get("priority"),
        "message"           : "Ticket escalated to a human specialist. Customer will be notified shortly.",
        "estimated_response": "2–4 hours for high/urgent; 24 hours for low/medium.",
    }