Genarrative/server-rs/crates/platform-oss/src/lib.rs

use std::{collections::BTreeMap, error::Error, fmt, future::Future, sync::Arc, time::Instant};

use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
use bytes::Bytes;
use hmac::{Hmac, Mac};
use reqwest::Method;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use time::{Duration, OffsetDateTime, format_description::well_known::Rfc3339};
use tokio::{sync::Semaphore, time::sleep};
use tracing::{info, warn};

pub mod client_downloads;
pub mod project_snapshots;

type HmacSha256 = Hmac<Sha256>;

pub const DEFAULT_POST_EXPIRE_SECONDS: u64 = 10 * 60;
pub const DEFAULT_READ_EXPIRE_SECONDS: u64 = 10 * 60;
pub const DEFAULT_POST_MAX_SIZE_BYTES: u64 = 20 * 1024 * 1024;
pub const DEFAULT_SUCCESS_ACTION_STATUS: u16 = 200;
pub const DEFAULT_METADATA_TOTAL_BYTES_LIMIT: usize = 8 * 1024;
pub const DEFAULT_IMMUTABLE_CACHE_CONTROL: &str = "public, max-age=31536000, immutable";
const OSS_V4_ALGORITHM: &str = "OSS4-HMAC-SHA256";
const OSS_V4_REQUEST: &str = "aliyun_v4_request";
const OSS_V4_SERVICE: &str = "oss";
const OSS_UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
const OSS_PROVIDER: &str = "aliyun-oss";

pub const LEGACY_PUBLIC_PREFIXES: [&str; 14] = [
    "generated-character-drafts",
    "generated-characters",
    "generated-animations",
    "generated-big-fish-assets",
    "generated-square-hole-assets",
    "generated-wooden-fish-assets",
    "generated-match3d-assets",
    "generated-puzzle-assets",
    "generated-puzzle-clear-assets",
    "generated-jump-hop-assets",
    "generated-custom-world-scenes",
    "generated-custom-world-covers",
    "generated-bark-battle-assets",
    "generated-qwen-sprites",
];

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OssObjectAccess {
    Public,
    Private,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LegacyAssetPrefix {
    EditorAgent,
    AgcErrorReports,
    CharacterDrafts,
    Characters,
    Animations,
    BigFishAssets,
    SquareHoleAssets,
    WoodenFishAssets,
    Match3DAssets,
    PuzzleAssets,
    PuzzleClearAssets,
    JumpHopAssets,
    CustomWorldScenes,
    CustomWorldCovers,
    BarkBattleAssets,
    QwenSprites,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssConfig {
    bucket: String,
    endpoint: String,
    access_key_id: String,
    access_key_secret: String,
    default_read_expire_seconds: u64,
    default_post_expire_seconds: u64,
    default_post_max_size_bytes: u64,
    default_success_action_status: u16,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssPostObjectRequest {
    pub prefix: LegacyAssetPrefix,
    pub path_segments: Vec<String>,
    pub file_name: String,
    pub content_type: Option<String>,
    pub access: OssObjectAccess,
    pub metadata: BTreeMap<String, String>,
    pub max_size_bytes: Option<u64>,
    pub expire_seconds: Option<u64>,
    pub success_action_status: Option<u16>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssSignedGetObjectUrlRequest {
    pub object_key: String,
    pub expire_seconds: Option<u64>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssHeadObjectRequest {
    pub object_key: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssGetObjectRequest {
    pub object_key: String,
    pub max_bytes: usize,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssDeleteObjectRequest {
    pub object_key: String,
}

/// 服务端专用内部对象写入请求。
///
/// 与 `OssPutObjectRequest` 的区别是对象键由调用方按内部前缀完整给出，不再走
/// `path_segments`/`file_name` 的低位规范化，因此可以保留项目内的原始大小写与
/// 目录层级。键必须落在 `normalize_internal_object_key` 允许的内部前缀下。
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssInternalPutObjectRequest {
    pub object_key: String,
    pub content_type: Option<String>,
    pub access: OssObjectAccess,
    pub metadata: BTreeMap<String, String>,
    pub body: Vec<u8>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssPutObjectRequest {
    pub prefix: LegacyAssetPrefix,
    pub path_segments: Vec<String>,
    pub file_name: String,
    pub content_type: Option<String>,
    pub access: OssObjectAccess,
    pub metadata: BTreeMap<String, String>,
    pub body: Vec<u8>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OssPostObjectResponse {
    #[serde(rename = "signatureVersion")]
    pub signature_version: &'static str,
    pub provider: &'static str,
    pub bucket: String,
    pub endpoint: String,
    pub host: String,
    #[serde(rename = "objectKey")]
    pub object_key: String,
    #[serde(rename = "legacyPublicPath")]
    pub legacy_public_path: String,
    #[serde(rename = "contentType", skip_serializing_if = "Option::is_none")]
    pub content_type: Option<String>,
    pub access: OssObjectAccess,
    #[serde(rename = "keyPrefix")]
    pub key_prefix: String,
    #[serde(rename = "expiresAt")]
    pub expires_at: String,
    #[serde(rename = "maxSizeBytes")]
    pub max_size_bytes: u64,
    #[serde(rename = "successActionStatus")]
    pub success_action_status: u16,
    #[serde(rename = "formFields")]
    pub form_fields: OssPostObjectFormFields,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OssSignedGetObjectUrlResponse {
    pub provider: &'static str,
    pub bucket: String,
    pub endpoint: String,
    pub host: String,
    #[serde(rename = "objectKey")]
    pub object_key: String,
    #[serde(rename = "expiresAt")]
    pub expires_at: String,
    #[serde(rename = "signedUrl")]
    pub signed_url: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssHeadObjectResponse {
    pub bucket: String,
    pub object_key: String,
    pub content_length: u64,
    pub content_type: Option<String>,
    pub etag: Option<String>,
    pub last_modified: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OssPutObjectResponse {
    pub provider: &'static str,
    pub bucket: String,
    pub endpoint: String,
    pub host: String,
    #[serde(rename = "objectKey")]
    pub object_key: String,
    #[serde(rename = "legacyPublicPath")]
    pub legacy_public_path: String,
    #[serde(rename = "contentType", skip_serializing_if = "Option::is_none")]
    pub content_type: Option<String>,
    #[serde(rename = "contentLength")]
    pub content_length: u64,
    pub access: OssObjectAccess,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub etag: Option<String>,
    #[serde(rename = "lastModified", skip_serializing_if = "Option::is_none")]
    pub last_modified: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OssPostObjectFormFields {
    pub key: String,
    pub policy: String,
    #[serde(rename = "x-oss-signature-version")]
    pub signature_version: String,
    #[serde(rename = "x-oss-credential")]
    pub credential: String,
    #[serde(rename = "x-oss-date")]
    pub date: String,
    #[serde(rename = "x-oss-signature")]
    pub signature: String,
    #[serde(rename = "success_action_status")]
    pub success_action_status: String,
    #[serde(rename = "Content-Type", skip_serializing_if = "Option::is_none")]
    pub content_type: Option<String>,
    #[serde(rename = "Cache-Control", skip_serializing_if = "Option::is_none")]
    pub cache_control: Option<String>,
    #[serde(flatten)]
    pub metadata: BTreeMap<String, String>,
}

#[derive(Clone, Debug)]
pub struct OssClient {
    config: OssConfig,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OssRequestOperation {
    Put,
    Head,
    Get,
    Delete,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OssRequestError {
    pub status: Option<u16>,
    pub timeout: bool,
    pub connect: bool,
    pub transport: bool,
    pub oss_code: Option<String>,
    pub oss_request_id: Option<String>,
    pub operation: OssRequestOperation,
    pub message: String,
}

#[derive(Debug, PartialEq, Eq)]
pub enum OssError {
    InvalidConfig(String),
    InvalidRequest(String),
    ObjectNotFound(String),
    Request(OssRequestError),
    SerializePolicy(String),
    Sign(String),
}

// 平台 OSS 错误只先归类，不在 platform 层绑定 HTTP status。
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OssErrorKind {
    InvalidConfig,
    InvalidRequest,
    ObjectNotFound,
    Request,
    SerializePolicy,
    Sign,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OssRequestAttemptContext {
    pub frame_index: usize,
    pub operation: &'static str,
}

const CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS: u32 = 3;
const CHARACTER_ANIMATION_OSS_RETRY_DELAYS_MS: [u64; 2] = [250, 500];
const CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES: usize = 16 * 1024;
const OSS_ERROR_CODE_MAX_BYTES: usize = 128;
const OSS_REQUEST_ID_MAX_BYTES: usize = 256;

struct PreparedHeadObject {
    object_key: String,
    target_url: reqwest::Url,
}

struct PreparedPutObject {
    object_key: String,
    target_url: reqwest::Url,
    content_type: Option<String>,
    headers: BTreeMap<String, String>,
    content_length: u64,
    access: OssObjectAccess,
    body: Bytes,
}

impl LegacyAssetPrefix {
    pub fn parse(raw: &str) -> Option<Self> {
        let normalized = raw
            .trim()
            .trim_start_matches('/')
            .trim_end_matches('/')
            .trim_end_matches('*')
            .trim_end_matches('/');

        match normalized {
            "editor-agent" => Some(Self::EditorAgent),
            "agc" => Some(Self::AgcErrorReports),
            "generated-character-drafts" => Some(Self::CharacterDrafts),
            "generated-characters" => Some(Self::Characters),
            "generated-animations" => Some(Self::Animations),
            "generated-big-fish-assets" => Some(Self::BigFishAssets),
            "generated-square-hole-assets" => Some(Self::SquareHoleAssets),
            "generated-wooden-fish-assets" => Some(Self::WoodenFishAssets),
            "generated-match3d-assets" => Some(Self::Match3DAssets),
            "generated-puzzle-assets" => Some(Self::PuzzleAssets),
            "generated-puzzle-clear-assets" => Some(Self::PuzzleClearAssets),
            "generated-jump-hop-assets" => Some(Self::JumpHopAssets),
            "generated-custom-world-scenes" => Some(Self::CustomWorldScenes),
            "generated-custom-world-covers" => Some(Self::CustomWorldCovers),
            "generated-bark-battle-assets" => Some(Self::BarkBattleAssets),
            "generated-qwen-sprites" => Some(Self::QwenSprites),
            _ => None,
        }
    }

    pub fn as_str(&self) -> &'static str {
        match self {
            Self::EditorAgent => "editor-agent",
            Self::AgcErrorReports => "agc",
            Self::CharacterDrafts => "generated-character-drafts",
            Self::Characters => "generated-characters",
            Self::Animations => "generated-animations",
            Self::BigFishAssets => "generated-big-fish-assets",
            Self::SquareHoleAssets => "generated-square-hole-assets",
            Self::WoodenFishAssets => "generated-wooden-fish-assets",
            Self::Match3DAssets => "generated-match3d-assets",
            Self::PuzzleAssets => "generated-puzzle-assets",
            Self::PuzzleClearAssets => "generated-puzzle-clear-assets",
            Self::JumpHopAssets => "generated-jump-hop-assets",
            Self::CustomWorldScenes => "generated-custom-world-scenes",
            Self::CustomWorldCovers => "generated-custom-world-covers",
            Self::BarkBattleAssets => "generated-bark-battle-assets",
            Self::QwenSprites => "generated-qwen-sprites",
        }
    }

    pub fn as_public_path_prefix(&self) -> String {
        format!("/{}", self.as_str())
    }

    pub fn from_object_key(raw: &str) -> Option<Self> {
        let normalized = raw.trim().trim_start_matches('/').trim();
        let prefix = normalized.split('/').next()?;
        match Self::parse(prefix) {
            // agc/error-reports is a server-only write prefix, never a caller-supplied object key.
            Some(Self::AgcErrorReports) => None,
            other => other,
        }
    }
}

impl OssConfig {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        bucket: String,
        endpoint: String,
        access_key_id: String,
        access_key_secret: String,
        default_read_expire_seconds: u64,
        default_post_expire_seconds: u64,
        default_post_max_size_bytes: u64,
        default_success_action_status: u16,
    ) -> Result<Self, OssError> {
        let bucket = normalize_required_value(bucket, "OSS bucket 不能为空")?;
        let endpoint = normalize_endpoint(&endpoint)?;
        let access_key_id = normalize_required_value(access_key_id, "OSS AccessKeyId 不能为空")?;
        let access_key_secret =
            normalize_required_value(access_key_secret, "OSS AccessKeySecret 不能为空")?;

        if default_read_expire_seconds == 0 {
            return Err(OssError::InvalidConfig(
                "OSS 私有读签名有效期必须大于 0".to_string(),
            ));
        }

        if default_post_expire_seconds == 0 {
            return Err(OssError::InvalidConfig(
                "OSS PostObject 签名有效期必须大于 0".to_string(),
            ));
        }

        if default_post_max_size_bytes == 0 {
            return Err(OssError::InvalidConfig(
                "OSS PostObject 最大上传大小必须大于 0".to_string(),
            ));
        }

        if !(100..=999).contains(&default_success_action_status) {
            return Err(OssError::InvalidConfig(
                "OSS success_action_status 必须是三位 HTTP 状态码".to_string(),
            ));
        }

        Ok(Self {
            bucket,
            endpoint,
            access_key_id,
            access_key_secret,
            default_read_expire_seconds,
            default_post_expire_seconds,
            default_post_max_size_bytes,
            default_success_action_status,
        })
    }

    pub fn upload_host(&self) -> String {
        format!("https://{}.{}", self.bucket, self.endpoint)
    }

    pub fn endpoint(&self) -> &str {
        &self.endpoint
    }

    pub fn bucket(&self) -> &str {
        &self.bucket
    }

    pub fn access_key_id(&self) -> &str {
        &self.access_key_id
    }

    pub fn access_key_secret(&self) -> &str {
        &self.access_key_secret
    }
}

impl OssClient {
    pub fn new(config: OssConfig) -> Self {
        Self { config }
    }

    pub fn config_bucket(&self) -> &str {
        self.config.bucket()
    }

    pub fn sign_post_object(
        &self,
        request: OssPostObjectRequest,
    ) -> Result<OssPostObjectResponse, OssError> {
        let started_at = Instant::now();
        let requested_prefix = request.prefix.as_str();
        let requested_content_type = request
            .content_type
            .as_deref()
            .map(str::trim)
            .unwrap_or("")
            .to_string();
        let requested_metadata_count = request.metadata.len();

        let result = (|| {
            let max_size_bytes = request
                .max_size_bytes
                .unwrap_or(self.config.default_post_max_size_bytes);
            let expire_seconds = request
                .expire_seconds
                .unwrap_or(self.config.default_post_expire_seconds);
            let success_action_status = request
                .success_action_status
                .unwrap_or(self.config.default_success_action_status);

            if max_size_bytes == 0 {
                return Err(OssError::InvalidRequest(
                    "maxSizeBytes 必须大于 0".to_string(),
                ));
            }

            if expire_seconds == 0 {
                return Err(OssError::InvalidRequest(
                    "expireSeconds 必须大于 0".to_string(),
                ));
            }

            if !(100..=999).contains(&success_action_status) {
                return Err(OssError::InvalidRequest(
                    "successActionStatus 必须是三位 HTTP 状态码".to_string(),
                ));
            }

            let sanitized_segments = request
                .path_segments
                .iter()
                .map(|segment| sanitize_path_segment(segment))
                .filter(|segment| !segment.is_empty())
                .collect::<Vec<_>>();
            let file_name = sanitize_file_name(&request.file_name)?;
            let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name);
            let legacy_public_path = format!("/{}", object_key);
            let content_type = normalize_optional_value(request.content_type);
            let metadata = normalize_metadata(request.metadata)?;
            let cache_control = Some(DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string());

            let expires_at = OffsetDateTime::now_utc()
                .checked_add(Duration::seconds(i64::try_from(expire_seconds).map_err(
                    |_| OssError::InvalidRequest("expireSeconds 超出可支持范围".to_string()),
                )?))
                .ok_or_else(|| {
                    OssError::InvalidRequest("expireSeconds 计算结果溢出".to_string())
                })?;
            let expires_at = expires_at.format(&Rfc3339).map_err(|error| {
                OssError::SerializePolicy(format!("格式化过期时间失败：{error}"))
            })?;

            let signed_at = OffsetDateTime::now_utc();
            let signature_scope = build_v4_signature_scope(&self.config.endpoint, signed_at)?;
            let signature_date = build_v4_signature_date(signed_at)?;
            let credential = format!("{}/{}", self.config.access_key_id, signature_scope);
            let policy_json = build_policy_json(
                &self.config.bucket,
                &object_key,
                &expires_at,
                max_size_bytes,
                success_action_status,
                content_type.as_deref(),
                cache_control.as_deref(),
                &metadata,
                &credential,
                &signature_date,
            );
            let policy = serde_json::to_string(&policy_json).map_err(|error| {
                OssError::SerializePolicy(format!("序列化 policy 失败：{error}"))
            })?;
            let encoded_policy = BASE64_STANDARD.encode(policy.as_bytes());
            let signature = sign_v4_content(
                &self.config.access_key_secret,
                &signature_scope,
                &encoded_policy,
            )?;

            Ok(OssPostObjectResponse {
                signature_version: "v4",
                provider: OSS_PROVIDER,
                bucket: self.config.bucket.clone(),
                endpoint: self.config.endpoint.clone(),
                host: self.config.upload_host(),
                object_key: object_key.clone(),
                legacy_public_path,
                content_type: content_type.clone(),
                access: request.access,
                key_prefix: build_key_prefix(request.prefix, &sanitized_segments),
                expires_at,
                max_size_bytes,
                success_action_status,
                form_fields: OssPostObjectFormFields {
                    key: object_key,
                    policy: encoded_policy,
                    signature_version: OSS_V4_ALGORITHM.to_string(),
                    credential,
                    date: signature_date,
                    signature,
                    success_action_status: success_action_status.to_string(),
                    content_type,
                    cache_control,
                    metadata,
                },
            })
        })();

        match &result {
            Ok(response) => info!(
                provider = OSS_PROVIDER,
                operation = "sign_post_object",
                bucket = %response.bucket,
                endpoint = %response.endpoint,
                object_key = %response.object_key,
                key_prefix = %response.key_prefix,
                access = oss_access_label(response.access),
                content_type = %response.content_type.as_deref().unwrap_or(""),
                max_size_bytes = response.max_size_bytes,
                success_action_status = response.success_action_status,
                metadata_count = response.form_fields.metadata.len(),
                expires_at = %response.expires_at,
                elapsed_ms = elapsed_ms(started_at),
                "OSS PostObject 签名完成"
            ),
            Err(error) => warn!(
                provider = OSS_PROVIDER,
                operation = "sign_post_object",
                bucket = %self.config.bucket(),
                endpoint = %self.config.endpoint(),
                key_prefix = requested_prefix,
                content_type = %requested_content_type,
                metadata_count = requested_metadata_count,
                error_kind = oss_error_kind_label(error),
                message = %error,
                elapsed_ms = elapsed_ms(started_at),
                "OSS PostObject 签名失败"
            ),
        }

        result
    }

    // 私有 bucket 的对象读取统一走短期签名 URL，避免把长期主凭证下发给浏览器。
    pub fn sign_get_object_url(
        &self,
        request: OssSignedGetObjectUrlRequest,
    ) -> Result<OssSignedGetObjectUrlResponse, OssError> {
        self.sign_get_object_url_with_normalizer(
            request,
            normalize_object_key,
            "sign_get_object_url",
        )
    }

    pub fn sign_internal_get_object_url(
        &self,
        request: OssSignedGetObjectUrlRequest,
    ) -> Result<OssSignedGetObjectUrlResponse, OssError> {
        self.sign_get_object_url_with_normalizer(
            request,
            normalize_editor_agent_messages_object_key,
            "sign_internal_get_object_url",
        )
    }

    fn sign_get_object_url_with_normalizer(
        &self,
        request: OssSignedGetObjectUrlRequest,
        normalize: fn(&str) -> Result<String, OssError>,
        operation: &'static str,
    ) -> Result<OssSignedGetObjectUrlResponse, OssError> {
        let started_at = Instant::now();
        let requested_object_key = request
            .object_key
            .trim()
            .trim_start_matches('/')
            .trim()
            .to_string();

        let result = (|| {
            let expire_seconds = request
                .expire_seconds
                .unwrap_or(self.config.default_read_expire_seconds);

            if expire_seconds == 0 {
                return Err(OssError::InvalidRequest(
                    "expireSeconds 必须大于 0".to_string(),
                ));
            }

            let object_key = normalize(&request.object_key)?;
            let expires_at = OffsetDateTime::now_utc()
                .checked_add(Duration::seconds(i64::try_from(expire_seconds).map_err(
                    |_| OssError::InvalidRequest("expireSeconds 超出可支持范围".to_string()),
                )?))
                .ok_or_else(|| {
                    OssError::InvalidRequest("expireSeconds 计算结果溢出".to_string())
                })?;
            let expires_at_text = expires_at
                .format(&Rfc3339)
                .map_err(|error| OssError::Sign(format!("格式化过期时间失败：{error}")))?;

            let signed_at = OffsetDateTime::now_utc();
            let signed_at_text = build_v4_signature_date(signed_at)?;
            let signature_scope = build_v4_signature_scope(&self.config.endpoint, signed_at)?;
            let credential = format!("{}/{}", self.config.access_key_id, signature_scope);
            let mut query = BTreeMap::from([
                ("x-oss-additional-headers".to_string(), "host".to_string()),
                (
                    "x-oss-signature-version".to_string(),
                    OSS_V4_ALGORITHM.to_string(),
                ),
                ("x-oss-credential".to_string(), credential),
                ("x-oss-date".to_string(), signed_at_text),
                ("x-oss-expires".to_string(), expire_seconds.to_string()),
            ]);
            let canonical_uri = build_v4_canonical_uri(&self.config.bucket, Some(&object_key));
            let object_url_path = format!("/{}", encode_url_path(&object_key));
            let additional_headers = "host";
            let canonical_headers =
                format!("host:{}.{}\n", self.config.bucket(), self.config.endpoint());
            let canonical_query = build_canonical_query_string(&query);
            let canonical_request = build_v4_canonical_request(
                Method::GET.as_str(),
                &canonical_uri,
                &canonical_query,
                &canonical_headers,
                additional_headers,
                OSS_UNSIGNED_PAYLOAD,
            );
            let string_to_sign = build_v4_string_to_sign(
                query["x-oss-date"].as_str(),
                &signature_scope,
                &canonical_request,
            );
            let signature = sign_v4_content(
                &self.config.access_key_secret,
                &signature_scope,
                &string_to_sign,
            )?;
            query.insert("x-oss-signature".to_string(), signature);
            let signed_url = format!(
                "{}{}?{}",
                self.config.upload_host(),
                object_url_path,
                build_canonical_query_string(&query)
            );

            Ok(OssSignedGetObjectUrlResponse {
                provider: OSS_PROVIDER,
                bucket: self.config.bucket.clone(),
                endpoint: self.config.endpoint.clone(),
                host: self.config.upload_host(),
                object_key,
                expires_at: expires_at_text,
                signed_url,
            })
        })();

        match &result {
            Ok(response) => info!(
                provider = OSS_PROVIDER,
                operation,
                bucket = %response.bucket,
                endpoint = %response.endpoint,
                object_key = %response.object_key,
                expires_at = %response.expires_at,
                elapsed_ms = elapsed_ms(started_at),
                "OSS GetObject 读签名完成"
            ),
            Err(error) => warn!(
                provider = OSS_PROVIDER,
                operation,
                bucket = %self.config.bucket(),
                endpoint = %self.config.endpoint(),
                object_key = %requested_object_key,
                error_kind = oss_error_kind_label(error),
                message = %error,
                elapsed_ms = elapsed_ms(started_at),
                "OSS GetObject 读签名失败"
            ),
        }

        result
    }

    // 上传完成确认前，服务端必须自己探测一次对象，不能只相信客户端回传的 object_key。
    pub async fn head_object(
        &self,
        client: &reqwest::Client,
        request: OssHeadObjectRequest,
    ) -> Result<OssHeadObjectResponse, OssError> {
        let started_at = Instant::now();
        let requested_object_key = request
            .object_key
            .trim()
            .trim_start_matches('/')
            .trim()
            .to_string();
        let mut response_status = None;

        let result = async {
            let object_key = normalize_object_key(&request.object_key)?;
            let target_url =
                build_object_url(&self.config.bucket, &self.config.endpoint, &object_key).map_err(
                    |error| {
                        request_error(
                            OssRequestOperation::Head,
                            &format!("构造 OSS 对象 URL 失败：{error}"),
                        )
                    },
                )?;
            let response = send_signed_request(
                client,
                &self.config,
                Method::HEAD,
                Some(&object_key),
                target_url,
                OssRequestOperation::Head,
            )
            .await?;
            response_status = Some(response.status().as_u16());

            if response.status() == reqwest::StatusCode::NOT_FOUND {
                return Err(OssError::ObjectNotFound(format!(
                    "OSS 对象不存在：{}",
                    request.object_key
                )));
            }

            if !response.status().is_success() {
                return Err(request_status_error(
                    OssRequestOperation::Head,
                    response.status().as_u16(),
                    format!("OSS HEAD Object 失败，状态码：{}", response.status()),
                ));
            }

            let headers = response.headers();
            let content_length = head_object_content_length(headers);
            let content_type = headers
                .get(reqwest::header::CONTENT_TYPE)
                .and_then(|value| value.to_str().ok())
                .map(|value| value.to_string());
            let etag = headers
                .get(reqwest::header::ETAG)
                .and_then(|value| value.to_str().ok())
                .map(|value| value.trim_matches('"').to_string());
            let last_modified = headers
                .get(reqwest::header::LAST_MODIFIED)
                .and_then(|value| value.to_str().ok())
                .map(|value| value.to_string());

            Ok(OssHeadObjectResponse {
                bucket: self.config.bucket.clone(),
                object_key,
                content_length,
                content_type,
                etag,
                last_modified,
            })
        }
        .await;

        match &result {
            Ok(response) => info!(
                provider = OSS_PROVIDER,
                operation = "head_object",
                bucket = %response.bucket,
                endpoint = %self.config.endpoint(),
                object_key = %response.object_key,
                status = response_status.unwrap_or(reqwest::StatusCode::OK.as_u16()),
                status_class = http_status_class_from_option(response_status),
                content_length = response.content_length,
                content_type = %response.content_type.as_deref().unwrap_or(""),
                etag_present = response.etag.is_some(),
                last_modified_present = response.last_modified.is_some(),
                elapsed_ms = elapsed_ms(started_at),
                "OSS HEAD Object 完成"
            ),
            Err(error) => warn!(
                provider = OSS_PROVIDER,
                operation = "head_object",
                bucket = %self.config.bucket(),
                endpoint = %self.config.endpoint(),
                object_key = %requested_object_key,
                status = response_status.unwrap_or_default(),
                status_class = http_status_class_from_option(response_status),
                error_kind = oss_error_kind_label(error),
                message = %error,
                elapsed_ms = elapsed_ms(started_at),
                "OSS HEAD Object 失败"
            ),
        }

        result
    }

    pub async fn get_object(
        &self,
        client: &reqwest::Client,
        request: OssGetObjectRequest,
    ) -> Result<Vec<u8>, OssError> {
        let key = normalize_internal_object_key(&request.object_key)?;
        let target = build_object_url(&self.config.bucket, &self.config.endpoint, &key)
            .map_err(|e| request_error(OssRequestOperation::Get, &e.to_string()))?;
        let mut response = send_signed_request(
            client,
            &self.config,
            Method::GET,
            Some(&key),
            target,
            OssRequestOperation::Get,
        )
        .await?;
        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(OssError::ObjectNotFound(format!("OSS 对象不存在：{key}")));
        }
        if !response.status().is_success() {
            return Err(request_status_error(
                OssRequestOperation::Get,
                response.status().as_u16(),
                format!("OSS GET Object 失败，状态码：{}", response.status()),
            ));
        }
        if response
            .content_length()
            .is_some_and(|length| length > request.max_bytes as u64)
        {
            return Err(OssError::InvalidRequest("OSS 对象超过读取上限".to_string()));
        }
        let mut bytes = Vec::new();
        while let Some(chunk) = response
            .chunk()
            .await
            .map_err(|e| request_error_from_reqwest(OssRequestOperation::Get, e))?
        {
            if bytes.len().saturating_add(chunk.len()) > request.max_bytes {
                return Err(OssError::InvalidRequest("OSS 对象超过读取上限".to_string()));
            }
            bytes.extend_from_slice(&chunk);
        }
        Ok(bytes)
    }

    pub async fn delete_object(
        &self,
        client: &reqwest::Client,
        request: OssDeleteObjectRequest,
    ) -> Result<(), OssError> {
        let key = normalize_internal_object_key(&request.object_key)?;
        let target = build_object_url(&self.config.bucket, &self.config.endpoint, &key)
            .map_err(|e| request_error(OssRequestOperation::Delete, &e.to_string()))?;
        let response = send_signed_request(
            client,
            &self.config,
            Method::DELETE,
            Some(&key),
            target,
            OssRequestOperation::Delete,
        )
        .await?;
        if response.status() == reqwest::StatusCode::NOT_FOUND || response.status().is_success() {
            return Ok(());
        }
        Err(request_status_error(
            OssRequestOperation::Delete,
            response.status().as_u16(),
            format!("OSS DELETE Object 失败，状态码：{}", response.status()),
        ))
    }

    /// 按内部前缀写入服务端专用对象；项目快照允许合法的零字节工程文件。
    pub async fn put_internal_object(
        &self,
        client: &reqwest::Client,
        request: OssInternalPutObjectRequest,
    ) -> Result<OssPutObjectResponse, OssError> {
        let object_key = normalize_internal_object_key(&request.object_key)?;
        if request.body.is_empty() && !project_snapshots::is_snapshot_file_key(&object_key) {
            return Err(OssError::InvalidRequest(
                "服务端内部对象内容不能为空".to_string(),
            ));
        }
        let content_type = normalize_optional_value(request.content_type);
        let headers = build_put_object_headers(request.metadata)?;
        let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key)
            .map_err(|error| {
                request_error(
                    OssRequestOperation::Put,
                    &format!("构造 OSS 对象 URL 失败：{error}"),
                )
            })?;
        let content_length = u64::try_from(request.body.len())
            .map_err(|_| OssError::InvalidRequest("上传对象大小超出可支持范围".to_string()))?;
        let builder = signed_request_builder(
            client,
            &self.config,
            Method::PUT,
            Some(&object_key),
            target_url,
            content_type.as_deref(),
            &headers,
        )?
        .header(reqwest::header::CONTENT_LENGTH, content_length)
        .body(request.body);
        let response = builder
            .send()
            .await
            .map_err(|error| request_error_from_reqwest(OssRequestOperation::Put, error))?;
        if !response.status().is_success() {
            return Err(request_status_error(
                OssRequestOperation::Put,
                response.status().as_u16(),
                format!("OSS PutObject 失败，状态码：{}", response.status()),
            ));
        }
        let headers = response.headers();
        let etag = headers
            .get(reqwest::header::ETAG)
            .and_then(|value| value.to_str().ok())
            .map(|value| value.trim_matches('"').to_string());
        let last_modified = headers
            .get(reqwest::header::LAST_MODIFIED)
            .and_then(|value| value.to_str().ok())
            .map(|value| value.to_string());
        Ok(OssPutObjectResponse {
            provider: OSS_PROVIDER,
            bucket: self.config.bucket.clone(),
            endpoint: self.config.endpoint.clone(),
            host: self.config.upload_host(),
            legacy_public_path: format!("/{object_key}"),
            object_key,
            content_type,
            content_length,
            access: request.access,
            etag,
            last_modified,
        })
    }

    /// 探测内部对象是否存在。`Ok(None)` 表示确定不存在，其余失败都按上游错误返回，
    /// 调用方不能把不确定当成"不存在"。
    pub async fn head_internal_object(
        &self,
        client: &reqwest::Client,
        object_key: &str,
    ) -> Result<Option<OssHeadObjectResponse>, OssError> {
        let object_key = normalize_internal_object_key(object_key)?;
        let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key)
            .map_err(|error| {
                request_error(
                    OssRequestOperation::Head,
                    &format!("构造 OSS 对象 URL 失败：{error}"),
                )
            })?;
        let response = send_signed_request(
            client,
            &self.config,
            Method::HEAD,
            Some(&object_key),
            target_url,
            OssRequestOperation::Head,
        )
        .await?;
        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Ok(None);
        }
        if !response.status().is_success() {
            return Err(request_status_error(
                OssRequestOperation::Head,
                response.status().as_u16(),
                format!("OSS HEAD Object 失败，状态码：{}", response.status()),
            ));
        }
        let headers = response.headers();
        Ok(Some(OssHeadObjectResponse {
            bucket: self.config.bucket.clone(),
            object_key,
            content_length: head_object_content_length(headers),
            content_type: headers
                .get(reqwest::header::CONTENT_TYPE)
                .and_then(|value| value.to_str().ok())
                .map(|value| value.to_string()),
            etag: headers
                .get(reqwest::header::ETAG)
                .and_then(|value| value.to_str().ok())
                .map(|value| value.trim_matches('"').to_string()),
            last_modified: headers
                .get(reqwest::header::LAST_MODIFIED)
                .and_then(|value| value.to_str().ok())
                .map(|value| value.to_string()),
        }))
    }

    // AI 生成资源默认由服务端上传 OSS，Web 端只拿签名读地址，不直接持有写权限。
    pub async fn put_object(
        &self,
        client: &reqwest::Client,
        request: OssPutObjectRequest,
    ) -> Result<OssPutObjectResponse, OssError> {
        let started_at = Instant::now();
        let requested_prefix = request.prefix.as_str();
        let requested_content_type = request
            .content_type
            .as_deref()
            .map(str::trim)
            .unwrap_or("")
            .to_string();
        let requested_content_length = request.body.len();
        let requested_metadata_count = request.metadata.len();
        let mut response_status = None;

        let result = async {
            if request.body.is_empty() {
                return Err(OssError::InvalidRequest(
                    "服务端上传对象内容不能为空".to_string(),
                ));
            }

            let sanitized_segments = request
                .path_segments
                .iter()
                .map(|segment| sanitize_path_segment(segment))
                .filter(|segment| !segment.is_empty())
                .collect::<Vec<_>>();
            let file_name = sanitize_file_name(&request.file_name)?;
            let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name);
            let content_type = normalize_optional_value(request.content_type);
            let headers = build_put_object_headers(request.metadata)?;
            let target_url =
                build_object_url(&self.config.bucket, &self.config.endpoint, &object_key).map_err(
                    |error| {
                        request_error(
                            OssRequestOperation::Put,
                            &format!("构造 OSS 对象 URL 失败：{error}"),
                        )
                    },
                )?;
            let content_length = u64::try_from(request.body.len())
                .map_err(|_| OssError::InvalidRequest("上传对象大小超出可支持范围".to_string()))?;
            let builder = signed_request_builder(
                client,
                &self.config,
                Method::PUT,
                Some(&object_key),
                target_url,
                content_type.as_deref(),
                &headers,
            )?
            .header(reqwest::header::CONTENT_LENGTH, content_length)
            .body(request.body);

            let response = builder
                .send()
                .await
                .map_err(|error| request_error_from_reqwest(OssRequestOperation::Put, error))?;
            response_status = Some(response.status().as_u16());

            if !response.status().is_success() {
                return Err(request_status_error(
                    OssRequestOperation::Put,
                    response.status().as_u16(),
                    format!("OSS PutObject 失败，状态码：{}", response.status()),
                ));
            }

            let headers = response.headers();
            let etag = headers
                .get(reqwest::header::ETAG)
                .and_then(|value| value.to_str().ok())
                .map(|value| value.trim_matches('"').to_string());
            let last_modified = headers
                .get(reqwest::header::LAST_MODIFIED)
                .and_then(|value| value.to_str().ok())
                .map(|value| value.to_string());

            Ok(OssPutObjectResponse {
                provider: OSS_PROVIDER,
                bucket: self.config.bucket.clone(),
                endpoint: self.config.endpoint.clone(),
                host: self.config.upload_host(),
                legacy_public_path: format!("/{object_key}"),
                object_key,
                content_type,
                content_length,
                access: request.access,
                etag,
                last_modified,
            })
        }
        .await;

        match &result {
            Ok(response) => info!(
                provider = OSS_PROVIDER,
                operation = "put_object",
                bucket = %response.bucket,
                endpoint = %response.endpoint,
                object_key = %response.object_key,
                access = oss_access_label(response.access),
                status = response_status.unwrap_or(reqwest::StatusCode::OK.as_u16()),
                status_class = http_status_class_from_option(response_status),
                content_length = response.content_length,
                content_type = %response.content_type.as_deref().unwrap_or(""),
                etag_present = response.etag.is_some(),
                last_modified_present = response.last_modified.is_some(),
                elapsed_ms = elapsed_ms(started_at),
                "OSS PutObject 上传完成"
            ),
            Err(error) => warn!(
                provider = OSS_PROVIDER,
                operation = "put_object",
                bucket = %self.config.bucket(),
                endpoint = %self.config.endpoint(),
                key_prefix = requested_prefix,
                content_length = requested_content_length,
                content_type = %requested_content_type,
                metadata_count = requested_metadata_count,
                status = response_status.unwrap_or_default(),
                status_class = http_status_class_from_option(response_status),
                error_kind = oss_error_kind_label(error),
                message = %error,
                elapsed_ms = elapsed_ms(started_at),
                "OSS PutObject 上传失败"
            ),
        }

        result
    }

    /// 角色动画帧专用的可重试 PUT。调用方传入进程级并发限制器，单次网络 attempt
    /// 独占一个 permit，退避等待期间不会占用 permit。
    pub async fn put_object_with_retry(
        &self,
        client: &reqwest::Client,
        request: OssPutObjectRequest,
        io_limiter: Arc<Semaphore>,
        attempt_context: OssRequestAttemptContext,
    ) -> Result<OssPutObjectResponse, OssError> {
        let prepared = self.prepare_put_object(request)?;
        let object_key = prepared.object_key.clone();

        run_animation_request_with_retry(io_limiter, attempt_context, &object_key, || {
            self.put_object_once(client, &prepared)
        })
        .await
    }

    /// 角色动画帧专用的可重试 HEAD。HEAD 与 PUT 独立重试，HEAD 失败不会重新上传 PUT。
    pub async fn head_object_with_retry(
        &self,
        client: &reqwest::Client,
        request: OssHeadObjectRequest,
        io_limiter: Arc<Semaphore>,
        attempt_context: OssRequestAttemptContext,
    ) -> Result<OssHeadObjectResponse, OssError> {
        let prepared = self.prepare_head_object(request)?;
        let object_key = prepared.object_key.clone();

        run_animation_request_with_retry(io_limiter, attempt_context, &object_key, || {
            self.head_object_once(client, &prepared)
        })
        .await
    }

    fn prepare_head_object(
        &self,
        request: OssHeadObjectRequest,
    ) -> Result<PreparedHeadObject, OssError> {
        let object_key = normalize_object_key(&request.object_key)?;
        let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key)
            .map_err(|error| {
                request_error(
                    OssRequestOperation::Head,
                    &format!("构造 OSS 对象 URL 失败：{error}"),
                )
            })?;
        Ok(PreparedHeadObject {
            object_key,
            target_url,
        })
    }

    async fn head_object_once(
        &self,
        client: &reqwest::Client,
        prepared: &PreparedHeadObject,
    ) -> Result<(OssHeadObjectResponse, u16), OssError> {
        let response = send_signed_request(
            client,
            &self.config,
            Method::HEAD,
            Some(&prepared.object_key),
            prepared.target_url.clone(),
            OssRequestOperation::Head,
        )
        .await?;

        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(OssError::ObjectNotFound(format!(
                "OSS 对象不存在：{}",
                prepared.object_key
            )));
        }
        if !response.status().is_success() {
            return Err(request_status_error(
                OssRequestOperation::Head,
                response.status().as_u16(),
                format!("OSS HEAD Object 失败，状态码：{}", response.status()),
            ));
        }

        let headers = response.headers();
        let content_length = headers
            .get(reqwest::header::CONTENT_LENGTH)
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.parse::<u64>().ok())
            .unwrap_or(0);
        let content_type = headers
            .get(reqwest::header::CONTENT_TYPE)
            .and_then(|value| value.to_str().ok())
            .map(str::to_string);
        let etag = headers
            .get(reqwest::header::ETAG)
            .and_then(|value| value.to_str().ok())
            .map(|value| value.trim_matches('"').to_string());
        let last_modified = headers
            .get(reqwest::header::LAST_MODIFIED)
            .and_then(|value| value.to_str().ok())
            .map(str::to_string);

        Ok((
            OssHeadObjectResponse {
                bucket: self.config.bucket.clone(),
                object_key: prepared.object_key.clone(),
                content_length,
                content_type,
                etag,
                last_modified,
            },
            response.status().as_u16(),
        ))
    }

    fn prepare_put_object(
        &self,
        request: OssPutObjectRequest,
    ) -> Result<PreparedPutObject, OssError> {
        if request.body.is_empty() {
            return Err(OssError::InvalidRequest(
                "服务端上传对象内容不能为空".to_string(),
            ));
        }

        let sanitized_segments = request
            .path_segments
            .iter()
            .map(|segment| sanitize_path_segment(segment))
            .filter(|segment| !segment.is_empty())
            .collect::<Vec<_>>();
        let file_name = sanitize_file_name(&request.file_name)?;
        let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name);
        let content_type = normalize_optional_value(request.content_type);
        let headers = build_put_object_headers(request.metadata)?;
        let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key)
            .map_err(|error| {
                request_error(
                    OssRequestOperation::Put,
                    &format!("构造 OSS 对象 URL 失败：{error}"),
                )
            })?;
        let content_length = u64::try_from(request.body.len())
            .map_err(|_| OssError::InvalidRequest("上传对象大小超出可支持范围".to_string()))?;

        Ok(PreparedPutObject {
            object_key,
            target_url,
            content_type,
            headers,
            content_length,
            access: request.access,
            body: Bytes::from(request.body),
        })
    }

    async fn put_object_once(
        &self,
        client: &reqwest::Client,
        prepared: &PreparedPutObject,
    ) -> Result<(OssPutObjectResponse, u16), OssError> {
        let response = signed_request_builder(
            client,
            &self.config,
            Method::PUT,
            Some(&prepared.object_key),
            prepared.target_url.clone(),
            prepared.content_type.as_deref(),
            &prepared.headers,
        )?
        .header(reqwest::header::CONTENT_LENGTH, prepared.content_length)
        .body(prepared.body.clone())
        .send()
        .await
        .map_err(|error| request_error_from_reqwest(OssRequestOperation::Put, error))?;

        if !response.status().is_success() {
            return Err(request_status_error_from_oss_put_response(response).await);
        }

        let headers = response.headers();
        let etag = headers
            .get(reqwest::header::ETAG)
            .and_then(|value| value.to_str().ok())
            .map(|value| value.trim_matches('"').to_string());
        let last_modified = headers
            .get(reqwest::header::LAST_MODIFIED)
            .and_then(|value| value.to_str().ok())
            .map(str::to_string);

        Ok((
            OssPutObjectResponse {
                provider: OSS_PROVIDER,
                bucket: self.config.bucket.clone(),
                endpoint: self.config.endpoint.clone(),
                host: self.config.upload_host(),
                legacy_public_path: format!("/{}", prepared.object_key),
                object_key: prepared.object_key.clone(),
                content_type: prepared.content_type.clone(),
                content_length: prepared.content_length,
                access: prepared.access,
                etag,
                last_modified,
            },
            response.status().as_u16(),
        ))
    }
}

fn request_error(operation: OssRequestOperation, message: &str) -> OssError {
    OssError::Request(OssRequestError {
        status: None,
        timeout: false,
        connect: false,
        transport: false,
        oss_code: None,
        oss_request_id: None,
        operation,
        message: message.to_string(),
    })
}

async fn request_status_error_from_oss_put_response(mut response: reqwest::Response) -> OssError {
    let status = response.status();
    let header_request_id = response
        .headers()
        .get("x-oss-request-id")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| normalize_oss_error_field(value, OSS_REQUEST_ID_MAX_BYTES));
    let body_read = if status == reqwest::StatusCode::BAD_REQUEST {
        read_bounded_oss_error_body(&mut response).await
    } else {
        OssErrorBodyRead {
            body: Vec::new(),
            read_failure: None,
        }
    };

    request_status_error_from_oss_parts(
        OssRequestOperation::Put,
        status.as_u16(),
        header_request_id,
        &body_read.body,
        body_read.read_failure,
    )
}

/// 400 错误响应体读取失败的原因。仅在部分响应体尚未解析出确定性
/// OSS 错误码时参与重试判定,否则只体现在 message 里。
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct OssErrorBodyReadFailure {
    timeout: bool,
}

struct OssErrorBodyRead {
    body: Vec<u8>,
    read_failure: Option<OssErrorBodyReadFailure>,
}

async fn read_bounded_oss_error_body(response: &mut reqwest::Response) -> OssErrorBodyRead {
    let mut body = Vec::new();
    let mut read_failure = None;
    while body.len() < CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES {
        let chunk = match response.chunk().await {
            Ok(Some(chunk)) => chunk,
            Ok(None) => break,
            Err(error) => {
                // 断流/超时不丢弃已读字节:部分响应体可能已含确定性错误码。
                read_failure = Some(OssErrorBodyReadFailure {
                    timeout: error.is_timeout(),
                });
                break;
            }
        };
        let remaining = CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES - body.len();
        body.extend_from_slice(&chunk[..chunk.len().min(remaining)]);
        if chunk.len() > remaining {
            break;
        }
    }
    OssErrorBodyRead { body, read_failure }
}

fn request_status_error_from_oss_parts(
    operation: OssRequestOperation,
    status: u16,
    header_request_id: Option<String>,
    body: &[u8],
    body_read_failure: Option<OssErrorBodyReadFailure>,
) -> OssError {
    let header_request_id = header_request_id
        .as_deref()
        .and_then(|value| normalize_oss_error_field(value, OSS_REQUEST_ID_MAX_BYTES));
    let bounded_body = &body[..body.len().min(CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES)];
    let oss_code = extract_oss_error_xml_field(bounded_body, "Code", OSS_ERROR_CODE_MAX_BYTES);
    let xml_request_id =
        extract_oss_error_xml_field(bounded_body, "RequestId", OSS_REQUEST_ID_MAX_BYTES);
    let oss_request_id = header_request_id.or(xml_request_id);
    // 确定性错误码优先:已解析出 Code 时,响应体读取失败只保留在 message 里,
    // 不改变重试语义;错误码缺失时才按读取失败归类为可重试的超时/传输错误。
    let unclassified_read_failure = if oss_code.is_some() {
        None
    } else {
        body_read_failure
    };
    let timeout = (status == reqwest::StatusCode::BAD_REQUEST.as_u16()
        && oss_code.as_deref() == Some("RequestTimeout"))
        || unclassified_read_failure.is_some_and(|failure| failure.timeout);
    let transport = unclassified_read_failure.is_some_and(|failure| !failure.timeout);
    let mut message = format!("OSS PutObject 失败，状态码：{status}");
    if let Some(oss_code) = oss_code.as_deref() {
        message.push_str(&format!("，OSS 错误码：{oss_code}"));
    }
    if let Some(oss_request_id) = oss_request_id.as_deref() {
        message.push_str(&format!("，OSS Request ID：{oss_request_id}"));
    }
    if body_read_failure.is_some() {
        message.push_str("，错误响应体读取失败");
    }

    OssError::Request(OssRequestError {
        status: Some(status),
        timeout,
        connect: false,
        transport,
        oss_code,
        oss_request_id,
        operation,
        message,
    })
}

fn extract_oss_error_xml_field(body: &[u8], field: &str, max_bytes: usize) -> Option<String> {
    let body = std::str::from_utf8(body).ok()?;
    let start_tag = format!("<{field}>");
    let end_tag = format!("</{field}>");
    let value_start = body.find(&start_tag)? + start_tag.len();
    let value_end = value_start + body[value_start..].find(&end_tag)?;
    normalize_oss_error_field(&body[value_start..value_end], max_bytes)
}

fn normalize_oss_error_field(value: &str, max_bytes: usize) -> Option<String> {
    let value = value.trim();
    if value.is_empty() || value.len() > max_bytes || value.chars().any(char::is_control) {
        return None;
    }
    Some(value.to_string())
}

async fn run_animation_request_with_retry<T, F, Fut>(
    io_limiter: Arc<Semaphore>,
    attempt_context: OssRequestAttemptContext,
    object_key: &str,
    mut attempt_request: F,
) -> Result<T, OssError>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<(T, u16), OssError>>,
{
    for attempt in 1..=CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS {
        let attempt_started_at = Instant::now();
        let permit_wait_started_at = Instant::now();
        let permit = io_limiter
            .acquire()
            .await
            .map_err(|_| OssError::InvalidConfig("角色动画 OSS 并发限制器已关闭".to_string()))?;
        let permit_wait_ms = elapsed_ms(permit_wait_started_at);
        let result = attempt_request().await;
        drop(permit);

        let retryable = result.as_ref().err().is_some_and(oss_error_is_retryable);
        let will_retry = retryable && attempt < CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS;
        let retry_delay_ms = if will_retry {
            CHARACTER_ANIMATION_OSS_RETRY_DELAYS_MS[(attempt - 1) as usize]
        } else {
            0
        };
        let success_status = result.as_ref().ok().map(|(_, status)| *status);
        log_animation_request_attempt(
            attempt_context,
            object_key,
            attempt,
            retryable,
            will_retry,
            retry_delay_ms,
            permit_wait_ms,
            elapsed_ms(attempt_started_at),
            success_status,
            result.as_ref().err(),
        );

        match result {
            Ok((response, _status)) => return Ok(response),
            Err(_error) if will_retry => {
                sleep(std::time::Duration::from_millis(retry_delay_ms)).await
            }
            Err(error) => return Err(error),
        }
    }

    unreachable!("角色动画 OSS 重试循环必须返回结果")
}

fn request_error_from_reqwest(operation: OssRequestOperation, error: reqwest::Error) -> OssError {
    let status = error.status().map(|status| status.as_u16());
    let timeout = error.is_timeout();
    let connect = error.is_connect();
    let transport = !timeout && !connect && (error.is_request() || error.is_body());

    OssError::Request(OssRequestError {
        status,
        timeout,
        connect,
        transport,
        oss_code: None,
        oss_request_id: None,
        operation,
        message: format!("请求 OSS 失败：{error}"),
    })
}

fn request_status_error(operation: OssRequestOperation, status: u16, message: String) -> OssError {
    OssError::Request(OssRequestError {
        status: Some(status),
        timeout: false,
        connect: false,
        transport: false,
        oss_code: None,
        oss_request_id: None,
        operation,
        message,
    })
}

fn oss_error_is_retryable(error: &OssError) -> bool {
    let OssError::Request(request_error) = error else {
        return false;
    };

    match (request_error.status, request_error.oss_code.as_deref()) {
        (Some(400), Some("RequestTimeout")) => true,
        // 400 是唯一会读取错误响应体的状态码:错误码缺失且响应体读取
        // 超时/断流时,无法证明是确定性 400,按传输错误重试。
        (Some(400), None) if request_error.timeout || request_error.transport => true,
        (Some(408 | 429 | 500..=599), _) => true,
        (Some(_), _) => false,
        (None, _) => request_error.timeout || request_error.connect || request_error.transport,
    }
}

fn request_error_details(
    error: Option<&OssError>,
) -> (Option<u16>, bool, bool, bool, Option<&str>, Option<&str>) {
    match error {
        Some(OssError::Request(request_error)) => (
            request_error.status,
            request_error.timeout,
            request_error.connect,
            request_error.transport,
            request_error.oss_code.as_deref(),
            request_error.oss_request_id.as_deref(),
        ),
        Some(OssError::ObjectNotFound(_)) => (Some(404), false, false, false, None, None),
        _ => (None, false, false, false, None, None),
    }
}

fn log_animation_request_attempt(
    context: OssRequestAttemptContext,
    object_key: &str,
    attempt: u32,
    retryable: bool,
    will_retry: bool,
    retry_delay_ms: u64,
    permit_wait_ms: u64,
    elapsed_ms: u64,
    success_status: Option<u16>,
    error: Option<&OssError>,
) {
    let (error_status, timeout, connect, transport, oss_code, oss_request_id) =
        request_error_details(error);
    let status = success_status.or(error_status);
    if error.is_none() {
        info!(
            provider = OSS_PROVIDER,
            frame_index = context.frame_index,
            object_key,
            operation = context.operation,
            attempt,
            max_attempts = CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS,
            retryable,
            will_retry,
            retry_delay_ms,
            permit_wait_ms,
            timeout,
            connect,
            transport,
            oss_code = oss_code.unwrap_or_default(),
            oss_request_id = oss_request_id.unwrap_or_default(),
            status = status.unwrap_or_default(),
            elapsed_ms,
            "角色动画 OSS 请求 attempt 完成"
        );
        return;
    }
    warn!(
        provider = OSS_PROVIDER,
        frame_index = context.frame_index,
        object_key,
        operation = context.operation,
        attempt,
        max_attempts = CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS,
        retryable,
        will_retry,
        retry_delay_ms,
        permit_wait_ms,
        timeout,
        connect,
        transport,
        oss_code = oss_code.unwrap_or_default(),
        oss_request_id = oss_request_id.unwrap_or_default(),
        status = status.unwrap_or_default(),
        elapsed_ms,
        error_kind = error.map(oss_error_kind_label),
        message = error.map(ToString::to_string).unwrap_or_default(),
        "角色动画 OSS 请求 attempt 完成"
    );
}

impl fmt::Display for OssError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidConfig(message)
            | Self::InvalidRequest(message)
            | Self::SerializePolicy(message)
            | Self::Sign(message) => f.write_str(message),
            Self::ObjectNotFound(message) => f.write_str(message),
            Self::Request(error) => f.write_str(&error.message),
        }
    }
}

impl Error for OssError {}

impl OssError {
    pub fn kind(&self) -> OssErrorKind {
        match self {
            Self::InvalidConfig(_) => OssErrorKind::InvalidConfig,
            Self::InvalidRequest(_) => OssErrorKind::InvalidRequest,
            Self::ObjectNotFound(_) => OssErrorKind::ObjectNotFound,
            Self::Request(_) => OssErrorKind::Request,
            Self::SerializePolicy(_) => OssErrorKind::SerializePolicy,
            Self::Sign(_) => OssErrorKind::Sign,
        }
    }
}

fn elapsed_ms(started_at: Instant) -> u64 {
    started_at.elapsed().as_millis().min(u64::MAX as u128) as u64
}

fn oss_access_label(access: OssObjectAccess) -> &'static str {
    match access {
        OssObjectAccess::Public => "public",
        OssObjectAccess::Private => "private",
    }
}

fn oss_error_kind_label(error: &OssError) -> &'static str {
    match error.kind() {
        OssErrorKind::InvalidConfig => "invalid_config",
        OssErrorKind::InvalidRequest => "invalid_request",
        OssErrorKind::ObjectNotFound => "object_not_found",
        OssErrorKind::Request => "request",
        OssErrorKind::SerializePolicy => "serialize_policy",
        OssErrorKind::Sign => "sign",
    }
}

fn http_status_class_from_option(status: Option<u16>) -> &'static str {
    status.map(http_status_class).unwrap_or("unknown")
}

fn http_status_class(status: u16) -> &'static str {
    match status {
        100..=199 => "1xx",
        200..=299 => "2xx",
        300..=399 => "3xx",
        400..=499 => "4xx",
        500..=599 => "5xx",
        _ => "unknown",
    }
}

fn build_policy_json(
    bucket: &str,
    object_key: &str,
    expires_at: &str,
    max_size_bytes: u64,
    success_action_status: u16,
    content_type: Option<&str>,
    cache_control: Option<&str>,
    metadata: &BTreeMap<String, String>,
    credential: &str,
    signature_date: &str,
) -> Value {
    let mut conditions = vec![
        json!({ "bucket": bucket }),
        json!(["eq", "$key", object_key]),
        json!(["content-length-range", 1, max_size_bytes]),
        json!([
            "eq",
            "$success_action_status",
            success_action_status.to_string()
        ]),
        json!(["eq", "$x-oss-signature-version", OSS_V4_ALGORITHM]),
        json!(["eq", "$x-oss-credential", credential]),
        json!(["eq", "$x-oss-date", signature_date]),
    ];

    if let Some(content_type) = content_type {
        conditions.push(json!(["eq", "$content-type", content_type]));
    }

    if let Some(cache_control) = cache_control {
        conditions.push(json!(["eq", "$Cache-Control", cache_control]));
    }

    for (key, value) in metadata {
        conditions.push(json!(["eq", format!("${key}"), value]));
    }

    json!({
        "expiration": expires_at,
        "conditions": conditions,
    })
}

/// HEAD 响应没有正文，`reqwest::Response::content_length()` 对 HEAD 恒为 0；
/// 对象大小只能读响应头，两个 HEAD 入口共用这一处解析。
fn head_object_content_length(headers: &reqwest::header::HeaderMap) -> u64 {
    headers
        .get(reqwest::header::CONTENT_LENGTH)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(0)
}

fn build_object_url(
    bucket: &str,
    endpoint: &str,
    object_key: &str,
) -> Result<reqwest::Url, String> {
    // 对象键是原始路径，#、% 和中文不能被 URL 解析器当作片段或已有转义。
    let path = encode_url_path(object_key.trim_start_matches('/'));
    reqwest::Url::parse(&format!("https://{bucket}.{endpoint}/{path}"))
        .map_err(|error| error.to_string())
}

fn build_object_key(
    prefix: LegacyAssetPrefix,
    path_segments: &[String],
    file_name: &str,
) -> String {
    let mut parts = Vec::with_capacity(path_segments.len() + 2);
    parts.push(prefix.as_str().to_string());
    parts.extend(path_segments.iter().cloned());
    parts.push(file_name.to_string());
    parts.join("/")
}

fn normalize_object_key(raw: &str) -> Result<String, OssError> {
    let normalized = raw.trim().trim_start_matches('/').trim().to_string();
    if normalized.is_empty() {
        return Err(OssError::InvalidRequest("objectKey 不能为空".to_string()));
    }

    match LegacyAssetPrefix::from_object_key(&normalized) {
        Some(LegacyAssetPrefix::EditorAgent) | None => {
            return Err(OssError::InvalidRequest(
                "objectKey 必须落在受支持的 OSS 前缀下".to_string(),
            ));
        }
        Some(_) => {}
    }

    validate_object_key_segments(&normalized)?;
    Ok(normalized)
}

fn normalize_editor_agent_messages_object_key(raw: &str) -> Result<String, OssError> {
    let normalized = raw.trim().trim_start_matches('/').trim().to_string();
    if normalized.is_empty() {
        return Err(OssError::InvalidRequest("objectKey 不能为空".to_string()));
    }
    validate_object_key_segments(&normalized)?;

    let mut segments = normalized.split('/');
    let prefix = segments.next();
    let file_name = segments.next();
    if prefix != Some(LegacyAssetPrefix::EditorAgent.as_str())
        || segments.next().is_some()
        || !file_name
            .map(|value| {
                value.starts_with("editor-agent-conv-")
                    && value.ends_with(".json")
                    && value.len() > "editor-agent-conv-.json".len()
            })
            .unwrap_or(false)
    {
        return Err(OssError::InvalidRequest(
            "objectKey 必须是 editor-agent 消息文档".to_string(),
        ));
    }

    Ok(normalized)
}

/// AGC 内部对象前缀：只允许服务端写入，客户端直传票据、公开对象键与 legacy
/// 公开路径都不覆盖这些前缀。
pub const AGC_ERROR_REPORTS_INTERNAL_PREFIX: &str = "agc/error-reports/v1/";
pub const AGC_PROJECT_SNAPSHOT_INTERNAL_PREFIX: &str = "agc/project-snapshots/v1/";

const AGC_INTERNAL_OBJECT_PREFIXES: [&str; 2] = [
    AGC_ERROR_REPORTS_INTERNAL_PREFIX,
    AGC_PROJECT_SNAPSHOT_INTERNAL_PREFIX,
];

/// 项目快照文件对象键：
/// `agc/project-snapshots/v1/{user}/{project}/files/{size}-{digest}/{relativePath}`。
///
/// 键里同时带字节数与内容摘要，既让同一内容重复提交落在同一个对象上，也让
/// "对象已存在且长度一致" 可以作为内容一致的判据；相对路径按原始大小写保留，
/// 不走 `put_object` 的低位规范化。
pub fn agc_project_snapshot_file_object_key(
    user_id: &str,
    project_id: &str,
    size_bytes: u64,
    checksum_digest: &str,
    relative_path: &str,
) -> Result<String, OssError> {
    let user = validate_internal_key_segment(user_id, "用户标识")?;
    let project = validate_internal_key_segment(project_id, "项目标识")?;
    let digest = validate_internal_checksum_digest(checksum_digest)?;
    let relative_path = validate_internal_relative_path(relative_path)?;
    Ok(format!(
        "{AGC_PROJECT_SNAPSHOT_INTERNAL_PREFIX}{user}/{project}/files/{size_bytes}-{digest}/{relative_path}"
    ))
}

/// 项目快照清单对象键：`agc/project-snapshots/v1/{user}/{project}/manifest.json`。
pub fn agc_project_snapshot_manifest_object_key(
    user_id: &str,
    project_id: &str,
) -> Result<String, OssError> {
    let user = validate_internal_key_segment(user_id, "用户标识")?;
    let project = validate_internal_key_segment(project_id, "项目标识")?;
    Ok(format!(
        "{AGC_PROJECT_SNAPSHOT_INTERNAL_PREFIX}{user}/{project}/manifest.json"
    ))
}

fn validate_internal_key_segment(raw: &str, label: &str) -> Result<String, OssError> {
    let trimmed = raw.trim();
    let allowed = !trimmed.is_empty()
        && trimmed.len() <= 128
        && trimmed != "."
        && trimmed != ".."
        && trimmed.chars().all(|character| {
            character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.')
        });
    if !allowed {
        return Err(OssError::InvalidRequest(format!(
            "{label}不能作为 OSS 对象键片段"
        )));
    }
    Ok(trimmed.to_string())
}

fn validate_internal_checksum_digest(raw: &str) -> Result<String, OssError> {
    let trimmed = raw.trim();
    if trimmed.is_empty()
        || trimmed.len() > 64
        || !trimmed
            .chars()
            .all(|character| character.is_ascii_hexdigit())
    {
        return Err(OssError::InvalidRequest(
            "对象摘要必须是 1 到 64 位十六进制".to_string(),
        ));
    }
    Ok(trimmed.to_ascii_lowercase())
}

fn validate_internal_relative_path(raw: &str) -> Result<String, OssError> {
    if raw.is_empty() || raw.len() > 1024 || raw.starts_with('/') || raw.contains('\\') {
        return Err(OssError::InvalidRequest(
            "对象相对路径必须是 1 到 1024 字节的正斜杠相对路径".to_string(),
        ));
    }
    for part in raw.split('/') {
        if part.is_empty() || part == "." || part == ".." || part.chars().any(char::is_control) {
            return Err(OssError::InvalidRequest(
                "对象相对路径包含非法片段".to_string(),
            ));
        }
    }
    Ok(raw.to_string())
}

fn normalize_internal_object_key(raw: &str) -> Result<String, OssError> {
    let normalized = raw.trim().trim_start_matches('/').trim().to_string();
    validate_object_key_segments(&normalized)?;
    if AGC_INTERNAL_OBJECT_PREFIXES
        .iter()
        .any(|prefix| normalized.starts_with(prefix))
    {
        Ok(normalized)
    } else {
        Err(OssError::InvalidRequest(
            "objectKey 不属于内部对象前缀".to_string(),
        ))
    }
}

fn validate_object_key_segments(normalized: &str) -> Result<(), OssError> {
    let segments = normalized.split('/').collect::<Vec<_>>();
    if segments.len() < 2 {
        return Err(OssError::InvalidRequest(
            "objectKey 至少需要包含前缀和文件名".to_string(),
        ));
    }

    for segment in &segments {
        if segment.is_empty() || *segment == "." || *segment == ".." {
            return Err(OssError::InvalidRequest(
                "objectKey 包含非法路径片段".to_string(),
            ));
        }

        if segment.contains('\\') {
            return Err(OssError::InvalidRequest(
                "objectKey 不能包含反斜杠".to_string(),
            ));
        }
    }

    Ok(())
}

fn build_key_prefix(prefix: LegacyAssetPrefix, path_segments: &[String]) -> String {
    let mut parts = Vec::with_capacity(path_segments.len() + 1);
    parts.push(prefix.as_str().to_string());
    parts.extend(path_segments.iter().cloned());
    parts.join("/")
}

fn normalize_metadata(
    metadata: BTreeMap<String, String>,
) -> Result<BTreeMap<String, String>, OssError> {
    let mut normalized = BTreeMap::new();

    for (key, value) in metadata {
        let key = key.trim();
        let value = value.trim();

        if key.is_empty() || value.is_empty() {
            continue;
        }

        let key = normalize_metadata_key(key);
        normalized.insert(key, value.to_string());
    }

    let total_bytes = normalized
        .iter()
        .map(|(key, value)| key.len() + value.len())
        .sum::<usize>();

    if total_bytes > DEFAULT_METADATA_TOTAL_BYTES_LIMIT {
        return Err(OssError::InvalidRequest(format!(
            "x-oss-meta-* 总大小不能超过 {} 字节",
            DEFAULT_METADATA_TOTAL_BYTES_LIMIT
        )));
    }

    Ok(normalized)
}

fn build_put_object_headers(
    metadata: BTreeMap<String, String>,
) -> Result<BTreeMap<String, String>, OssError> {
    // 中文注释：生成资产 object key 含会话与 asset id，内容不可变，适合交给浏览器/CDN 长缓存。
    let mut headers = BTreeMap::from([(
        "Cache-Control".to_string(),
        DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(),
    )]);
    headers.extend(normalize_metadata(metadata)?);
    Ok(headers)
}

fn normalize_metadata_key(raw: &str) -> String {
    let stripped = raw
        .trim()
        .trim_start_matches("x-oss-meta-")
        .trim()
        .to_ascii_lowercase();
    let sanitized = stripped
        .chars()
        .map(|character| match character {
            'a'..='z' | '0'..='9' | '-' => character,
            '_' | ' ' | '/' | '.' => '-',
            _ => '-',
        })
        .collect::<String>();
    let sanitized = collapse_dashes(&sanitized);

    format!(
        "x-oss-meta-{}",
        if sanitized.is_empty() {
            "metadata".to_string()
        } else {
            sanitized
        }
    )
}

fn sanitize_path_segment(raw: &str) -> String {
    let normalized = raw
        .trim()
        .to_ascii_lowercase()
        .chars()
        .map(|character| match character {
            'a'..='z' | '0'..='9' | '-' | '_' => character,
            _ => '-',
        })
        .collect::<String>();

    collapse_dashes(&normalized)
}

fn sanitize_file_name(raw: &str) -> Result<String, OssError> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return Err(OssError::InvalidRequest("fileName 不能为空".to_string()));
    }

    let file_name = trimmed.rsplit(['/', '\\']).next().unwrap_or(trimmed).trim();

    if file_name.is_empty() {
        return Err(OssError::InvalidRequest("fileName 不能为空".to_string()));
    }

    let (raw_stem, raw_extension) = match file_name.rsplit_once('.') {
        Some((stem, extension)) if !stem.trim().is_empty() && !extension.trim().is_empty() => {
            (stem, Some(extension))
        }
        _ => (file_name, None),
    };

    let stem = raw_stem
        .trim()
        .to_ascii_lowercase()
        .chars()
        .map(|character| match character {
            'a'..='z' | '0'..='9' | '-' | '_' => character,
            _ => '-',
        })
        .collect::<String>();
    let stem = collapse_dashes(&stem);

    let stem = if stem.is_empty() {
        "file".to_string()
    } else {
        stem
    };

    let extension = raw_extension
        .map(|extension| {
            extension
                .trim()
                .to_ascii_lowercase()
                .chars()
                .filter(|character| character.is_ascii_alphanumeric())
                .collect::<String>()
        })
        .filter(|extension| !extension.is_empty());

    Ok(match extension {
        Some(extension) => format!("{stem}.{extension}"),
        None => stem,
    })
}

fn normalize_required_value(value: String, message: &str) -> Result<String, OssError> {
    let value = value.trim().to_string();
    if value.is_empty() {
        return Err(OssError::InvalidConfig(message.to_string()));
    }

    Ok(value)
}

fn normalize_optional_value(value: Option<String>) -> Option<String> {
    value.and_then(|value| {
        let value = value.trim().to_string();
        if value.is_empty() { None } else { Some(value) }
    })
}

fn normalize_endpoint(raw: &str) -> Result<String, OssError> {
    let endpoint = raw
        .trim()
        .trim_start_matches("https://")
        .trim_start_matches("http://")
        .trim_matches('/')
        .to_string();

    if endpoint.is_empty() {
        return Err(OssError::InvalidConfig("OSS endpoint 不能为空".to_string()));
    }

    Ok(endpoint)
}

fn collapse_dashes(value: &str) -> String {
    value
        .chars()
        .fold(
            (String::new(), false),
            |(mut output, last_is_dash), character| {
                let is_dash = character == '-';
                if is_dash && last_is_dash {
                    return (output, true);
                }

                output.push(character);
                (output, is_dash)
            },
        )
        .0
        .trim_matches('-')
        .to_string()
}

async fn send_signed_request(
    client: &reqwest::Client,
    config: &OssConfig,
    method: Method,
    object_key: Option<&str>,
    target_url: reqwest::Url,
    operation: OssRequestOperation,
) -> Result<reqwest::Response, OssError> {
    signed_request_builder(
        client,
        config,
        method,
        object_key,
        target_url,
        None,
        &BTreeMap::new(),
    )?
    .send()
    .await
    .map_err(|error| request_error_from_reqwest(operation, error))
}

fn signed_request_builder(
    client: &reqwest::Client,
    config: &OssConfig,
    method: Method,
    object_key: Option<&str>,
    target_url: reqwest::Url,
    content_type: Option<&str>,
    oss_headers: &BTreeMap<String, String>,
) -> Result<reqwest::RequestBuilder, OssError> {
    let signed_at = OffsetDateTime::now_utc();
    let signed_at_text = build_v4_signature_date(signed_at)?;
    let signature_scope = build_v4_signature_scope(config.endpoint(), signed_at)?;
    let object_path = object_key.map(str::trim).filter(|value| !value.is_empty());
    let canonical_uri = build_v4_canonical_uri(config.bucket(), object_path);
    let body_sha256 = OSS_UNSIGNED_PAYLOAD.to_string();
    let mut signed_headers = BTreeMap::from([
        (
            "host".to_string(),
            format!("{}.{}", config.bucket(), config.endpoint()),
        ),
        ("x-oss-content-sha256".to_string(), body_sha256.clone()),
        ("x-oss-date".to_string(), signed_at_text.clone()),
    ]);
    if let Some(content_type) = content_type {
        signed_headers.insert("content-type".to_string(), content_type.to_string());
    }
    for (key, value) in oss_headers {
        signed_headers.insert(key.to_ascii_lowercase(), value.trim().to_string());
    }

    let canonical_headers = build_v4_canonical_headers(&signed_headers);
    let additional_headers = build_v4_additional_headers(&signed_headers);
    let query = target_url
        .query_pairs()
        .into_owned()
        .collect::<BTreeMap<_, _>>();
    let canonical_query = build_canonical_query_string(&query);
    let canonical_request = build_v4_canonical_request(
        method.as_str(),
        &canonical_uri,
        &canonical_query,
        &canonical_headers,
        &additional_headers,
        &body_sha256,
    );
    let string_to_sign =
        build_v4_string_to_sign(&signed_at_text, &signature_scope, &canonical_request);
    let signature = sign_v4_content(
        config.access_key_secret(),
        &signature_scope,
        &string_to_sign,
    )?;
    let mut builder = client
        .request(method, target_url)
        .header("x-oss-content-sha256", body_sha256)
        .header("x-oss-date", signed_at_text)
        .header(
            "Authorization",
            format!(
                "{OSS_V4_ALGORITHM} Credential={}/{},AdditionalHeaders={},Signature={}",
                config.access_key_id(),
                signature_scope,
                additional_headers,
                signature
            ),
        );

    if let Some(content_type) = content_type {
        builder = builder.header(reqwest::header::CONTENT_TYPE, content_type);
    }

    for (key, value) in oss_headers {
        builder = builder.header(key.as_str(), value.as_str());
    }

    Ok(builder)
}

fn build_v4_signature_scope(endpoint: &str, signed_at: OffsetDateTime) -> Result<String, OssError> {
    let date = format_v4_signature_scope_date(signed_at);
    let region = extract_oss_region(endpoint)?;

    Ok(format!("{date}/{region}/{OSS_V4_SERVICE}/{OSS_V4_REQUEST}"))
}

fn build_v4_signature_date(signed_at: OffsetDateTime) -> Result<String, OssError> {
    // 中文注释：time::Time 的 Display 在小时小于 10 时不会稳定补零，OSS V4 必须使用固定宽度 UTC 时间。
    Ok(format!(
        "{}T{:02}{:02}{:02}Z",
        format_v4_signature_scope_date(signed_at),
        signed_at.hour(),
        signed_at.minute(),
        signed_at.second()
    ))
}

fn format_v4_signature_scope_date(signed_at: OffsetDateTime) -> String {
    format!(
        "{:04}{:02}{:02}",
        signed_at.year(),
        signed_at.month() as u8,
        signed_at.day()
    )
}

fn build_v4_canonical_uri(bucket: &str, object_key: Option<&str>) -> String {
    match object_key.map(str::trim).filter(|value| !value.is_empty()) {
        Some(object_key) => format!(
            "/{}/{}",
            encode_url_query_value(bucket),
            encode_url_path(object_key.trim_start_matches('/'))
        ),
        None => format!("/{}/", encode_url_query_value(bucket)),
    }
}

fn extract_oss_region(endpoint: &str) -> Result<String, OssError> {
    endpoint
        .trim()
        .trim_start_matches("https://")
        .trim_start_matches("http://")
        .split('.')
        .next()
        .and_then(|segment| segment.strip_prefix("oss-"))
        .map(str::to_string)
        .filter(|region| !region.is_empty())
        .ok_or_else(|| {
            OssError::InvalidConfig(format!("OSS endpoint 无法解析 region，当前值：{endpoint}"))
        })
}

fn sign_v4_content(
    access_key_secret: &str,
    signature_scope: &str,
    content: &str,
) -> Result<String, OssError> {
    let signing_key = build_v4_signing_key(access_key_secret, signature_scope)?;
    Ok(hex_sha256_hmac(&signing_key, content.as_bytes()))
}

fn build_v4_signing_key(
    access_key_secret: &str,
    signature_scope: &str,
) -> Result<Vec<u8>, OssError> {
    let mut parts = signature_scope.split('/');
    let date = parts
        .next()
        .ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少日期".to_string()))?;
    let region = parts
        .next()
        .ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 region".to_string()))?;
    let service = parts
        .next()
        .ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 service".to_string()))?;
    let request = parts
        .next()
        .ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 request".to_string()))?;

    let date_key = hmac_sha256_raw(format!("aliyun_v4{access_key_secret}").as_bytes(), date)?;
    let region_key = hmac_sha256_raw(&date_key, region)?;
    let service_key = hmac_sha256_raw(&region_key, service)?;
    hmac_sha256_raw(&service_key, request)
}

fn hmac_sha256_raw(key: &[u8], content: &str) -> Result<Vec<u8>, OssError> {
    let mut signer = HmacSha256::new_from_slice(key)
        .map_err(|error| OssError::Sign(format!("初始化 HMAC-SHA256 失败：{error}")))?;
    signer.update(content.as_bytes());
    Ok(signer.finalize().into_bytes().to_vec())
}

fn hex_sha256_hmac(key: &[u8], content: &[u8]) -> String {
    let mut signer = HmacSha256::new_from_slice(key).expect("HMAC-SHA256 accepts keys of any size");
    signer.update(content);
    hex_lower(&signer.finalize().into_bytes())
}

fn build_v4_canonical_request(
    method: &str,
    canonical_uri: &str,
    canonical_query: &str,
    canonical_headers: &str,
    signed_headers: &str,
    payload_hash: &str,
) -> String {
    format!(
        "{method}\n{canonical_uri}\n{canonical_query}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
    )
}

fn build_v4_string_to_sign(
    signature_date: &str,
    signature_scope: &str,
    canonical_request: &str,
) -> String {
    format!(
        "{OSS_V4_ALGORITHM}\n{signature_date}\n{signature_scope}\n{}",
        sha256_hex(canonical_request.as_bytes())
    )
}

fn sha256_hex(content: &[u8]) -> String {
    let mut hasher = Sha256::new();
    hasher.update(content);
    hex_lower(&hasher.finalize())
}

fn hex_lower(bytes: &[u8]) -> String {
    bytes
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<String>()
}

fn build_v4_canonical_headers(headers: &BTreeMap<String, String>) -> String {
    headers
        .iter()
        .map(|(key, value)| format!("{}:{}\n", key.to_ascii_lowercase(), value.trim()))
        .collect::<String>()
}

fn build_v4_additional_headers(headers: &BTreeMap<String, String>) -> String {
    let mut additional_headers = headers
        .keys()
        .map(|key| key.to_ascii_lowercase())
        .filter(|key| key != "content-type" && key != "content-md5" && !key.starts_with("x-oss-"))
        .collect::<Vec<_>>();
    additional_headers.sort();
    additional_headers.join(";")
}

fn build_canonical_query_string(params: &BTreeMap<String, String>) -> String {
    params
        .iter()
        .map(|(key, value)| {
            format!(
                "{}={}",
                encode_url_query_value(key),
                encode_url_query_value(value)
            )
        })
        .collect::<Vec<_>>()
        .join("&")
}

fn encode_url_path(path: &str) -> String {
    path.split('/')
        .map(encode_url_query_value)
        .collect::<Vec<_>>()
        .join("/")
}

fn encode_url_query_value(value: &str) -> String {
    let mut encoded = String::with_capacity(value.len());

    for byte in value.bytes() {
        match byte {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                encoded.push(byte as char)
            }
            _ => {
                use std::fmt::Write as _;

                let _ = write!(&mut encoded, "%{byte:02X}");
            }
        }
    }

    encoded
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    fn retryable_request_error(
        status: Option<u16>,
        timeout: bool,
        connect: bool,
        transport: bool,
    ) -> OssError {
        OssError::Request(OssRequestError {
            status,
            timeout,
            connect,
            transport,
            oss_code: None,
            oss_request_id: None,
            operation: OssRequestOperation::Put,
            message: "mock request failure".to_string(),
        })
    }

    fn animation_attempt_context(operation: &'static str) -> OssRequestAttemptContext {
        OssRequestAttemptContext {
            frame_index: 1,
            operation,
        }
    }

    #[test]
    fn oss_error_kind_is_stable_for_adapter_mapping() {
        assert_eq!(
            OssError::InvalidConfig("bad config".to_string()).kind(),
            OssErrorKind::InvalidConfig
        );
        assert_eq!(
            OssError::ObjectNotFound("missing".to_string()).kind(),
            OssErrorKind::ObjectNotFound
        );
        assert_eq!(
            OssError::Request(OssRequestError {
                status: None,
                timeout: false,
                connect: false,
                transport: true,
                oss_code: None,
                oss_request_id: None,
                operation: OssRequestOperation::Put,
                message: "network".to_string(),
            })
            .kind(),
            OssErrorKind::Request
        );
    }

    #[test]
    fn object_not_found_attempt_log_details_keep_404_status() {
        let error = OssError::ObjectNotFound("missing".to_string());
        assert_eq!(
            request_error_details(Some(&error)),
            (Some(404), false, false, false, None, None)
        );
        assert_eq!(oss_error_kind_label(&error), "object_not_found");
        assert!(!oss_error_is_retryable(&error));
    }

    #[test]
    fn oss_request_timeout_400_is_retryable_and_prefers_header_request_id() {
        let body = br#"<?xml version="1.0" encoding="UTF-8"?>
<Error><Code>RequestTimeout</Code><RequestId>xml-request-id</RequestId></Error>"#;
        let error = request_status_error_from_oss_parts(
            OssRequestOperation::Put,
            400,
            Some("header-request-id".to_string()),
            body,
            None,
        );
        let OssError::Request(request_error) = &error else {
            panic!("OSS status failure should remain a request error");
        };

        assert_eq!(request_error.status, Some(400));
        assert!(request_error.timeout);
        assert_eq!(request_error.oss_code.as_deref(), Some("RequestTimeout"));
        assert_eq!(
            request_error.oss_request_id.as_deref(),
            Some("header-request-id")
        );
        assert!(oss_error_is_retryable(&error));
    }

    #[test]
    fn oss_request_timeout_400_uses_xml_request_id_when_header_is_missing() {
        let body = br#"<Error>
<Code>RequestTimeout</Code><RequestId>xml-request-id</RequestId>
</Error>"#;
        let error =
            request_status_error_from_oss_parts(OssRequestOperation::Put, 400, None, body, None);
        let OssError::Request(request_error) = &error else {
            panic!("OSS status failure should remain a request error");
        };

        assert_eq!(
            request_error.oss_request_id.as_deref(),
            Some("xml-request-id")
        );
        assert!(oss_error_is_retryable(&error));
    }

    #[test]
    fn other_oss_400_errors_and_malformed_xml_are_not_retryable() {
        for body in [
            b"<Error><Code>InvalidArgument</Code></Error>".as_slice(),
            b"<Error><Code>RequestTimeout".as_slice(),
            b"not xml".as_slice(),
        ] {
            let error = request_status_error_from_oss_parts(
                OssRequestOperation::Put,
                400,
                None,
                body,
                None,
            );
            assert!(!oss_error_is_retryable(&error));
        }

        let error = request_status_error_from_oss_parts(
            OssRequestOperation::Put,
            403,
            None,
            b"<Error><Code>RequestTimeout</Code></Error>",
            None,
        );
        assert!(!oss_error_is_retryable(&error));
    }

    #[test]
    fn oss_error_xml_fields_beyond_body_limit_are_ignored() {
        let mut body = vec![b' '; CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES];
        body.extend_from_slice(
            b"<Error><Code>RequestTimeout</Code><RequestId>late</RequestId></Error>",
        );
        let error =
            request_status_error_from_oss_parts(OssRequestOperation::Put, 400, None, &body, None);
        let OssError::Request(request_error) = &error else {
            panic!("OSS status failure should remain a request error");
        };

        assert_eq!(request_error.oss_code, None);
        assert_eq!(request_error.oss_request_id, None);
        assert!(!request_error.timeout);
        assert!(!oss_error_is_retryable(&error));
    }

    #[test]
    fn oss_400_without_code_and_broken_body_read_is_retryable() {
        for (read_timeout, expect_transport) in [(true, false), (false, true)] {
            let error = request_status_error_from_oss_parts(
                OssRequestOperation::Put,
                400,
                None,
                b"<Error><Code>Request",
                Some(OssErrorBodyReadFailure {
                    timeout: read_timeout,
                }),
            );
            let OssError::Request(request_error) = &error else {
                panic!("OSS status failure should remain a request error");
            };

            assert_eq!(request_error.status, Some(400));
            assert_eq!(request_error.oss_code, None);
            assert_eq!(request_error.timeout, read_timeout);
            assert_eq!(request_error.transport, expect_transport);
            assert!(request_error.message.contains("错误响应体读取失败"));
            assert!(oss_error_is_retryable(&error));
        }
    }

    #[test]
    fn oss_400_with_parsed_code_keeps_deterministic_semantics_on_read_failure() {
        let error = request_status_error_from_oss_parts(
            OssRequestOperation::Put,
            400,
            None,
            b"<Error><Code>InvalidArgument</Code><RequestId>partial",
            Some(OssErrorBodyReadFailure { timeout: true }),
        );
        let OssError::Request(request_error) = &error else {
            panic!("OSS status failure should remain a request error");
        };

        assert_eq!(request_error.oss_code.as_deref(), Some("InvalidArgument"));
        assert!(!request_error.timeout);
        assert!(!request_error.transport);
        assert!(request_error.message.contains("错误响应体读取失败"));
        assert!(!oss_error_is_retryable(&error));

        let error = request_status_error_from_oss_parts(
            OssRequestOperation::Put,
            400,
            None,
            b"<Error><Code>RequestTimeout</Code>",
            Some(OssErrorBodyReadFailure { timeout: false }),
        );
        assert!(oss_error_is_retryable(&error));
    }

    const MOCK_PUT_BODY: &[u8] = b"animation-frame-bytes";

    /// 极简 HTTP/1.1 mock:读完整个 PUT 请求后返回 400 与部分 XML 响应体
    /// (Content-Length 大于实际发送字节),`stall_before_close` 决定挂住
    /// 连接触发客户端读超时,还是直接断开触发传输错误。
    async fn spawn_broken_error_body_server(stall_before_close: bool) -> std::net::SocketAddr {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("mock server should bind");
        let addr = listener
            .local_addr()
            .expect("mock server should expose its addr");
        tokio::spawn(async move {
            let Ok((mut socket, _)) = listener.accept().await else {
                return;
            };
            let mut received = Vec::new();
            let mut buffer = [0u8; 4096];
            while !received.ends_with(MOCK_PUT_BODY) {
                match socket.read(&mut buffer).await {
                    Ok(0) | Err(_) => return,
                    Ok(read) => received.extend_from_slice(&buffer[..read]),
                }
            }
            let response = "HTTP/1.1 400 Bad Request\r\n\
                x-oss-request-id: mock-request-id\r\n\
                Content-Type: application/xml\r\n\
                Content-Length: 4096\r\n\
                \r\n\
                <Error><Code>Request";
            if socket.write_all(response.as_bytes()).await.is_err() {
                return;
            }
            let _ = socket.flush().await;
            if stall_before_close {
                tokio::time::sleep(std::time::Duration::from_secs(5)).await;
            }
        });
        addr
    }

    #[tokio::test]
    async fn oss_400_with_broken_error_body_stream_is_retryable_transport() {
        let addr = spawn_broken_error_body_server(false).await;
        let response = reqwest::Client::new()
            .put(format!("http://{addr}/generated-animations/frame01.png"))
            .body(MOCK_PUT_BODY.to_vec())
            .send()
            .await
            .expect("response headers should arrive before the body breaks");
        assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);

        let error = request_status_error_from_oss_put_response(response).await;
        let OssError::Request(request_error) = &error else {
            panic!("OSS status failure should remain a request error");
        };

        assert_eq!(request_error.status, Some(400));
        assert_eq!(request_error.oss_code, None);
        assert!(!request_error.timeout);
        assert!(request_error.transport);
        assert_eq!(
            request_error.oss_request_id.as_deref(),
            Some("mock-request-id")
        );
        assert!(request_error.message.contains("错误响应体读取失败"));
        assert!(oss_error_is_retryable(&error));
    }

    #[tokio::test]
    async fn oss_400_with_stalled_error_body_stream_is_retryable_timeout() {
        let addr = spawn_broken_error_body_server(true).await;
        let client = reqwest::Client::builder()
            .timeout(std::time::Duration::from_millis(300))
            .build()
            .expect("test client should build");
        let response = client
            .put(format!("http://{addr}/generated-animations/frame01.png"))
            .body(MOCK_PUT_BODY.to_vec())
            .send()
            .await
            .expect("response headers should arrive before the body stalls");
        assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);

        let error = request_status_error_from_oss_put_response(response).await;
        let OssError::Request(request_error) = &error else {
            panic!("OSS status failure should remain a request error");
        };

        assert_eq!(request_error.status, Some(400));
        assert_eq!(request_error.oss_code, None);
        assert!(request_error.timeout);
        assert!(!request_error.transport);
        assert!(oss_error_is_retryable(&error));
    }

    #[tokio::test]
    async fn reqwest_builder_error_is_not_retryable_transport() {
        let error = reqwest::Client::new()
            .put("https://example.com")
            .header("x-oss-meta-invalid", "first line\nsecond line")
            .send()
            .await
            .expect_err("invalid header must fail while building the request");
        assert!(error.is_builder());

        let error = request_error_from_reqwest(OssRequestOperation::Put, error);
        let OssError::Request(request_error) = &error else {
            panic!("builder failure should remain an OSS request error");
        };

        assert_eq!(request_error.status, None);
        assert!(!request_error.timeout);
        assert!(!request_error.connect);
        assert!(!request_error.transport);
        assert!(!oss_error_is_retryable(&error));
    }

    #[tokio::test]
    async fn animation_retry_retries_transport_then_succeeds() {
        let attempts = Arc::new(AtomicUsize::new(0));
        let attempts_for_request = attempts.clone();
        let result = run_animation_request_with_retry(
            Arc::new(Semaphore::new(8)),
            animation_attempt_context("source_put"),
            "generated-animations/editor/layer/task/green-screen-frame01.png",
            move || {
                let attempts = attempts_for_request.clone();
                async move {
                    if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
                        Err(retryable_request_error(None, false, false, true))
                    } else {
                        Ok(("uploaded", 201))
                    }
                }
            },
        )
        .await;

        assert_eq!(result.expect("second attempt should succeed"), "uploaded");
        assert_eq!(attempts.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn animation_retry_retries_timeout_and_retryable_statuses() {
        for (status, timeout, connect, transport) in [
            (None, true, false, false),
            (Some(408), false, false, false),
            (Some(429), false, false, false),
            (Some(500), false, false, false),
            (Some(502), false, false, false),
            (Some(503), false, false, false),
            (Some(504), false, false, false),
        ] {
            let attempts = Arc::new(AtomicUsize::new(0));
            let attempts_for_request = attempts.clone();
            let result = run_animation_request_with_retry(
                Arc::new(Semaphore::new(8)),
                animation_attempt_context("final_put"),
                "generated-animations/editor/layer/task/frame01.png",
                move || {
                    let attempts = attempts_for_request.clone();
                    async move {
                        if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
                            Err(retryable_request_error(status, timeout, connect, transport))
                        } else {
                            Ok(((), 204))
                        }
                    }
                },
            )
            .await;

            assert!(result.is_ok(), "status={status:?} should retry");
            assert_eq!(attempts.load(Ordering::SeqCst), 2, "status={status:?}");
        }
    }

    #[tokio::test]
    async fn animation_retry_does_not_retry_deterministic_statuses() {
        for status in [400, 403, 404] {
            let attempts = Arc::new(AtomicUsize::new(0));
            let attempts_for_request = attempts.clone();
            let result = run_animation_request_with_retry(
                Arc::new(Semaphore::new(8)),
                animation_attempt_context("final_head"),
                "generated-animations/editor/layer/task/frame01.png",
                move || {
                    let attempts = attempts_for_request.clone();
                    async move {
                        attempts.fetch_add(1, Ordering::SeqCst);
                        Err::<((), u16), _>(retryable_request_error(
                            Some(status),
                            false,
                            false,
                            false,
                        ))
                    }
                },
            )
            .await;

            assert!(result.is_err(), "status={status} should fail");
            assert_eq!(attempts.load(Ordering::SeqCst), 1, "status={status}");
        }
    }

    #[tokio::test]
    async fn animation_retry_returns_final_error_after_three_attempts() {
        let attempts = Arc::new(AtomicUsize::new(0));
        let attempts_for_request = attempts.clone();
        let result = run_animation_request_with_retry(
            Arc::new(Semaphore::new(8)),
            animation_attempt_context("final_put"),
            "generated-animations/editor/layer/task/frame01.png",
            move || {
                let attempts = attempts_for_request.clone();
                async move {
                    attempts.fetch_add(1, Ordering::SeqCst);
                    Err::<((), u16), _>(retryable_request_error(Some(503), false, false, false))
                }
            },
        )
        .await;

        assert!(result.is_err());
        assert_eq!(attempts.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn animation_retry_keeps_head_retry_independent_from_successful_put() {
        let put_attempts = Arc::new(AtomicUsize::new(0));
        let put_attempts_for_request = put_attempts.clone();
        run_animation_request_with_retry(
            Arc::new(Semaphore::new(8)),
            animation_attempt_context("final_put"),
            "generated-animations/editor/layer/task/frame01.png",
            move || {
                put_attempts_for_request.fetch_add(1, Ordering::SeqCst);
                async { Ok::<_, OssError>(((), 204)) }
            },
        )
        .await
        .expect("PUT should succeed once");

        let head_attempts = Arc::new(AtomicUsize::new(0));
        let head_attempts_for_request = head_attempts.clone();
        run_animation_request_with_retry(
            Arc::new(Semaphore::new(8)),
            animation_attempt_context("final_head"),
            "generated-animations/editor/layer/task/frame01.png",
            move || {
                let head_attempts = head_attempts_for_request.clone();
                async move {
                    if head_attempts.fetch_add(1, Ordering::SeqCst) == 0 {
                        Err(retryable_request_error(Some(503), false, false, false))
                    } else {
                        Ok(((), 204))
                    }
                }
            },
        )
        .await
        .expect("HEAD should succeed on its retry");

        assert_eq!(put_attempts.load(Ordering::SeqCst), 1);
        assert_eq!(head_attempts.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn animation_retry_keeps_permit_bound_at_eight_in_flight_requests() {
        let limiter = Arc::new(Semaphore::new(8));
        let current = Arc::new(AtomicUsize::new(0));
        let maximum = Arc::new(AtomicUsize::new(0));
        let mut tasks = Vec::new();

        for frame_index in 0..32 {
            let limiter = limiter.clone();
            let current_for_request = current.clone();
            let maximum_for_request = maximum.clone();
            tasks.push(tokio::spawn(async move {
                run_animation_request_with_retry(
                    limiter,
                    OssRequestAttemptContext {
                        frame_index: frame_index + 1,
                        operation: "source_put",
                    },
                    "generated-animations/editor/layer/task/green-screen-frame01.png",
                    move || {
                        let current = current_for_request.clone();
                        let maximum = maximum_for_request.clone();
                        async move {
                            let in_flight = current.fetch_add(1, Ordering::SeqCst) + 1;
                            maximum.fetch_max(in_flight, Ordering::SeqCst);
                            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
                            current.fetch_sub(1, Ordering::SeqCst);
                            Ok::<_, OssError>(((), 204))
                        }
                    },
                )
                .await
            }));
        }

        for task in tasks {
            task.await
                .expect("mock request task should join")
                .expect("request should pass");
        }
        assert!(maximum.load(Ordering::SeqCst) <= 8);
    }

    #[tokio::test]
    async fn animation_retry_reuses_object_key_and_body_across_attempts() {
        let client = OssClient::new(
            OssConfig::new(
                "bucket".to_string(),
                "oss-cn-shanghai.aliyuncs.com".to_string(),
                "access-key".to_string(),
                "access-secret".to_string(),
                60,
                60,
                1024,
                204,
            )
            .expect("test OSS config should be valid"),
        );
        let prepared = client
            .prepare_put_object(OssPutObjectRequest {
                prefix: LegacyAssetPrefix::Animations,
                path_segments: vec![
                    "editor".to_string(),
                    "layer".to_string(),
                    "task".to_string(),
                ],
                file_name: "frame01.png".to_string(),
                content_type: Some("image/png".to_string()),
                access: OssObjectAccess::Private,
                metadata: BTreeMap::new(),
                body: vec![1, 2, 3, 4],
            })
            .expect("test request should be prepared");
        let expected_key = prepared.object_key.clone();
        let expected_body = prepared.body.clone();
        let expected_key_for_request = expected_key.clone();
        let attempts = Arc::new(AtomicUsize::new(0));
        let attempts_for_request = attempts.clone();
        let result = run_animation_request_with_retry(
            Arc::new(Semaphore::new(8)),
            animation_attempt_context("final_put"),
            &expected_key_for_request,
            move || {
                let attempts = attempts_for_request.clone();
                let key = prepared.object_key.clone();
                let body = prepared.body.clone();
                let expected_key = expected_key.clone();
                let expected_body = expected_body.clone();
                async move {
                    assert_eq!(key, expected_key);
                    assert_eq!(body, expected_body);
                    if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
                        Err(retryable_request_error(Some(503), false, false, false))
                    } else {
                        Ok(((), 204))
                    }
                }
            },
        )
        .await;

        assert!(result.is_ok());
        assert_eq!(attempts.load(Ordering::SeqCst), 2);
    }

    #[test]
    fn structured_log_labels_are_stable() {
        assert_eq!(
            oss_error_kind_label(&OssError::InvalidRequest("bad input".to_string())),
            "invalid_request"
        );
        assert_eq!(oss_access_label(OssObjectAccess::Private), "private");
        assert_eq!(http_status_class(204), "2xx");
        assert_eq!(http_status_class(404), "4xx");
        assert_eq!(http_status_class_from_option(None), "unknown");
    }

    fn build_client() -> OssClient {
        OssClient::new(
            OssConfig::new(
                "genarrative-assets".to_string(),
                "oss-cn-shanghai.aliyuncs.com".to_string(),
                "test-access-key-id".to_string(),
                "test-access-key-secret".to_string(),
                DEFAULT_READ_EXPIRE_SECONDS,
                DEFAULT_POST_EXPIRE_SECONDS,
                DEFAULT_POST_MAX_SIZE_BYTES,
                DEFAULT_SUCCESS_ACTION_STATUS,
            )
            .expect("OSS config should be valid"),
        )
    }

    #[test]
    fn parse_legacy_prefix_accepts_public_style_path() {
        assert_eq!(
            LegacyAssetPrefix::parse("/generated-characters/*"),
            Some(LegacyAssetPrefix::Characters)
        );
        assert_eq!(
            LegacyAssetPrefix::parse("/generated-puzzle-assets/*"),
            Some(LegacyAssetPrefix::PuzzleAssets)
        );
        assert_eq!(
            LegacyAssetPrefix::parse("/generated-match3d-assets/*"),
            Some(LegacyAssetPrefix::Match3DAssets)
        );
        assert_eq!(
            LegacyAssetPrefix::parse("/generated-wooden-fish-assets/*"),
            Some(LegacyAssetPrefix::WoodenFishAssets)
        );
        assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-puzzle-assets"));
        assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-match3d-assets"));
        assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-bark-battle-assets"));
        assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-wooden-fish-assets"));
        assert_eq!(LegacyAssetPrefix::parse("unknown"), None);
    }

    #[test]
    fn build_v4_signature_date_zero_pads_single_digit_time_parts() {
        let signed_at =
            OffsetDateTime::from_unix_timestamp(1_771_477_389).expect("timestamp should be valid");

        assert_eq!(
            build_v4_signature_date(signed_at).expect("date should format"),
            "20260219T050309Z"
        );
        assert_eq!(
            build_v4_signature_scope("oss-cn-shanghai.aliyuncs.com", signed_at)
                .expect("scope should format"),
            "20260219/cn-shanghai/oss/aliyun_v4_request"
        );
    }

    #[test]
    fn sign_post_object_returns_bucket_and_object_key_for_private_storage_truth() {
        let client = build_client();
        let mut metadata = BTreeMap::new();
        metadata.insert("asset-kind".to_string(), "character-visual".to_string());
        metadata.insert("origin".to_string(), "browser-upload".to_string());

        let response = client
            .sign_post_object(OssPostObjectRequest {
                prefix: LegacyAssetPrefix::Characters,
                path_segments: vec![
                    "Hero_001".to_string(),
                    "Visual".to_string(),
                    "Asset_01".to_string(),
                ],
                file_name: "Master.PNG".to_string(),
                content_type: Some("image/png".to_string()),
                access: OssObjectAccess::Public,
                metadata,
                max_size_bytes: Some(5 * 1024 * 1024),
                expire_seconds: Some(300),
                success_action_status: Some(200),
            })
            .expect("post object signature should build");

        assert_eq!(
            response.object_key,
            "generated-characters/hero_001/visual/asset_01/master.png"
        );
        assert_eq!(
            response.legacy_public_path,
            "/generated-characters/hero_001/visual/asset_01/master.png"
        );
        assert_eq!(response.bucket, "genarrative-assets".to_string());
        assert_eq!(
            response.form_fields.signature_version,
            OSS_V4_ALGORITHM.to_string()
        );
        assert!(
            response
                .form_fields
                .credential
                .starts_with("test-access-key-id/")
        );
        assert!(
            response
                .form_fields
                .credential
                .ends_with("/cn-shanghai/oss/aliyun_v4_request")
        );
        assert_eq!(response.form_fields.date.len(), "20260507T120000Z".len());
        assert_eq!(
            response.form_fields.metadata.get("x-oss-meta-asset-kind"),
            Some(&"character-visual".to_string())
        );
    }

    #[test]
    fn sign_post_object_embeds_policy_constraints() {
        let client = build_client();
        let response = client
            .sign_post_object(OssPostObjectRequest {
                prefix: LegacyAssetPrefix::QwenSprites,
                path_segments: vec!["_drafts".to_string(), "master".to_string()],
                file_name: "candidate-01.png".to_string(),
                content_type: Some("image/png".to_string()),
                access: OssObjectAccess::Private,
                metadata: BTreeMap::new(),
                max_size_bytes: Some(1024),
                expire_seconds: Some(60),
                success_action_status: Some(200),
            })
            .expect("post object signature should build");

        let decoded_policy = BASE64_STANDARD
            .decode(response.form_fields.policy.as_bytes())
            .expect("policy should be valid base64");
        let policy: Value =
            serde_json::from_slice(&decoded_policy).expect("policy should be valid json");

        assert_eq!(
            policy["conditions"][0]["bucket"],
            Value::String("genarrative-assets".to_string())
        );
        assert_eq!(
            policy["conditions"][1],
            json!([
                "eq",
                "$key",
                "generated-qwen-sprites/_drafts/master/candidate-01.png"
            ])
        );
        assert_eq!(
            policy["conditions"][2],
            json!(["content-length-range", 1, 1024])
        );
        assert_eq!(
            policy["conditions"][3],
            json!(["eq", "$success_action_status", "200"])
        );
        assert_eq!(
            policy["conditions"][4],
            json!(["eq", "$x-oss-signature-version", "OSS4-HMAC-SHA256"])
        );
        assert_eq!(
            policy["conditions"][5],
            json!(["eq", "$x-oss-credential", response.form_fields.credential])
        );
        assert_eq!(
            policy["conditions"][6],
            json!(["eq", "$x-oss-date", response.form_fields.date])
        );
        assert_eq!(
            policy["conditions"][7],
            json!(["eq", "$content-type", "image/png"])
        );
        assert_eq!(
            policy["conditions"][8],
            json!(["eq", "$Cache-Control", DEFAULT_IMMUTABLE_CACHE_CONTROL])
        );
        assert_eq!(
            response.form_fields.cache_control,
            Some(DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string())
        );
        assert_eq!(response.bucket, "genarrative-assets".to_string());
    }

    #[test]
    fn sanitize_file_name_rejects_empty_input() {
        let error = sanitize_file_name("   ").expect_err("empty file name should fail");

        assert_eq!(
            error,
            OssError::InvalidRequest("fileName 不能为空".to_string())
        );
    }

    #[test]
    fn sanitize_file_name_falls_back_when_stem_has_no_ascii_body() {
        assert_eq!(
            sanitize_file_name("剪贴板素材.png").expect("file name should sanitize"),
            "file.png"
        );
    }

    #[test]
    fn sign_get_object_url_returns_signed_private_read_url() {
        let client = build_client();

        let response = client
            .sign_get_object_url(OssSignedGetObjectUrlRequest {
                object_key: "generated-characters/hero_001/visual/asset_01/master.png".to_string(),
                expire_seconds: Some(300),
            })
            .expect("signed get url should build");

        assert_eq!(response.bucket, "genarrative-assets".to_string());
        assert_eq!(
            response.object_key,
            "generated-characters/hero_001/visual/asset_01/master.png".to_string()
        );
        assert!(response
            .signed_url
            .starts_with("https://genarrative-assets.oss-cn-shanghai.aliyuncs.com/generated-characters/hero_001/visual/asset_01/master.png?"));
        assert!(
            response
                .signed_url
                .contains("x-oss-signature-version=OSS4-HMAC-SHA256")
        );
        assert!(
            response
                .signed_url
                .contains("x-oss-credential=test-access-key-id%2F")
        );
        assert!(response.signed_url.contains("&x-oss-expires=300"));
        assert!(response.signed_url.contains("&x-oss-signature="));
    }

    #[test]
    fn sign_get_object_url_uses_square_hole_object_key_without_bucket_prefix() {
        let client = OssClient::new(
            OssConfig::new(
                "xushi-dev".to_string(),
                "oss-cn-shanghai.aliyuncs.com".to_string(),
                "test-access-key-id".to_string(),
                "test-access-key-secret".to_string(),
                DEFAULT_READ_EXPIRE_SECONDS,
                DEFAULT_POST_EXPIRE_SECONDS,
                DEFAULT_POST_MAX_SIZE_BYTES,
                DEFAULT_SUCCESS_ACTION_STATUS,
            )
            .expect("OSS config should be valid"),
        );

        let response = client
            .sign_get_object_url(OssSignedGetObjectUrlRequest {
                object_key: "generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png".to_string(),
                expire_seconds: Some(300),
            })
            .expect("square hole object key should build signed url");

        assert_eq!(response.bucket, "xushi-dev".to_string());
        assert_eq!(
            response.object_key,
            "generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png".to_string()
        );
        assert!(response
            .signed_url
            .starts_with("https://xushi-dev.oss-cn-shanghai.aliyuncs.com/generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png?"));
    }

    #[test]
    fn sign_get_object_url_rejects_unsupported_prefix() {
        let client = build_client();

        let error = client
            .sign_get_object_url(OssSignedGetObjectUrlRequest {
                object_key: "workflow-cache/task-1.json".to_string(),
                expire_seconds: Some(300),
            })
            .expect_err("unsupported prefix should fail");

        assert_eq!(
            error,
            OssError::InvalidRequest("objectKey 必须落在受支持的 OSS 前缀下".to_string())
        );
    }

    #[test]
    fn sign_internal_get_object_url_allows_editor_agent_messages_document() {
        let client = build_client();

        let response = client
            .sign_internal_get_object_url(OssSignedGetObjectUrlRequest {
                object_key: "editor-agent/editor-agent-conv-1.json".to_string(),
                expire_seconds: Some(300),
            })
            .expect("editor agent messages document should build signed url");

        assert_eq!(
            response.object_key,
            "editor-agent/editor-agent-conv-1.json".to_string()
        );
        assert!(response.signed_url.starts_with(
            "https://genarrative-assets.oss-cn-shanghai.aliyuncs.com/editor-agent/editor-agent-conv-1.json?"
        ));
    }

    #[test]
    fn sign_internal_get_object_url_rejects_non_editor_agent_messages_document() {
        let client = build_client();

        for object_key in [
            "workflow-cache/task-1.json",
            "generated-characters/hero/master.png",
            "editor-agent/editor-agent-conv-1/part.json",
            "editor-agent/other.json",
            "editor-agent/editor-agent-conv-1.png",
        ] {
            let error = client
                .sign_internal_get_object_url(OssSignedGetObjectUrlRequest {
                    object_key: object_key.to_string(),
                    expire_seconds: Some(300),
                })
                .expect_err("non editor agent messages document should fail");

            assert_eq!(
                error,
                OssError::InvalidRequest("objectKey 必须是 editor-agent 消息文档".to_string())
            );
        }
    }

    #[test]
    fn legacy_prefix_can_be_resolved_from_object_key() {
        assert_eq!(
            LegacyAssetPrefix::from_object_key(
                "generated-custom-world-scenes/profile_01/landmark_01/scene.png"
            ),
            Some(LegacyAssetPrefix::CustomWorldScenes)
        );
        assert_eq!(
            LegacyAssetPrefix::from_object_key(
                "generated-wooden-fish-assets/session/profile/hit_object/asset/image.png"
            ),
            Some(LegacyAssetPrefix::WoodenFishAssets)
        );
        assert_eq!(
            LegacyAssetPrefix::from_object_key("workflow-cache/demo.json"),
            None
        );
        assert_eq!(
            LegacyAssetPrefix::from_object_key(
                "agc/project-snapshots/v1/user-1/project-1/manifest.json"
            ),
            None,
            "AGC 内部前缀不能经由通用对象键解析变成客户端可写前缀"
        );
    }

    #[test]
    fn head_object_content_length_reads_the_response_header_not_the_head_body() {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            reqwest::header::CONTENT_LENGTH,
            "32".parse().expect("content length header value"),
        );
        assert_eq!(
            head_object_content_length(&headers),
            32,
            "HEAD 的对象大小必须来自响应头；reqwest 对 HEAD 的 body 长度恒为 0"
        );
        assert_eq!(
            head_object_content_length(&reqwest::header::HeaderMap::new()),
            0
        );
    }

    #[test]
    fn agc_project_snapshot_object_keys_preserve_case_and_reject_traversal() {
        let file_key = agc_project_snapshot_file_object_key(
            "user-1",
            "gameagent-1a2b3c4d",
            1234,
            "0123456789ABCDEF",
            "Game/Scenes/Main.HTML",
        )
        .expect("file key");
        assert_eq!(
            file_key,
            "agc/project-snapshots/v1/user-1/gameagent-1a2b3c4d/files/1234-0123456789abcdef/Game/Scenes/Main.HTML"
        );
        assert_eq!(
            agc_project_snapshot_manifest_object_key("user-1", "gameagent-1a2b3c4d")
                .expect("manifest key"),
            "agc/project-snapshots/v1/user-1/gameagent-1a2b3c4d/manifest.json"
        );

        for (user, project, path) in [
            ("../escape", "project-1", "game/index.html"),
            ("user-1", "../escape", "game/index.html"),
            ("user-1", "project-1", "../outside.txt"),
            ("user-1", "project-1", "game/../../outside.txt"),
            ("user-1", "project-1", "game\\index.html"),
        ] {
            assert!(
                agc_project_snapshot_file_object_key(user, project, 1, "abcdef", path).is_err(),
                "越界键片段必须被拒绝：{user} {project} {path}"
            );
        }
        assert!(
            agc_project_snapshot_file_object_key("user-1", "project-1", 1, "not-hex", "game/a.txt")
                .is_err(),
            "摘要必须是十六进制"
        );
    }

    #[test]
    fn internal_object_prefixes_cover_agc_snapshots_but_reject_everything_else() {
        let file_key = agc_project_snapshot_file_object_key(
            "user-1",
            "project-1",
            7,
            "abcdef",
            "game/index.html",
        )
        .expect("file key");
        assert_eq!(
            normalize_internal_object_key(&file_key).expect("snapshot key is internal"),
            file_key
        );
        assert_eq!(
            normalize_internal_object_key("agc/error-reports/v1/batch.zip")
                .expect("error report key stays internal"),
            "agc/error-reports/v1/batch.zip"
        );
        for key in [
            "generated-characters/hero/master.png",
            "agc/other-purpose/v1/file.json",
        ] {
            assert!(
                normalize_internal_object_key(key).is_err(),
                "非内部前缀不能走服务端内部写入：{key}"
            );
        }
    }

    #[test]
    fn put_object_request_reuses_generated_object_key_contract() {
        let request = OssPutObjectRequest {
            prefix: LegacyAssetPrefix::CustomWorldCovers,
            path_segments: vec!["Profile 001".to_string(), "asset_01".to_string()],
            file_name: "Cover.PNG".to_string(),
            content_type: Some(" image/png ".to_string()),
            access: OssObjectAccess::Private,
            metadata: BTreeMap::from([
                ("asset_kind".to_string(), "custom_world_cover".to_string()),
                ("source job id".to_string(), "job_001".to_string()),
            ]),
            body: b"cover-bytes".to_vec(),
        };
        let sanitized_segments = request
            .path_segments
            .iter()
            .map(|segment| sanitize_path_segment(segment))
            .filter(|segment| !segment.is_empty())
            .collect::<Vec<_>>();
        let file_name = sanitize_file_name(&request.file_name).expect("file name should sanitize");
        let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name);
        let metadata = normalize_metadata(request.metadata).expect("metadata should normalize");

        assert_eq!(
            object_key,
            "generated-custom-world-covers/profile-001/asset_01/cover.png"
        );
        assert_eq!(
            metadata.get("x-oss-meta-asset-kind"),
            Some(&"custom_world_cover".to_string())
        );
        assert_eq!(
            metadata.get("x-oss-meta-source-job-id"),
            Some(&"job_001".to_string())
        );
    }

    #[test]
    fn canonicalized_oss_headers_matches_sorted_v4_header_shape() {
        let headers = BTreeMap::from([
            (
                "Cache-Control".to_string(),
                DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(),
            ),
            (
                "x-oss-meta-source-job-id".to_string(),
                " job_001 ".to_string(),
            ),
            (
                "x-oss-meta-asset-kind".to_string(),
                "character_visual".to_string(),
            ),
        ]);

        assert_eq!(
            build_v4_canonical_headers(&headers),
            "cache-control:public, max-age=31536000, immutable\nx-oss-meta-asset-kind:character_visual\nx-oss-meta-source-job-id:job_001\n"
        );
    }

    #[test]
    fn additional_headers_include_plain_headers_and_skip_oss_managed_headers() {
        let headers = BTreeMap::from([
            (
                "host".to_string(),
                "genarrative-assets.oss-cn-beijing.aliyuncs.com".to_string(),
            ),
            ("content-type".to_string(), "image/png".to_string()),
            (
                "Cache-Control".to_string(),
                DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(),
            ),
            ("x-oss-date".to_string(), "20260507T120000Z".to_string()),
            (
                "x-oss-meta-asset-kind".to_string(),
                "puzzle-cover".to_string(),
            ),
        ]);

        assert_eq!(build_v4_additional_headers(&headers), "cache-control;host");
    }

    #[test]
    fn put_object_headers_include_immutable_cache_control_for_generated_assets() {
        let headers = build_put_object_headers(BTreeMap::from([(
            "asset-kind".to_string(),
            "puzzle-cover".to_string(),
        )]))
        .expect("headers should build");

        assert_eq!(
            headers.get("Cache-Control"),
            Some(&DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string())
        );
        assert_eq!(
            headers.get("x-oss-meta-asset-kind"),
            Some(&"puzzle-cover".to_string())
        );
    }

    #[tokio::test]
    async fn put_object_rejects_empty_body_before_calling_oss() {
        let client = build_client();
        let error = client
            .put_object(
                &reqwest::Client::new(),
                OssPutObjectRequest {
                    prefix: LegacyAssetPrefix::Characters,
                    path_segments: vec!["hero".to_string()],
                    file_name: "master.png".to_string(),
                    content_type: Some("image/png".to_string()),
                    access: OssObjectAccess::Private,
                    metadata: BTreeMap::new(),
                    body: Vec::new(),
                },
            )
            .await
            .expect_err("empty server upload should fail before network");

        assert_eq!(
            error,
            OssError::InvalidRequest("服务端上传对象内容不能为空".to_string())
        );
    }
}