🚚 Shipment Exception Desk#

Shipment Exception Desk is an operations-focused AI workflow for logistics exception handling. Instead of directly drafting responses from raw complaint text, the system follows a structured triage pipeline: classify issue type, calculate policy compensation, evaluate escalation need, generate the right communication, and record everything in a session ledger.

The design deliberately combines LLM reasoning with deterministic business rules. Classification and narrative drafting are delegated to LangChain model chains, while monetary calculations and escalation thresholds remain exact and auditable in Python.

1. Hybrid architecture: probabilistic AI + deterministic policy

The project uses the model for language-heavy tasks (classification and writing) and uses plain code for exact rule execution. This split keeps behavior stable for compliance-sensitive operations while still benefiting from LLM flexibility.

2. Tier-aware escalation gating

Premium and standard customers use different escalation thresholds. Unknown reports auto-escalate. This makes escalation decisions both explainable and easy to tune:

• Standard: escalate above $100 compensation.
• Premium: escalate above $50 compensation.
• Unknown: escalate immediately for manual review.

3. Session-level operational analytics

Each processed claim is appended to an in-memory ledger. A daily summary endpoint aggregates totals, escalation rate, and costliest category by cumulative payout, turning each run into an auditable operations snapshot.

Triage pipeline for Northwind Logistics Shipment Exception Desk.

"""Triage pipeline for Northwind Logistics Shipment Exception Desk.

Coordinates:
1. LLM classification of incoming exception report
2. Rule-based compensation calculation
3. Tier-based escalation evaluation (lower threshold for premium clients, auto-escalation for unknown)
4. LLM drafting of internal manager briefing (if escalated) or customer email (if resolved)
5. Session logging for daily aggregation
"""

from typing import Dict, Any, List

try:
    from .chains import classify_chain, escalate_chain, draft_email_chain
    from .tools import (
        calculate_delay_compensation,
        calculate_damage_compensation,
        calculate_lost_compensation,
        calculate_unknown_compensation,
    )
    from .session import log_exception
except ImportError:
    from chains import classify_chain, escalate_chain, draft_email_chain
    from tools import (
        calculate_delay_compensation,
        calculate_damage_compensation,
        calculate_lost_compensation,
        calculate_unknown_compensation,
    )
    from session import log_exception

# ① set compensation escalation thresholds for manager review
# Premium customers receive manager intervention at a lower dollar threshold ($50 vs $100)
ESCALATION_THRESHOLDS = {
    "standard": 100.0,
    "premium": 50.0,
}


def evaluate_escalation(
    category: str,
    compensation_amount: float,
    customer_tier: str,
) -> tuple[bool, str]:
    """Evaluate whether an exception requires manager escalation.

    Rules:
    - Unknown or unclassifiable exceptions escalate automatically regardless of value.
    - Escalates if compensation exceeds tier threshold (Standard: $100, Premium: $50).
    """
    # ① normalize the tier and choose its escalation threshold
    tier_normalized = customer_tier.strip().lower()
    threshold = ESCALATION_THRESHOLDS.get(tier_normalized, 100.0)

    # ② escalate unclear reports so a human can review them
    if category == "unknown":
        return True, "Unclassifiable or garbled report requires manual operations review."

    # ③ escalate payouts that exceed the tier approval limit
    if compensation_amount > threshold:
        return (
            True,
            f"Compensation amount (${compensation_amount:.2f}) exceeds {tier_normalized.capitalize()} tier threshold (${threshold:.2f}).",
        )

    # ④ auto-approve payouts that stay inside the tier limit
    return (
        False,
        f"Compensation (${compensation_amount:.2f}) is within {tier_normalized.capitalize()} tier auto-approval limit (${threshold:.2f}).",
    )


def process_exception(
    report_text: str,
    shipment_value: float,
    customer_tier: str = "standard",
    log_to_session: bool = True,
) -> Dict[str, Any]:
    """Process an incoming shipment exception report through the full triage pipeline."""
    # ① prepare the decision trail and normalize inputs for consistent rules
    steps: List[str] = []
    customer_tier = customer_tier.strip().lower()
    shipment_value = float(shipment_value)

    # ② classify report with the classifier chain
    category = classify_chain.invoke({"report_text": report_text})
    steps.append(f"Step 1 [Classify]: Report classified as '{category.upper()}' via LLM.")

    # ③ route to compensation calculator by category
    if category == "delayed":
        comp_result = calculate_delay_compensation(shipment_value)
    elif category == "damaged":
        comp_result = calculate_damage_compensation(shipment_value)
    elif category == "lost":
        comp_result = calculate_lost_compensation(shipment_value)
    else:
        comp_result = calculate_unknown_compensation(shipment_value)

    comp_amount = float(comp_result.get("amount", 0.0))
    comp_reason = comp_result.get("reason", "")
    steps.append(
        f"Step 2 [Compensate]: Calculated compensation ${comp_amount:.2f} ({comp_reason})."
    )

    # ④ make the escalation decision from tier and payout
    escalated, escalation_reason = evaluate_escalation(
        category=category,
        compensation_amount=comp_amount,
        customer_tier=customer_tier,
    )
    action_taken = "Escalated to Manager" if escalated else "Auto-Resolved"
    steps.append(f"Step 3 [Escalation Check]: {action_taken} — {escalation_reason}")

    # ⑤ draft appropriate message for manager or customer
    if escalated:
        draft = escalate_chain.invoke(
            {
                "customer_tier": customer_tier.capitalize(),
                "shipment_value": f"{shipment_value:.2f}",
                "category": category,
                "compensation_amount": f"{comp_amount:.2f}",
                "escalation_reason": escalation_reason,
                "report_text": report_text,
            }
        )
        steps.append("Step 4 [Draft]: Generated internal Manager Escalation Briefing.")
    else:
        draft = draft_email_chain.invoke(
            {
                "customer_tier": customer_tier.capitalize(),
                "shipment_value": f"{shipment_value:.2f}",
                "category": category,
                "compensation_amount": f"{comp_amount:.2f}",
                "compensation_reason": comp_reason,
                "report_text": report_text,
            }
        )
        steps.append("Step 4 [Draft]: Generated customer resolution & apology email.")

    # ⑥ package every pipeline decision into the API/CLI result
    result = {
        "report_text": report_text,
        "shipment_value": shipment_value,
        "customer_tier": customer_tier,
        "category": category,
        "compensation": comp_result,
        "compensation_amount": comp_amount,
        "escalated": escalated,
        "escalation_reason": escalation_reason,
        "action_taken": action_taken,
        "draft": draft,
        "steps": steps,
    }

    # ⑦ log to session for daily aggregation when requested
    if log_to_session:
        log_exception(result)
        steps.append("Step 5 [Session]: Exception logged into daily triage ledger.")

    return result

LangChain chains for Northwind Logistics Shipment Exception Desk.

"""LangChain chains for Northwind Logistics Shipment Exception Desk.

Implements the standard prompt | model | parser chain pattern for:
1. classify_chain: Categorizes exception reports into delayed, damaged, lost, or unknown.
2. escalate_chain: Drafts internal escalation notes for managers.
3. draft_email_chain: Drafts customer resolution emails for auto-resolved claims.
"""

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

try:
    from .llm import llm
except ImportError:
    from llm import llm


# ① classification chain: define the prompt that maps reports to one category
_classify_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are an expert logistics triage classifier at Northwind Logistics.\n"
            "Given an incoming customer exception report, classify it into EXACTLY one category:\n"
            "- delayed: shipment was late, held up, stuck at a hub, or missed delivery deadline.\n"
            "- damaged: package or contents arrived broken, crushed, leaking, cracked, or ruined.\n"
            "- lost: package marked delivered but missing, lost in transit, or untraceable by carrier.\n"
            "- unknown: report is garbled, unreadable, meaningless gibberish, or cannot be reliably determined.\n\n"
            "Respond with ONLY one lowercase word: delayed, damaged, lost, or unknown.",
        ),
        (
            "human",
            "Exception Report:\n{report_text}\n\nCategory:",
        ),
    ]
)


def _clean_category(raw_output: str) -> str:
    """Normalize and validate the classification output."""
    # ① normalize raw model text before comparing it to known categories
    cleaned = raw_output.strip().lower().replace('"', "").replace("'", "")
    # ② check for known categories in the output
    for cat in ["delayed", "damaged", "lost", "unknown"]:
        if cat in cleaned:
            return cat
    # ③ fall back to unknown when the model output is not trustworthy
    return "unknown"


# ② connect prompt, model, parser, and cleanup into one classifier
classify_chain = (
    _classify_prompt
    | llm
    | StrOutputParser()
    | RunnableLambda(_clean_category)
)


# ① escalation chain: define the prompt for internal manager briefings
_escalate_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a logistics operations analyst at Northwind Logistics.\n"
            "Draft a concise, professional internal escalation briefing to an Operations Manager.\n"
            "The briefing must include:\n"
            "- Subject line: [ESCALATION REQUIRED] followed by Category and Tier\n"
            "- Incident Summary\n"
            "- Customer Tier & Value at Risk\n"
            "- Compensation Calculated\n"
            "- Escalation Trigger / Reason\n"
            "- Recommended Operational Action",
        ),
        (
            "human",
            "Customer Tier: {customer_tier}\n"
            "Shipment Value: ${shipment_value}\n"
            "Category: {category}\n"
            "Calculated Compensation: ${compensation_amount}\n"
            "Escalation Reason: {escalation_reason}\n\n"
            "Customer Report:\n{report_text}\n\n"
            "Draft Internal Escalation Briefing:",
        ),
    ]
)

# ② connect the escalation prompt to the model and string parser
escalate_chain = _escalate_prompt | llm | StrOutputParser()


# ① draft email chain: define the customer prompt for auto-resolved claims
_draft_email_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a Senior Customer Care Specialist at Northwind Logistics.\n"
            "Draft an empathetic, professional resolution email to the customer regarding their shipment exception.\n"
            "Guidelines:\n"
            "- Acknowledge the issue ({category}) and apologize sincerely for the inconvenience.\n"
            "- Clearly state the approved compensation of ${compensation_amount} under our Northwind Guarantee.\n"
            "- Mention that credit will reflect within 2-3 business days.\n"
            "- Maintain an empathetic, helpful tone.\n"
            "- Sign off as: 'Northwind Logistics Customer Care Team'.",
        ),
        (
            "human",
            "Customer Tier: {customer_tier}\n"
            "Shipment Value: ${shipment_value}\n"
            "Category: {category}\n"
            "Approved Compensation: ${compensation_amount}\n"
            "Compensation Reason: {compensation_reason}\n\n"
            "Customer Report:\n{report_text}\n\n"
            "Draft Customer Resolution Email:",
        ),
    ]
)

# ② connect the email prompt to the model and string parser
draft_email_chain = _draft_email_prompt | llm | StrOutputParser()

import os
from pathlib import Path
from dotenv import load_dotenv, find_dotenv
from langchain_openai import ChatOpenAI

# ① load .env file from project root or current working directory
dotenv_path = find_dotenv(usecwd=True)
if not dotenv_path:
    # ② try looking in parent directory if executing from within src
    parent_env = Path(__file__).resolve().parent.parent / ".env"
    if parent_env.exists():
        dotenv_path = str(parent_env)

# ③ load environment variables before creating the model client
load_dotenv(dotenv_path)

# ④ initialize the ChatOpenAI instance used by all chains
llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
    temperature=0.0,
    api_key=os.getenv("OPENAI_API_KEY"),
)

Compensation calculation tools for Northwind Logistics Shipment Exception Desk.

"""Compensation calculation tools for Northwind Logistics Shipment Exception Desk.

Each function calculates compensation based on policy rules and returns
a structured dictionary containing:
  - category: The exception category (delayed, damaged, lost, unknown)
  - amount: Calculated compensation dollar amount (float)
  - currency: "USD"
  - reason: Explanation of how the compensation was determined
"""

from typing import Dict, Any


def calculate_delay_compensation(
    shipment_value: float, days_delayed: int = 1
) -> Dict[str, Any]:
    """Calculate compensation for delayed shipments.

    Policy:
    - 20% of shipment value
    - Minimum courtesy credit of $15.00
    - Capped at 100% of shipment value
    """
    # ① reject invalid negative values before applying policy
    if shipment_value < 0:
        raise ValueError("Shipment value cannot be negative.")

    # ② return no payout when the shipment has no declared value
    if shipment_value == 0:
        return {
            "category": "delayed",
            "amount": 0.0,
            "currency": "USD",
            "reason": "Shipment value is $0.00; no compensation issued.",
        }

    base_compensation = shipment_value * 0.20
    # ③ apply minimum courtesy credit, but never exceed shipment value
    compensation = max(base_compensation, 15.0)
    compensation = min(compensation, shipment_value)

    # ④ return rounded delay payout details for the pipeline
    return {
        "category": "delayed",
        "amount": round(compensation, 2),
        "currency": "USD",
        "reason": (
            f"Delay compensation: 20% of ${shipment_value:.2f} "
            f"(with $15.00 minimum credit, capped at shipment value)"
        ),
    }


def calculate_damage_compensation(
    shipment_value: float, damage_severity: str = "partial"
) -> Dict[str, Any]:
    """Calculate compensation for damaged shipments.

    Policy:
    - Partial damage: 50% of shipment value
    - Total/severe damage: 100% of shipment value
    """
    # ① reject invalid negative values before applying damage policy
    if shipment_value < 0:
        raise ValueError("Shipment value cannot be negative.")

    # ② normalize severity and pick full or partial reimbursement rate
    severity = damage_severity.strip().lower()
    rate = 1.0 if severity in ("total", "severe", "complete") else 0.50
    compensation = shipment_value * rate

    # ③ return rounded damage payout details for the pipeline
    return {
        "category": "damaged",
        "amount": round(compensation, 2),
        "currency": "USD",
        "reason": (
            f"Damage compensation: {int(rate * 100)}% of shipment value "
            f"(${shipment_value:.2f}) for {severity} damage"
        ),
    }


def calculate_lost_compensation(shipment_value: float) -> Dict[str, Any]:
    """Calculate compensation for lost in transit shipments.

    Policy:
    - 100% full replacement value of the shipment
    """
    # ① reject invalid negative values before replacement payout
    if shipment_value < 0:
        raise ValueError("Shipment value cannot be negative.")

    # ② return full replacement value for a lost shipment
    return {
        "category": "lost",
        "amount": round(shipment_value, 2),
        "currency": "USD",
        "reason": f"Lost shipment compensation: 100% full replacement value of ${shipment_value:.2f}",
    }


def calculate_unknown_compensation(shipment_value: float) -> Dict[str, Any]:
    """Fallback compensation calculator for unclassified or garbled reports."""
    # ① return zero automatic payout so manual review handles unclear reports
    return {
        "category": "unknown",
        "amount": 0.0,
        "currency": "USD",
        "reason": "Unclassified exception: no automatic compensation calculated; routed for manual review",
    }

Session tracking and daily aggregation for Northwind Logistics Shipment Exception Desk.

"""Session tracking and daily aggregation for Northwind Logistics Shipment Exception Desk.

Aggregates session records to calculate:
- Total compensation paid across all processed exceptions
- Overall escalation rate
- Named costliest category by cumulative compensation paid
"""

from datetime import datetime
from typing import List, Dict, Any, Optional

# In-memory session ledger
_SESSION_RECORDS: List[Dict[str, Any]] = []


def log_exception(record: Dict[str, Any]) -> Dict[str, Any]:
    """Append an exception triage result to the daily session ledger."""
    # ① copy the result so logging does not mutate the caller's object
    entry = dict(record)
    # ② add a timestamp when the pipeline did not provide one
    if "timestamp" not in entry:
        entry["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    # ③ store the entry in the in-memory ledger and return it
    _SESSION_RECORDS.append(entry)
    return entry


def get_triage_log() -> List[Dict[str, Any]]:
    """Return all exception records logged during the current session."""
    return list(_SESSION_RECORDS)


def clear_session() -> None:
    """Clear the session ledger."""
    global _SESSION_RECORDS
    _SESSION_RECORDS.clear()


def generate_daily_summary() -> Dict[str, Any]:
    """Perform real aggregation on the session triage log.

    Calculates:
    - total_exceptions
    - total_compensation
    - escalation_rate
    - costliest_category (by total dollars, where multiple small payouts can beat a single large one)
    - detailed category breakdown
    - formatted markdown summary
    """
    # ① count logged exceptions before aggregating metrics
    total_exceptions = len(_SESSION_RECORDS)

    # ② return an empty summary when no claims have been processed
    if total_exceptions == 0:
        return {
            "total_exceptions": 0,
            "total_compensation": 0.0,
            "escalated_count": 0,
            "resolved_count": 0,
            "escalation_rate": 0.0,
            "costliest_category": "None",
            "costliest_category_amount": 0.0,
            "category_breakdown": {},
            "markdown_summary": (
                "### Daily Triage Summary\n"
                "_No exception reports processed in this session yet._"
            ),
        }

    # ③ total compensation and escalation outcomes across the session
    total_compensation = sum(
        float(r.get("compensation_amount", 0.0)) for r in _SESSION_RECORDS
    )
    escalated_count = sum(1 for r in _SESSION_RECORDS if r.get("escalated"))
    resolved_count = total_exceptions - escalated_count
    escalation_rate = (escalated_count / total_exceptions) * 100.0

    # ④ aggregate by category for claim counts, payouts, and escalations
    categories = ["delayed", "damaged", "lost", "unknown"]
    category_breakdown: Dict[str, Dict[str, Any]] = {}

    for cat in categories:
        cat_records = [r for r in _SESSION_RECORDS if r.get("category") == cat]
        count = len(cat_records)
        cat_total = sum(float(r.get("compensation_amount", 0.0)) for r in cat_records)
        escalated_cat = sum(1 for r in cat_records if r.get("escalated"))
        category_breakdown[cat] = {
            "count": count,
            "total_compensation": round(cat_total, 2),
            "escalated": escalated_cat,
        }

    # ⑤ determine costliest category by total compensation paid
    # Find category with highest cumulative payout; fallback to "None" if total is 0
    payout_per_category = {
        cat: category_breakdown[cat]["total_compensation"] for cat in categories
    }

    max_payout = max(payout_per_category.values())
    if max_payout > 0:
        costliest_category = max(
            payout_per_category, key=lambda k: payout_per_category[k]
        )
        costliest_amount = payout_per_category[costliest_category]
    else:
        # If no compensation paid anywhere, pick category with most claims or "None"
        costliest_category = "None"
        costliest_amount = 0.0

    # ⑥ build formatted Markdown summary for the dashboard
    summary_lines = [
        "### 📊 Northwind Logistics — Daily Triage Summary",
        f"- **Total Exceptions Processed**: {total_exceptions}",
        f"- **Total Compensation Paid**: ${total_compensation:,.2f}",
        f"- **Escalation Rate**: {escalation_rate:.1f}% ({escalated_count} escalated / {resolved_count} auto-resolved)",
        f"- **Costliest Category**: **{costliest_category.upper()}** (${costliest_amount:,.2f} total paid)",
        "",
        "#### Category Breakdown",
        "| Category | Total Claims | Total Compensation | Escalated | Payout Share |",
        "| :--- | :--- | :--- | :--- | :--- |",
    ]

    # ⑦ append each category row with its share of total payouts
    for cat in sorted(
        categories,
        key=lambda c: category_breakdown[c]["total_compensation"],
        reverse=True,
    ):
        data = category_breakdown[cat]
        share = (
            (data["total_compensation"] / total_compensation * 100.0)
            if total_compensation > 0
            else 0.0
        )
        summary_lines.append(
            f"| {cat.capitalize()} | {data['count']} | ${data['total_compensation']:,.2f} | {data['escalated']} | {share:.1f}% |"
        )

    markdown_summary = "\n".join(summary_lines)

    # ⑧ return machine-readable metrics plus the formatted markdown summary
    return {
        "total_exceptions": total_exceptions,
        "total_compensation": round(total_compensation, 2),
        "escalated_count": escalated_count,
        "resolved_count": resolved_count,
        "escalation_rate": round(escalation_rate, 2),
        "costliest_category": costliest_category,
        "costliest_category_amount": round(costliest_amount, 2),
        "category_breakdown": category_breakdown,
        "markdown_summary": markdown_summary,
    }

Flask web server for Shipment Exception Desk.

"""Flask web server for Shipment Exception Desk.

Serves the dashboard UI and provides API endpoints for:
- /api/triage: process exception reports through the pipeline
- /api/log: fetch the current session triage ledger
- /api/summary: fetch aggregated KPI summary
- /api/reset: clear current in-memory session ledger
"""

import os
from pathlib import Path

from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS

from pipeline import process_exception
from rate_limiter import check_rate_limit
from session import clear_session, generate_daily_summary, get_triage_log

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")

STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)
bp = Blueprint("main", __name__)


@bp.before_request
def enforce_rate_limit():
    """Enforce strict 10 requests per hour limit on all POST endpoints."""
    # ① apply rate limits only to state-changing post requests
    if request.method == "POST":
        # ② ask the limiter whether this client has exceeded the hourly quota
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=10, window_seconds=3600
        )
        # ③ return http 429 with retry timing when the request is blocked
        if blocked:
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp



@bp.route("/")
def index():
    """Serve index.html with API base injected for local/prod parity."""
    # ① read the static dashboard page from the configured static folder
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    # ② inject the deployment path prefix so browser api calls use the right base
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ③ return the modified page as html
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    """Serve stylesheets from src/css."""
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    """Serve scripts from src/js."""
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/api/health", methods=["GET"])
def health_check():
    """Health check endpoint."""
    return jsonify({"status": "ok", "service": "shipment-exception-desk"})


@bp.route("/api/triage", methods=["POST"])
def triage_report():
    """Process an incoming shipment exception report."""
    # ① parse and normalize request fields from the json body
    data = request.get_json(force=True) or {}
    report_text = (data.get("report_text") or "").strip()
    shipment_value = data.get("shipment_value")
    customer_tier = (data.get("customer_tier") or "standard").strip().lower()

    # ② reject missing report text before invoking the ai workflow
    if not report_text:
        return jsonify({"detail": "Report text cannot be empty."}), 400

    # ③ convert shipment value to a number or return validation error
    try:
        shipment_value = float(shipment_value)
    except (TypeError, ValueError):
        return jsonify({"detail": "Shipment value must be a valid number."}), 400

    # ④ reject negative values because compensation cannot be negative
    if shipment_value < 0:
        return jsonify({"detail": "Shipment value cannot be negative."}), 400

    # ⑤ accept only supported service tiers for policy lookup
    if customer_tier not in {"standard", "premium"}:
        return jsonify({"detail": "Customer tier must be standard or premium."}), 400

    # ⑥ run the triage pipeline and return its structured result
    try:
        result = process_exception(
            report_text=report_text,
            shipment_value=shipment_value,
            customer_tier=customer_tier,
            log_to_session=True,
        )
        return jsonify(result)
    # ⑦ return pipeline errors as client-readable json
    except Exception as exc:
        return jsonify({"detail": str(exc)}), 500


@bp.route("/api/log", methods=["GET"])
def fetch_log():
    """Return all triage records logged in the current session."""
    return jsonify(get_triage_log())


@bp.route("/api/summary", methods=["GET"])
def fetch_summary():
    """Return aggregated summary metrics and category breakdown."""
    return jsonify(generate_daily_summary())


@bp.route("/api/reset", methods=["POST"])
def reset_session():
    """Clear session records."""
    # ① clear the in-memory ledger for a fresh demo session
    clear_session()
    # ② confirm the reset so the dashboard can refresh state
    return jsonify({"status": "session_cleared"})


app.register_blueprint(bp, url_prefix=PATH_PREFIX)


if __name__ == "__main__":
    # ① start flask when this file is executed directly
    app.run(host="0.0.0.0", port=5000)

Command-line interface runner for Northwind Logistics Shipment Exception Desk.

"""Command-line interface runner for Northwind Logistics Shipment Exception Desk.

Allows quick terminal testing of single exception reports or running demo presets.
"""

import sys
import argparse
from pathlib import Path

src_dir = Path(__file__).resolve().parent
if str(src_dir) not in sys.path:
    sys.path.insert(0, str(src_dir))

from pipeline import process_exception
from session import generate_daily_summary, get_triage_log


def print_result(res: dict) -> None:
    # ① print the high-level report and policy inputs
    print("\n" + "=" * 65)
    print("NORTHWIND LOGISTICS — EXCEPTION TRIAGE RESULT")
    print("=" * 65)
    print(f"Report Snippet     : {res['report_text'][:80]}...")
    print(f"Shipment Value     : ${res['shipment_value']:.2f}")
    print(f"Customer Tier      : {res['customer_tier'].capitalize()}")
    print("-" * 65)
    # ② print classification, compensation, and escalation details
    print(f"Classified Category: {res['category'].upper()}")
    print(f"Compensation Amount: ${res['compensation_amount']:.2f}")
    print(f"Compensation Reason: {res['compensation'].get('reason', 'N/A')}")
    print(f"Escalation Status  : {'[ESCALATED]' if res['escalated'] else '[AUTO-RESOLVED]'}")
    print(f"Escalation Details : {res['escalation_reason']}")
    print("-" * 65)
    # ③ replay the pipeline decision trail for the operator
    print("Decision Trail:")
    for step in res["steps"]:
        print(f"  • {step}")
    print("-" * 65)
    # ④ label the draft based on whether this case escalated
    action_label = (
        "INTERNAL MANAGER ESCALATION BRIEFING"
        if res["escalated"]
        else "CUSTOMER RESOLUTION EMAIL DRAFT"
    )
    # ⑤ print the generated draft for review
    print(f"Generated Draft ({action_label}):\n")
    print(res["draft"])
    print("=" * 65 + "\n")


def main() -> None:
    # ① define command-line options for report, value, tier, and demo mode
    parser = argparse.ArgumentParser(
        description="Northwind Logistics Shipment Exception Desk CLI"
    )
    parser.add_argument(
        "--report",
        type=str,
        help="Exception report text submitted by customer",
    )
    parser.add_argument(
        "--value",
        type=float,
        default=100.0,
        help="Shipment monetary value in USD (default: 100.0)",
    )
    parser.add_argument(
        "--tier",
        type=str,
        choices=["standard", "premium"],
        default="standard",
        help="Customer account tier (standard or premium, default: standard)",
    )
    parser.add_argument(
        "--demo",
        action="store_true",
        help="Run a standard delay demo scenario",
    )

    # ② parse user arguments before choosing an input source
    args = parser.parse_args()

    # ③ use the supplied report when the caller provides one
    if args.report:
        report_text = args.report
        value = args.value
        tier = args.tier
    # ④ run the default demo when no report is supplied
    elif args.demo or len(sys.argv) == 1:
        print("Running demo exception report...")
        report_text = (
            "Hi, my package was supposed to arrive 3 days ago. "
            "Tracking has been stuck on 'In Transit - Weather Delay' with no updates."
        )
        value = 120.0
        tier = "standard"
    else:
        # ⑤ otherwise collect the report details interactively
        report_text = input("Enter shipment exception report: ").strip()
        val_input = input("Enter shipment value in USD (default 100): ").strip()
        value = float(val_input) if val_input else 100.0
        tier_input = input("Enter customer tier (standard/premium, default standard): ").strip()
        tier = tier_input.lower() if tier_input in ["standard", "premium"] else "standard"

    # ⑥ run the triage pipeline and log the case in the session
    res = process_exception(
        report_text=report_text,
        shipment_value=value,
        customer_tier=tier,
        log_to_session=True,
    )
    # ⑦ print the result in a learner-friendly cli format
    print_result(res)


if __name__ == "__main__":
    # ① run the cli entry point when invoked as a script
    main()

Test harness for Northwind Logistics Shipment Exception Desk.

"""Test harness for Northwind Logistics Shipment Exception Desk.

Exercises the 4 required canned scenarios:
1. Mild delay -> category: delayed, escalated: False (Auto-Resolved)
2. High-value loss -> category: lost, escalated: True (Escalated to Manager)
3. Minor damage claim -> category: damaged, escalated: False (Auto-Resolved)
4. Garbled unclassifiable report -> category: unknown, escalated: True (Auto-Escalated)

Also validates session aggregation and costliest-category calculation.
"""

import sys
from pathlib import Path

# ① ensure src directory is in python path for local imports
src_dir = Path(__file__).resolve().parent
if str(src_dir) not in sys.path:
    sys.path.insert(0, str(src_dir))

from pipeline import process_exception
from session import clear_session, generate_daily_summary, get_triage_log

# Canned test scenarios
SCENARIOS = [
    {
        "name": "Scenario 1: Mild delay",
        "report_text": (
            "My shipment was scheduled for delivery yesterday afternoon. "
            "Tracking now indicates a weather delay at the regional depot "
            "and delivery is rescheduled for tomorrow."
        ),
        "shipment_value": 50.00,
        "customer_tier": "standard",
        "expected_category": "delayed",
        "expected_escalated": False,
    },
    {
        "name": "Scenario 2: High-value loss",
        "report_text": (
            "Our pallet of high-end consumer electronics was marked delivered, "
            "but our warehouse never received it. The carrier has officially confirmed "
            "the cargo was lost in transit."
        ),
        "shipment_value": 500.00,
        "customer_tier": "standard",
        "expected_category": "lost",
        "expected_escalated": True,
    },
    {
        "name": "Scenario 3: Minor damage claim",
        "report_text": (
            "The parcel arrived on time, but the exterior box was crushed "
            "and one of the glass items inside is cracked."
        ),
        "shipment_value": 60.00,
        "customer_tier": "standard",
        "expected_category": "damaged",
        "expected_escalated": False,
    },
    {
        "name": "Scenario 4: Garbled unclassifiable report",
        "report_text": "asdf1234 !!@@##$$ order ?? xx zz 998234 lkjasdf",
        "shipment_value": 10.00,
        "customer_tier": "standard",
        "expected_category": "unknown",
        "expected_escalated": True,
    },
]


def run_checks() -> bool:
    # ① print the verification banner
    print("=" * 70)
    print("NORTHWIND LOGISTICS — TRIAGE PIPELINE VERIFICATION")
    print("=" * 70)

    # ② reset session state before running canned scenarios
    clear_session()
    all_passed = True

    # ③ run each canned scenario through the full pipeline
    for i, s in enumerate(SCENARIOS, 1):
        print(f"\n[{i}/4] Testing: {s['name']}")
        print(f"    Value: ${s['shipment_value']:.2f} | Tier: {s['customer_tier']}")
        print(f"    Report: \"{s['report_text'][:60]}...\"")

        res = process_exception(
            report_text=s["report_text"],
            shipment_value=s["shipment_value"],
            customer_tier=s["customer_tier"],
            log_to_session=True,
        )

        # ④ compare category and escalation results with expectations
        cat_match = res["category"] == s["expected_category"]
        esc_match = res["escalated"] == s["expected_escalated"]

        print(f"    Result Category  : {res['category']} (Expected: {s['expected_category']}) -> {'✓' if cat_match else '✗'}")
        print(f"    Result Escalated : {res['escalated']} (Expected: {s['expected_escalated']}) -> {'✓' if esc_match else '✗'}")
        print(f"    Compensation     : ${res['compensation_amount']:.2f}")
        print(f"    Action Taken     : {res['action_taken']}")
        print(f"    Draft Preview    :\n      {res['draft'].strip().splitlines()[0]}")

        # ⑤ mark the run failed if a scenario misses either expectation
        if not (cat_match and esc_match):
            all_passed = False
            print("    >>> FAILED SCENARIO <<<")

    print("\n" + "=" * 70)
    print("VERIFYING DAILY SESSION AGGREGATION")
    print("=" * 70)

    # ⑥ load aggregate metrics from the session ledger
    summary = generate_daily_summary()
    total_exceptions = summary["total_exceptions"]
    total_comp = summary["total_compensation"]
    escalation_rate = summary["escalation_rate"]
    costliest_category = summary["costliest_category"]

    print(f"Total Exceptions Processed : {total_exceptions} (Expected: 4)")
    print(f"Total Compensation Paid    : ${total_comp:,.2f}")
    print(f"Escalation Rate            : {escalation_rate:.1f}% (Expected: 50.0%)")
    print(f"Costliest Category         : {costliest_category} (Expected: lost)")

    # ⑦ compare session summary values with expected outcomes
    if total_exceptions != 4:
        all_passed = False
        print("✗ Session total exceptions mismatch!")
    if escalation_rate != 50.0:
        all_passed = False
        print("✗ Escalation rate mismatch!")
    if costliest_category != "lost":
        all_passed = False
        print("✗ Costliest category mismatch!")

    print("\n" + "=" * 70)
    # ⑧ print final pass/fail result and return it
    if all_passed:
        print("🎉 ALL 4 SCENARIOS & SESSION AGGREGATIONS PASSED SUCCESSFULLY!")
        print("=" * 70)
        return True
    else:
        print("❌ SOME CHECKS FAILED. Please review the errors above.")
        print("=" * 70)
        return False


if __name__ == "__main__":
    # ① run checks and map the boolean result to a process exit code
    success = run_checks()
    sys.exit(0 if success else 1)