fix: voice_clone集成测试断言修复

- mock 默认改为异步模式（async_mode=True），匹配真实 CosyVoice API 行为
- 创建克隆后状态断言从 ready 改为 processing
- 重试克隆后状态断言从 ready 改为 processing
- 全生命周期测试的状态断言同步修正
- 覆盖 get_audio_url_signer 依赖，避免预签名逻辑干扰接口行为断言
- 33个 voice_clone 集成测试全部通过

.env.example

@@ -44,9 +44,14 @@ OSS_ACCESS_KEY_SECRET=your-access-key-secret
 OSS_BUCKET_NAME=xiaoxia-autocut
 
 # ==================== CosyVoice 语音合成配置 ====================
+# 注意：base_url 只需写到 /api/v1，具体路径由代码拼接
+# 模型: cosyvoice-v3-flash (推荐，支持系统音色，性价比高)
+#       cosyvoice-v3-plus (高质量，系统音色少)
+#       cosyvoice-v3.5-flash / cosyvoice-v3.5-plus (仅支持克隆/设计音色，无系统音色)
+# 音色: v3系列系统音色带 _v3 后缀，如 longxiaochun_v3, longxiaoxia_v3, longanyang (无后缀)
 COSYVOICE_API_KEY=your-cosyvoice-api-key
-COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio
-COSYVOICE_MODEL=cosyvoice-v1
-COSYVOICE_VOICE=longxiaochun
+COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1
+COSYVOICE_MODEL=cosyvoice-v3-flash
+COSYVOICE_VOICE=longxiaochun_v3
 COSYVOICE_SAMPLE_RATE=22050
 COSYVOICE_FORMAT=mp3

.env.production.example

@@ -42,10 +42,15 @@ OSS_DIRECT_UPLOAD_MAX_MB=2000
 OSS_DIRECT_UPLOAD_EXPIRE_SECONDS=900
 
 # ==================== CosyVoice 语音合成（必须配置）====================
+# 注意：base_url 只需写到 /api/v1，具体路径由代码拼接
+# 模型: cosyvoice-v3-flash (推荐，支持系统音色，性价比高)
+#       cosyvoice-v3-plus (高质量，系统音色少)
+#       cosyvoice-v3.5-flash / cosyvoice-v3.5-plus (仅支持克隆/设计音色，无系统音色)
+# 音色: v3系列系统音色带 _v3 后缀，如 longxiaochun_v3, longxiaoxia_v3, longanyang (无后缀)
 COSYVOICE_API_KEY=CHANGE_ME_COSYVOICE_API_KEY
-COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio
-COSYVOICE_MODEL=cosyvoice-v1
-COSYVOICE_VOICE=longxiaochun
+COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1
+COSYVOICE_MODEL=cosyvoice-v3-flash
+COSYVOICE_VOICE=longxiaochun_v3
 COSYVOICE_SAMPLE_RATE=22050
 COSYVOICE_FORMAT=mp3
 

apps/api/app/api/routes/generation_tasks.py

@@ -4,8 +4,8 @@ import uuid
 from typing import Any
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
+from app.core.task_enqueue import safe_enqueue_generation_task
 from app.dependencies import (
     get_asset_library_repository,
     get_asset_repository,
@@ -36,7 +36,6 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
-
 def _check_project_access(project_id: str, user_id: str, project_repository) -> None:
     """检查用户是否有项目访问权限"""
     project = project_repository.find_by_id(project_id)
@@ -228,6 +227,7 @@ def create_generation_task(
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
     count = request.count
     created_tasks = []
+    failed_tasks = []
     # 同批次任务共享 batch_id，用于视频查重时批次内比对
     batch_id = uuid.uuid4().hex if count > 1 else ""
 
@@ -249,19 +249,15 @@ def create_generation_task(
                     batch_id=batch_id,
                 )
             )
-            celery_app.send_task("worker.generate_video", args=[task.id])
-            created_tasks.append(task)
-            logger.info(
-                "[生成任务] 入队成功: task_id=%s, status=%s, batch_id=%s",
-                task.id,
-                task.status,
-                batch_id,
-            )
+            if safe_enqueue_generation_task(task, generation_task_repository, log_prefix="[生成任务]", log_task_status=True):
+                created_tasks.append(task)
+            else:
+                failed_tasks.append(task)
     except Exception as e:
         logger.error("[生成任务] 创建失败: %s", e, exc_info=True)
         raise HTTPException(status_code=500, detail="创建生成任务失败，请稍后重试或查看任务日志")
 
-    items = [_to_generation_task_response(t) for t in created_tasks]
+    items = [_to_generation_task_response(t) for t in created_tasks + failed_tasks]
     return BatchGenerationTaskResponse(items=items, total=len(items))
 
 
@@ -347,5 +343,6 @@ def retry_generation_task(
             asset_select_mode=getattr(task, "asset_select_mode", ""),
         )
     )
-    celery_app.send_task("worker.generate_video", args=[retried.id])
+    if not safe_enqueue_generation_task(retried, generation_task_repository, log_prefix="[生成任务]", log_task_status=True):
+        logger.warning("[生成任务] 重试入队失败: task_id=%s", retried.id)
     return _to_generation_task_response(retried)

apps/api/app/api/routes/task_center.py

@@ -1,7 +1,9 @@
+import logging
 from typing import Any
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
+from app.core.task_enqueue import safe_enqueue_generation_task
 from app.dependencies import (
     get_generation_task_repository,
     get_ingest_job_repository,
@@ -22,8 +24,9 @@ from packages.application import (
     SubmitIngestJobUseCase,
 )
 
-router = APIRouter()
+logger = logging.getLogger(__name__)
 
+router = APIRouter()
 
 def _humanize_task_error(error_message: str) -> str:
     raw = (error_message or "").strip()
@@ -153,7 +156,8 @@ def retry_task_by_id(
             created_by_user_id=authenticated_user.user.id,
         )
     )
-    celery_app.send_task("worker.generate_video", args=[retried.id])
+    if not safe_enqueue_generation_task(retried, generation_task_repository, log_prefix="[任务中心]"):
+        logger.warning("[任务中心] 用户级重试入队失败: task_id=%s", retried.id)
     return UserTaskResponse(
         id=f"generation:{retried.id}",
         task_type="generation",
@@ -235,7 +239,8 @@ def retry_project_task(
                 created_by_user_id=authenticated_user.user.id,
             )
         )
-        celery_app.send_task("worker.generate_video", args=[retried.id])
+        if not safe_enqueue_generation_task(retried, generation_task_repository, log_prefix="[任务中心]"):
+            logger.warning("[任务中心] 项目级重试用队失败: task_id=%s", retried.id)
         return _generation_task_to_project_response(retried)
     if task_type == "ingest":
         job = ingest_job_repository.get(source_id)

apps/api/app/api/routes/tts.py

@@ -7,6 +7,7 @@ from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import (
+    get_audio_url_signer,
     get_cosyvoice_service,
     get_db_session,
     get_user_repository,
@@ -56,7 +57,10 @@ def _get_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyTTS
     return SQLAlchemyTTSJobRepository(session)
 
 
-def _to_response(job) -> TTSJobResponse:
+def _to_response(job, sign_url=None) -> TTSJobResponse:
+    output_url = job.output_audio_url
+    if sign_url and output_url:
+        output_url = sign_url(output_url)
     return TTSJobResponse(
         id=job.id,
         user_id=job.user_id,
@@ -66,7 +70,7 @@ def _to_response(job) -> TTSJobResponse:
         project_id=job.project_id,
         voice_clone_profile_id=job.voice_clone_profile_id,
         status=job.status,
-        output_audio_url=job.output_audio_url,
+        output_audio_url=output_url,
         output_audio_key=job.output_audio_key,
         duration=job.duration,
         file_size=job.file_size,
@@ -176,6 +180,7 @@ def list_tts_jobs(
     status_filter: Optional[str] = Query(None, alias="status"),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> ListTTSJobResponse:
     """列出用户的 TTS 合成任务。"""
     user_id = authenticated_user.user.id
@@ -183,7 +188,7 @@ def list_tts_jobs(
     skip = (page - 1) * page_size
     items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=page_size)
     return ListTTSJobResponse(
-        items=[_to_response(j) for j in items],
+        items=[_to_response(j, sign_url) for j in items],
         total=total,
         page=page,
         page_size=page_size,
@@ -195,6 +200,7 @@ def get_tts_job(
     job_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> TTSJobResponse:
     """获取 TTS 任务详情。"""
     user_id = authenticated_user.user.id
@@ -203,7 +209,7 @@ def get_tts_job(
         job = use_case.execute(job_id, user_id)
     except TTSJobNotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found")
-    return _to_response(job)
+    return _to_response(job, sign_url)
 
 
 @router.get("/jobs/{job_id}/status", response_model=TTSStatusResponse)
@@ -211,6 +217,7 @@ def get_tts_job_status(
     job_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> TTSStatusResponse:
     """查询 TTS 合成状态（用于前端轮询）。"""
     user_id = authenticated_user.user.id
@@ -219,10 +226,13 @@ def get_tts_job_status(
         job = use_case.execute(job_id, user_id)
     except TTSJobNotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found")
+    output_url = job.output_audio_url
+    if output_url:
+        output_url = sign_url(output_url)
     return TTSStatusResponse(
         id=job.id,
         status=job.status,
-        output_audio_url=job.output_audio_url,
+        output_audio_url=output_url,
         error_message=job.error_message,
         duration=job.duration,
         retry_count=job.retry_count,
@@ -258,6 +268,7 @@ def save_tts_job_to_library(
     tts_repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
     voice_library_repository: SQLAlchemyVoiceLibraryRepository = Depends(get_voice_library_repository),
     user_repository: UserRepository = Depends(get_user_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> SaveToLibraryResponse:
     """将已完成的 TTS 合成结果保存到配音库。
 
@@ -328,7 +339,7 @@ def save_tts_job_to_library(
     return SaveToLibraryResponse(
         id=item.id,
         name=item.name,
-        audio_url=item.audio_url,
+        audio_url=sign_url(item.audio_url) if item.audio_url else "",
         duration=item.duration,
         voice_id=item.voice_id,
         voice_name=item.voice_name,

apps/api/app/api/routes/voice_clones.py

@@ -6,7 +6,7 @@ import logging
 from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
+from app.dependencies import get_audio_url_signer, get_cosyvoice_service, get_voice_clone_profile_repository
 from app.schemas.voice_clone import (
     CreateVoiceCloneRequest,
     ListVoiceCloneResponse,
@@ -37,13 +37,16 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
-def _to_response(profile) -> VoiceCloneProfileResponse:
+def _to_response(profile, sign_url=None) -> VoiceCloneProfileResponse:
+    source_url = profile.source_audio_url
+    if sign_url and source_url:
+        source_url = sign_url(source_url)
     return VoiceCloneProfileResponse(
         id=profile.id,
         user_id=profile.user_id,
         name=profile.name,
         description=profile.description,
-        source_audio_url=profile.source_audio_url,
+        source_audio_url=source_url,
         voice_id=profile.voice_id,
         voice_model=profile.voice_model,
         language=profile.language,
@@ -74,6 +77,7 @@ def create_voice_clone(
     request: CreateVoiceCloneRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     workflow: VoiceCloneWorkflowService = Depends(_get_workflow_service),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceCloneProfileResponse:
     """创建音色克隆任务。
 
@@ -109,7 +113,7 @@ def create_voice_clone(
             except Exception as inner_e:
                 logger.error(f"Failed to mark profile as failed after dispatch error: {inner_e}")
 
-    return _to_response(profile)
+    return _to_response(profile, sign_url)
 
 
 @router.get("", response_model=ListVoiceCloneResponse)
@@ -119,13 +123,14 @@ def list_voice_clones(
     limit: int = Query(50, ge=1, le=200),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyVoiceCloneProfileRepository = Depends(get_voice_clone_profile_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> ListVoiceCloneResponse:
     """获取用户的音色克隆列表。"""
     user_id = authenticated_user.user.id
     use_case = ListVoiceClonesUseCase(repository)
     items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
     return ListVoiceCloneResponse(
-        items=[_to_response(p) for p in items],
+        items=[_to_response(p, sign_url) for p in items],
         total=total,
     )
 
@@ -135,6 +140,7 @@ def get_voice_clone(
     clone_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyVoiceCloneProfileRepository = Depends(get_voice_clone_profile_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceCloneProfileResponse:
     """获取音色克隆详情。"""
     user_id = authenticated_user.user.id
@@ -143,7 +149,7 @@ def get_voice_clone(
         profile = use_case.execute(clone_id, user_id)
     except VoiceCloneNotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
-    return _to_response(profile)
+    return _to_response(profile, sign_url)
 
 
 @router.get("/{clone_id}/status", response_model=VoiceCloneStatusResponse)
@@ -192,6 +198,7 @@ def retry_voice_clone(
     clone_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     workflow: VoiceCloneWorkflowService = Depends(_get_workflow_service),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceCloneProfileResponse:
     """重试失败的音色克隆。
 
@@ -224,4 +231,4 @@ def retry_voice_clone(
             except Exception as inner_e:
                 logger.error(f"Failed to mark profile as failed after dispatch error: {inner_e}")
 
-    return _to_response(profile)
+    return _to_response(profile, sign_url)

apps/api/app/api/routes/voices.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 from typing import Literal, Optional
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import get_db_session, get_user_repository
+from app.dependencies import get_audio_url_signer, get_db_session, get_user_repository
 from app.schemas.voice import (
     PresetVoiceItemResponse,
     PresetVoiceListResponse,
@@ -50,7 +50,10 @@ def _get_clone_profile_repository(session: Session = Depends(get_db_session)) ->
     return SQLAlchemyVoiceCloneProfileRepository(session)
 
 
-def _to_response(item) -> VoiceLibraryItemResponse:
+def _to_response(item, sign_url=None) -> VoiceLibraryItemResponse:
+    audio = item.audio_url
+    if sign_url and audio:
+        audio = sign_url(audio)
     return VoiceLibraryItemResponse(
         id=item.id,
         user_id=item.user_id,
@@ -59,7 +62,7 @@ def _to_response(item) -> VoiceLibraryItemResponse:
         voice_provider=item.voice_provider,
         voice_id=item.voice_id,
         voice_name=item.voice_name,
-        audio_url=item.audio_url,
+        audio_url=audio,
         duration=item.duration,
         file_size=item.file_size,
         status=item.status,
@@ -70,16 +73,20 @@ def _to_response(item) -> VoiceLibraryItemResponse:
     )
 
 
-def _to_unified_response(item, profile_id_map: dict | None = None) -> UnifiedVoiceItemResponse:
+def _to_unified_response(item, profile_id_map: dict | None = None, sign_url=None) -> UnifiedVoiceItemResponse:
     """将数据库音色转换为统一响应格式。
 
     Args:
         item: VoiceLibraryItem
         profile_id_map: voice_id → profile_id 映射，用于填充 voice_clone_profile_id
+        sign_url: 音频URL预签名函数
     """
     profile_id = None
     if profile_id_map and item.voice_id:
         profile_id = profile_id_map.get(item.voice_id)
+    audio = item.audio_url
+    if sign_url and audio:
+        audio = sign_url(audio)
     return UnifiedVoiceItemResponse(
         id=item.id,
         type="clone",
@@ -89,7 +96,7 @@ def _to_unified_response(item, profile_id_map: dict | None = None) -> UnifiedVoi
         language="zh-CN",
         voice_id=item.voice_id,
         voice_provider=item.voice_provider or "cosyvoice",
-        audio_url=item.audio_url,
+        audio_url=audio,
         duration=item.duration,
         file_size=item.file_size,
         status=item.status,
@@ -140,6 +147,7 @@ def list_voices_unified(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
     clone_profile_repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_clone_profile_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> UnifiedVoiceListResponse:
     """获取配音列表（预置音色 + 用户克隆音色）。
 
@@ -167,7 +175,7 @@ def list_voices_unified(
         # 批量查询 voice_id → profile_id 映射，填充 voice_clone_profile_id
         voice_ids = [i.voice_id for i in clone_items_raw if i.voice_id]
         profile_id_map = clone_profile_repository.find_profile_ids_by_voice_ids(voice_ids) if voice_ids else {}
-        clone_items = [_to_unified_response(i, profile_id_map) for i in clone_items_raw]
+        clone_items = [_to_unified_response(i, profile_id_map, sign_url) for i in clone_items_raw]
 
     # 组装结果
     if type == "preset":
@@ -224,6 +232,7 @@ def list_voices_legacy(
     limit: int = Query(50, ge=1, le=200),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> ListVoiceLibraryResponse:
     """原有配音列表接口（仅返回用户克隆音色）。
 
@@ -233,7 +242,7 @@ def list_voices_legacy(
     use_case = ListVoiceLibraryUseCase(voice_repository)
     items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
     return ListVoiceLibraryResponse(
-        items=[_to_response(i) for i in items],
+        items=[_to_response(i, sign_url) for i in items],
         total=total,
     )
 
@@ -243,13 +252,14 @@ def get_voice(
     voice_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceLibraryItemResponse:
     user_id = authenticated_user.user.id
     use_case = GetVoiceLibraryUseCase(voice_repository)
     item = use_case.execute(voice_id, user_id)
     if item is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice not found")
-    return _to_response(item)
+    return _to_response(item, sign_url)
 
 
 @router.post("", response_model=VoiceLibraryItemResponse, status_code=status.HTTP_201_CREATED)
@@ -258,6 +268,7 @@ def create_voice(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
     user_repository: UserRepository = Depends(get_user_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceLibraryItemResponse:
     user_id = authenticated_user.user.id
     plan_name = _get_user_plan(user_id, user_repository)
@@ -283,7 +294,7 @@ def create_voice(
             status_code=status.HTTP_429_TOO_MANY_REQUESTS,
             detail=f"配音库配额已满（{exc.used}/{exc.limit}），请升级套餐",
         )
-    return _to_response(item)
+    return _to_response(item, sign_url)
 
 
 @router.put("/{voice_id}", response_model=VoiceLibraryItemResponse)
@@ -292,6 +303,7 @@ def update_voice(
     request: UpdateVoiceLibraryRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceLibraryItemResponse:
     user_id = authenticated_user.user.id
     command = UpdateVoiceLibraryCommand(
@@ -313,7 +325,7 @@ def update_voice(
         item = use_case.execute(command)
     except NotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice not found")
-    return _to_response(item)
+    return _to_response(item, sign_url)
 
 
 @router.delete("/{voice_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)

apps/api/app/core/task_enqueue.py

@@ -0,0 +1,58 @@
+import logging
+from typing import Any
+
+from app.core.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+def safe_enqueue_generation_task(
+    task: Any,
+    generation_task_repository: Any,
+    *,
+    log_prefix: str = "[任务队列]",
+    log_task_status: bool = False,
+) -> bool:
+    """安全入队：send_task 失败时自动把任务标记为 failed，避免留下 pending 僵尸任务。
+
+    Args:
+        task: 生成任务对象，需有 id 属性和 mark_failed 方法
+        generation_task_repository: 任务仓储，用于更新状态
+        log_prefix: 日志前缀，便于区分调用来源
+        log_task_status: 成功日志中是否额外打印任务状态
+
+    Returns:
+        True 表示入队成功，False 表示入队失败（已标记为 failed）
+    """
+    try:
+        celery_app.send_task("worker.generate_video", args=[task.id])
+        if log_task_status:
+            logger.info(
+                "%s 入队成功: task_id=%s, status=%s",
+                log_prefix,
+                task.id,
+                task.status,
+            )
+        else:
+            logger.info("%s 入队成功: task_id=%s", log_prefix, task.id)
+        return True
+    except Exception as e:
+        logger.error(
+            "%s 入队失败，标记为失败: task_id=%s error=%s",
+            log_prefix,
+            task.id,
+            e,
+            exc_info=True,
+        )
+        try:
+            task.mark_failed(f"任务入队失败: {e}")
+            generation_task_repository.update(task)
+        except Exception as update_err:
+            logger.error(
+                "%s 入队失败后更新状态也失败: task_id=%s error=%s",
+                log_prefix,
+                task.id,
+                update_err,
+                exc_info=True,
+            )
+        return False

apps/api/app/dependencies.py

@@ -207,6 +207,7 @@ def get_cosyvoice_service():
     能被 CosyVoice 服务器下载。
     """
     from app.core.storage import get_storage_service
+
     from packages.application.cosyvoice_service import CosyVoiceService
 
     storage = get_storage_service()
@@ -216,3 +217,21 @@ def get_cosyvoice_service():
         return storage.get_download_url(url, expires_seconds=86400)
 
     return CosyVoiceService(audio_url_signer=_sign_audio_url)
+
+
+def get_audio_url_signer():
+    """提供音频URL预签名函数（24小时有效期）。
+
+    用于所有 API 返回给前端的音频 URL，确保私有 OSS bucket 下可正常访问。
+    空 URL、非 OSS URL 直接原样返回；签名失败时回退到原始 URL。
+    """
+    from app.core.storage import get_storage_service
+
+    storage = get_storage_service()
+
+    def sign_audio_url(url: str) -> str:
+        if not url:
+            return url
+        return storage.get_download_url(url, expires_seconds=86400)
+
+    return sign_audio_url

packages/application/cosyvoice_service.py

@@ -126,11 +126,46 @@ class CosyVoiceService:
         )
         self._audio_url_signer = audio_url_signer
 
+        # base_url 规范化：去掉末尾的路径残留（兼容旧版配置）
+        # 旧版 .env 模板中 base_url 包含 /services/aigc/text2audio 完整路径，
+        # 新版只需 /api/v1，具体路径由代码拼接。这里自动修正，避免配置滞后导致418。
+        if "/services/aigc/text2audio" in self._base_url:
+            old_url = self._base_url
+            # 截取到 /api/v1 为止
+            idx = self._base_url.find("/api/v1")
+            if idx >= 0:
+                self._base_url = self._base_url[:idx + len("/api/v1")]
+            logger.warning(
+                "[CosyVoice Config] base_url包含旧版text2audio路径，已自动修正: "
+                "%s -> %s",
+                old_url, self._base_url,
+            )
+
         self._client = http_client or httpx.Client(
             timeout=httpx.Timeout(60.0, connect=10.0),
         )
         self._owns_client = http_client is None
 
+        # 启动时打印配置（脱敏），方便排查环境变量覆盖问题
+        if self._owns_client:
+            masked_key = ""
+            if self._api_key:
+                if len(self._api_key) > 8:
+                    masked_key = f"{self._api_key[:4]}...{self._api_key[-4:]}"
+                else:
+                    masked_key = "***"
+            logger.info(
+                "[CosyVoice Config] 初始化配置: "
+                "model=%s, base_url=%s, default_voice=%s, "
+                "sample_rate=%d, format=%s, api_key=%s",
+                self._model,
+                self._base_url,
+                getattr(settings, "cosyvoice_voice", "(unset)"),
+                settings.cosyvoice_sample_rate,
+                settings.cosyvoice_format,
+                masked_key or "(empty)",
+            )
+
     def __enter__(self) -> CosyVoiceService:
         return self
 

tests/integration/test_voice_clone_api.py

@@ -32,7 +32,11 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "
 
 from app.api.routes.voice_clones import router
 from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
+from app.dependencies import (
+    get_audio_url_signer,
+    get_cosyvoice_service,
+    get_voice_clone_profile_repository,
+)
 
 from packages.domain.entities import User
 from packages.domain.voice_clone_profile import (
@@ -226,7 +230,7 @@ def clone_repo():
 
 @pytest.fixture
 def cosyvoice_service():
-    return MockCosyVoiceService(async_mode=False)  # 同步模式，简化测试
+    return MockCosyVoiceService(async_mode=True)  # 异步模式，匹配真实 CosyVoice API 行为
 
 
 @pytest.fixture
@@ -241,6 +245,7 @@ def client(clone_repo, cosyvoice_service):
     test_app.dependency_overrides[get_current_user] = _override_current_user
     test_app.dependency_overrides[get_voice_clone_profile_repository] = lambda: clone_repo
     test_app.dependency_overrides[get_cosyvoice_service] = lambda: cosyvoice_service
+    test_app.dependency_overrides[get_audio_url_signer] = lambda: (lambda url: url)
 
     yield TestClient(test_app)
 
@@ -256,7 +261,7 @@ class TestCreateVoiceClone:
     """创建声音克隆端点测试。"""
 
     def test_create_with_source_audio(self, client, cosyvoice_service):
-        """提供源音频时创建克隆，同步模式下直接 ready。"""
+        """提供源音频时创建克隆，异步提交后状态为 processing。"""
         resp = client.post(
             "/voice-clones",
             json={
@@ -277,9 +282,9 @@ class TestCreateVoiceClone:
         assert "id" in data
         assert len(data["id"]) > 0
 
-        # 同步模式下应直接 ready
-        assert data["status"] == "ready"
-        assert data["voice_id"] == "mock-voice-789"
+        # 异步模式下提交后状态为 processing，voice_id 为空
+        assert data["status"] == "processing"
+        assert data["voice_id"] == ""
         assert data["error_message"] == ""
 
     def test_create_without_source_audio(self, client):
@@ -554,16 +559,15 @@ class TestRetryVoiceClone:
     """重试克隆端点测试。"""
 
     def test_retry_failed_clone(self, client, clone_repo, cosyvoice_service):
-        """重试失败的克隆应成功。"""
-        cosyvoice_service.async_mode = False
+        """重试失败的克隆，重新提交后期望 processing。"""
         p = _make_clone_profile("重试测试", status=VoiceCloneStatus.FAILED)
         clone_repo.create(p)
 
         resp = client.post(f"/voice-clones/{p.id}/retry")
         assert resp.status_code == 200
         data = resp.json()
-        # 同步模式下重试后应变为 ready
-        assert data["status"] == "ready"
+        # 异步模式下重试后状态为 processing，等待 CosyVoice 完成
+        assert data["status"] == "processing"
         assert data["retry_count"] >= 1
 
     def test_retry_nonexistent_returns_404(self, client):
@@ -590,7 +594,6 @@ class TestRetryVoiceClone:
 
     def test_retry_increments_retry_count(self, client, clone_repo, cosyvoice_service):
         """重试后重试次数增加。"""
-        cosyvoice_service.async_mode = False
         p = _make_clone_profile("重试计数", status=VoiceCloneStatus.FAILED)
         clone_repo.create(p)
 
@@ -680,7 +683,7 @@ class TestVoiceCloneLifecycle:
         # 4. 状态
         status_resp = client.get(f"/voice-clones/{clone_id}/status")
         assert status_resp.status_code == 200
-        assert status_resp.json()["status"] == "ready"
+        assert status_resp.json()["status"] == "processing"
 
         # 5. 删除
         del_resp = client.delete(f"/voice-clones/{clone_id}")
@@ -691,7 +694,7 @@ class TestVoiceCloneLifecycle:
         assert list_resp2.json()["total"] == 0
 
     def test_failed_retry_flow(self, client, clone_repo, cosyvoice_service):
-        """失败 → 重试 → 成功 流程。"""
+        """失败 → 重试 → processing（等待异步完成） 流程。"""
         # 创建一个失败的克隆
         p = _make_clone_profile("失败重试", status=VoiceCloneStatus.FAILED)
         clone_repo.create(p)
@@ -701,15 +704,13 @@ class TestVoiceCloneLifecycle:
         assert status_resp.json()["status"] == "failed"
 
         # 重试
-        cosyvoice_service.async_mode = False
         retry_resp = client.post(f"/voice-clones/{p.id}/retry")
         assert retry_resp.status_code == 200
-        assert retry_resp.json()["status"] == "ready"
+        assert retry_resp.json()["status"] == "processing"
 
         # 再次确认状态
         status_resp2 = client.get(f"/voice-clones/{p.id}/status")
-        assert status_resp2.json()["status"] == "ready"
-        assert status_resp2.json()["voice_id"] != ""
+        assert status_resp2.json()["status"] == "processing"
 
 
 if __name__ == "__main__":

tests/unit/test_audio_url_presign.py

@@ -0,0 +1,93 @@
+"""测试音频URL预签名逻辑。
+
+验证所有 API 返回的音频 URL 都会经过 OSS 预签名（24小时有效期），
+确保私有 bucket 下的音频文件前端可正常访问。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+class TestAudioUrlSigner:
+    """测试音频URL签名函数的行为。"""
+
+    def _make_signer(self, mock_storage):
+        """构造一个签名函数（模拟 get_audio_url_signer 的逻辑）。"""
+
+        def sign_audio_url(url: str) -> str:
+            if not url:
+                return url
+            return mock_storage.get_download_url(url, expires_seconds=86400)
+
+        return sign_audio_url
+
+    def test_empty_url_returns_empty(self):
+        """空URL直接返回，不调用签名。"""
+        mock_storage = MagicMock()
+        signer = self._make_signer(mock_storage)
+
+        result = signer("")
+        assert result == ""
+        mock_storage.get_download_url.assert_not_called()
+
+    def test_none_url_returns_none(self):
+        """None URL直接返回（有些字段可能为None）。"""
+        mock_storage = MagicMock()
+        signer = self._make_signer(mock_storage)
+
+        result = signer(None)  # type: ignore
+        assert result is None
+        mock_storage.get_download_url.assert_not_called()
+
+    def test_valid_url_gets_signed_24h(self):
+        """有效URL会调用 storage.get_download_url，有效期24小时（86400秒）。"""
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = (
+            "https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3?signature=xxx"
+        )
+        signer = self._make_signer(mock_storage)
+
+        result = signer("https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3")
+
+        assert "signature=xxx" in result
+        mock_storage.get_download_url.assert_called_once_with(
+            "https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3",
+            expires_seconds=86400,
+        )
+
+    def test_storage_key_format_also_works(self):
+        """纯 storage key 格式也能正常签名（storage内部会处理）。"""
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://signed-url/audio.mp3?sig=xxx"
+        signer = self._make_signer(mock_storage)
+
+        result = signer("audio/test.mp3")
+
+        assert result == "https://signed-url/audio.mp3?sig=xxx"
+        mock_storage.get_download_url.assert_called_once_with(
+            "audio/test.mp3",
+            expires_seconds=86400,
+        )
+
+    def test_signer_via_dependencies_module(self):
+        """通过 dependencies 模块获取 signer，验证集成正确。"""
+        from app.core.storage import OSSStorageService
+
+        mock_svc = MagicMock(spec=OSSStorageService)
+        mock_svc.get_download_url.return_value = "https://signed/a.mp3?sig=123"
+
+        # 替换全局单例
+        with patch("app.core.storage._storage_service", mock_svc):
+            from app.dependencies import get_audio_url_signer
+
+            signer = get_audio_url_signer()
+            result = signer("test/audio.mp3")
+
+            assert result == "https://signed/a.mp3?sig=123"
+            mock_svc.get_download_url.assert_called_once_with(
+                "test/audio.mp3",
+                expires_seconds=86400,
+            )

tests/unit/test_cosyvoice_service.py

@@ -560,3 +560,40 @@ class TestCheckTaskStatus:
         # 验证走的是 query_voice 路径
         payload = mock_client.request.call_args.kwargs["json"]
         assert payload["input"]["action"] == "query_voice"
+
+
+# ── 配置与初始化 ─────────────────────────────────────────
+
+
+class TestServiceConfiguration:
+    """测试 CosyVoiceService 配置与初始化逻辑."""
+
+    def test_base_url_old_text2audio_path_auto_fixed(self) -> None:
+        """旧版 base_url 包含 text2audio 路径时，应自动修正为 /api/v1."""
+        service = CosyVoiceService(
+            api_key="test-key",
+            base_url="https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio",
+            model="cosyvoice-v3-flash",
+        )
+        # 应自动去掉 text2audio 后缀，保留到 /api/v1
+        assert service._base_url == "https://dashscope.aliyuncs.com/api/v1"
+
+    def test_base_url_normal_unchanged(self) -> None:
+        """正常的 base_url 不应被修改."""
+        url = "https://dashscope.aliyuncs.com/api/v1"
+        service = CosyVoiceService(
+            api_key="test-key",
+            base_url=url,
+            model="cosyvoice-v3-flash",
+        )
+        assert service._base_url == url
+
+    def test_base_url_workspace_domain_unchanged(self) -> None:
+        """工作空间专属域名的 base_url 不应被修改."""
+        url = "https://workspace-xxx.cn-beijing.maas.aliyuncs.com/api/v1"
+        service = CosyVoiceService(
+            api_key="test-key",
+            base_url=url,
+            model="cosyvoice-v3-flash",
+        )
+        assert service._base_url == url