spiral-sdk 0.1.12

// Generated by tools/service-contracts/rust.py. Do not edit.
#![doc = "Generated reqwest client with typed errors and streaming bodies."]

use async_trait::async_trait;
use futures::{Stream, TryStreamExt, stream};
use reqwest::Url;
use reqwest_middleware::ClientWithMiddleware;
use spiral_contract::auth::AuthHeader;
use spiral_contract::call::{CallError, ServiceFailure};
use spiral_contract::error::ErrorEnvelope;
use uuid::Uuid;

use crate::api::tasks::TasksApi;
use crate::api::tasks::objects::*;
use crate::http::default_http_client;
/// HTTP client for the tasks service.
#[derive(Clone)]
pub struct TasksClient {
    http: ClientWithMiddleware,
    base: Url,
}

impl std::fmt::Debug for TasksClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TasksClient")
            .field("base", &self.base)
            .finish_non_exhaustive()
    }
}

impl TasksClient {
    /// Create a redirect-disabled client; base may contain a deployment prefix.
    pub fn new(base: &str) -> Result<Self, CallError<TasksError>> {
        let http = default_http_client().map_err(CallError::transport)?;
        Self::from_http(base, http)
    }
    /// Use an existing connection pool with redirects disabled.
    pub fn from_http(
        base: &str,
        http: impl Into<ClientWithMiddleware>,
    ) -> Result<Self, CallError<TasksError>> {
        Self::from_transport(base, http.into())
    }
    /// Use a host-owned connection pool and instrumentation. Redirects must be disabled.
    pub fn from_transport(
        base: &str,
        http: ClientWithMiddleware,
    ) -> Result<Self, CallError<TasksError>> {
        let base =
            Url::parse(base).map_err(|_| CallError::<TasksError>::protocol("invalid base URL"))?;
        if !matches!(base.scheme(), "http" | "https")
            || base.host_str().is_none()
            || !base.username().is_empty()
            || base.password().is_some()
            || base.query().is_some()
            || base.fragment().is_some()
        {
            return Err(CallError::<TasksError>::protocol("invalid base URL"));
        }
        Ok(Self { http, base })
    }
    /// Replace the host-owned connection pool and instrumentation. Redirects must be disabled.
    pub fn with_transport(mut self, http: ClientWithMiddleware) -> Self {
        self.http = http;
        self
    }
    /// Supply transport instrumentation or custom connection policy.
    ///
    /// The caller must disable redirects to protect bearer credentials.
    pub fn with_http(mut self, http: impl Into<ClientWithMiddleware>) -> Self {
        self.http = http.into();
        self
    }
    /// Activate one Worker pool.
    pub async fn activate_worker_pool_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_pool_id: Uuid,
    ) -> Result<WorkerPoolResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-pools");
            let segment = worker_pool_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("activate");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Cancel a Task through a Project. Deprecated; use cancel_task, which derives the Project.
    pub async fn cancel_project_task_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        task_id: Uuid,
    ) -> Result<Task, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("tasks");
            let segment = task_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("cancel");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Request cancellation of a Task. The Task's own Project authorizes the request; a Task the
    /// caller cannot view is not found.
    pub async fn cancel_task_http(
        &self,
        auth_header: AuthHeader,
        task_id: Uuid,
    ) -> Result<Task, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            let segment = task_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("cancel");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Claim worker task.
    pub async fn claim_worker_task_http(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
        task_id: Uuid,
    ) -> Result<PollTaskResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("worker-sessions");
            let segment = worker_session_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("tasks");
            let segment = task_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("claim");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Create a Task from a specification ID and JSON input. Deprecated; use submit_task.
    pub async fn create_task_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: CreateTaskRequest,
    ) -> Result<CreateTaskResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("tasks");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Create trigger.
    pub async fn create_trigger_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: TriggerDefinition,
    ) -> Result<TaskTrigger, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-triggers");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Create worker pool.
    pub async fn create_worker_pool_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: CreateWorkerPoolRequest,
    ) -> Result<WorkerPoolResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-pools");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Delete trigger.
    pub async fn delete_trigger_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
    ) -> Result<(), CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-triggers");
            let segment = trigger_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
        }
        let builder = self.http.delete(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Drain worker.
    pub async fn drain_worker_http(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
    ) -> Result<DrainWorkerResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("worker-sessions");
            let segment = worker_session_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("drain");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Finish worker attempt.
    pub async fn finish_worker_attempt_http(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
        attempt_id: Uuid,
        request: FinishAttemptRequest,
    ) -> Result<FinishAttemptResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("worker-sessions");
            let segment = worker_session_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("attempts");
            let segment = attempt_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("finish");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Heartbeat worker.
    pub async fn heartbeat_worker_http(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
        request: HeartbeatWorkerRequest,
    ) -> Result<HeartbeatWorkerResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("worker-sessions");
            let segment = worker_session_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("heartbeat");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// List task specs.
    pub async fn list_task_specs_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
    ) -> Result<ListTaskSpecsResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-specs");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// List Tasks by descending Task ID. Pages default to 100 items with a maximum of 1000.
    /// Continuations expire after one hour and are scoped to the caller, target, operation, filters
    /// and page size. This is a live listing; concurrent updates can change later pages.
    pub async fn list_tasks_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: ListTasksRequest,
    ) -> Result<ListTasksResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("tasks");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder.query(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Fetch pages lazily, retaining the request filters and stopping when the token is absent.
    /// Dropping the stream cancels an active request; no later page is prefetched.
    pub fn list_tasks_pages<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        request: ListTasksRequest,
    ) -> impl Stream<Item = Result<ListTasksResponse, CallError<TasksError>>> + 'a {
        stream::try_unfold(Some(request), move |request| {
            #[allow(
                clippy::clone_on_copy,
                reason = "Hosts may implement Copy for generated aliases; pagination also supports Clone aliases."
            )]
            let project_id = project_id.clone();
            async move {
                let Some(mut request) = request else {
                    return Ok(None);
                };
                let page = self
                    .list_tasks_http(auth_header.clone(), project_id, request.clone())
                    .await?;
                let next = page.next_page_token.clone();
                if next.as_ref().is_some_and(|token| {
                    token.is_empty() || Some(token) == request.page_token.as_ref()
                }) {
                    return Err(CallError::<TasksError>::protocol(
                        "invalid pagination token",
                    ));
                }
                request.page_token = next;
                let next_request = request.page_token.is_some().then_some(request);
                Ok(Some((page, next_request)))
            }
        })
    }
    /// Yield items lazily, continuing across empty pages that have a continuation token.
    pub fn list_tasks_items<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        request: ListTasksRequest,
    ) -> impl Stream<Item = Result<Task, CallError<TasksError>>> + 'a {
        self.list_tasks_pages(auth_header, project_id, request)
            .map_ok(|page| stream::iter(page.tasks.into_iter().map(Ok::<_, CallError<TasksError>>)))
            .try_flatten()
    }
    /// List invocations by descending submission time and Task ID in fixed pages of 10.
    /// Continuations expire after one hour and are scoped to the caller, target, operation, filters
    /// and page size. This is a live listing; concurrent updates can change later pages.
    pub async fn list_trigger_invocations_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: InvocationPageRequest,
    ) -> Result<InvocationPage, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-triggers");
            let segment = trigger_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("invocations");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder.query(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Fetch pages lazily, retaining the request filters and stopping when the token is absent.
    /// Dropping the stream cancels an active request; no later page is prefetched.
    pub fn list_trigger_invocations_pages<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: InvocationPageRequest,
    ) -> impl Stream<Item = Result<InvocationPage, CallError<TasksError>>> + 'a {
        stream::try_unfold(Some(request), move |request| {
            #[allow(
                clippy::clone_on_copy,
                reason = "Hosts may implement Copy for generated aliases; pagination also supports Clone aliases."
            )]
            let project_id = project_id.clone();
            async move {
                let Some(mut request) = request else {
                    return Ok(None);
                };
                let page = self
                    .list_trigger_invocations_http(
                        auth_header.clone(),
                        project_id,
                        trigger_id,
                        request.clone(),
                    )
                    .await?;
                let next = page.next_page_token.clone();
                if next.as_ref().is_some_and(|token| {
                    token.is_empty() || Some(token) == request.page_token.as_ref()
                }) {
                    return Err(CallError::<TasksError>::protocol(
                        "invalid pagination token",
                    ));
                }
                request.page_token = next;
                let next_request = request.page_token.is_some().then_some(request);
                Ok(Some((page, next_request)))
            }
        })
    }
    /// Yield items lazily, continuing across empty pages that have a continuation token.
    pub fn list_trigger_invocations_items<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: InvocationPageRequest,
    ) -> impl Stream<Item = Result<TriggerInvocation, CallError<TasksError>>> + 'a {
        self.list_trigger_invocations_pages(auth_header, project_id, trigger_id, request)
            .map_ok(|page| stream::iter(page.items.into_iter().map(Ok::<_, CallError<TasksError>>)))
            .try_flatten()
    }
    /// List triggers by descending trigger ID in fixed pages of 50. Continuations expire after one
    /// hour and are scoped to the caller, target, operation, filters and page size. This is a live
    /// listing; concurrent updates can change later pages.
    pub async fn list_triggers_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: TriggerPageRequest,
    ) -> Result<TriggerPage, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-triggers");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder.query(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Fetch pages lazily, retaining the request filters and stopping when the token is absent.
    /// Dropping the stream cancels an active request; no later page is prefetched.
    pub fn list_triggers_pages<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        request: TriggerPageRequest,
    ) -> impl Stream<Item = Result<TriggerPage, CallError<TasksError>>> + 'a {
        stream::try_unfold(Some(request), move |request| {
            #[allow(
                clippy::clone_on_copy,
                reason = "Hosts may implement Copy for generated aliases; pagination also supports Clone aliases."
            )]
            let project_id = project_id.clone();
            async move {
                let Some(mut request) = request else {
                    return Ok(None);
                };
                let page = self
                    .list_triggers_http(auth_header.clone(), project_id, request.clone())
                    .await?;
                let next = page.next_page_token.clone();
                if next.as_ref().is_some_and(|token| {
                    token.is_empty() || Some(token) == request.page_token.as_ref()
                }) {
                    return Err(CallError::<TasksError>::protocol(
                        "invalid pagination token",
                    ));
                }
                request.page_token = next;
                let next_request = request.page_token.is_some().then_some(request);
                Ok(Some((page, next_request)))
            }
        })
    }
    /// Yield items lazily, continuing across empty pages that have a continuation token.
    pub fn list_triggers_items<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        request: TriggerPageRequest,
    ) -> impl Stream<Item = Result<TaskTrigger, CallError<TasksError>>> + 'a {
        self.list_triggers_pages(auth_header, project_id, request)
            .map_ok(|page| stream::iter(page.items.into_iter().map(Ok::<_, CallError<TasksError>>)))
            .try_flatten()
    }
    /// List Attempts by descending Attempt ID. Pages default to 100 items with a maximum of 100.
    /// Continuations expire after one hour and are scoped to the caller, target, operation, filters
    /// and page size. This is a live listing; concurrent updates can change later pages.
    pub async fn list_worker_attempts_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_session_id: Uuid,
        request: ListWorkerAttemptsRequest,
    ) -> Result<ListWorkerAttemptsResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-sessions");
            let segment = worker_session_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("attempts");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder.query(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Fetch pages lazily, retaining the request filters and stopping when the token is absent.
    /// Dropping the stream cancels an active request; no later page is prefetched.
    pub fn list_worker_attempts_pages<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        worker_session_id: Uuid,
        request: ListWorkerAttemptsRequest,
    ) -> impl Stream<Item = Result<ListWorkerAttemptsResponse, CallError<TasksError>>> + 'a {
        stream::try_unfold(Some(request), move |request| {
            #[allow(
                clippy::clone_on_copy,
                reason = "Hosts may implement Copy for generated aliases; pagination also supports Clone aliases."
            )]
            let project_id = project_id.clone();
            async move {
                let Some(mut request) = request else {
                    return Ok(None);
                };
                let page = self
                    .list_worker_attempts_http(
                        auth_header.clone(),
                        project_id,
                        worker_session_id,
                        request.clone(),
                    )
                    .await?;
                let next = page.next_page_token.clone();
                if next.as_ref().is_some_and(|token| {
                    token.is_empty() || Some(token) == request.page_token.as_ref()
                }) {
                    return Err(CallError::<TasksError>::protocol(
                        "invalid pagination token",
                    ));
                }
                request.page_token = next;
                let next_request = request.page_token.is_some().then_some(request);
                Ok(Some((page, next_request)))
            }
        })
    }
    /// Yield items lazily, continuing across empty pages that have a continuation token.
    pub fn list_worker_attempts_items<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        worker_session_id: Uuid,
        request: ListWorkerAttemptsRequest,
    ) -> impl Stream<Item = Result<WorkerAttempt, CallError<TasksError>>> + 'a {
        self.list_worker_attempts_pages(auth_header, project_id, worker_session_id, request)
            .map_ok(|page| {
                stream::iter(
                    page.attempts
                        .into_iter()
                        .map(Ok::<_, CallError<TasksError>>),
                )
            })
            .try_flatten()
    }
    /// List Worker pools and aligned statistics by ascending pool ID. Pages default to 100 items
    /// with a maximum of 1000. Continuations expire after one hour and are scoped to the caller,
    /// target, operation, filters and page size. This is a live listing; concurrent updates can
    /// change later pages.
    pub async fn list_worker_pools_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: ListWorkerPoolsRequest,
    ) -> Result<ListWorkerPoolsResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-pools");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder.query(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Fetch pages lazily, retaining the request filters and stopping when the token is absent.
    /// Dropping the stream cancels an active request; no later page is prefetched.
    pub fn list_worker_pools_pages<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        request: ListWorkerPoolsRequest,
    ) -> impl Stream<Item = Result<ListWorkerPoolsResponse, CallError<TasksError>>> + 'a {
        stream::try_unfold(Some(request), move |request| {
            #[allow(
                clippy::clone_on_copy,
                reason = "Hosts may implement Copy for generated aliases; pagination also supports Clone aliases."
            )]
            let project_id = project_id.clone();
            async move {
                let Some(mut request) = request else {
                    return Ok(None);
                };
                let page = self
                    .list_worker_pools_http(auth_header.clone(), project_id, request.clone())
                    .await?;
                let next = page.next_page_token.clone();
                if next.as_ref().is_some_and(|token| {
                    token.is_empty() || Some(token) == request.page_token.as_ref()
                }) {
                    return Err(CallError::<TasksError>::protocol(
                        "invalid pagination token",
                    ));
                }
                request.page_token = next;
                let next_request = request.page_token.is_some().then_some(request);
                Ok(Some((page, next_request)))
            }
        })
    }
    /// Yield items lazily, continuing across empty pages that have a continuation token.
    pub fn list_worker_pools_items<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        request: ListWorkerPoolsRequest,
    ) -> impl Stream<Item = Result<WorkerPool, CallError<TasksError>>> + 'a {
        self.list_worker_pools_pages(auth_header, project_id, request)
            .map_ok(|page| {
                stream::iter(
                    page.worker_pools
                        .into_iter()
                        .map(Ok::<_, CallError<TasksError>>),
                )
            })
            .try_flatten()
    }
    /// List Worker sessions by descending session ID. Pages default to 100 items with a maximum of
    /// 1000. Continuations expire after one hour and are scoped to the caller, target, operation,
    /// filters and page size. This is a live listing; concurrent updates can change later pages.
    pub async fn list_worker_sessions_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: ListWorkerSessionsRequest,
    ) -> Result<ListWorkerSessionsResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-sessions");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder.query(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Fetch pages lazily, retaining the request filters and stopping when the token is absent.
    /// Dropping the stream cancels an active request; no later page is prefetched.
    pub fn list_worker_sessions_pages<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        request: ListWorkerSessionsRequest,
    ) -> impl Stream<Item = Result<ListWorkerSessionsResponse, CallError<TasksError>>> + 'a {
        stream::try_unfold(Some(request), move |request| {
            #[allow(
                clippy::clone_on_copy,
                reason = "Hosts may implement Copy for generated aliases; pagination also supports Clone aliases."
            )]
            let project_id = project_id.clone();
            async move {
                let Some(mut request) = request else {
                    return Ok(None);
                };
                let page = self
                    .list_worker_sessions_http(auth_header.clone(), project_id, request.clone())
                    .await?;
                let next = page.next_page_token.clone();
                if next.as_ref().is_some_and(|token| {
                    token.is_empty() || Some(token) == request.page_token.as_ref()
                }) {
                    return Err(CallError::<TasksError>::protocol(
                        "invalid pagination token",
                    ));
                }
                request.page_token = next;
                let next_request = request.page_token.is_some().then_some(request);
                Ok(Some((page, next_request)))
            }
        })
    }
    /// Yield items lazily, continuing across empty pages that have a continuation token.
    pub fn list_worker_sessions_items<'a>(
        &'a self,
        auth_header: &'a AuthHeader,
        project_id: String,
        request: ListWorkerSessionsRequest,
    ) -> impl Stream<Item = Result<WorkerSession, CallError<TasksError>>> + 'a {
        self.list_worker_sessions_pages(auth_header, project_id, request)
            .map_ok(|page| {
                stream::iter(
                    page.worker_sessions
                        .into_iter()
                        .map(Ok::<_, CallError<TasksError>>),
                )
            })
            .try_flatten()
    }
    /// Pause one Worker pool.
    pub async fn pause_worker_pool_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_pool_id: Uuid,
    ) -> Result<WorkerPoolResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-pools");
            let segment = worker_pool_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("pause");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Poll worker.
    pub async fn poll_worker_http(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
    ) -> Result<PollTaskResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("worker-sessions");
            let segment = worker_session_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("poll");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Read a Task through a Project. Deprecated; use read_task, which derives the Project.
    pub async fn read_project_task_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        task_id: Uuid,
    ) -> Result<TaskDetails, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("tasks");
            let segment = task_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Read resource usage.
    pub async fn read_resource_usage_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_session_id: Uuid,
        request: ResourceUsageRequest,
    ) -> Result<ResourceUsageHistory, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-sessions");
            let segment = worker_session_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("resource-usage");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder.query(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Read a Task with its Attempts and resource references. The Task's own Project authorizes the
    /// read; a Task the caller cannot view is not found.
    pub async fn read_task_http(
        &self,
        auth_header: AuthHeader,
        task_id: Uuid,
    ) -> Result<TaskDetails, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            let segment = task_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Read task stats.
    pub async fn read_task_stats_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: TaskStatsRequest,
    ) -> Result<TaskStats, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-stats");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder.query(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Read trigger.
    pub async fn read_trigger_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
    ) -> Result<TaskTrigger, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-triggers");
            let segment = trigger_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Read worker pool.
    pub async fn read_worker_pool_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_pool_id: Uuid,
    ) -> Result<WorkerPoolDetails, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-pools");
            let segment = worker_pool_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Read worker session.
    pub async fn read_worker_session_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_session_id: Uuid,
    ) -> Result<WorkerSession, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-sessions");
            let segment = worker_session_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Register worker.
    pub async fn register_worker_http(
        &self,
        auth_header: AuthHeader,
        request: RegisterWorkerRequest,
    ) -> Result<RegisterWorkerResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("worker-sessions");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Retire one Worker pool.
    pub async fn retire_worker_pool_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_pool_id: Uuid,
    ) -> Result<WorkerPoolResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("worker-pools");
            let segment = worker_pool_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("retire");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Run trigger.
    pub async fn run_trigger_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: RunTriggerRequest,
    ) -> Result<TriggerInvocation, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-triggers");
            let segment = trigger_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("run");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Submit a Task from its definition. The response is accepted with the Task, or skipped when
    /// the definition's overlap policy keeps an equivalent Task already waiting.
    pub async fn submit_task_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: SubmitTaskRequest,
    ) -> Result<CreateTaskResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("submissions");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Test trigger.
    pub async fn test_trigger_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: TestTriggerRequest,
    ) -> Result<TestTriggerResponse, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-triggers");
            let segment = trigger_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("test");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Update trigger.
    pub async fn update_trigger_http(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: UpdateTriggerRequest,
    ) -> Result<TaskTrigger, CallError<TasksError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<TasksError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("api");
            path.push("tasks");
            path.push("projects");
            let segment = project_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
            path.push("task-triggers");
            let segment = trigger_id.to_string();
            if matches!(segment.as_str(), "." | "..") {
                return Err(CallError::<TasksError>::protocol("dot path segment"));
            }
            path.push(&segment);
        }
        let builder = self.http.put(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    async fn send(
        &self,
        builder: reqwest_middleware::RequestBuilder,
        expected_media: &str,
    ) -> Result<reqwest::Response, CallError<TasksError>> {
        let response = builder
            .timeout(std::time::Duration::from_secs(60))
            .send()
            .await
            .map_err(CallError::transport)?;
        let status = response.status();
        if !status.is_success() {
            let status = status.as_u16();
            let content_type = response
                .headers()
                .get("content-type")
                .and_then(|value| value.to_str().ok())
                .map(str::to_owned);
            let (bytes, truncated) =
                bounded_error_body::<TasksError>(response, 64 * 1024, status, content_type.clone())
                    .await?;
            if truncated {
                return Err(CallError::unknown_remote(status, content_type, bytes, true));
            }
            return match serde_json::from_slice::<ErrorEnvelope>(&bytes) {
                Ok(envelope) => Err(CallError::Service(ServiceFailure::remote(status, envelope))),
                Err(_) => Err(CallError::unknown_remote(
                    status,
                    content_type,
                    bytes,
                    false,
                )),
            };
        }
        if !media_matches(&response, expected_media) {
            return Err(CallError::protocol_response(
                status.as_u16(),
                "unexpected response media type",
            ));
        }
        Ok(response)
    }
}

fn media_matches(response: &reqwest::Response, expected: &str) -> bool {
    response
        .headers()
        .get("content-type")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.split(';').next())
        .is_some_and(|value| value.trim().eq_ignore_ascii_case(expected))
}

async fn bounded_body(
    mut response: reqwest::Response,
    limit: usize,
) -> Result<Vec<u8>, CallError<TasksError>> {
    let status = response.status().as_u16();
    let mut result = Vec::new();
    while let Some(chunk) = response
        .chunk()
        .await
        .map_err(|error| CallError::transport_response(status, error))?
    {
        if chunk.len() > limit.saturating_sub(result.len()) {
            return Err(CallError::protocol_response(
                status,
                "response exceeds JSON size limit",
            ));
        }
        result.extend_from_slice(&chunk);
    }
    Ok(result)
}

async fn bounded_error_body<E>(
    mut response: reqwest::Response,
    limit: usize,
    status: u16,
    content_type: Option<String>,
) -> Result<(Vec<u8>, bool), CallError<E>> {
    let mut result = Vec::new();
    loop {
        let chunk = response.chunk().await.map_err(|error| {
            CallError::unreadable_remote(status, content_type.clone(), result.clone(), error)
        })?;
        let Some(chunk) = chunk else {
            return Ok((result, false));
        };
        let remaining = limit.saturating_sub(result.len());
        if chunk.len() > remaining {
            result.extend_from_slice(&chunk[..remaining]);
            return Ok((result, true));
        }
        result.extend_from_slice(&chunk);
    }
}
#[async_trait]
impl TasksApi for TasksClient {
    async fn activate_worker_pool(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_pool_id: Uuid,
    ) -> Result<WorkerPoolResponse, CallError<TasksError>> {
        let response =
            Self::activate_worker_pool_http(self, auth_header, project_id, worker_pool_id).await?;
        Ok(response)
    }
    async fn cancel_project_task(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        task_id: Uuid,
    ) -> Result<Task, CallError<TasksError>> {
        let response =
            Self::cancel_project_task_http(self, auth_header, project_id, task_id).await?;
        Ok(response)
    }
    async fn cancel_task(
        &self,
        auth_header: AuthHeader,
        task_id: Uuid,
    ) -> Result<Task, CallError<TasksError>> {
        let response = Self::cancel_task_http(self, auth_header, task_id).await?;
        Ok(response)
    }
    async fn claim_worker_task(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
        task_id: Uuid,
    ) -> Result<PollTaskResponse, CallError<TasksError>> {
        let response =
            Self::claim_worker_task_http(self, auth_header, worker_session_id, task_id).await?;
        Ok(response)
    }
    async fn create_task(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: CreateTaskRequest,
    ) -> Result<CreateTaskResponse, CallError<TasksError>> {
        let response = Self::create_task_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn create_trigger(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: TriggerDefinition,
    ) -> Result<TaskTrigger, CallError<TasksError>> {
        let response = Self::create_trigger_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn create_worker_pool(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: CreateWorkerPoolRequest,
    ) -> Result<WorkerPoolResponse, CallError<TasksError>> {
        let response =
            Self::create_worker_pool_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn delete_trigger(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
    ) -> Result<(), CallError<TasksError>> {
        let response = Self::delete_trigger_http(self, auth_header, project_id, trigger_id).await?;
        Ok(response)
    }
    async fn drain_worker(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
    ) -> Result<DrainWorkerResponse, CallError<TasksError>> {
        let response = Self::drain_worker_http(self, auth_header, worker_session_id).await?;
        Ok(response)
    }
    async fn finish_worker_attempt(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
        attempt_id: Uuid,
        request: FinishAttemptRequest,
    ) -> Result<FinishAttemptResponse, CallError<TasksError>> {
        let response = Self::finish_worker_attempt_http(
            self,
            auth_header,
            worker_session_id,
            attempt_id,
            request,
        )
        .await?;
        Ok(response)
    }
    async fn heartbeat_worker(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
        request: HeartbeatWorkerRequest,
    ) -> Result<HeartbeatWorkerResponse, CallError<TasksError>> {
        let response =
            Self::heartbeat_worker_http(self, auth_header, worker_session_id, request).await?;
        Ok(response)
    }
    async fn list_task_specs(
        &self,
        auth_header: AuthHeader,
        project_id: String,
    ) -> Result<ListTaskSpecsResponse, CallError<TasksError>> {
        let response = Self::list_task_specs_http(self, auth_header, project_id).await?;
        Ok(response)
    }
    async fn list_tasks(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: ListTasksRequest,
    ) -> Result<ListTasksResponse, CallError<TasksError>> {
        let response = Self::list_tasks_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn list_trigger_invocations(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: InvocationPageRequest,
    ) -> Result<InvocationPage, CallError<TasksError>> {
        let response =
            Self::list_trigger_invocations_http(self, auth_header, project_id, trigger_id, request)
                .await?;
        Ok(response)
    }
    async fn list_triggers(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: TriggerPageRequest,
    ) -> Result<TriggerPage, CallError<TasksError>> {
        let response = Self::list_triggers_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn list_worker_attempts(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_session_id: Uuid,
        request: ListWorkerAttemptsRequest,
    ) -> Result<ListWorkerAttemptsResponse, CallError<TasksError>> {
        let response = Self::list_worker_attempts_http(
            self,
            auth_header,
            project_id,
            worker_session_id,
            request,
        )
        .await?;
        Ok(response)
    }
    async fn list_worker_pools(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: ListWorkerPoolsRequest,
    ) -> Result<ListWorkerPoolsResponse, CallError<TasksError>> {
        let response = Self::list_worker_pools_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn list_worker_sessions(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: ListWorkerSessionsRequest,
    ) -> Result<ListWorkerSessionsResponse, CallError<TasksError>> {
        let response =
            Self::list_worker_sessions_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn pause_worker_pool(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_pool_id: Uuid,
    ) -> Result<WorkerPoolResponse, CallError<TasksError>> {
        let response =
            Self::pause_worker_pool_http(self, auth_header, project_id, worker_pool_id).await?;
        Ok(response)
    }
    async fn poll_worker(
        &self,
        auth_header: AuthHeader,
        worker_session_id: Uuid,
    ) -> Result<PollTaskResponse, CallError<TasksError>> {
        let response = Self::poll_worker_http(self, auth_header, worker_session_id).await?;
        Ok(response)
    }
    async fn read_project_task(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        task_id: Uuid,
    ) -> Result<TaskDetails, CallError<TasksError>> {
        let response = Self::read_project_task_http(self, auth_header, project_id, task_id).await?;
        Ok(response)
    }
    async fn read_resource_usage(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_session_id: Uuid,
        request: ResourceUsageRequest,
    ) -> Result<ResourceUsageHistory, CallError<TasksError>> {
        let response = Self::read_resource_usage_http(
            self,
            auth_header,
            project_id,
            worker_session_id,
            request,
        )
        .await?;
        Ok(response)
    }
    async fn read_task(
        &self,
        auth_header: AuthHeader,
        task_id: Uuid,
    ) -> Result<TaskDetails, CallError<TasksError>> {
        let response = Self::read_task_http(self, auth_header, task_id).await?;
        Ok(response)
    }
    async fn read_task_stats(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: TaskStatsRequest,
    ) -> Result<TaskStats, CallError<TasksError>> {
        let response = Self::read_task_stats_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn read_trigger(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
    ) -> Result<TaskTrigger, CallError<TasksError>> {
        let response = Self::read_trigger_http(self, auth_header, project_id, trigger_id).await?;
        Ok(response)
    }
    async fn read_worker_pool(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_pool_id: Uuid,
    ) -> Result<WorkerPoolDetails, CallError<TasksError>> {
        let response =
            Self::read_worker_pool_http(self, auth_header, project_id, worker_pool_id).await?;
        Ok(response)
    }
    async fn read_worker_session(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_session_id: Uuid,
    ) -> Result<WorkerSession, CallError<TasksError>> {
        let response =
            Self::read_worker_session_http(self, auth_header, project_id, worker_session_id)
                .await?;
        Ok(response)
    }
    async fn register_worker(
        &self,
        auth_header: AuthHeader,
        request: RegisterWorkerRequest,
    ) -> Result<RegisterWorkerResponse, CallError<TasksError>> {
        let response = Self::register_worker_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn retire_worker_pool(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        worker_pool_id: Uuid,
    ) -> Result<WorkerPoolResponse, CallError<TasksError>> {
        let response =
            Self::retire_worker_pool_http(self, auth_header, project_id, worker_pool_id).await?;
        Ok(response)
    }
    async fn run_trigger(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: RunTriggerRequest,
    ) -> Result<TriggerInvocation, CallError<TasksError>> {
        let response =
            Self::run_trigger_http(self, auth_header, project_id, trigger_id, request).await?;
        Ok(response)
    }
    async fn submit_task(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        request: SubmitTaskRequest,
    ) -> Result<CreateTaskResponse, CallError<TasksError>> {
        let response = Self::submit_task_http(self, auth_header, project_id, request).await?;
        Ok(response)
    }
    async fn test_trigger(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: TestTriggerRequest,
    ) -> Result<TestTriggerResponse, CallError<TasksError>> {
        let response =
            Self::test_trigger_http(self, auth_header, project_id, trigger_id, request).await?;
        Ok(response)
    }
    async fn update_trigger(
        &self,
        auth_header: AuthHeader,
        project_id: String,
        trigger_id: Uuid,
        request: UpdateTriggerRequest,
    ) -> Result<TaskTrigger, CallError<TasksError>> {
        let response =
            Self::update_trigger_http(self, auth_header, project_id, trigger_id, request).await?;
        Ok(response)
    }
}