apify-client 0.14.0

//! Offline unit tests for the HTTP layer (retries, backoff, error parsing, 404 handling).
//!
//! These use a mock [`HttpBackend`] and require no network access or `APIFY_TOKEN`, so the
//! most bug-prone logic is always exercised in CI.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use apify_client::http_client::{HttpBackend, HttpRequest, HttpResponse};
use apify_client::{
    ActorStartOptions, ApifyClient, ApifyClientError, LastRunOptions, RequestCompression,
    WaitForResources,
};
use async_trait::async_trait;
use futures_util::StreamExt;

/// A scripted backend that returns a queued sequence of responses and counts calls.
#[derive(Debug)]
struct MockBackend {
    responses: Mutex<Vec<MockOutcome>>,
    calls: AtomicUsize,
    last_url: Mutex<Option<String>>,
    /// Every request URL in call order, so tests can assert per-page pagination wiring.
    urls: Mutex<Vec<String>>,
    last_headers: Mutex<std::collections::HashMap<String, String>>,
    last_body: Mutex<Option<Vec<u8>>>,
}

#[derive(Debug, Clone)]
enum MockOutcome {
    Status(u16, Vec<u8>),
    NetworkError,
}

impl MockBackend {
    fn new(responses: Vec<MockOutcome>) -> Arc<Self> {
        Arc::new(Self {
            responses: Mutex::new(responses),
            calls: AtomicUsize::new(0),
            last_url: Mutex::new(None),
            urls: Mutex::new(Vec::new()),
            last_headers: Mutex::new(std::collections::HashMap::new()),
            last_body: Mutex::new(None),
        })
    }

    fn call_count(&self) -> usize {
        self.calls.load(Ordering::SeqCst)
    }

    fn last_url(&self) -> Option<String> {
        self.last_url.lock().unwrap().clone()
    }

    /// All request URLs in call order.
    fn urls(&self) -> Vec<String> {
        self.urls.lock().unwrap().clone()
    }

    /// Case-insensitive lookup of the last request's header value.
    fn last_header(&self, name: &str) -> Option<String> {
        let headers = self.last_headers.lock().unwrap();
        headers
            .iter()
            .find(|(k, _)| k.eq_ignore_ascii_case(name))
            .map(|(_, v)| v.clone())
    }

    fn last_body(&self) -> Option<Vec<u8>> {
        self.last_body.lock().unwrap().clone()
    }
}

#[async_trait]
impl HttpBackend for MockBackend {
    async fn send(&self, request: HttpRequest) -> Result<HttpResponse, ApifyClientError> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        *self.last_url.lock().unwrap() = Some(request.url.clone());
        self.urls.lock().unwrap().push(request.url.clone());
        *self.last_headers.lock().unwrap() = request.headers.clone();
        *self.last_body.lock().unwrap() = request.body.clone();
        let mut queue = self.responses.lock().unwrap();
        let outcome = if queue.len() > 1 {
            queue.remove(0)
        } else {
            // Repeat the last scripted outcome for any further calls.
            queue
                .first()
                .cloned()
                .expect("at least one scripted response")
        };
        match outcome {
            MockOutcome::Status(status, body) => Ok(HttpResponse {
                status,
                headers: Default::default(),
                body,
            }),
            MockOutcome::NetworkError => Err(ApifyClientError::Http("boom".to_string())),
        }
    }
}

fn client_with(backend: Arc<MockBackend>, max_retries: u32) -> ApifyClient {
    ApifyClient::builder()
        .token("test-token")
        .max_retries(max_retries)
        // Keep backoff tiny so tests run fast.
        .min_delay_between_retries(Duration::from_millis(1))
        .http_backend(backend)
        .build()
}

/// Builds a client with an explicit request-compression algorithm and no retries.
fn client_with_compression(
    backend: Arc<MockBackend>,
    compression: RequestCompression,
) -> ApifyClient {
    ApifyClient::builder()
        .token("test-token")
        .max_retries(0)
        .min_delay_between_retries(Duration::from_millis(1))
        .request_compression(compression)
        .http_backend(backend)
        .build()
}

/// A successful GET unwraps the `data` envelope and makes exactly one call.
#[tokio::test]
async fn success_single_call() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"abc","username":"u"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 3);
    let user = client.me().get().await.expect("ok").expect("some");
    assert_eq!(user.id, "abc");
    assert_eq!(backend.call_count(), 1);
}

/// A `429` is retried up to `max_retries + 1` total attempts, then surfaces the API error.
#[tokio::test]
async fn rate_limit_is_retried() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        429,
        br#"{"error":{"type":"rate-limit-exceeded","message":"slow down"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 2);
    let err = client.me().get().await.expect_err("should fail");
    assert_eq!(backend.call_count(), 3, "1 initial + 2 retries");
    let api = err.as_api_error().expect("api error");
    assert_eq!(api.status_code, 429);
    assert_eq!(api.error_type.as_deref(), Some("rate-limit-exceeded"));
    assert_eq!(api.attempt, 3);
}

/// A `5xx` is retried.
#[tokio::test]
async fn server_error_is_retried() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        503,
        br#"{"error":{"type":"server-error","message":"oops"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 1);
    let _ = client.me().get().await.expect_err("should fail");
    assert_eq!(backend.call_count(), 2, "1 initial + 1 retry");
}

/// A non-429 `4xx` is NOT retried (caller error).
#[tokio::test]
async fn client_error_not_retried() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        400,
        br#"{"error":{"type":"invalid-request","message":"bad"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 5);
    let err = client.me().get().await.expect_err("should fail");
    assert_eq!(backend.call_count(), 1, "4xx must not be retried");
    assert_eq!(err.status_code(), Some(400));
}

/// Network errors are retried, then surface.
#[tokio::test]
async fn network_error_is_retried() {
    let backend = MockBackend::new(vec![MockOutcome::NetworkError]);
    let client = client_with(backend.clone(), 3);
    let err = client.me().get().await.expect_err("should fail");
    assert_eq!(backend.call_count(), 4, "1 initial + 3 retries");
    assert!(matches!(err, ApifyClientError::Http(_)));
}

/// Retries eventually succeed when a later attempt returns 200.
#[tokio::test]
async fn retry_then_success() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(500, b"{}".to_vec()),
        MockOutcome::Status(500, b"{}".to_vec()),
        MockOutcome::Status(200, br#"{"data":{"id":"ok"}}"#.to_vec()),
    ]);
    let client = client_with(backend.clone(), 5);
    let user = client.me().get().await.expect("ok").expect("some");
    assert_eq!(user.id, "ok");
    assert_eq!(backend.call_count(), 3);
}

/// A `404 record-not-found` on a `get` maps to `Ok(None)`, not an error.
#[tokio::test]
async fn not_found_maps_to_none() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        404,
        br#"{"error":{"type":"record-not-found","message":"missing"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 3);
    let actor = client.actor("nope").get().await.expect("ok");
    assert!(actor.is_none(), "404 record-not-found should be Ok(None)");
    assert_eq!(backend.call_count(), 1, "404 is not retried");
}

/// The error body is parsed into the structured `ApiError` fields.
#[tokio::test]
async fn error_body_is_parsed() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        400,
        br#"{"error":{"type":"invalid-value","message":"Comments required"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);
    let err = client.me().get().await.expect_err("should fail");
    let api = err.as_api_error().expect("api error");
    assert_eq!(api.status_code, 400);
    assert_eq!(api.error_type.as_deref(), Some("invalid-value"));
    assert_eq!(api.message, "Comments required");
    assert_eq!(api.http_method.as_deref(), Some("GET"));
    assert!(api.path.as_deref().unwrap().contains("/users/me"));
}

/// The API returns `null` for some fields (`actVersion.buildTag`, `restartOnError`,
/// `isOneTimeEvent`); such a response must deserialize, with the value kept in the model's
/// `extra` map.
#[tokio::test]
async fn null_valued_fields_still_deserialize() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"b1","actId":"a1","status":"SUCCEEDED","actVersion":{"versionNumber":"0.1","buildTag":null}}}"#
            .to_vec(),
    )]);
    let client = client_with(backend, 0);
    let build = client
        .build("b1")
        .get()
        .await
        .expect("a build with a null buildTag should deserialize")
        .expect("build should exist");
    assert_eq!(
        build.extra["actVersion"].get("buildTag"),
        Some(&serde_json::Value::Null)
    );

    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"a1","restartOnError":null,"pricingInfos":[{"pricingModel":"PAY_PER_EVENT","pricingPerEvent":{"actorChargeEvents":{"e":{"eventTitle":"E","isOneTimeEvent":null}}}}]}}"#
            .to_vec(),
    )]);
    let client = client_with(backend, 0);
    let actor = client
        .actor("a1")
        .get()
        .await
        .expect("an Actor with null fields should deserialize")
        .expect("actor should exist");
    assert_eq!(
        actor.extra.get("restartOnError"),
        Some(&serde_json::Value::Null)
    );
    assert_eq!(
        actor.extra["pricingInfos"][0]["pricingPerEvent"]["actorChargeEvents"]["e"]
            .get("isOneTimeEvent"),
        Some(&serde_json::Value::Null)
    );
}

/// A successful response whose body does not parse is reported on the first attempt: the body
/// arrived in full, so a retry would receive the same body.
#[tokio::test]
async fn unparsable_success_body_is_not_retried() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        b"<html>oops</html>".to_vec(),
    )]);
    let client = client_with(backend.clone(), 3);
    let err = client.me().get().await.expect_err("should fail to parse");
    assert!(
        matches!(err, ApifyClientError::Serde(_)),
        "expected a deserialization error, got {err:?}"
    );
    assert_eq!(
        backend.call_count(),
        1,
        "an unparsable success body is not retried"
    );
}

/// An error status whose body is not the API's JSON error (such as an HTML page from a proxy) is
/// reported as an `ApiError` carrying the raw body, and is retried by its status code.
#[tokio::test]
async fn unparsable_error_body_is_retried_by_status_code() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        502,
        b"<html>Bad Gateway</html>".to_vec(),
    )]);
    let client = client_with(backend.clone(), 2);
    let err = client.me().get().await.expect_err("should fail");
    let api = err.as_api_error().expect("api error");
    assert_eq!(api.status_code, 502);
    assert!(
        api.message.contains("Bad Gateway"),
        "raw body is kept in the message"
    );
    assert_eq!(
        backend.call_count(),
        3,
        "a 502 is retried until the retries run out"
    );
}

/// `ApiError`'s `is_*` status-classification predicates (the idiomatic equivalent of the
/// reference client's `NotFoundError`/`RateLimitError`/... subclasses) match the response status,
/// and only that status's predicate is `true`.
#[tokio::test]
async fn api_error_status_predicates_classify_by_status_code() {
    async fn error_for(status: u16) -> apify_client::ApiError {
        let backend = MockBackend::new(vec![MockOutcome::Status(status, b"{}".to_vec())]);
        let client = client_with(backend, 0);
        client
            .me()
            .get()
            .await
            .expect_err("should fail")
            .as_api_error()
            .expect("api error")
            .clone()
    }

    let not_found = error_for(404).await;
    assert!(not_found.is_not_found());
    assert!(!not_found.is_invalid_request());
    assert!(!not_found.is_unauthorized());
    assert!(!not_found.is_forbidden());
    assert!(!not_found.is_conflict());
    assert!(!not_found.is_rate_limited());
    assert!(!not_found.is_server_error());

    assert!(error_for(400).await.is_invalid_request());
    assert!(error_for(401).await.is_unauthorized());
    assert!(error_for(403).await.is_forbidden());
    assert!(error_for(409).await.is_conflict());
    assert!(error_for(429).await.is_rate_limited());
    assert!(error_for(500).await.is_server_error());
    assert!(error_for(503).await.is_server_error());
}

/// `RunClient::metamorph_raw` sends the given bytes verbatim (no JSON serialization), defaulting
/// to `application/octet-stream` when the caller does not set `options.content_type`, and still
/// forwards `targetActorId`/`build` as query parameters like `metamorph`.
#[tokio::test]
async fn metamorph_raw_sends_bytes_verbatim_with_default_content_type() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"run1"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let input = b"raw metamorph input".to_vec();
    client
        .run("run1")
        .metamorph_raw(
            "other/actor",
            &input,
            apify_client::RunMetamorphOptions {
                build: Some("1.0".to_string()),
                ..Default::default()
            },
        )
        .await
        .expect("ok");

    assert_eq!(
        backend.last_header("Content-Type").as_deref(),
        Some("application/octet-stream")
    );
    assert_eq!(backend.last_body().expect("a body was sent"), input);
    let url = backend.last_url().expect("a url was sent");
    assert!(url.contains("targetActorId=other~actor"));
    assert!(url.contains("build=1.0"));
}

/// `DatasetClient::create_items_public_url_with_format` adds a `format` query parameter to the
/// generated URL, and `create_items_public_url` (unchanged) omits it, defaulting to the
/// endpoint's own `json` default.
#[tokio::test]
async fn create_items_public_url_with_format_adds_format_param() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"ds1"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let url = client
        .dataset("ds1")
        .create_items_public_url_with_format(
            Default::default(),
            None,
            Some(apify_client::DownloadItemsFormat::Csv),
        )
        .await
        .expect("ok");
    assert!(url.contains("format=csv"), "url was: {url}");

    let url = client
        .dataset("ds1")
        .create_items_public_url(Default::default(), None)
        .await
        .expect("ok");
    assert!(
        !url.contains("format="),
        "create_items_public_url must not set a format param, url was: {url}"
    );
}

/// `max_retries = 0` means exactly one attempt.
#[tokio::test]
async fn zero_retries_single_attempt() {
    let backend = MockBackend::new(vec![MockOutcome::Status(500, b"{}".to_vec())]);
    let client = client_with(backend.clone(), 0);
    let _ = client.me().get().await.expect_err("should fail");
    assert_eq!(backend.call_count(), 1);
}

/// Regression guard for the validate-input envelope-skip fix: the `validate-input` endpoint
/// returns a **bare** `{"valid":...}` object with no `{"data":...}` envelope, so it must NOT be
/// routed through `parse_data_envelope` (which would fail with `missing field 'data'`). This is
/// token-free, so a future refactor that re-introduces envelope unwrapping is caught even in a
/// run without `APIFY_TOKEN`.
#[tokio::test]
async fn validate_input_does_not_unwrap_data_envelope() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"valid":true}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 3);
    let result = client
        .actor("me~some-actor")
        .validate_input(&serde_json::json!({}))
        .await
        .expect("validate_input should parse a bare {valid} body");
    assert_eq!(
        result.get("valid").and_then(|v| v.as_bool()),
        Some(true),
        "the bare body must be returned verbatim, not unwrapped from a `data` envelope"
    );
    assert_eq!(backend.call_count(), 1);
}

/// `validate_input_for_build` sends the spec's optional `build` query parameter, and the
/// no-arg `validate_input` omits it.
#[tokio::test]
async fn validate_input_sends_build_query_param() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"valid":true}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    // With a build tag -> `build=latest` present in the URL.
    client
        .actor("me~some-actor")
        .validate_input_for_build(&serde_json::json!({}), Some("latest"))
        .await
        .expect("ok");
    let url = backend.last_url().expect("a request was sent");
    assert!(
        url.contains("/validate-input") && url.contains("build=latest"),
        "expected build=latest in {url}"
    );

    // Without a build -> no `build` param.
    client
        .actor("me~some-actor")
        .validate_input(&serde_json::json!({}))
        .await
        .expect("ok");
    let url = backend.last_url().expect("a request was sent");
    assert!(
        !url.contains("build="),
        "default validate_input must not send a build param, got {url}"
    );
}

/// `LogClient::get_with_options` sends the spec's optional `raw` query parameter, and the
/// no-arg `get` omits it.
#[tokio::test]
async fn log_get_sends_raw_query_param() {
    let backend = MockBackend::new(vec![MockOutcome::Status(200, b"some log output".to_vec())]);
    let client = client_with(backend.clone(), 0);

    // With raw -> `raw=1` present in the run log URL.
    client
        .run("some-run-id")
        .log()
        .get_with_options(apify_client::LogOptions { raw: Some(true) })
        .await
        .expect("ok");
    let url = backend.last_url().expect("a request was sent");
    assert!(
        url.contains("/log") && url.contains("raw=1"),
        "expected raw=1 on the run log request, got {url}"
    );

    // Without options -> no `raw` param.
    client.run("some-run-id").log().get().await.expect("ok");
    let url = backend.last_url().expect("a request was sent");
    assert!(
        !url.contains("raw="),
        "default log get must not send a raw param, got {url}"
    );
}

/// `log().stream()` goes through the configured `HttpBackend` — including a fully custom one,
/// not just `ReqwestBackend` — and applies the same auth/user-agent headers as a buffered call.
/// `MockBackend` does not override `send_streaming`, so this also exercises
/// `HttpBackend::send_streaming`'s default (buffer-via-`send`-and-yield-one-chunk) fallback.
#[tokio::test]
async fn log_stream_goes_through_the_configured_backend() {
    let backend = MockBackend::new(vec![MockOutcome::Status(200, b"some log output".to_vec())]);
    let client = client_with(backend.clone(), 0);

    let mut stream = client
        .run("some-run-id")
        .log()
        .stream()
        .await
        .expect("stream should open");
    let mut collected = Vec::new();
    while let Some(chunk) = stream.next().await {
        collected.extend(chunk.expect("chunk"));
    }

    assert_eq!(
        collected, b"some log output",
        "the mock backend's response must reach the stream consumer"
    );
    assert_eq!(
        backend.last_header("Authorization").as_deref(),
        Some("Bearer test-token"),
        "a streaming request must carry the same bearer auth as a buffered one"
    );
    assert!(
        backend.last_header("User-Agent").is_some(),
        "a streaming request must carry the User-Agent header"
    );
}

/// A failed `log().stream()` through a custom `HttpBackend` (not just `ReqwestBackend`) still
/// surfaces the same structured `ApiError` a buffered call would, built from the mock's 404
/// response — proving the error path also goes through the configured backend.
#[tokio::test]
async fn log_stream_through_mock_backend_surfaces_structured_error() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        404,
        br#"{"error":{"type":"record-not-found","message":"Run was not found"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let err = client
        .run("missing-run-id")
        .log()
        .stream()
        .await
        .map(|_| ())
        .expect_err("streaming a 404 should fail");

    let api_error = err
        .as_api_error()
        .expect("error should be a structured ApiError");
    assert_eq!(api_error.status_code, 404);
    assert_eq!(api_error.error_type.as_deref(), Some("record-not-found"));
}

/// A request body at or above the 1024-byte threshold is brotli-compressed: the backend sees a
/// `Content-Encoding: br` header and a body that is smaller than (and different from) the original
/// and decodes back to the original bytes.
#[tokio::test]
async fn large_request_body_is_brotli_compressed() {
    let backend = MockBackend::new(vec![MockOutcome::Status(200, b"".to_vec())]);
    let client = client_with(backend.clone(), 0);

    // A highly compressible 4 KiB payload, comfortably over the 1024-byte threshold.
    let original = vec![b'a'; 4096];
    client
        .key_value_store("me~store")
        .set_record_raw("key", original.clone(), "text/plain")
        .await
        .expect("ok");

    assert_eq!(
        backend.last_header("Content-Encoding").as_deref(),
        Some("br"),
        "large bodies must be sent with Content-Encoding: br"
    );
    let sent = backend.last_body().expect("a body was sent");
    assert!(
        sent.len() < original.len(),
        "compressed body ({}) must be smaller than the original ({})",
        sent.len(),
        original.len()
    );
    assert_ne!(sent, original, "the sent body must actually be encoded");

    // The sent bytes must decode back to the original via brotli.
    let mut decompressed = Vec::new();
    let mut decoder = brotli::DecompressorWriter::new(&mut decompressed, 4096);
    std::io::Write::write_all(&mut decoder, &sent).expect("decode");
    drop(decoder);
    assert_eq!(decompressed, original, "brotli round-trip must be lossless");
}

/// With the gzip encoding selected, a large request body is gzip-compressed: the backend sees a
/// `Content-Encoding: gzip` header and a body that is smaller than (and different from) the original
/// and decodes back to the original bytes. This exercises the gzip path end-to-end.
#[tokio::test]
async fn large_request_body_is_gzip_compressed_when_selected() {
    let backend = MockBackend::new(vec![MockOutcome::Status(200, b"".to_vec())]);
    let client = client_with_compression(backend.clone(), RequestCompression::Gzip);

    // A highly compressible 4 KiB payload, comfortably over the 1024-byte threshold.
    let original = vec![b'a'; 4096];
    client
        .key_value_store("me~store")
        .set_record_raw("key", original.clone(), "text/plain")
        .await
        .expect("ok");

    assert_eq!(
        backend.last_header("Content-Encoding").as_deref(),
        Some("gzip"),
        "large bodies must be sent with Content-Encoding: gzip when gzip is selected"
    );
    let sent = backend.last_body().expect("a body was sent");
    assert!(
        sent.len() < original.len(),
        "compressed body ({}) must be smaller than the original ({})",
        sent.len(),
        original.len()
    );
    assert_ne!(sent, original, "the sent body must actually be encoded");

    // The sent bytes must decode back to the original via gzip.
    let mut decoder = flate2::read::GzDecoder::new(&sent[..]);
    let mut decompressed = Vec::new();
    std::io::Read::read_to_end(&mut decoder, &mut decompressed).expect("decode");
    assert_eq!(decompressed, original, "gzip round-trip must be lossless");
}

/// When `request_compression()` is never called on the builder, large bodies are brotli-compressed:
/// the builder's default encoding is brotli, so the backend sees `Content-Encoding: br`. This
/// exercises the builder default via the public builder path (not the `Default` derive), guarding
/// against the default silently changing.
#[tokio::test]
async fn builder_default_request_compression_is_brotli() {
    let backend = MockBackend::new(vec![MockOutcome::Status(200, b"".to_vec())]);
    // `client_with` builds a client without ever calling `.request_compression(...)`.
    let client = client_with(backend.clone(), 0);

    let original = vec![b'a'; 4096];
    client
        .key_value_store("me~store")
        .set_record_raw("key", original.clone(), "text/plain")
        .await
        .expect("ok");

    assert_eq!(
        backend.last_header("Content-Encoding").as_deref(),
        Some("br"),
        "the default compression must be brotli"
    );
}

/// A request body below the 1024-byte threshold is sent verbatim, with no `Content-Encoding`.
#[tokio::test]
async fn small_request_body_is_not_compressed() {
    let backend = MockBackend::new(vec![MockOutcome::Status(200, b"".to_vec())]);
    let client = client_with(backend.clone(), 0);

    let original = vec![b'a'; 100];
    client
        .key_value_store("me~store")
        .set_record_raw("key", original.clone(), "text/plain")
        .await
        .expect("ok");

    assert!(
        backend.last_header("Content-Encoding").is_none(),
        "small bodies must not be compressed"
    );
    assert_eq!(
        backend.last_body().expect("a body was sent"),
        original,
        "small bodies must be sent verbatim"
    );
}

/// A large body whose content type already carries its own compression (e.g. a ZIP archive) is
/// sent uncompressed, with no `Content-Encoding`: compressing it again would burn CPU for little
/// to no size reduction. Exercises the raw-bytes path via `ActorClient::start_raw`, which is the
/// main way a caller sets a non-JSON content type on a large body.
#[tokio::test]
async fn already_compressed_content_type_is_not_compressed() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"run1"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let original = vec![b'a'; 4096];
    client
        .actor("me/actor")
        .start_raw(
            &original,
            apify_client::ActorStartOptions {
                content_type: Some("application/zip".to_string()),
                ..Default::default()
            },
        )
        .await
        .expect("ok");

    assert!(
        backend.last_header("Content-Encoding").is_none(),
        "an already-compressed content type must not be compressed"
    );
    assert_eq!(
        backend.last_body().expect("a body was sent"),
        original,
        "an already-compressed body must be sent verbatim, not JSON-re-encoded"
    );
}

/// `set_record_raw_with_content_encoding` sends the caller's `Content-Encoding` verbatim and
/// does not compress the body a second time, even above the auto-compression size threshold:
/// `maybe_compress_request` skips any body that already carries a `Content-Encoding` header.
#[tokio::test]
async fn set_record_with_content_encoding_is_not_compressed_again() {
    let backend = MockBackend::new(vec![MockOutcome::Status(201, b"{}".to_vec())]);
    let client = client_with(backend.clone(), 0);

    // Pretend this is already-gzipped data; it does not need to actually be valid gzip for this
    // test, which only checks that the client leaves it untouched and sets the right header.
    let pre_compressed = [0x1fu8, 0x8b, 0x08, 0x00].repeat(512); // well over the 1024-byte threshold
    client
        .key_value_store("me~store")
        .set_record_raw_with_content_encoding(
            "key",
            pre_compressed.clone(),
            "application/octet-stream",
            "gzip",
        )
        .await
        .expect("ok");

    assert_eq!(
        backend.last_header("Content-Encoding").as_deref(),
        Some("gzip"),
        "the caller's Content-Encoding must be forwarded verbatim"
    );
    assert_eq!(
        backend.last_body().expect("a body was sent"),
        pre_compressed,
        "a pre-compressed body must be sent byte-for-byte, not compressed again"
    );
}

/// `image/bmp` sits under the already-compressed `image/` prefix but is itself raw, uncompressed
/// data, so it is still compressed. Exercises the exact-media-type exception list.
#[tokio::test]
async fn compressible_exception_under_already_compressed_prefix_is_compressed() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"run1"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let original = vec![b'a'; 4096];
    client
        .actor("me/actor")
        .start_raw(
            &original,
            apify_client::ActorStartOptions {
                content_type: Some("image/bmp".to_string()),
                ..Default::default()
            },
        )
        .await
        .expect("ok");

    assert_eq!(
        backend.last_header("Content-Encoding").as_deref(),
        Some("br"),
        "image/bmp is raw despite the image/ prefix and should still be compressed"
    );
}

/// `ActorClient::start_raw` defaults to `application/octet-stream` when the caller does not set
/// `options.content_type`, and sends the bytes verbatim (not as a JSON-encoded byte array).
#[tokio::test]
async fn start_raw_defaults_content_type_and_sends_bytes_verbatim() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"run1"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let input = b"not json, just bytes: \x00\x01\x02".to_vec();
    client
        .actor("me/actor")
        .start_raw(&input, Default::default())
        .await
        .expect("ok");

    assert_eq!(
        backend.last_header("Content-Type").as_deref(),
        Some("application/octet-stream")
    );
    assert_eq!(backend.last_body().expect("a body was sent"), input);
}

/// The status-only `last_run(status)` and the options-based `last_run_with_options` thread the
/// optional `status` and `origin` filters into the last-run request as query parameters, matching
/// the JS reference's `lastRun({ status, origin })`. Leaving a filter as `None` omits it.
#[tokio::test]
async fn last_run_sends_status_and_origin_query_params() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"id":"run1","status":"SUCCEEDED"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    // The non-breaking status-only convenience sends just `status`.
    client
        .actor("me~some-actor")
        .last_run(Some("SUCCEEDED"))
        .get()
        .await
        .expect("ok");
    let url = backend.last_url().expect("a request was sent");
    assert!(
        url.contains("/runs/last") && url.contains("status=SUCCEEDED"),
        "expected status=SUCCEEDED on the actor last-run request, got {url}"
    );
    assert!(
        !url.contains("origin="),
        "last_run(Some(..)) must not send an origin param, got {url}"
    );

    // Both filters set via options on the actor last run -> both query params present.
    client
        .actor("me~some-actor")
        .last_run_with_options(LastRunOptions {
            status: Some("SUCCEEDED".to_owned()),
            origin: Some("API".to_owned()),
        })
        .get()
        .await
        .expect("ok");
    let url = backend.last_url().expect("a request was sent");
    assert!(
        url.contains("/runs/last")
            && url.contains("status=SUCCEEDED")
            && url.contains("origin=API"),
        "expected status=SUCCEEDED and origin=API on the actor last-run request, got {url}"
    );

    // Only `origin` set via options on the task last run -> origin present, status absent.
    client
        .task("me~some-task")
        .last_run_with_options(LastRunOptions {
            status: None,
            origin: Some("SCHEDULER".to_owned()),
        })
        .get()
        .await
        .expect("ok");
    let url = backend.last_url().expect("a request was sent");
    assert!(
        url.contains("/runs/last") && url.contains("origin=SCHEDULER"),
        "expected origin=SCHEDULER on the task last-run request, got {url}"
    );
    assert!(
        !url.contains("status="),
        "task last_run_with_options without status must not send a status param, got {url}"
    );

    // Neither filter set -> no `status`/`origin` params.
    client
        .actor("me~some-actor")
        .last_run(None)
        .get()
        .await
        .expect("ok");
    let url = backend.last_url().expect("a request was sent");
    assert!(
        !url.contains("status=") && !url.contains("origin="),
        "default last_run must not send status/origin params, got {url}"
    );
}

/// `iterate_items` must keep paging even when the dataset-items response omits the
/// `X-Apify-Pagination-Total` header. With the header absent, `list_items` reports `total = 0`
/// ("unknown"), so the iterator falls through to the short-page/empty-page backstop and walks
/// every page. The `MockBackend` returns no headers, reproducing the missing-total case. Before
/// the fix, `list_items` fell back to `total = count`, which the iterator read as a completed
/// total after the first full page (`next_offset == total`) and silently dropped every later
/// item — this test scripts three data pages then an empty one and asserts all items are yielded.
#[tokio::test]
async fn iterate_items_without_total_header_walks_all_pages() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(200, br#"[{"i":0},{"i":1}]"#.to_vec()),
        MockOutcome::Status(200, br#"[{"i":2},{"i":3}]"#.to_vec()),
        MockOutcome::Status(200, br#"[{"i":4}]"#.to_vec()),
        MockOutcome::Status(200, b"[]".to_vec()),
    ]);
    let client = client_with(backend.clone(), 0);

    let mut it = client
        .dataset("some-dataset-id")
        .iterate_items::<serde_json::Value>(Default::default());
    let mut seen = Vec::new();
    while let Some(item) = it.next().await.expect("ok") {
        seen.push(item["i"].as_i64().expect("i field"));
    }

    assert_eq!(
        seen,
        vec![0, 1, 2, 3, 4],
        "all items must be yielded even without a total header"
    );
    assert_eq!(
        backend.call_count(),
        4,
        "three data pages plus the terminating empty page"
    );
}

/// Hermetic coverage of the `list_iterator!` macro wiring used by every offset/limit collection
/// `iterate()`. A total-item cap plus a smaller `with_chunk_size` must (a) yield exactly the cap
/// and (b) drive the per-page request window correctly: the first page requests
/// `min(remaining, chunk_size)` at `offset=0`, and the second advances `offset` by the items
/// received and requests only the remaining budget. Previously this path was exercised only by
/// `APIFY_TOKEN`-gated integration tests, so it was skipped offline.
#[tokio::test]
async fn iterate_macro_caps_and_pages_correctly() {
    let backend = MockBackend::new(vec![
        // Endpoint reports a large total (10) so termination is driven by the cap, not by total.
        MockOutcome::Status(
            200,
            br#"{"data":{"total":10,"items":[{"id":"a0"},{"id":"a1"}]}}"#.to_vec(),
        ),
        MockOutcome::Status(
            200,
            br#"{"data":{"total":10,"items":[{"id":"a2"}]}}"#.to_vec(),
        ),
    ]);
    let client = client_with(backend.clone(), 0);

    let mut it = client
        .actors()
        .iterate(apify_client::ActorListOptions {
            limit: Some(3),
            ..Default::default()
        })
        .with_chunk_size(2);
    let mut ids = Vec::new();
    while let Some(actor) = it.next().await.expect("ok") {
        ids.push(actor.id);
    }

    assert_eq!(ids, vec!["a0", "a1", "a2"], "cap of 3 must yield exactly 3");
    assert_eq!(backend.call_count(), 2, "two pages suffice under the cap");
    let urls = backend.urls();
    assert!(
        urls[0].contains("offset=0") && urls[0].contains("limit=2"),
        "first page requests min(remaining=3, chunk=2)=2 at offset 0, got {}",
        urls[0]
    );
    assert!(
        urls[1].contains("offset=2") && urls[1].contains("limit=1"),
        "second page advances to offset 2 and requests the remaining budget of 1, got {}",
        urls[1]
    );
}

/// Hermetic coverage of `RunCollectionClient::iterate`, which builds its iterator directly
/// (not via the `list_iterator!` macro) because it threads a separate `filter` argument. The
/// iterator must walk every page — ignoring the reported `total`, which can grow as more runs
/// start — until a page scans no rows, and must forward the `status` filter on every page
/// request. Offline-only; the integration suite gates this on a live token.
#[tokio::test]
async fn run_collection_iterate_walks_pages_and_forwards_filter() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(
            200,
            br#"{"data":{"total":3,"limit":2,"items":[{"id":"r0"},{"id":"r1"}]}}"#.to_vec(),
        ),
        MockOutcome::Status(
            200,
            br#"{"data":{"total":3,"limit":2,"items":[{"id":"r2"}]}}"#.to_vec(),
        ),
        MockOutcome::Status(
            200,
            br#"{"data":{"total":3,"limit":2,"items":[]}}"#.to_vec(),
        ),
    ]);
    let client = client_with(backend.clone(), 0);

    let mut it = client.runs().iterate(
        Default::default(),
        apify_client::RunListOptions {
            status: vec!["SUCCEEDED".to_owned()],
            ..Default::default()
        },
    );
    let mut ids = Vec::new();
    while let Some(run) = it.next().await.expect("ok") {
        ids.push(run.id);
    }

    assert_eq!(
        ids,
        vec!["r0", "r1", "r2"],
        "all runs across pages must be yielded"
    );
    assert_eq!(
        backend.call_count(),
        3,
        "two data pages then the terminating empty page; `total` is never consulted"
    );
    for url in backend.urls() {
        assert!(
            url.contains("status=SUCCEEDED"),
            "the status filter must be forwarded on every page, got {url}"
        );
    }
}

/// Hermetic coverage of the cursor-based `KeyValueStoreClient::iterate_keys`. Key-value stores
/// paginate by `exclusiveStartKey`/`nextExclusiveStartKey` (not offset), so this walks two pages
/// and asserts: every key is yielded, the second request carries the previous page's
/// `nextExclusiveStartKey` as `exclusiveStartKey`, and iteration stops once the API returns a
/// null next cursor.
#[tokio::test]
async fn iterate_keys_walks_cursor_pages() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(
            200,
            br#"{"data":{"items":[{"key":"k0"},{"key":"k1"}],"isTruncated":true,"nextExclusiveStartKey":"k1"}}"#.to_vec(),
        ),
        MockOutcome::Status(
            200,
            br#"{"data":{"items":[{"key":"k2"}],"isTruncated":false,"nextExclusiveStartKey":null}}"#.to_vec(),
        ),
    ]);
    let client = client_with(backend.clone(), 0);

    let mut it = client
        .key_value_store("some-store")
        .iterate_keys(Default::default());
    let mut keys = Vec::new();
    while let Some(key) = it.next().await.expect("ok") {
        keys.push(key.key);
    }

    assert_eq!(
        keys,
        vec!["k0", "k1", "k2"],
        "all keys across pages must be yielded"
    );
    assert_eq!(backend.call_count(), 2, "two pages then null-cursor stop");
    let urls = backend.urls();
    assert!(
        !urls[0].contains("exclusiveStartKey="),
        "first page has no start cursor, got {}",
        urls[0]
    );
    assert!(
        urls[1].contains("exclusiveStartKey=k1"),
        "second page must carry the previous nextExclusiveStartKey as the cursor, got {}",
        urls[1]
    );
}

/// `iterate_keys` terminates on the model's `isTruncated == false` flag (the authoritative
/// end-of-data signal), not merely on a null cursor. If the API sends the final page with
/// `isTruncated:false` but still carries a non-null `nextExclusiveStartKey`, the walk must stop
/// and not perform a wasted extra fetch. (The mock repeats its last scripted page, so a regression
/// that followed the stale cursor would loop; the iteration guard below catches that.)
#[tokio::test]
async fn iterate_keys_stops_on_is_truncated_even_with_trailing_cursor() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(
            200,
            br#"{"data":{"items":[{"key":"k0"},{"key":"k1"}],"isTruncated":true,"nextExclusiveStartKey":"k1"}}"#.to_vec(),
        ),
        // Final page: no more data (isTruncated:false) yet a non-null trailing cursor.
        MockOutcome::Status(
            200,
            br#"{"data":{"items":[{"key":"k2"}],"isTruncated":false,"nextExclusiveStartKey":"k2"}}"#.to_vec(),
        ),
    ]);
    let client = client_with(backend.clone(), 0);

    let mut it = client
        .key_value_store("some-store")
        .iterate_keys(Default::default());
    let mut keys = Vec::new();
    let mut guard = 0;
    while let Some(key) = it.next().await.expect("ok") {
        keys.push(key.key);
        guard += 1;
        assert!(
            guard < 100,
            "iterate_keys did not terminate on isTruncated=false"
        );
    }

    assert_eq!(
        keys,
        vec!["k0", "k1", "k2"],
        "all keys across pages must be yielded"
    );
    assert_eq!(
        backend.call_count(),
        2,
        "isTruncated:false ends the walk on the second page; no wasted fetch despite the trailing cursor"
    );
}

/// The `iterate_keys` `limit` is a total cap that also bounds the first page's request size
/// (reference parity): with `limit=2` the first request asks for `limit=2` and, once two keys
/// are consumed, the walk stops without a second fetch even though the API advertised more keys.
#[tokio::test]
async fn iterate_keys_limit_caps_the_walk() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        200,
        br#"{"data":{"items":[{"key":"k0"},{"key":"k1"}],"isTruncated":true,"nextExclusiveStartKey":"k1"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let mut it = client
        .key_value_store("some-store")
        .iterate_keys(apify_client::ListKeysOptions {
            limit: Some(2),
            ..Default::default()
        });
    let mut keys = Vec::new();
    while let Some(key) = it.next().await.expect("ok") {
        keys.push(key.key);
    }

    assert_eq!(
        keys,
        vec!["k0", "k1"],
        "yields exactly the first (capped) page"
    );
    assert_eq!(
        backend.call_count(),
        1,
        "the cap is reached on the first page, so no second fetch despite isTruncated"
    );
    assert!(
        backend.urls()[0].contains("limit=2"),
        "the cap bounds the first page's request size, got {}",
        backend.urls()[0]
    );
}

/// `iterate_keys` with `limit = Some(0)` means "iterate everything" — the first request must NOT
/// send `limit=0` (out of the endpoint's `minimum: 1` range); the `0` is normalized to no limit.
#[tokio::test]
async fn iterate_keys_zero_limit_sends_no_limit_and_walks_all() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(
            200,
            br#"{"data":{"items":[{"key":"k0"}],"isTruncated":true,"nextExclusiveStartKey":"k0"}}"#.to_vec(),
        ),
        MockOutcome::Status(
            200,
            br#"{"data":{"items":[{"key":"k1"}],"isTruncated":false,"nextExclusiveStartKey":null}}"#.to_vec(),
        ),
    ]);
    let client = client_with(backend.clone(), 0);

    let mut it = client
        .key_value_store("some-store")
        .iterate_keys(apify_client::ListKeysOptions {
            limit: Some(0),
            ..Default::default()
        });
    let mut keys = Vec::new();
    while let Some(key) = it.next().await.expect("ok") {
        keys.push(key.key);
    }

    assert_eq!(keys, vec!["k0", "k1"], "0 must iterate the whole store");
    assert!(
        !backend.urls()[0].contains("limit="),
        "limit=0 must be normalized to no limit param, got {}",
        backend.urls()[0]
    );
}

/// `RunClient::charge` always sends the `idempotency-key` header (required by the spec) and the
/// `eventName`/`count` body, whether or not an explicit key is supplied. This is hermetic because
/// a real charge can only be made from inside a running pay-per-event Actor with that run's own
/// token, which the integration suite (running as an external account-level client) cannot do.
#[tokio::test]
async fn charge_sends_required_idempotency_key_header() {
    let backend = MockBackend::new(vec![MockOutcome::Status(201, b"{}".to_vec())]);
    let client = client_with(backend.clone(), 0);

    // No explicit key -> one is still auto-generated and sent.
    client
        .run("some-run-id")
        .charge(apify_client::RunChargeOptions {
            event_name: "ANALYZE_PAGE".to_owned(),
            count: Some(2),
            idempotency_key: None,
            request_id: None,
        })
        .await
        .expect("charge should succeed");
    assert!(
        backend
            .last_header("idempotency-key")
            .is_some_and(|k| !k.is_empty()),
        "an idempotency-key header must always be sent, even when none is supplied"
    );
    let url = backend.last_url().expect("a request was sent");
    assert!(
        url.contains("/actor-runs/some-run-id/charge"),
        "charge must POST to the run's /charge endpoint, got {url}"
    );
    let body = backend.last_body().expect("a body was sent");
    let parsed: serde_json::Value = serde_json::from_slice(&body).expect("valid JSON body");
    assert_eq!(parsed["eventName"], "ANALYZE_PAGE");
    assert_eq!(parsed["count"], 2);
    assert!(
        parsed.get("requestId").is_none(),
        "requestId must be omitted from the body when not supplied"
    );

    // Explicit key -> sent verbatim (not overwritten by an auto-generated one).
    client
        .run("some-run-id")
        .charge(apify_client::RunChargeOptions {
            event_name: "ANALYZE_PAGE".to_owned(),
            count: None,
            idempotency_key: Some("my-explicit-key".to_owned()),
            request_id: Some("some-request-id".to_owned()),
        })
        .await
        .expect("charge should succeed");
    assert_eq!(
        backend.last_header("idempotency-key").as_deref(),
        Some("my-explicit-key"),
        "an explicit idempotency key must be sent verbatim"
    );
    let body = backend.last_body().expect("a body was sent");
    let parsed: serde_json::Value = serde_json::from_slice(&body).expect("valid JSON body");
    assert_eq!(parsed["count"], 1, "count defaults to 1 when omitted");
    assert_eq!(
        parsed["requestId"], "some-request-id",
        "a supplied request ID must be sent in the body"
    );
}

/// A finite `iterate_keys` cap larger than one page must be clamped to the endpoint maximum per
/// request (so it does not 400) while still yielding across pages until the cap or the store is
/// exhausted.
#[tokio::test]
async fn iterate_keys_large_cap_clamps_page_size() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(
            200,
            br#"{"data":{"items":[{"key":"k0"},{"key":"k1"}],"isTruncated":true,"nextExclusiveStartKey":"k1"}}"#.to_vec(),
        ),
        MockOutcome::Status(
            200,
            br#"{"data":{"items":[{"key":"k2"}],"isTruncated":false,"nextExclusiveStartKey":null}}"#.to_vec(),
        ),
    ]);
    let client = client_with(backend.clone(), 0);

    // Cap of 1500 > the endpoint max (1000): each request must ask for at most 1000.
    let mut it = client
        .key_value_store("some-store")
        .iterate_keys(apify_client::ListKeysOptions {
            limit: Some(1500),
            ..Default::default()
        });
    let mut keys = Vec::new();
    while let Some(key) = it.next().await.expect("ok") {
        keys.push(key.key);
    }

    assert_eq!(
        keys,
        vec!["k0", "k1", "k2"],
        "walk continues under the large cap"
    );
    let urls = backend.urls();
    assert!(
        urls[0].contains("limit=1000") && urls[1].contains("limit=1000"),
        "each request must be clamped to the endpoint max of 1000, got {urls:?}"
    );
    assert!(
        urls[1].contains("exclusiveStartKey=k1"),
        "the cursor must still advance across the clamped pages, got {}",
        urls[1]
    );
}

/// `wait_for_finish` must not poll a persistently-missing run forever when `wait_secs` is `None`:
/// past the 3-second not-found grace period it surfaces the `404` as an error instead. Uses
/// paused Tokio time so the grace period and poll interval elapse instantly in wall-clock time.
#[tokio::test(start_paused = true)]
async fn wait_for_finish_errors_after_persistent_not_found() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        404,
        br#"{"error":{"type":"record-not-found","message":"Run was not found"}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let err = client
        .run("ghost-run-id")
        .wait_for_finish(None)
        .await
        .expect_err("a run that never appears must error out, not poll forever");
    let api = err.as_api_error().expect("api error");
    assert!(api.is_not_found());
    assert!(
        backend.call_count() > 1,
        "must have retried within the grace period, not failed on the very first 404"
    );
}

/// A run/build that 404s once (e.g. a replica lagging just behind the write that created it) and
/// then starts answering normally must not be treated as persistently missing: the grace-period
/// clock resets on the first successful fetch.
#[tokio::test(start_paused = true)]
async fn wait_for_finish_recovers_from_a_transient_not_found() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(
            404,
            br#"{"error":{"type":"record-not-found","message":"Run was not found"}}"#.to_vec(),
        ),
        MockOutcome::Status(
            200,
            br#"{"data":{"id":"run-id","status":"SUCCEEDED"}}"#.to_vec(),
        ),
    ]);
    let client = client_with(backend.clone(), 0);

    let run = client
        .run("run-id")
        .wait_for_finish(None)
        .await
        .expect("a transient 404 followed by success must not error");
    assert_eq!(run.status.as_deref(), Some("SUCCEEDED"));
}

/// `ActorStartOptions::wait_for_resources` retries a start rejected for lack of resources
/// (`actor-memory-limit-exceeded` / `concurrent-runs-limit-exceeded`) every 10 seconds until it
/// succeeds. Paused time collapses the two 10-second cooldowns to no wall-clock time.
#[tokio::test(start_paused = true)]
async fn wait_for_resources_retries_until_success() {
    let backend =
        MockBackend::new(vec![
        MockOutcome::Status(
            402,
            br#"{"error":{"type":"actor-memory-limit-exceeded","message":"not enough memory"}}"#
                .to_vec(),
        ),
        MockOutcome::Status(
            402,
            br#"{"error":{"type":"concurrent-runs-limit-exceeded","message":"too many runs"}}"#
                .to_vec(),
        ),
        MockOutcome::Status(201, br#"{"data":{"id":"run-id","status":"READY"}}"#.to_vec()),
    ]);
    let client = client_with(backend.clone(), 0);

    let run = client
        .actor("some-actor")
        .start::<serde_json::Value>(
            None,
            ActorStartOptions {
                wait_for_resources: Some(WaitForResources::Forever),
                ..Default::default()
            },
        )
        .await
        .expect("must eventually succeed once resources free up");
    assert_eq!(run.id, "run-id");
    assert_eq!(backend.call_count(), 3, "two rejections, then the success");
}

/// Without `wait_for_resources`, a resource-limit rejection is returned immediately, unretried —
/// matching every other `ApiError`.
#[tokio::test]
async fn wait_for_resources_defaults_to_a_single_attempt() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        402,
        br#"{"error":{"type":"actor-memory-limit-exceeded","message":"not enough memory"}}"#
            .to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let err = client
        .actor("some-actor")
        .start::<serde_json::Value>(None, ActorStartOptions::default())
        .await
        .expect_err("must fail without the option");
    assert_eq!(
        err.as_api_error().unwrap().error_type.as_deref(),
        Some("actor-memory-limit-exceeded")
    );
    assert_eq!(backend.call_count(), 1);
}

/// `WaitForResources::ForSecs` bounds the retrying: attempts happen at 0s, 10s, 20s and 25s (the
/// last cooldown cut short by the bound), then the last rejection is returned.
#[tokio::test(start_paused = true)]
async fn wait_for_resources_for_secs_bounds_retrying_and_returns_last_error() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        402,
        br#"{"error":{"type":"actor-memory-limit-exceeded","message":"not enough memory"}}"#
            .to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let err = client
        .actor("some-actor")
        .start::<serde_json::Value>(
            None,
            ActorStartOptions {
                wait_for_resources: Some(WaitForResources::ForSecs(25.0)),
                ..Default::default()
            },
        )
        .await
        .expect_err("must give up once the bound elapses");
    assert_eq!(
        err.as_api_error().unwrap().error_type.as_deref(),
        Some("actor-memory-limit-exceeded")
    );
    assert_eq!(backend.call_count(), 4, "attempts at 0s, 10s, 20s and 25s");
}

/// Any error other than a resource-limit rejection is returned right away, never retried, even
/// with `wait_for_resources` set.
#[tokio::test]
async fn wait_for_resources_does_not_retry_unrelated_errors() {
    let backend = MockBackend::new(vec![MockOutcome::Status(
        400,
        br#"{"error":{"type":"invalid-input","message":"Input is not valid."}}"#.to_vec(),
    )]);
    let client = client_with(backend.clone(), 0);

    let err = client
        .actor("some-actor")
        .start::<serde_json::Value>(
            None,
            ActorStartOptions {
                wait_for_resources: Some(WaitForResources::Forever),
                ..Default::default()
            },
        )
        .await
        .expect_err("a non-resource-limit error must not be retried");
    assert_eq!(
        err.as_api_error().unwrap().error_type.as_deref(),
        Some("invalid-input")
    );
    assert_eq!(backend.call_count(), 1);
}

/// `TaskClient::start` shares the same `wait_for_resources` retry behavior as `ActorClient::start`
/// (both funnel through the same helper — DRY).
#[tokio::test(start_paused = true)]
async fn task_wait_for_resources_retries_until_success() {
    let backend = MockBackend::new(vec![
        MockOutcome::Status(
            402,
            br#"{"error":{"type":"concurrent-runs-limit-exceeded","message":"too many runs"}}"#
                .to_vec(),
        ),
        MockOutcome::Status(
            201,
            br#"{"data":{"id":"run-id","status":"READY"}}"#.to_vec(),
        ),
    ]);
    let client = client_with(backend.clone(), 0);

    let run = client
        .task("some-task")
        .start::<serde_json::Value>(
            None,
            ActorStartOptions {
                wait_for_resources: Some(WaitForResources::Forever),
                ..Default::default()
            },
        )
        .await
        .expect("must eventually succeed once resources free up");
    assert_eq!(run.id, "run-id");
    assert_eq!(backend.call_count(), 2);
}

/// Time spent retrying a resource-limited start inside `call()` must not count against
/// `wait_secs`: that budget only starts once the run has actually started.
#[tokio::test(start_paused = true)]
async fn call_wait_for_resources_does_not_count_toward_wait_secs() {
    let backend =
        MockBackend::new(vec![
        MockOutcome::Status(
            402,
            br#"{"error":{"type":"actor-memory-limit-exceeded","message":"not enough memory"}}"#
                .to_vec(),
        ),
        MockOutcome::Status(201, br#"{"data":{"id":"run-id","status":"READY"}}"#.to_vec()),
        MockOutcome::Status(200, br#"{"data":{"id":"run-id","status":"SUCCEEDED"}}"#.to_vec()),
    ]);
    let client = client_with(backend.clone(), 0);

    // `wait_secs: Some(1)` is a tight budget for *finishing*; the resource-limit retry's 10-second
    // cooldown happens entirely before that budget's clock starts, in `start()`.
    let run = client
        .actor("some-actor")
        .call::<serde_json::Value>(
            None,
            ActorStartOptions {
                wait_for_resources: Some(WaitForResources::Forever),
                ..Default::default()
            },
            Some(1),
        )
        .await
        .expect("the resource-limit retry must not eat into the wait_for_finish budget");
    assert_eq!(run.status.as_deref(), Some("SUCCEEDED"));
}