PROTHAM
Initial commit: AdaptiveWorld Env v2.0 (Round 2)
7043cc6
Raw
History Blame Contribute Delete
19.4 kB
"""
mock_api.py β€” Dynamic mock API for AdaptiveWorld environment.
All business endpoints read their field names, endpoint paths, status values,
and behaviour from DYNAMIC_CONFIG. The /_admin/mutate endpoint lets the
environment silently mutate DYNAMIC_CONFIG mid-episode to simulate API drift.
Domains covered:
e-commerce β†’ /orders, /products, /cart, /discount
hotel β†’ /rooms/book, /v2/rooms/book
flight β†’ /flights/search
insurance β†’ /claims, /claims/v2
"""
import time
import uuid
from fastapi import APIRouter, Header, HTTPException, Request, Query
from pydantic import BaseModel
from typing import Optional
router = APIRouter(prefix="/mock_api", tags=["mock"])
# ── In-memory state ────────────────────────────────────────────────────────────
_issued_tokens: set = set()
_request_log: dict = {}
_items_db: dict = {}
_orders_db: dict = {}
_bookings_db: dict = {}
_claims_db: dict = {}
# ── World configuration β€” mutated mid-episode via /_admin/mutate ───────────────
DYNAMIC_CONFIG: dict = {
# e-commerce / order fields
"order_field": "qty", # field_rename drift changes this
"required_extra": None, # set to "customer_id" after field_rename drift
"order_status": "confirmed", # silent_semantic drift changes this to "approved"
"noisy_errors": False, # level 2+ escalation: vague error messages
# hotel booking
"rooms_endpoint": "/mock_api/rooms/book", # endpoint_version drift changes this
# flight
"auth_scheme": "Bearer", # auth drift changes this to "ApiKey"
# insurance claims
"claims_endpoint": "/mock_api/claims",
"claims_id_field": "policy_id",
"claims_amount_field": "amount",
"claims_require_date": False,
# e-commerce discount
"discount_requires_membership": False, # policy_change drift flips this
# product search response shape
"product_key": "products", # response_structure drift changes to "items"
"product_id_field": "id", # changes to "product_id"
"product_price_field": "price", # changes to "cost"
# product category
"product_category": "electronics", # cascading_invalidation drift changes to "tech"
# rate limiting
"max_rooms_per_request": 0, # 0 = no limit; set to 1 after rate_limit drift
"bulk_booking_allowed": True,
# auth / token (carried from Round 1)
"demo_token": "demo_token_123",
"client_id": "abc",
"client_secret": "xyz",
}
LOG_PAGES = {
None: {"items": list(range(10)), "next_cursor": "cur_abc", "has_more": True},
"cur_abc": {"items": list(range(10, 20)), "next_cursor": "cur_def", "has_more": True},
"cur_def": {"items": list(range(20, 25)), "next_cursor": None, "has_more": False},
}
# ── Admin endpoints ────────────────────────────────────────────────────────────
@router.post("/_admin/mutate")
async def admin_mutate(config: Optional[dict] = None):
"""
Called by AdaptiveWorldEnvironment to inject drift mid-episode.
Silently updates DYNAMIC_CONFIG β€” no notification to the agent.
"""
_issued_tokens.clear()
_request_log.clear()
_items_db.clear()
_orders_db.clear()
_bookings_db.clear()
_claims_db.clear()
if config:
DYNAMIC_CONFIG.update(config)
return {"status": "mutated", "applied": DYNAMIC_CONFIG}
@router.post("/_admin/reset")
async def admin_reset(config: Optional[dict] = None):
"""Alias for mutate β€” also used for episode resets."""
return await admin_mutate(config)
# ── Auth ───────────────────────────────────────────────────────────────────────
class TokenRequest(BaseModel):
client_id: str
client_secret: str
@router.post("/auth/token")
async def get_token(body: TokenRequest):
if (body.client_id == DYNAMIC_CONFIG["client_id"] and
body.client_secret == DYNAMIC_CONFIG["client_secret"]):
token = f"tok_{int(time.time())}"
_issued_tokens.add(token)
return {"access_token": token, "expires_in": 3600}
raise HTTPException(status_code=401, detail="Invalid client credentials")
def _check_auth(authorization: Optional[str]) -> bool:
scheme = DYNAMIC_CONFIG.get("auth_scheme", "Bearer")
dt = DYNAMIC_CONFIG["demo_token"]
if not authorization:
return False
if not authorization.startswith(f"{scheme} "):
return False
token = authorization.split(" ", 1)[1]
return token in _issued_tokens or token == dt
# ── Users (carried from Round 1) ──────────────────────────────────────────────
@router.get("/users")
async def get_users(authorization: Optional[str] = Header(default=None)):
scheme = DYNAMIC_CONFIG.get("auth_scheme", "Bearer")
dt = DYNAMIC_CONFIG["demo_token"]
if not _check_auth(authorization):
raise HTTPException(
status_code=401,
detail=f"Authorization header missing or malformed. "
f"Expected format: {scheme} <token>. Use '{dt}' for testing."
)
return {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}
# ── E-commerce: Orders ─────────────────────────────────────────────────────────
@router.post("/orders")
async def create_order(request: Request):
ct = request.headers.get("content-type", "")
if "application/json" not in ct:
raise HTTPException(status_code=415, detail="Content-Type must be application/json")
body = await request.json()
# Dynamic field name β€” field_rename drift changes "qty" to "quantity"
required_field = DYNAMIC_CONFIG.get("order_field", "qty")
extra_field = DYNAMIC_CONFIG.get("required_extra")
if required_field not in body:
detail = f"Missing required field: '{required_field}'"
if DYNAMIC_CONFIG.get("noisy_errors"):
detail = "Validation error in request body" # ambiguous at level 2+
raise HTTPException(status_code=422, detail=detail)
if extra_field and extra_field not in body:
detail = f"Missing required field: '{extra_field}'"
if DYNAMIC_CONFIG.get("noisy_errors"):
detail = "Validation error in request body"
raise HTTPException(status_code=422, detail=detail)
order_id = str(uuid.uuid4())
qty_value = body.get(required_field, body.get("qty", body.get("quantity", 1)))
# Dynamic status value β€” silent_semantic drift changes "confirmed" to "approved"
status_value = DYNAMIC_CONFIG.get("order_status", "confirmed")
order = {
"order_id": order_id,
"status": status_value,
"product_id": body.get("product_id"),
required_field: qty_value,
}
_orders_db[order_id] = order
return order
@router.get("/orders/{order_id}")
async def get_order(order_id: str):
if order_id not in _orders_db:
raise HTTPException(status_code=404, detail="Order not found")
return _orders_db[order_id]
# ── E-commerce: Products ───────────────────────────────────────────────────────
@router.get("/products")
async def search_products(q: Optional[str] = Query(default=None)):
"""
Returns product list. response_structure drift changes the key and field names.
"""
product_key = DYNAMIC_CONFIG.get("product_key", "products")
id_field = DYNAMIC_CONFIG.get("product_id_field", "id")
price_field = DYNAMIC_CONFIG.get("product_price_field", "price")
category = DYNAMIC_CONFIG.get("product_category", "electronics")
products = [
{id_field: 1, price_field: 100, "name": "Widget A", "category": category},
{id_field: 2, price_field: 250, "name": "Widget B", "category": category},
{id_field: 5, price_field: 75, "name": "Widget C", "category": category},
]
if q:
products = [p for p in products if q.lower() in p.get("name", "").lower()]
return {product_key: products, "total": len(products)}
# ── E-commerce: Cart ───────────────────────────────────────────────────────────
@router.post("/cart/add")
async def add_to_cart(request: Request):
ct = request.headers.get("content-type", "")
if "application/json" not in ct:
raise HTTPException(status_code=415, detail="Content-Type must be application/json")
body = await request.json()
id_field = DYNAMIC_CONFIG.get("product_id_field", "id")
pid = body.get("product_id") or body.get(id_field)
if not pid:
raise HTTPException(status_code=422, detail="Missing product_id in body")
return {"cart_id": str(uuid.uuid4()), "product_id": pid, "added": True}
# ── E-commerce: Discount ───────────────────────────────────────────────────────
@router.post("/discount/apply")
async def apply_discount(request: Request):
ct = request.headers.get("content-type", "")
if "application/json" not in ct:
raise HTTPException(status_code=415, detail="Content-Type must be application/json")
body = await request.json()
code = body.get("code") or body.get("discount_code")
if not code:
raise HTTPException(status_code=422, detail="Missing discount code")
# policy_change drift: discount now requires membership_tier: "gold"
if DYNAMIC_CONFIG.get("discount_requires_membership"):
tier = body.get("membership_tier")
if tier != "gold":
raise HTTPException(
status_code=403,
detail={
"error": "Discount ineligible",
"reason": "Discount codes now require membership_tier: 'gold' in the request body.",
"policy": "membership_required_for_discounts"
}
)
return {"discount_applied": True, "code": code, "amount": 20}
# ── Hotel: Room Booking ────────────────────────────────────────────────────────
@router.post("/rooms/book")
async def book_room_v1(request: Request):
"""
V1 booking endpoint. After endpoint_version drift, this returns 404
because the environment will call /v2/rooms/book which is the new path.
The environment checks rooms_endpoint to decide which is active.
"""
current_endpoint = DYNAMIC_CONFIG.get("rooms_endpoint", "/mock_api/rooms/book")
if current_endpoint != "/mock_api/rooms/book":
raise HTTPException(
status_code=404,
detail="This endpoint has moved. Please check the API documentation."
)
return await _do_room_booking(request)
@router.post("/v2/rooms/book")
async def book_room_v2(request: Request):
"""V2 booking endpoint β€” active after endpoint_version drift."""
current_endpoint = DYNAMIC_CONFIG.get("rooms_endpoint", "/mock_api/rooms/book")
if current_endpoint != "/mock_api/v2/rooms/book":
raise HTTPException(
status_code=404,
detail="Endpoint not found."
)
return await _do_room_booking(request)
async def _do_room_booking(request: Request):
ct = request.headers.get("content-type", "")
if "application/json" not in ct:
raise HTTPException(status_code=415, detail="Content-Type must be application/json")
body = await request.json()
nights = body.get("nights")
room_type = body.get("room_type", "standard")
# rate_limit drift: max 1 room per request (qty check)
qty = body.get("qty", body.get("quantity", 1))
max_rooms = DYNAMIC_CONFIG.get("max_rooms_per_request", 0)
if max_rooms and int(qty) > max_rooms:
raise HTTPException(
status_code=429,
detail={
"error": "Booking limit exceeded",
"reason": f"Maximum {max_rooms} room(s) per request. Bulk booking has been removed.",
"max_per_request": max_rooms,
}
)
if not nights:
raise HTTPException(status_code=422, detail="Missing required field: 'nights'")
booking_id = str(uuid.uuid4())
booking = {"booking_id": booking_id, "nights": nights, "room_type": room_type,
"confirmation": f"CONF-{booking_id[:8].upper()}"}
_bookings_db[booking_id] = booking
return booking
# ── Flight: Search ─────────────────────────────────────────────────────────────
@router.get("/flights/search")
async def search_flights(
origin: Optional[str] = Query(default=None),
destination: Optional[str] = Query(default=None),
depart: Optional[str] = Query(default=None),
authorization: Optional[str] = Header(default=None),
):
"""Requires auth. Auth scheme changes on auth drift."""
if not _check_auth(authorization):
scheme = DYNAMIC_CONFIG.get("auth_scheme", "Bearer")
raise HTTPException(
status_code=401,
detail=f"Unauthorized. Expected Authorization: {scheme} <token>. "
f"Auth scheme may have changed β€” check /openapi.json."
)
if not origin or not destination:
raise HTTPException(status_code=422, detail="Missing required params: origin, destination")
return {
"flights": [
{"flight_id": "FL001", "origin": origin, "destination": destination,
"depart": depart or "tomorrow", "price": 4500, "seats": 12},
{"flight_id": "FL002", "origin": origin, "destination": destination,
"depart": depart or "tomorrow", "price": 5200, "seats": 4},
]
}
# ── Insurance: Claims ──────────────────────────────────────────────────────────
@router.post("/claims")
async def file_claim_v1(request: Request):
"""V1 claims endpoint. Moves to /claims/v2 after endpoint_version drift."""
current_endpoint = DYNAMIC_CONFIG.get("claims_endpoint", "/mock_api/claims")
if current_endpoint != "/mock_api/claims":
raise HTTPException(status_code=404, detail="Claims endpoint has moved. Check API docs.")
return await _do_file_claim(request)
@router.post("/claims/v2")
async def file_claim_v2(request: Request):
"""V2 claims endpoint β€” active after drift."""
current_endpoint = DYNAMIC_CONFIG.get("claims_endpoint", "/mock_api/claims")
if current_endpoint != "/mock_api/claims/v2":
raise HTTPException(status_code=404, detail="Endpoint not found.")
return await _do_file_claim(request)
async def _do_file_claim(request: Request):
ct = request.headers.get("content-type", "")
if "application/json" not in ct:
raise HTTPException(status_code=415, detail="Content-Type must be application/json")
body = await request.json()
id_field = DYNAMIC_CONFIG.get("claims_id_field", "policy_id")
amount_field = DYNAMIC_CONFIG.get("claims_amount_field", "amount")
require_date = DYNAMIC_CONFIG.get("claims_require_date", False)
if id_field not in body:
raise HTTPException(status_code=422,
detail=f"Missing required field: '{id_field}'")
if amount_field not in body:
raise HTTPException(status_code=422,
detail=f"Missing required field: '{amount_field}'")
if require_date and "incident_date" not in body:
raise HTTPException(status_code=422,
detail="Missing required field: 'incident_date'")
claim_id = str(uuid.uuid4())
claim = {
"claim_id": claim_id,
id_field: body[id_field],
amount_field: body[amount_field],
"status": "pending",
}
_claims_db[claim_id] = claim
return claim
@router.get("/claims/{claim_id}")
async def get_claim(claim_id: str):
if claim_id not in _claims_db:
raise HTTPException(status_code=404, detail="Claim not found")
return _claims_db[claim_id]
# ── Utility: Search, Rate limiting, Logs (carried from Round 1) ────────────────
@router.get("/search")
async def search(q: Optional[str] = Query(default=None)):
if not q:
raise HTTPException(status_code=422,
detail=[{"loc": ["query", "q"], "msg": "field required"}])
return {"results": [{"title": f"Result for {q}", "score": 0.95}], "total": 1}
@router.get("/rate_limited")
async def rate_limited(request: Request,
x_retry_after: Optional[str] = Header(default=None)):
client_id = request.client.host if request.client else "default"
now = time.time()
window = [t for t in _request_log.get(client_id, []) if now - t < 5]
if len(window) >= 3 and not x_retry_after:
_request_log[client_id] = window
raise HTTPException(
status_code=429,
detail={"error": "Rate limit exceeded",
"hint": "Add X-Retry-After header with value 2 to override."},
headers={"Retry-After": "2"}
)
window.append(now)
_request_log[client_id] = window
return {"data": "rate_limited_resource", "requests_in_window": len(window)}
@router.get("/logs")
async def get_logs(cursor: Optional[str] = Query(default=None)):
if cursor not in LOG_PAGES:
raise HTTPException(status_code=400, detail=f"Invalid cursor: {cursor}")
return LOG_PAGES[cursor]
@router.get("/openapi-schema")
async def get_schema_summary():
"""Returns a simplified summary of the current API contract."""
return {
"order_field": DYNAMIC_CONFIG.get("order_field", "qty"),
"required_extra": DYNAMIC_CONFIG.get("required_extra"),
"order_status": DYNAMIC_CONFIG.get("order_status", "confirmed"),
"rooms_endpoint": DYNAMIC_CONFIG.get("rooms_endpoint", "/mock_api/rooms/book"),
"auth_scheme": DYNAMIC_CONFIG.get("auth_scheme", "Bearer"),
"claims_endpoint": DYNAMIC_CONFIG.get("claims_endpoint", "/mock_api/claims"),
"claims_id_field": DYNAMIC_CONFIG.get("claims_id_field", "policy_id"),
"claims_amount_field": DYNAMIC_CONFIG.get("claims_amount_field", "amount"),
"product_key": DYNAMIC_CONFIG.get("product_key", "products"),
"product_id_field": DYNAMIC_CONFIG.get("product_id_field", "id"),
"product_price_field": DYNAMIC_CONFIG.get("product_price_field", "price"),
"max_rooms_per_request": DYNAMIC_CONFIG.get("max_rooms_per_request", 0),
}