abi/screenshot-to-code · error · ValueError

Invalid prediction status response.

Error message

Invalid prediction status response.

What it means

During prediction polling in the Replicate client, _poll_prediction raises ValueError("Invalid prediction status response") when the GET to /predictions/{id} succeeds at the HTTP level but the parsed JSON body is not an object (dict). This guards against HTML error pages from proxies, JSON arrays, or string bodies that httpx's .json() happily parses but that carry no `status` field.

Source

Thrown at backend/image_generation/replicate.py:61

def _extract_prediction_id(response_json: Mapping[str, Any]) -> str:
    prediction_id = response_json.get("id")
    if not isinstance(prediction_id, str) or not prediction_id:
        raise ValueError("Prediction ID not found in initial response.")
    return prediction_id


async def _poll_prediction(
    client: httpx.AsyncClient, prediction_id: str, headers: dict[str, str]
) -> dict[str, Any]:
    status_check_url = f"{REPLICATE_API_BASE_URL}/predictions/{prediction_id}"

    for _ in range(MAX_POLLS):
        await asyncio.sleep(POLL_INTERVAL_SECONDS)
        status_response = await client.get(status_check_url, headers=headers)
        status_response.raise_for_status()
        status_response_raw: Any = status_response.json()
        if not isinstance(status_response_raw, dict):
            raise ValueError("Invalid prediction status response.")
        status_response_json = cast(dict[str, Any], status_response_raw)

        status = status_response_json.get("status")
        if status == "succeeded":
            return cast(dict[str, Any], status_response_json)
        if status == "error":
            error_message = str(status_response_json.get("error", "Unknown error"))
            raise ValueError(f"Inference errored out: {error_message}")
        if status == "failed":
            raise ValueError("Inference failed")

    raise TimeoutError("Inference timed out")


async def _run_prediction(
    endpoint_url: str, payload: dict[str, Any], api_token: str
) -> Any:
    headers = _build_headers(api_token)

View on GitHub (pinned to d026163f58)

Solutions

1. Capture and log status_response.text when the body is not a dict to identify the interceptor.
2. Ensure direct egress to api.replicate.com without a rewriting proxy.
3. If the shape legitimately changed, adapt the isinstance check to the new envelope.

from typing import Any, Mapping

def is_prediction_status_payload(body: Any) -> bool:
    return isinstance(body, Mapping) and "status" in body

try:
    output = await call_replicate_model(model, input, token)
except ValueError as e:
    if "Invalid prediction status response" in str(e):
        log.error("Polling got a non-dict body — proxy interference likely")
    raise

When it happens

Trigger: Polling GET /predictions/{id} returns 200 with body "null", a JSON array, a bare string/number, or a proxy-injected HTML page that happens to parse as JSON scalar. Any non-dict JSON triggers it immediately, aborting the poll loop.

Common situations: Reverse proxies or CDNs intercepting long-running polls; Replicate returning an unexpected envelope during incidents; API version drift.

Related errors

• Prediction ID not found in initial response.
• Invalid prediction creation response.
• Inference errored out: {error_message}
• Inference timed out
• HTTP error occurred: {exc}

AI-assisted analysis of abi/screenshot-to-code@d026163f58 (2026-08-14). Data as JSON: /api/errors/ee62f2af9a5e5a7f. Report an issue: GitHub.