Catch Visual Bugs in CI with the Decisions API

httpx==0.28.1
playwright==1.64.0
pytest==9.1.1

[tool.ruff]
line-length = 150
target-version = "py310"

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP", "C4", "SIM"]

[tool.mypy]
strict = true
ignore_missing_imports = true

[tool.pytest.ini_options]
addopts = "-q"

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Ledgerline: invoicing for small teams</title>
  <link rel="stylesheet" href="/styles.css">
  <link rel="stylesheet" href="/theme.css">
</head>
<body>
  <header class="nav">
    <a class="brand" href="/">Ledgerline</a>
    <nav><a href="/pricing.html">Pricing</a><a href="#features">Features</a><a href="#" class="button small">Sign in</a></nav>
  </header>
  <main>
    <section class="hero">
      <h1>Invoices out in minutes, paid in days</h1>
      <p>Ledgerline sends, tracks, and reconciles invoices for teams of two to fifty. No spreadsheets, no chasing.</p>
      <a href="/pricing.html" class="button primary">Start free trial</a>
    </section>
    <section id="features" class="features">
      <article><h2>Send</h2><p>Branded invoices with your terms, due dates, and payment links.</p></article>
      <article><h2>Track</h2><p>See who opened what, and get a nudge before an invoice goes late.</p></article>
      <article><h2>Reconcile</h2><p>Match payments to invoices automatically and export to your books.</p></article>
    </section>
  </main>
  <footer>Ledgerline Inc. All amounts shown in USD.</footer>
</body>
</html>

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Ledgerline pricing</title>
  <link rel="stylesheet" href="/styles.css">
  <link rel="stylesheet" href="/theme.css">
</head>
<body>
  <header class="nav">
    <a class="brand" href="/">Ledgerline</a>
    <nav><a href="/pricing.html">Pricing</a><a href="/#features">Features</a><a href="#" class="button small">Sign in</a></nav>
  </header>
  <main>
    <section class="hero compact">
      <h1>Simple pricing, billed monthly</h1>
      <p>Every plan includes unlimited invoices and payment links. Cancel any time.</p>
    </section>
    <section class="plans">
      <article class="plan">
        <h2>Starter</h2>
        <p class="price">$12<span>/month</span></p>
        <ul><li>1 user</li><li>Unlimited invoices</li><li>Email reminders</li></ul>
        <a href="#" class="button">Choose Starter</a>
      </article>
      <article class="plan featured">
        <h2>Team</h2>
        <p class="price">$39<span>/month</span></p>
        <ul><li>Up to 10 users</li><li>Automatic reconciliation</li><li>Accounting export</li></ul>
        <a href="#" class="button primary">Choose Team</a>
      </article>
      <article class="plan">
        <h2>Business</h2>
        <p class="price">$99<span>/month</span></p>
        <ul><li>Up to 50 users</li><li>Approval workflows</li><li>Priority support</li></ul>
        <a href="#" class="button">Choose Business</a>
      </article>
    </section>
  </main>
  <footer>Ledgerline Inc. All amounts shown in USD.</footer>
</body>
</html>

/* Shared layout. The variant file served as /theme.css adds the responsive rules and the look. */
* { box-sizing: border-box; }
body { margin: 0; font: 16px/1.5 -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif; color: #1f2933; background: #fff; }
.nav { display: flex; justify-content: space-between; align-items: center; padding: 16px 24px; border-bottom: 1px solid #e4e7eb; }
.nav nav { display: flex; gap: 20px; align-items: center; }
.nav a { color: #1f2933; text-decoration: none; }
.brand { font-weight: 700; font-size: 20px; }
main { max-width: 1040px; margin: 0 auto; padding: 0 24px; }
.hero { text-align: center; padding: 72px 0 48px; }
.hero.compact { padding: 48px 0 24px; }
.hero h1 { font-size: 44px; line-height: 1.1; margin: 0 0 16px; }
.hero p { font-size: 18px; color: #52606d; max-width: 560px; margin: 0 auto 28px; }
.button { display: inline-block; padding: 12px 22px; border-radius: 8px; border: 1px solid #1f2933; text-decoration: none; color: #1f2933; font-weight: 600; }
.button.primary { background: #0b57d0; border-color: #0b57d0; color: #fff; }
.button.small { padding: 6px 14px; }
.features { display: grid; grid-template-columns: repeat(auto-fit, minmax(240px, 1fr)); gap: 24px; padding: 24px 0 64px; }
.features article { padding: 24px; border: 1px solid #e4e7eb; border-radius: 12px; }
.features h2 { margin: 0 0 8px; font-size: 22px; }
.plans { display: grid; grid-template-columns: repeat(3, minmax(0, 1fr)); gap: 24px; padding: 24px 0 64px; }
.plan { min-width: 280px; padding: 28px; border: 1px solid #e4e7eb; border-radius: 12px; background: #fff; }
.plan.featured { border-color: #0b57d0; box-shadow: 0 8px 24px rgba(11, 87, 208, 0.12); }
.plan h2 { margin: 0 0 8px; font-size: 24px; }
.price { font-size: 40px; font-weight: 700; margin: 0 0 16px; }
.price span { font-size: 16px; font-weight: 400; color: #52606d; }
.plan ul { padding-left: 20px; margin: 0 0 24px; color: #52606d; }
footer { text-align: center; color: #7b8794; padding: 24px; border-top: 1px solid #e4e7eb; font-size: 14px; }

/* What is deployed today. The media query stacks the plan columns on narrow screens. */
@media (max-width: 720px) {
  .hero h1 { font-size: 32px; }
  .plans { grid-template-columns: 1fr; }
  .plan { min-width: 0; }
}

/* The pull request under test. A cleanup removed the "unused" media query from main.css, so the
   three plan columns keep their 280 px minimum width at 390 px and draw on top of each other. */
.hero h1 { letter-spacing: -0.01em; }

/* A deliberate redesign: dark hero, new headline weight, rounder buttons. Nothing is broken. */
@media (max-width: 720px) {
  .hero h1 { font-size: 32px; }
  .plans { grid-template-columns: 1fr; }
  .plan { min-width: 0; }
}
.hero { background: #0f172a; color: #fff; border-radius: 16px; padding: 64px 24px 56px; margin-top: 24px; }
.hero.compact { padding: 48px 24px 32px; }
.hero h1 { font-weight: 800; }
.hero p { color: #cbd5e1; }
.button { border-radius: 999px; }
.button.primary { background: #22c55e; border-color: #22c55e; color: #052e16; }
.plan.featured { border-color: #22c55e; box-shadow: 0 8px 24px rgba(34, 197, 94, 0.18); }

# Save as .github/workflows/visual-qa.yml. The step summary on the job page shows the report
# that visual_qa.py writes to $GITHUB_STEP_SUMMARY; the screenshots are uploaded as an artifact.
name: Visual QA
on:
  pull_request:

permissions:
  contents: read

jobs:
  visual-qa:
    if: github.event.pull_request.head.repo.full_name == github.repository
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
        with:
          python-version: "3.12"
      - name: Install
        run: |
          python -m pip install -r requirements.txt
          python -m playwright install --with-deps chromium
      - name: Visual QA gate
        env:
          PERPLEXITY_API_KEY: ${{ secrets.PERPLEXITY_API_KEY }}
        # The bundled site: the pull request's stylesheet against main's. For your own site, build or deploy
        # a preview first and use --url "$PREVIEW_URL" --baseline-url "$PRODUCTION_URL" instead.
        run: python visual_qa.py --variant pr --baseline main
      - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
        if: always()
        with:
          name: visual-qa
          path: out/

"""Offline tests for the policy, the report, and the requests. No API key, network, or browser needed."""

from __future__ import annotations

import json
import random
import urllib.request
from pathlib import Path
from typing import Any

import httpx
import pytest

import decisions_client as dc
import policy
import visual_qa as vq

CLEAN = {"text_clipped": 0.03, "elements_overlap": 0.02, "cta_visible": 0.97, "low_contrast": 0.01, "layout_broken": 0.04, "error_state": 0.0}


def test_policy_on_a_synthetic_distribution_bounds_misses_and_false_alarms() -> None:
    """Spread risk across outcomes: 85% clean screenshots with a noisy tail, 15% defects of which a quarter are subtle.
    The gate must miss few defects and flag few clean screenshots. Review-everything misses nothing but flags everything."""
    rng = random.Random(7)
    shots: list[tuple[bool, dict[str, float]]] = []
    for _ in range(20_000):
        p = {q: rng.betavariate(1, 20) if rng.random() < 0.9 else rng.betavariate(2, 6) for q in policy.DEFECT_QUESTIONS}
        p["cta_visible"] = rng.betavariate(20, 1)
        defect = rng.random() < 0.15
        if defect:
            q = rng.choice(policy.DEFECT_QUESTIONS + ["cta_visible"])
            v = rng.betavariate(9, 1.5) if rng.random() < 0.75 else rng.betavariate(3, 3)  # obvious, or subtle and near 50%
            p[q] = 1 - v if q == "cta_visible" else v
        shots.append((defect, p))
    outcomes = [(defect, policy.judge_single("x", p).outcome) for defect, p in shots]
    defects = [o for d, o in outcomes if d]
    clean = [o for d, o in outcomes if not d]
    missed = defects.count("pass") / len(defects)
    clean_failed = clean.count("fail") / len(clean)
    clean_flagged = 1 - clean.count("pass") / len(clean)
    flagged = 1 - [o for _, o in outcomes].count("pass") / len(outcomes)
    assert missed <= 0.10, f"missed {missed:.1%} of defects"
    assert clean_failed <= 0.01, f"failed {clean_failed:.1%} of clean screenshots"
    assert clean_flagged <= 0.10, f"flagged {clean_flagged:.1%} of clean screenshots"
    assert flagged < 0.25  # review-everything flags 100%; the gate has to do much better to be worth running


def test_pair_policy_on_a_synthetic_distribution() -> None:
    """10% regressions (content lost, not deliberate), 15% redesigns (deliberate, some remove content), 75% unchanged."""
    rng = random.Random(11)
    results: dict[str, list[str]] = {"regression": [], "redesign": [], "same": []}
    for _ in range(20_000):
        kind = rng.random()
        if kind < 0.10:
            p, bucket = {"content_lost": rng.betavariate(9, 1.5), "intentional": rng.betavariate(1.5, 9)}, "regression"
        elif kind < 0.25:
            lost = rng.betavariate(6, 3) if rng.random() < 0.3 else rng.betavariate(1.5, 9)
            p, bucket = {"content_lost": lost, "intentional": rng.betavariate(9, 1.5)}, "redesign"
        else:
            p, bucket = {"content_lost": rng.betavariate(1, 20), "intentional": rng.random()}, "same"
        results[bucket].append(policy.judge_pair("x", p).outcome)
    assert results["regression"].count("pass") / len(results["regression"]) <= 0.10
    assert results["redesign"].count("fail") / len(results["redesign"]) <= 0.01
    assert 1 - results["same"].count("pass") / len(results["same"]) <= 0.10


def test_judge_single_fails_warns_and_passes_at_the_cutoffs() -> None:
    assert policy.judge_single("x", CLEAN).outcome == "pass"
    assert policy.judge_single("x", {**CLEAN, "elements_overlap": 0.80}).rule.startswith("elements_overlap 80% is at or above 80%")
    assert policy.judge_single("x", {**CLEAN, "text_clipped": 0.79}).outcome == "warn"
    assert policy.judge_single("x", {**CLEAN, "text_clipped": 0.39}).outcome == "pass"
    assert policy.judge_single("x", {**CLEAN, "cta_visible": 0.20}).outcome == "fail"
    assert policy.judge_single("x", {**CLEAN, "cta_visible": 0.55}).outcome == "warn"
    worst = policy.judge_single("x", {**CLEAN, "layout_broken": 0.85, "text_clipped": 0.95})
    assert worst.rule.startswith("text_clipped 95%")  # the worst question names the rule


def test_judge_pair_fails_regressions_and_only_warns_on_deliberate_removals() -> None:
    assert policy.judge_pair("x", {"content_lost": 0.92, "intentional": 0.08}).outcome == "fail"
    assert policy.judge_pair("x", {"content_lost": 0.92, "intentional": 0.85}).outcome == "warn"
    assert policy.judge_pair("x", {"content_lost": 0.45, "intentional": 0.10}).outcome == "warn"
    assert policy.judge_pair("x", {"content_lost": 0.05, "intentional": 0.95}).outcome == "pass"


def test_overall_is_the_worst_verdict() -> None:
    verdicts = [policy.Verdict("a", "pass", ""), policy.Verdict("b", "warn", ""), policy.Verdict("c", "fail", "")]
    assert policy.overall(verdicts) == "fail"
    assert policy.overall(verdicts[:2]) == "warn"
    assert policy.overall([]) == "pass"


def test_render_report_lists_every_verdict_with_its_rule_and_probabilities() -> None:
    verdicts = [
        policy.judge_single("pricing at 390 px", {**CLEAN, "elements_overlap": 0.91}, "out/pr-pricing-mobile.png"),
        policy.judge_pair("pricing at 390 px (before/after)", {"content_lost": 0.9, "intentional": 0.1}),
    ]
    report = policy.render_report(verdicts, "pr", "main", 12_345, "pplx-decider-v1.2")
    assert report.startswith("## Visual QA: FAIL")
    assert "Checked `pr` against baseline `main` with `pplx-decider-v1.2`: 2 verdicts, 12,345 input tokens." in report
    assert "| pricing at 390 px | fail | elements_overlap 91% is at or above 80% |" in report
    assert "- **pricing at 390 px**: text_clipped 3%, elements_overlap 91%" in report
    assert "(out/pr-pricing-mobile.png)" in report
    assert "intentional 10%" in report


def test_single_body_carries_a_caption_the_image_and_six_questions() -> None:
    body = dc.single_body(b"png-bytes", "pricing at 390 px")
    assert body["model"] == "pplx-decider-v1.2"
    assert body["state"][0] == "Screenshot of pricing at 390 px."
    assert body["state"][1]["type"] == "image_url"
    assert body["state"][1]["image_url"]["url"] == "data:image/png;base64,cG5nLWJ5dGVz"
    assert list(body["questions"]) == ["text_clipped", "elements_overlap", "cta_visible", "low_contrast", "layout_broken", "error_state"]
    assert all(q["type"] == "noul" for q in body["questions"].values())


def test_pair_body_puts_before_then_after_each_behind_its_caption() -> None:
    body = dc.pair_body(b"old", b"new", "index at 1280 px")
    kinds = [item if isinstance(item, str) else item["type"] for item in body["state"]]
    assert kinds == ["Two screenshots of index at 1280 px.", "BEFORE (baseline):", "image_url", "AFTER (change under review):", "image_url"]
    assert body["state"][2]["image_url"]["url"].endswith("b2xk")  # "old"
    assert body["state"][4]["image_url"]["url"].endswith("bmV3")  # "new"
    assert list(body["questions"]) == ["content_lost", "intentional"]


def fake_client(responses: list[httpx.Response], calls: list[dict[str, Any]]) -> httpx.Client:
    """An httpx.Client that answers with the given responses in order and records each request body in calls."""

    def handler(request: httpx.Request) -> httpx.Response:
        calls.append(json.loads(request.content))
        return responses[len(calls) - 1]

    return httpx.Client(transport=httpx.MockTransport(handler))


OK: dict[str, Any] = {
    "model": "pplx-decider-v1.2",
    "answers": {name: {"type": "noul", "noul": p} for name, p in CLEAN.items()},
    "usage": {"input_tokens": 1234, "output_tokens": 6},
}


def test_ask_returns_probabilities_tokens_and_request_id_and_retries_a_429() -> None:
    calls: list[dict[str, Any]] = []
    responses = [
        httpx.Response(429, json={"error": "slow down"}, headers={"retry-after": "0"}),
        httpx.Response(200, json=OK, headers={"x-request-id": "rid-1"}),
    ]
    with fake_client(responses, calls) as client:
        probs, tokens, rid = dc.ask(client, dc.single_body(b"png", "x"))
    assert (probs, tokens, rid, len(calls)) == (CLEAN, 1234, "rid-1", 2)
    assert calls[0] == calls[1]


def test_ask_raises_on_a_400_and_on_a_200_missing_an_answer() -> None:
    calls: list[dict[str, Any]] = []
    with fake_client([httpx.Response(400, json={"error": {"message": "bad state"}})], calls) as client:
        try:
            dc.ask(client, dc.single_body(b"png", "x"))
            raise AssertionError("expected DecisionsError")
        except dc.DecisionsError as exc:
            assert "400" in str(exc) and "bad state" in str(exc)
    partial = {**OK, "answers": {k: v for k, v in OK["answers"].items() if k != "error_state"}}
    with fake_client([httpx.Response(200, json=partial)], []) as client:
        try:
            dc.ask(client, dc.single_body(b"png", "x"))
            raise AssertionError("expected DecisionsError")
        except dc.DecisionsError as exc:
            assert "cannot read" in str(exc)


def test_max_height_keeps_every_width_inside_the_tile_budget() -> None:
    for width in (390, 1280, 1440, 2048):
        assert round(width / 32) * (vq.max_height(width) // 32) <= vq.MAX_TILES
    assert vq.max_height(1280) == 1632
    assert vq.slug("/pricing.html") == "pricing" and vq.slug("/") == "index"


@pytest.mark.parametrize("invalid", [float("nan"), float("inf"), -0.1, 1.1])
def test_invalid_probabilities_cannot_pass_the_gate(invalid: float) -> None:
    body = {**OK, "answers": {**OK["answers"], "text_clipped": {"noul": invalid}}}
    response = httpx.Response(200, text=json.dumps(body))
    with fake_client([response], []) as client, pytest.raises(dc.DecisionsError, match="cannot read"):
        dc.ask(client, dc.single_body(b"png", "x"))


def test_oversized_page_is_not_silently_cropped() -> None:
    from unittest.mock import MagicMock

    browser = MagicMock()
    page = browser.new_context.return_value.new_page.return_value
    page.evaluate.return_value = 5000
    with pytest.raises(ValueError, match="split it into sections"):
        vq.screenshot(browser, "http://localhost/", 1280, 800)
    page.screenshot.assert_not_called()
    browser.new_context.return_value.close.assert_called_once()


def test_screenshot_preserves_viewport_width_and_full_height() -> None:
    from unittest.mock import MagicMock

    browser = MagicMock()
    page = browser.new_context.return_value.new_page.return_value
    page.evaluate.return_value = 1200
    vq.screenshot(browser, "http://localhost/", 390, 844)
    page.screenshot.assert_called_once_with(full_page=True, clip={"x": 0, "y": 0, "width": 390, "height": 1200}, animations="disabled")


def test_run_appends_report_to_job_summary(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    from unittest.mock import MagicMock

    monkeypatch.setattr(vq, "sync_playwright", MagicMock())
    monkeypatch.setattr(vq, "screenshot", lambda *args: b"png")
    monkeypatch.setattr(dc, "ask", lambda *args: (dict(CLEAN), 123, "rid"))
    summary = tmp_path / "summary.md"
    summary.write_text("Existing summary\n")
    monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary))
    args = vq.parse_args(["--url", "https://example.com", "--pages", "/", "--out", str(tmp_path / "out")])
    assert vq.run(args, "unused-test-key") == 0
    report = (tmp_path / "out" / "report.md").read_text()
    assert summary.read_text() == "Existing summary\n" + report


def test_site_server_serves_the_variant_as_theme_css() -> None:
    server, url = vq.serve("pr")
    try:
        theme = urllib.request.urlopen(f"{url}/theme.css", timeout=5).read().decode()
        page = urllib.request.urlopen(f"{url}/pricing.html", timeout=5).read().decode()
    finally:
        server.shutdown()
    assert theme == (vq.SITE_DIR / "pr.css").read_text()
    assert '<link rel="stylesheet" href="/theme.css">' in page

httpx==0.28.1
playwright==1.64.0
pytest==9.1.1

[tool.ruff]
line-length = 150
target-version = "py310"

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP", "C4", "SIM"]

[tool.mypy]
strict = true
ignore_missing_imports = true

[tool.pytest.ini_options]
addopts = "-q"

"""The part of the gate that talks to the Decisions API: the questions, the request bodies, and
bounded retries. Nothing in this file knows about browsers, pages, or cutoffs."""

from __future__ import annotations

import base64
import time
from typing import Any

import httpx

API_URL = "https://api.perplexity.ai/v1/decisions"
MODEL = "pplx-decider-v1.2"  # groups the questions and bills the screenshot once
REQUEST_TIMEOUT_S = 90.0  # a 1,000-tile screenshot answers in about a second; this covers a slow one
RETRY_STATUSES = {429, 500, 502, 503, 504}
MAX_RETRIES = 2  # two more tries for those and for dropped connections, waiting Retry-After seconds (or 1 s, then 2 s)
MAX_RETRY_DELAY_S = 30.0  # if Retry-After asks for more than this, stop with an error instead of waiting

# Six yes-or-no questions about one screenshot. The answer to each is the probability of yes.
SINGLE_QUESTIONS: dict[str, dict[str, Any]] = {
    "text_clipped": {"type": "noul", "instructions": "Is any text clipped, cut off mid-word, or drawn on top of other text?"},
    "elements_overlap": {
        "type": "noul",
        "instructions": "Do any elements such as cards, images, or buttons overlap or cover other content that should be visible?",
    },
    "cta_visible": {
        "type": "noul",
        "instructions": "Is a primary call-to-action button, such as Start free trial, Choose a plan, Get started, or Buy, "
        "fully visible and readable?",
    },
    "low_contrast": {"type": "noul", "instructions": "Is any text hard to read because it has too little contrast with its background?"},
    "layout_broken": {
        "type": "noul",
        "instructions": "Is the layout broken, with elements pushed off-screen, misaligned, stacked wrongly, or a horizontal scrollbar?",
    },
    "error_state": {"type": "noul", "instructions": "Does the page look like a loading, blank, or error state rather than a finished page?"},
}

# Two questions about a before-and-after pair. BEFORE is the baseline, AFTER is the change under review.
PAIR_QUESTIONS: dict[str, dict[str, Any]] = {
    "content_lost": {
        "type": "noul",
        "instructions": "Is content that is present in the BEFORE screenshot (text, buttons, or whole sections) missing or hidden "
        "in the AFTER screenshot?",
    },
    "intentional": {
        "type": "noul",
        "instructions": "Do the differences between BEFORE and AFTER look like a deliberate design change rather than a rendering bug?",
    },
}


class DecisionsError(Exception):
    """The Decisions API returned an error response, or a response the script cannot read."""


def make_client(api_key: str) -> httpx.Client:
    """One HTTP client for the whole run: it carries the key, the timeout, and a reused connection."""
    return httpx.Client(headers={"Authorization": f"Bearer {api_key}"}, timeout=REQUEST_TIMEOUT_S)


def image_part(png: bytes) -> dict[str, Any]:
    """One screenshot as the OpenAI-style image part the Decisions API reads inside state."""
    return {"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(png).decode()}}


def single_body(png: bytes, label: str) -> dict[str, Any]:
    """The request for one screenshot: a caption, the image, and the six questions."""
    return {"model": MODEL, "state": [f"Screenshot of {label}.", image_part(png)], "questions": SINGLE_QUESTIONS}


def pair_body(before: bytes, after: bytes, label: str) -> dict[str, Any]:
    """The request for a before-and-after pair. Order matters: the model reads state top to bottom,
    so each caption comes right before its image."""
    return {
        "model": MODEL,
        "state": [f"Two screenshots of {label}.", "BEFORE (baseline):", image_part(before), "AFTER (change under review):", image_part(after)],
        "questions": PAIR_QUESTIONS,
    }


def ask(client: httpx.Client, body: dict[str, Any]) -> tuple[dict[str, float], int, str | None]:
    """Send one request and return the probability of yes per question, the input tokens, and the request id.
    Retries 429, 5xx, and dropped connections a bounded number of times; raises DecisionsError on any other failure."""
    for attempt in range(MAX_RETRIES + 1):
        try:
            response = client.post(API_URL, json=body)
        except httpx.TransportError:  # dropped connection or read timeout
            if attempt == MAX_RETRIES:
                raise
            time.sleep(float(2**attempt))
            continue
        rid = response.headers.get("x-request-id")
        if response.status_code == 200:
            return _parse(response, body["questions"], rid)
        if response.status_code not in RETRY_STATUSES or attempt == MAX_RETRIES:
            break
        delay = _retry_delay(response, attempt)
        if delay > MAX_RETRY_DELAY_S:
            raise DecisionsError(f"Decisions API asked to wait {delay:.0f} s before retrying ({response.status_code}); stopping instead")
        time.sleep(delay)
    raise DecisionsError(f"Decisions API failed: {response.status_code} {response.text.strip()}" + (f" (x-request-id {rid})" if rid else ""))


def _parse(response: httpx.Response, questions: dict[str, Any], rid: str | None) -> tuple[dict[str, float], int, str | None]:
    """The probabilities out of a 200, checked for every question this script asked. Anything else is a DecisionsError."""
    try:
        data = response.json()
        probabilities = {name: float(data["answers"][name]["noul"]) for name in questions}
        if any(not 0 <= p <= 1 for p in probabilities.values()):
            raise ValueError("probabilities must be finite numbers from 0 to 1")
        return probabilities, int(data["usage"]["input_tokens"]), rid
    except (ValueError, KeyError, TypeError) as exc:
        raise DecisionsError(
            f"Decisions API returned a 200 this script cannot read ({exc!r}): {response.text[:200]!r}" + (f" (x-request-id {rid})" if rid else "")
        ) from exc


def _retry_delay(response: httpx.Response, attempt: int) -> float:
    """Seconds to wait before retrying: the Retry-After header if it is a number, else 1 s then 2 s."""
    try:
        return max(0.0, float(response.headers.get("retry-after", "")))
    except ValueError:
        return float(2**attempt)