Build a Resilient Chain of Clue-Driven HTTP Requests

Problem Statement

Modern financial integrations frequently require chaining multiple asynchronous or cursor-based API calls across internal and external partner endpoints. Implement a resilient resolver, ClueChainResolver, that navigates a linked sequence of HTTP endpoints to retrieve a terminal result.

Each successful step returns a JSON payload containing either a pointer to the next clue or the terminal result:

• Step response: {"status": "continue", "next_url": "", "auth_token": ""}
• Terminal response: {"status": "complete", "result": ""}

Your client must safely traverse these steps while handling real-world distributed systems failures.

Requirements

1. Loop Detection & Hop Limits: Terminate with a TraversalError if a URL has already been visited in the current chain, or if total hops exceed max_hops.
2. Security & Origin Boundaries: The initial request uses a base auth_header. If a subsequent next_url belongs to a different origin (scheme + host + port), do not forward the initial auth_header unless a new auth_token was explicitly provided for that hop. Never leak credentials across origins.
3. Deadlines & Retry Budgets: Enforce an overall execution deadline (deadline_seconds) across the entire traversal. For transient HTTP errors (429, 502, 503, 504), retry using exponential backoff with a cap on attempts per step (max_retries_per_step). Deduct elapsed backoff time from the global deadline.
4. Defensive Validation: Validate that responses are valid JSON and contain expected keys. Enforce max_response_bytes to prevent unbounded memory consumption from malicious or corrupted endpoints.

Interface

from typing import Protocol, Optional, Dict, Any

class HttpResponse:
    status_code: int
    headers: Dict[str, str]
    content: bytes

class Transport(Protocol):
    def get(self, url: str, headers: Dict[str, str], timeout_seconds: float) -> HttpResponse:
        """Executes a GET request. Raises NetworkError on socket-level failures."""
        ...

class ClueChainResolver:
    def __init__(
        self,
        transport: Transport,
        max_hops: int = 15,
        max_retries_per_step: int = 3,
        deadline_seconds: float = 30.0,
        max_response_bytes: int = 1_000_000,
    ):
        ...

    def resolve(self, initial_url: str, initial_auth: Optional[str] = None) -> str:
        """Executes the chain traversal and returns the terminal result string."""
        ...

Examples

Example 1: Clean 2-Hop Traversal

# Step 1: GET https://api.ramp.com/clue/1 with header {"Authorization": "Bearer init_token"}
# Response 200: {"status": "continue", "next_url": "https://api.ramp.com/clue/2"}
# Step 2: GET https://api.ramp.com/clue/2 with header {"Authorization": "Bearer init_token"}
# Response 200: {"status": "complete", "result": "TX_SUCCESS_9481"}

resolver.resolve("https://api.ramp.com/clue/1", initial_auth="Bearer init_token")
# Output: "TX_SUCCESS_9481"

Example 2: Cross-Origin Boundary & Loop Protection

# Step 1: GET https://api.ramp.com/clue/1 -> next_url: "https://partner.bank.com/probe"
# Step 2: GET https://partner.bank.com/probe -> must NOT send "Bearer init_token"
# If https://partner.bank.com/probe returns next_url: "https://api.ramp.com/clue/1"
# Traversal raises TraversalError("Cycle detected")

Edge Cases

• Server returns 429 with a Retry-After header.
• Non-transient errors (e.g., 401 Unauthorized, 404 Not Found) must fail immediately without retry.
• Relative vs. absolute URLs in next_url (normalize them relative to current URL).