Capstone 4 — Domain B: B2B Order Lifecycle Pipeline

mkdir order-lifecycle-pipeline && cd order-lifecycle-pipeline
python3 -m venv venv && source venv/bin/activate
pip install "anthropic>=0.30.0" pytest
export ANTHROPIC_API_KEY=your-key-here

order-lifecycle-pipeline/
├── mock_tools.py           # All mock tool implementations (ERP, inventory, carrier, messaging)
├── pipeline.py             # All 4 agents + state + circuit breaker + orchestrator
└── test_pipeline.py        # Test suite (5 scenarios, pytest)

{
  "po_number": "PO-2024-11234",
  "stage": "intake",
  "created_at": "2024-03-10T08:00:00Z",

  "intake_output": {
    "validated": true,
    "customer_verified": true,
    "credit_check_passed": true,
    "duplicate_detected": false,
    "line_items_enriched": [
      {"sku": "SKU-4892", "qty": 50, "unit_price": 495.00,
       "contract_price_verified": true, "available": true},
      {"sku": "SKU-7721", "qty": 200, "unit_price": 125.00,
       "contract_price_verified": true, "available": false}
    ],
    "total_value": 49750.00,
    "priority": "standard"
  },

  "fulfillment_output": {
    "plan_id": "FP-2024-11234",
    "requires_split": true,
    "shipments": [
      {"warehouse": "East", "skus": ["SKU-4892"], "value": 24750.00,
       "estimated_ship_date": "2024-03-12", "carrier": "UPS"},
      {"warehouse": "West", "skus": ["SKU-7721"], "value": 25000.00,
       "estimated_ship_date": "2024-03-15", "carrier": "FedEx"}
    ],
    "total_value": 49750.00,
    "human_approval_required": false
  },

  "exception_output": null,
  "communication_output": null,

  "circuit_breaker": {
    "carrier_api_failures": 0,
    "window_start": null,
    "threshold": 5,
    "window_minutes": 10,
    "status": "healthy"
  }
}

"""mock_tools.py — Tools for all 4 pipeline agents."""

from datetime import datetime, timedelta

# ═══ INTAKE AGENT ══════════════════════════════════════════════

CUSTOMER_CREDIT = {
    "Acme Manufacturing": {"limit": 100000, "outstanding": 35000, "terms": "Net 45"},
    "Beta Industrial": {"limit": 50000, "outstanding": 48000, "terms": "Net 30"},
}
PROCESSED_POS = set()

def validate_purchase_order(raw_po: dict) -> dict:
    required = ["po_number", "customer", "line_items"]
    missing = [f for f in required if f not in raw_po]
    if missing:
        return {"validated": False, "errors": missing}
    if raw_po["po_number"] in PROCESSED_POS:
        return {"validated": False, "errors": ["DUPLICATE_PO"]}
    PROCESSED_POS.add(raw_po["po_number"])
    return {"validated": True, "errors": [], "normalized_po": raw_po}

def verify_customer_credit(customer_id: str, order_value: float) -> dict:
    cust = CUSTOMER_CREDIT.get(customer_id)
    if not cust:
        return {"error": "CUSTOMER_NOT_FOUND"}
    available = cust["limit"] - cust["outstanding"]
    return {"credit_approved": available >= order_value,
            "credit_limit": cust["limit"], "available_credit": available,
            "payment_terms": cust["terms"]}

def enrich_line_items(line_items: list, customer_id: str) -> list:
    """Add contract pricing and availability to each line item."""
    enriched = []
    for item in line_items:
        enriched.append({**item,
            "contract_price_verified": True,
            "current_availability": item["sku"] != "SKU-7721",
            "lead_time": 14 if item["sku"] == "SKU-4892" else 7})
    return enriched

# ═══ FULFILLMENT AGENT ═════════════════════════════════════════

WAREHOUSE_INVENTORY = {
    ("SKU-4892", "East"): {"available": 120, "reserved": 0},
    ("SKU-4892", "West"): {"available": 30, "reserved": 0},
    ("SKU-7721", "East"): {"available": 0, "reserved": 0},
    ("SKU-7721", "West"): {"available": 250, "reserved": 0},
}

def check_inventory_allocation(sku: str, quantity: int, warehouse: str = None) -> dict:
    if warehouse:
        inv = WAREHOUSE_INVENTORY.get((sku, warehouse))
        if not inv:
            return {"error": "SKU_NOT_FOUND"}
        return {"sku": sku, "warehouse": warehouse,
                "available": inv["available"], "allocated": min(quantity, inv["available"]),
                "backorder_qty": max(0, quantity - inv["available"])}
    # Check all warehouses
    results = []
    for (s, w), inv in WAREHOUSE_INVENTORY.items():
        if s == sku:
            results.append({"warehouse": w, "available": inv["available"]})
    return {"sku": sku, "warehouses": results}

def create_shipment_plan(po_number: str, line_items: list,
                          shipping_priority: str = "standard") -> dict:
    shipments = []
    for item in line_items:
        best_wh = None
        for (s, w), inv in WAREHOUSE_INVENTORY.items():
            if s == item["sku"] and inv["available"] >= item.get("qty", item.get("quantity", 0)):
                best_wh = w
                break
        if best_wh:
            shipments.append({"warehouse": best_wh, "skus": [item["sku"]],
                "value": item.get("qty", item.get("quantity", 0)) * item.get("unit_price", 0),
                "estimated_ship_date": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
                "carrier": "UPS" if best_wh == "East" else "FedEx"})
    requires_split = len(set(s["warehouse"] for s in shipments)) > 1
    total = sum(s["value"] for s in shipments)
    return {"plan_id": f"FP-{po_number}", "shipments": shipments,
            "requires_split": requires_split, "total_value": total,
            "human_approval_required": requires_split and total > 50000}

def request_split_shipment_approval(po_number: str, plan_id: str,
                                     total_value: float, reason: str) -> dict:
    return {"approval_id": f"APR-{po_number}",
            "queue_position": 1, "estimated_response_time": "15 minutes"}

# ═══ EXCEPTION MONITOR ═════════════════════════════════════════

def poll_shipment_status(tracking_numbers: list) -> list:
    statuses = []
    for tn in tracking_numbers:
        statuses.append({"tracking_number": tn, "status": "in-transit",
                         "carrier": "UPS", "eta": "2024-03-12"})
    return statuses

def detect_exceptions(shipment_statuses: list, original_plan: dict) -> list:
    exceptions = []
    for s in shipment_statuses:
        if s.get("status") == "exception":
            exceptions.append({"type": "carrier_delay",
                "severity": "medium", "tracking": s["tracking_number"],
                "recommended_action": "Notify customer with revised ETA"})
    return exceptions

# ═══ COMMUNICATION AGENT ═══════════════════════════════════════

def draft_order_update(po_number: str, update_type: str, details: dict) -> dict:
    templates = {
        "confirmation": f"Your order {po_number} has been confirmed and is being processed.",
        "shipment": f"Your order {po_number} has shipped! Tracking: {details.get('tracking', 'N/A')}",
        "delay": f"We're writing to let you know about a delay on order {po_number}.",
        "delivery": f"Great news! Order {po_number} has been delivered.",
    }
    return {"draft_id": f"DRF-{po_number}-{update_type}",
            "subject": f"Order Update: {po_number}",
            "body": templates.get(update_type, "Order update."),
            "urgency": "high" if update_type == "delay" else "standard"}

def send_customer_update(draft_id: str, channel: str, customer_contact: str) -> dict:
    return {"message_id": f"MSG-{draft_id}",
            "sent_at": datetime.now().isoformat(),
            "delivery_status": "delivered"}

"""pipeline.py — B2B Order Lifecycle Pipeline (Capstone 4-B)

Usage: export ANTHROPIC_API_KEY=... && python pipeline.py

Agents 1 (Intake) and 4 (Communication) use Claude API calls with tool use.
Agents 2 (Fulfillment) and 3 (Exception Monitor) are deterministic — they
don't need LLM reasoning because inventory allocation and status polling
are rule-based operations.
"""

import json, anthropic
from datetime import datetime, timedelta
from dataclasses import dataclass, field, asdict

# ── Time-Windowed Circuit Breaker ──────────────────────────────
@dataclass
class CarrierCircuitBreaker:
    """Trips after 5 failures within a 10-minute window."""
    failures: list = field(default_factory=list)  # timestamps
    threshold: int = 5
    window_minutes: int = 10
    status: str = "healthy"  # healthy | tripped | fallback

    def record_failure(self):
        now = datetime.now()
        self.failures.append(now)
        # Prune old failures outside the window
        cutoff = now - timedelta(minutes=self.window_minutes)
        self.failures = [f for f in self.failures if f > cutoff]
        if len(self.failures) >= self.threshold:
            self.status = "tripped"
            print(f"[CIRCUIT BREAKER] TRIPPED! {len(self.failures)} failures in {self.window_minutes} min.")
            print("[CIRCUIT BREAKER] Switching to fallback polling mode.")

    def record_success(self):
        # Don't clear — let the window expire naturally
        pass

    def is_tripped(self) -> bool:
        return self.status == "tripped"

    def reset(self):
        self.failures.clear()
        self.status = "healthy"
        print("[CIRCUIT BREAKER] Reset to healthy.")


# ── Pipeline State ─────────────────────────────────────────────
@dataclass
class OrderPipelineState:
    po_number: str
    stage: str = "intake"
    raw_po: dict = field(default_factory=dict)
    intake_output: dict = field(default_factory=dict)
    fulfillment_output: dict = field(default_factory=dict)
    exception_output: dict = field(default_factory=dict)
    communication_output: dict = field(default_factory=dict)


# ── HITL for Split Shipments ───────────────────────────────────
def split_shipment_review(state: OrderPipelineState) -> dict:
    plan = state.fulfillment_output
    print("\n" + "=" * 50)
    print("  OPERATIONS MANAGER REVIEW REQUIRED")
    print("=" * 50)
    print(f"  PO: {state.po_number}")
    print(f"  Total Value: ${plan.get('total_value', 0):,.2f}")
    print(f"  Split: {len(plan.get('shipments', []))} shipments")
    for i, s in enumerate(plan.get("shipments", []), 1):
        print(f"    Ship {i}: {s['warehouse']} — SKUs: {s['skus']} — ${s['value']:,.2f}")
    print(f"\n  [1] Approve split  [2] Wait for consolidation  [3] Partial ship")
    choice = input("  Decision (1-3): ").strip()
    options = {"1": "approve_split", "2": "wait_consolidation", "3": "partial_ship"}
    return {"decision": options.get(choice, "approve_split"),
            "reviewer": "ops_manager_01",
            "reviewed_at": datetime.now().isoformat()}


# ── HITL for Credit/Refund Resolutions ────────────────────────
def credit_resolution_review(state: OrderPipelineState, exception: dict) -> dict:
    """Requires ops manager approval when exception resolution involves credit/refund."""
    print("\n" + "=" * 50)
    print("  OPERATIONS MANAGER REVIEW REQUIRED — CREDIT/REFUND")
    print("=" * 50)
    print(f"  PO: {state.po_number}")
    print(f"  Exception Type: {exception.get('type', 'unknown')}")
    print(f"  Severity: {exception.get('severity', 'unknown')}")
    print(f"  Recommended Action: {exception.get('recommended_action', 'N/A')}")
    print(f"\n  [1] Approve credit/refund  [2] Deny  [3] Escalate to finance")
    choice = input("  Decision (1-3): ").strip()
    options = {"1": "approve", "2": "deny", "3": "escalate_finance"}
    return {"decision": options.get(choice, "approve"),
            "reviewer": "ops_manager_01",
            "reviewed_at": datetime.now().isoformat(),
            "resolution_type": "credit_refund"}


# ── Main Pipeline ──────────────────────────────────────────────
def run_order_pipeline(raw_po: dict) -> OrderPipelineState:
    from mock_tools import (validate_purchase_order, verify_customer_credit,
        enrich_line_items, create_shipment_plan, poll_shipment_status,
        detect_exceptions, draft_order_update, send_customer_update)

    client = anthropic.Anthropic()
    state = OrderPipelineState(po_number=raw_po["po_number"], raw_po=raw_po)
    cb = CarrierCircuitBreaker()

    # ── Agent 1: Intake (Claude-powered) ──────────────────────
    print(f"\n[INTAKE] Processing {state.po_number}...")
    intake_tools = [
        {"name": "validate_order", "description": "Validate PO format and required fields",
         "input_schema": {"type": "object", "properties": {"po_number": {"type": "string"}}, "required": ["po_number"]}},
        {"name": "check_credit", "description": "Check customer credit availability",
         "input_schema": {"type": "object", "properties": {"customer_id": {"type": "string"}, "order_total": {"type": "number"}}, "required": ["customer_id", "order_total"]}}
    ]
    intake_msgs = [{"role": "user", "content": f"Validate this order and check credit: {json.dumps(raw_po, default=str)}"}]

    response = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=1024,
        system="You are an order intake agent. Validate the order format and check customer credit. Use the available tools.",
        tools=intake_tools, messages=intake_msgs
    )

    # Process tool calls in an agentic loop
    while response.stop_reason == "tool_use":
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                if block.name == "validate_order":
                    result = validate_purchase_order(raw_po)
                elif block.name == "check_credit":
                    result = verify_customer_credit(block.input["customer_id"], block.input["order_total"])
                else:
                    result = {"error": f"Unknown tool: {block.name}"}
                tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result, default=str)})
        intake_msgs.append({"role": "assistant", "content": response.content})
        intake_msgs.append({"role": "user", "content": tool_results})
        response = client.messages.create(
            model="claude-sonnet-4-6", max_tokens=1024,
            system="You are an order intake agent. Validate the order format and check customer credit.",
            tools=intake_tools, messages=intake_msgs
        )

    # Extract Claude's text summary
    intake_text = ""
    for block in response.content:
        if hasattr(block, "text"):
            intake_text = block.text
            break

    # Check if validation/credit actually passed (from tool results)
    validation = validate_purchase_order(raw_po)
    if not validation.get("validated"):
        print(f"[INTAKE] FAILED: {validation.get('errors')}")
        state.stage = "error"
        return state
    order_total = sum(li.get("qty", 0) * li.get("unit_price", 0) for li in raw_po.get("line_items", []))
    credit = verify_customer_credit(raw_po.get("customer", ""), order_total)
    if not credit.get("credit_approved"):
        print(f"[INTAKE] Credit check failed. Available: ${credit.get('available_credit', 0):,.2f}")
        state.stage = "error"
        return state
    enriched = enrich_line_items(raw_po["line_items"], raw_po["customer"])
    state.intake_output = {"validated": True, "credit_passed": True,
        "line_items": enriched, "total_value": order_total,
        "claude_summary": intake_text}
    state.stage = "fulfillment"
    print("[INTAKE] Claude validated order and extracted key fields.")
    print("[INTAKE] Complete.")

    # ── Agent 2: Fulfillment ───────────────────────────────────
    print("[FULFILLMENT] Planning shipments...")
    plan = create_shipment_plan(state.po_number, raw_po["line_items"])
    state.fulfillment_output = plan

    if plan.get("human_approval_required"):
        override = split_shipment_review(state)
        state.fulfillment_output["manager_override"] = override
        if override["decision"] == "wait_consolidation":
            print("[FULFILLMENT] Waiting for consolidation. Pipeline paused.")
            state.stage = "awaiting_consolidation"
            return state
    state.stage = "monitoring"
    print(f"[FULFILLMENT] Complete. Split: {plan.get('requires_split')}.")

    # ── Agent 3: Exception Monitor ─────────────────────────────
    print("[MONITOR] Checking shipment status...")
    tracking_numbers = [s.get("tracking", "TRK-MOCK") for s in plan.get("shipments", [])]
    if not cb.is_tripped():
        try:
            statuses = poll_shipment_status(tracking_numbers)
            cb.record_success()
            exceptions = detect_exceptions(statuses, plan)
            state.exception_output = {"statuses": statuses, "exceptions": exceptions}

            # HITL: credit/refund resolutions require ops manager approval
            for exc in exceptions:
                resolution_type = exc.get("recommended_action", "").lower()
                requires_approval = ("credit" in resolution_type
                                     or "refund" in resolution_type)
                if requires_approval:
                    print(f"[MONITOR] Credit/refund resolution detected. Requesting ops manager approval.")
                    approval = credit_resolution_review(state, exc)
                    exc["manager_override"] = approval
                    if approval["decision"] == "deny":
                        exc["resolution_blocked"] = True
                        print("[MONITOR] Credit resolution denied by ops manager.")
        except Exception:
            cb.record_failure()
            state.exception_output = {"fallback_mode": True}
    else:
        state.exception_output = {"fallback_mode": True,
            "message": "Circuit breaker tripped. Using fallback polling."}
    state.stage = "communication"
    print("[MONITOR] Complete.")

    # ── Agent 4: Communication (Claude-powered) ────────────────
    print("[COMMUNICATION] Drafting updates...")
    update_type = "shipment" if not state.exception_output.get("exceptions") else "delay"
    comms_tools = [
        {"name": "draft_update", "description": "Draft an order status email to the customer",
         "input_schema": {"type": "object", "properties": {"po_number": {"type": "string"}, "update_type": {"type": "string", "enum": ["confirmation", "shipment", "delay", "delivery"]}, "details": {"type": "object"}}, "required": ["po_number", "update_type"]}},
        {"name": "send_update", "description": "Send the drafted update to the customer",
         "input_schema": {"type": "object", "properties": {"draft_id": {"type": "string"}, "channel": {"type": "string"}, "customer_contact": {"type": "string"}}, "required": ["draft_id", "channel", "customer_contact"]}}
    ]
    comms_context = {"po_number": state.po_number, "update_type": update_type,
        "tracking": tracking_numbers[0] if tracking_numbers else "N/A",
        "customer": raw_po["customer"], "shipments": plan.get("shipments", [])}
    comms_msgs = [{"role": "user", "content": f"Draft and send a {update_type} notification for this order: {json.dumps(comms_context, default=str)}"}]

    comms_response = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=1024,
        system="You are a customer communication agent for B2B orders. Draft professional, clear status updates. Use draft_update to create the message, then send_update to deliver it.",
        tools=comms_tools, messages=comms_msgs
    )

    # Process tool calls in an agentic loop
    draft_result = None
    while comms_response.stop_reason == "tool_use":
        tool_results = []
        for block in comms_response.content:
            if block.type == "tool_use":
                if block.name == "draft_update":
                    result = draft_order_update(block.input["po_number"], block.input["update_type"], block.input.get("details", {}))
                    draft_result = result
                elif block.name == "send_update":
                    result = send_customer_update(block.input["draft_id"], block.input["channel"], block.input["customer_contact"])
                else:
                    result = {"error": f"Unknown tool: {block.name}"}
                tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result, default=str)})
        comms_msgs.append({"role": "assistant", "content": comms_response.content})
        comms_msgs.append({"role": "user", "content": tool_results})
        comms_response = client.messages.create(
            model="claude-sonnet-4-6", max_tokens=1024,
            system="You are a customer communication agent for B2B orders. Draft professional, clear status updates.",
            tools=comms_tools, messages=comms_msgs
        )

    # Fallback if Claude didn't use tools
    if not draft_result:
        draft_result = draft_order_update(state.po_number, update_type,
            {"tracking": tracking_numbers[0] if tracking_numbers else "N/A"})
        send_customer_update(draft_result["draft_id"], "email", f"{raw_po['customer']}@example.com")

    state.communication_output = {"draft_id": draft_result["draft_id"],
        "sent_via": "email", "sent_at": datetime.now().isoformat()}
    state.stage = "complete"
    print("[COMMUNICATION] Claude drafted customer notification.")
    print(f"[PIPELINE] Complete! Order {state.po_number} processed.")
    return state


def main():
    print("=" * 60)
    print("  B2B Order Lifecycle Pipeline — Capstone 4-B")
    print("  Type 'demo' for sample PO, 'quit' to exit.")
    print("=" * 60)
    sample = {"po_number": "PO-2024-11234", "customer": "Acme Manufacturing",
        "line_items": [
            {"sku": "SKU-4892", "qty": 50, "unit_price": 495.00},
            {"sku": "SKU-7721", "qty": 200, "unit_price": 125.00}]}
    while True:
        cmd = input("\nCommand: ").strip()
        if cmd.lower() in ("quit", "exit"): break
        if cmd.lower() == "demo":
            state = run_order_pipeline(sample)
            print(f"\nFinal state:\n{json.dumps(asdict(state), indent=2, default=str)}")

if __name__ == "__main__":
    main()

"""test_pipeline.py — 5 test scenarios for B2B Order Lifecycle Pipeline."""

import pytest
from unittest.mock import patch, MagicMock
from pipeline import (
    run_order_pipeline, CarrierCircuitBreaker, OrderPipelineState
)
from mock_tools import PROCESSED_POS


def _stub_claude(*args, **kwargs):
    """Return a mock Claude response that skips tool use."""
    mock_resp = MagicMock()
    mock_resp.stop_reason = "end_turn"
    text_block = MagicMock()
    text_block.type = "text"
    text_block.text = "Order validated successfully."
    mock_resp.content = [text_block]
    return mock_resp


@pytest.fixture(autouse=True)
def reset_processed_pos():
    """Clear duplicate-detection set between tests."""
    PROCESSED_POS.clear()
    yield
    PROCESSED_POS.clear()


# ── Test 1: Happy path ──────────────────────────────────────
@patch("pipeline.anthropic.Anthropic")
def test_happy_path_standard_order(mock_anthropic_cls):
    mock_client = MagicMock()
    mock_client.messages.create.side_effect = _stub_claude
    mock_anthropic_cls.return_value = mock_client

    po = {"po_number": "PO-TEST-001", "customer": "Acme Manufacturing",
          "line_items": [{"sku": "SKU-4892", "qty": 10, "unit_price": 100.00}]}
    state = run_order_pipeline(po)

    assert state.stage == "complete"
    assert state.intake_output["validated"] is True
    assert state.communication_output["sent_via"] == "email"


# ── Test 2: Credit check failure ─────────────────────────────
@patch("pipeline.anthropic.Anthropic")
def test_credit_check_failure(mock_anthropic_cls):
    mock_client = MagicMock()
    mock_client.messages.create.side_effect = _stub_claude
    mock_anthropic_cls.return_value = mock_client

    # Beta Industrial has only $2K available credit
    po = {"po_number": "PO-TEST-002", "customer": "Beta Industrial",
          "line_items": [{"sku": "SKU-4892", "qty": 50, "unit_price": 495.00}]}
    state = run_order_pipeline(po)

    assert state.stage == "error", "Pipeline should stop when credit is insufficient"


# ── Test 3: HITL escalation (high-value split >$50K) ─────────
@patch("builtins.input", return_value="1")  # auto-approve
@patch("pipeline.anthropic.Anthropic")
def test_hitl_escalation_high_value(mock_anthropic_cls, mock_input):
    mock_client = MagicMock()
    mock_client.messages.create.side_effect = _stub_claude
    mock_anthropic_cls.return_value = mock_client

    po = {"po_number": "PO-TEST-003", "customer": "Acme Manufacturing",
          "line_items": [
              {"sku": "SKU-4892", "qty": 60, "unit_price": 495.00},
              {"sku": "SKU-7721", "qty": 200, "unit_price": 125.00}]}
    state = run_order_pipeline(po)

    assert state.stage == "complete"
    assert state.fulfillment_output.get("human_approval_required") is True
    assert "manager_override" in state.fulfillment_output


# ── Test 4: Circuit breaker trip ────���────────────────────────
def test_circuit_breaker_trip():
    cb = CarrierCircuitBreaker()
    assert cb.status == "healthy"

    for _ in range(5):
        cb.record_failure()

    assert cb.status == "tripped"
    assert cb.is_tripped() is True


# ── Test 5: Invalid input (missing fields) ───────────────────
@patch("pipeline.anthropic.Anthropic")
def test_invalid_input_missing_fields(mock_anthropic_cls):
    mock_client = MagicMock()
    mock_client.messages.create.side_effect = _stub_claude
    mock_anthropic_cls.return_value = mock_client

    # Missing 'customer' and 'line_items' fields
    po = {"po_number": "PO-TEST-005"}
    state = run_order_pipeline(po)

    assert state.stage == "error", "Pipeline should reject orders with missing fields"

python -c "
from pipeline import run_order_pipeline
from dataclasses import asdict
import json

sample = {'po_number': 'PO-2024-11234', 'customer': 'Acme Manufacturing',
    'line_items': [
        {'sku': 'SKU-4892', 'qty': 50, 'unit_price': 495.00},
        {'sku': 'SKU-7721', 'qty': 200, 'unit_price': 125.00}]}
state = run_order_pipeline(sample)
print(json.dumps(asdict(state), indent=2, default=str))
" && pytest test_pipeline.py -v