Merge pull request #240 from yiyabo/feat/dashscope-integration

Add DashScope (Alibaba Cloud Bailian) provider: image gen + TTS + ASR

.agents/skills/dashscope/SKILL.md

@@ -0,0 +1,136 @@
+---
+name: dashscope
+description: DashScope (Alibaba Cloud Bailian / 阿里云百炼) integration — image generation (qwen-image-2.0-pro), text-to-speech (qwen3-tts-flash), and ASR with word-level timestamps (qwen3-asr-flash-filetrans). Use when generating images via Qwen-Image, narrating via Qwen-TTS, or transcribing with word-level timestamps via Qwen-ASR.
+---
+
+# DashScope
+
+Requires `DASHSCOPE_API_KEY` in `.env`. Get one at https://dashscope.aliyun.com/.
+
+## Current API
+
+**CRITICAL:** DashScope's `/compatible-mode/v1/` only supports `/chat/completions` and `/embeddings`. Image generation, TTS, and ASR all use **DashScope-native endpoints** — not OpenAI-compatible paths.
+
+All three tools use `Authorization: Bearer $DASHSCOPE_API_KEY`.
+
+### Image Generation
+
+```text
+POST https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation
+```
+
+- Model: `qwen-image-2.0-pro` (default), `qwen-image-max`, `wan2.7-image`, `z-image-turbo`
+- Body: `{model, input: {messages: [{role: "user", content: [{text: "prompt"}]}]}, parameters: {size: "W*H", n, prompt_extend, watermark}}`
+- **Size format uses asterisk:** `"1024*1024"` not `"1024x1024"`
+- Response: `output.choices[0].message.content[0].image` (URL, valid ~24h) — must download separately
+
+### Text-to-Speech
+
+```text
+POST https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation
+```
+
+Same endpoint as image gen, different body.
+
+- Model: `qwen3-tts-flash` (default), `qwen3-tts-instruct-flash`, `qwen-tts-2025-05-22`
+- Body: `{model, input: {text, voice: "Cherry", language_type: "Auto"}}`
+- Response: `output.audio.url` (WAV, valid ~24h) — must download separately
+
+### ASR with Word-Level Timestamps
+
+```text
+POST https://dashscope.aliyuncs.com/api/v1/services/audio/asr/transcription
+Header: X-DashScope-Async: enable
+```
+
+- Model: `qwen3-asr-flash-filetrans` (NOT `qwen3-asr-flash` — the sync version has no word timestamps)
+- Body: `{model, input: {file_url: "https://public-url/audio.mp3"}, parameters: {enable_words: true, language_hints: ["zh","en"]}}`
+- Returns `task_id` → poll `GET /api/v1/tasks/{task_id}` until `SUCCEEDED` → download `output.result.transcription_url` → JSON with `transcripts[].sentences[].words[]`
+- Timestamps in `begin_time`/`end_time` are in **milliseconds** — the tool normalizes to seconds
+
+## OpenMontage Usage
+
+### Image via selector
+
+```python
+from tools.graphics.image_selector import ImageSelector
+
+result = ImageSelector().execute({
+    "preferred_provider": "dashscope",
+    "prompt": "一只猫坐在沙发上",
+    "output_path": "projects/my-video/assets/images/cat.png",
+})
+```
+
+### TTS via selector
+
+```python
+from tools.audio.tts_selector import TTSSelector
+
+result = TTSSelector().execute({
+    "preferred_provider": "dashscope",
+    "text": "如果 AI 真的会改变未来，普通人到底该怎么参与？",
+    "voice": "Cherry",
+    "output_path": "projects/my-video/assets/audio/narration.wav",
+})
+```
+
+### ASR directly (word timestamps for subtitles)
+
+```python
+from tools.analysis.dashscope_asr import DashscopeAsr
+
+result = DashscopeAsr().execute({
+    "audio_url": "https://example.com/narration.wav",
+    "output_path": "projects/my-video/assets/audio/transcription.json",
+})
+
+# result.data["words"] is a flat list of {text, begin_time_seconds, end_time_seconds}
+```
+
+## Recommended Workflow
+
+1. **Image:** Generate a sample first. Check `prompt_extend: true` (default) — DashScope rewrites your prompt for better results. Disable if you need literal prompt adherence.
+2. **TTS:** Generate a 10-15 second sample before full narration. Approve voice and pacing before committing to full generation.
+3. **ASR:** Audio must be at a **publicly accessible URL**. Upload to any public host (S3, etc.) first. Local paths are rejected with a clear error.
+4. **Subtitles:** Build from `result.data["words"]` — each word has `begin_time_seconds` and `end_time_seconds`. Group words into caption phrases by language semantics, not fixed character count.
+
+## Parameters
+
+### Image (`dashscope_image`)
+- `prompt` (required): text prompt
+- `model`: default `qwen-image-2.0-pro`
+- `size`: default `"1024*1024"` — **asterisk separator, not "x"**
+- `n`: 1-6 images
+- `negative_prompt`: things to avoid (max 500 chars)
+- `prompt_extend`: default `true` — auto-rewrite prompt for better results
+- `watermark`: default `false`
+- `seed`: for reproducibility
+
+### TTS (`dashscope_tts`)
+- `text` (required): text to synthesize (max 600 chars for qwen3-tts-flash)
+- `model`: default `qwen3-tts-flash`
+- `voice`: default `"Cherry"` — other voices: `"Ethan"`, `"Chelsie"`, etc.
+- `language_type`: default `"Auto"` — `"Chinese"`, `"English"`, `"Japanese"`, `"Korean"`
+- `instructions`: natural language delivery instructions (only for `qwen3-tts-instruct-flash`)
+
+### ASR (`dashscope_asr`)
+- `audio_url` (required): **must be publicly accessible URL**
+- `model`: `qwen3-asr-flash-filetrans` (only model that supports word timestamps)
+- `language_hints`: default `["zh", "en"]`
+- `enable_words`: default `true` — required for word-level timestamps
+- `poll_interval_seconds`: default `5.0`
+- `timeout_seconds`: default `300`
+
+## Troubleshooting
+
+- **Image size error:** Use `"W*H"` with asterisk, not `"WxH"`. Example: `"2048*2048"`.
+- **TTS no audio URL:** Check `output.audio.url` — if empty, the model name or voice may be wrong.
+- **ASR "file not accessible":** `audio_url` must be publicly reachable. DashScope servers fetch the file; local paths and auth-gated URLs don't work.
+- **ASR poll timeout:** Increase `timeout_seconds` (default 300). Long audio files take longer to transcribe.
+- **ASR no word timestamps:** Ensure `enable_words: true` and model is `qwen3-asr-flash-filetrans` (not the sync `qwen3-asr-flash`).
+- **Auth error (401):** Verify `DASHSCOPE_API_KEY` is set. Use `Authorization: Bearer $KEY` header.
+
+## Safety
+
+Never print or write the API key to logs, metadata, patches, or project artifacts. `.env.example` should contain only empty variable names. The tool's `_safe_error()` method redacts the key from error messages.

.env.example

@@ -22,6 +22,10 @@ DOUBAO_SPEECH_API_KEY=       # Volcengine Doubao Speech TTS (new console API Key
 DOUBAO_SPEECH_VOICE_TYPE=    # Default Doubao speaker/voice type, e.g. zh_female_vv_uranus_bigtts
 # Piper local voices do not require env vars; install `piper-tts` via pip
 
+# --- DashScope (Alibaba Cloud Bailian) ---
+DASHSCOPE_API_KEY=           # Qwen image gen (qwen-image-2.0-pro), TTS (qwen3-tts-flash), ASR with word timestamps (qwen3-asr-flash-filetrans)
+                              # Get one at https://dashscope.aliyun.com/
+
 # --- Music ---
 SUNO_API_KEY=                # Suno AI music generation (full songs, instrumentals, any genre)
 

docs/PROVIDERS.md

@@ -41,6 +41,7 @@ OPENAI_API_KEY=              # OpenAI TTS + GPT Image 2 images
 XAI_API_KEY=                 # xAI Grok image generation/editing + Grok video generation
 DOUBAO_SPEECH_API_KEY=       # Volcengine Doubao Speech TTS (strong Mandarin narration)
 DOUBAO_SPEECH_VOICE_TYPE=    # Default Doubao speaker/voice type
+DASHSCOPE_API_KEY=           # Alibaba DashScope (Qwen image gen, TTS, ASR with word timestamps)
 
 # MULTI-MODEL GATEWAY (one key, 6+ tools)
 FAL_KEY=                     # FLUX, Recraft, Kling, Veo, MiniMax video
@@ -94,6 +95,43 @@ OpenMontage now uses those published rates in the Grok tool estimators.
 
 ---
 
+### Alibaba DashScope — Qwen Image + TTS + ASR
+
+> **Best for Chinese-language production.** One key unlocks Qwen-Image generation, Qwen-TTS Mandarin narration, and Qwen-ASR with word-level timestamps — the only DashScope path that provides word-level granularity for subtitle alignment.
+
+**Tools unlocked:** `dashscope_image`, `dashscope_tts`, `dashscope_asr`
+**Env var:** `DASHSCOPE_API_KEY`
+
+#### Setup
+
+1. Go to [dashscope.aliyun.com](https://dashscope.aliyun.com/)
+2. Create an Alibaba Cloud account if you don't have one
+3. Generate an API key in the DashScope console
+4. Add to `.env`: `DASHSCOPE_API_KEY=sk-...`
+
+#### What it's best for
+
+- Chinese-language image generation with strong prompt understanding (Qwen-Image)
+- Natural Mandarin narration (Qwen-TTS, Cherry voice)
+- Word-level timestamp transcription for subtitle alignment (Qwen-ASR filetrans)
+- Replacing the broken `whisperx` slot for ASR
+
+#### API notes
+
+DashScope's `/compatible-mode/v1/` only supports `/chat/completions` and `/embeddings`. Image gen, TTS, and ASR all use DashScope-native endpoints with nested `{model, input, parameters}` request shape — not OpenAI-compatible paths.
+
+The ASR tool (`qwen3-asr-flash-filetrans`) uses an async submit-poll pattern. Audio must be at a publicly accessible URL (local files are not supported). Word timestamps are in milliseconds, normalized to seconds by the tool.
+
+#### Pricing
+
+| Model | Price |
+|------|-------|
+| `qwen-image-2.0-pro` | ~$0.02 per image (check console for current rates) |
+| `qwen3-tts-flash` | ~$0.000015 per character |
+| `qwen3-asr-flash-filetrans` | Per-minute billing (check console) |
+
+---
+
 ### fal.ai — Multi-Model Gateway
 
 > **Broad single-key coverage.** One API key unlocks image and video providers across multiple models.

tests/contracts/test_dashscope_tools.py

@@ -0,0 +1,709 @@
+"""Contract tests for DashScope (Alibaba Cloud Bailian) provider tools.
+
+These tests verify that the tools satisfy the BaseTool contract without
+requiring a real DashScope API key or making any API calls. They check
+class attributes, schemas, status reporting, cost estimates, and the
+Layer 3 skill file existence.
+
+Run: pytest tests/contracts/test_dashscope_tools.py -v
+"""
+
+from pathlib import Path
+
+import pytest
+
+from tools.base_tool import (
+    BaseTool,
+    ExecutionMode,
+    ToolRuntime,
+    ToolStability,
+    ToolStatus,
+    ToolTier,
+)
+from tools.graphics.dashscope_image import DashscopeImage
+from tools.audio.dashscope_tts import DashscopeTTS
+from tools.analysis.dashscope_asr import DashscopeAsr
+
+TOOLS = [DashscopeImage, DashscopeTTS, DashscopeAsr]
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+
+EXPECTED_TIER = {
+    DashscopeImage: ToolTier.GENERATE,
+    DashscopeTTS: ToolTier.VOICE,
+    DashscopeAsr: ToolTier.ANALYZE,
+}
+EXPECTED_CAPABILITY = {
+    DashscopeImage: "image_generation",
+    DashscopeTTS: "tts",
+    DashscopeAsr: "analysis",
+}
+EXPECTED_EXECUTION_MODE = {
+    DashscopeImage: ExecutionMode.SYNC,
+    DashscopeTTS: ExecutionMode.SYNC,
+    DashscopeAsr: ExecutionMode.ASYNC,
+}
+
+
+# ------------------------------------------------------------------
+# Contract compliance (parametrized over all 3 tools)
+# ------------------------------------------------------------------
+
+@pytest.mark.parametrize("cls", TOOLS, ids=lambda c: c.name)
+class TestContract:
+
+    def test_inherits_base_tool(self, cls):
+        assert issubclass(cls, BaseTool)
+
+    def test_has_required_identity(self, cls):
+        tool = cls()
+        assert tool.name
+        assert tool.version
+        assert tool.provider == "dashscope"
+        assert tool.capability == EXPECTED_CAPABILITY[cls]
+        assert tool.tier == EXPECTED_TIER[cls]
+        assert tool.stability == ToolStability.EXPERIMENTAL
+        assert tool.runtime == ToolRuntime.API
+
+    def test_has_input_schema(self, cls):
+        tool = cls()
+        schema = tool.input_schema
+        assert schema.get("type") == "object"
+        props = schema.get("properties", {})
+        required = schema.get("required", [])
+        # Each tool has at least one required field
+        assert len(required) >= 1
+        for field in required:
+            assert field in props
+
+    def test_has_capabilities(self, cls):
+        tool = cls()
+        assert len(tool.capabilities) > 0
+
+    def test_has_agent_skills(self, cls):
+        tool = cls()
+        assert tool.agent_skills
+        assert "dashscope" in tool.agent_skills
+
+    def test_dashscope_layer3_skill_exists(self, cls):
+        skill_path = (
+            PROJECT_ROOT / ".agents" / "skills" / "dashscope" / "SKILL.md"
+        )
+        assert skill_path.exists(), f"Missing Layer 3 skill: {skill_path}"
+        content = skill_path.read_text(encoding="utf-8")
+        assert "DASHSCOPE_API_KEY" in content
+
+    def test_has_fallbacks(self, cls):
+        tool = cls()
+        assert tool.fallback or tool.fallback_tools
+
+    def test_has_install_instructions(self, cls):
+        tool = cls()
+        assert tool.install_instructions
+        assert "DASHSCOPE_API_KEY" in tool.install_instructions
+
+    def test_get_info_returns_dict(self, cls):
+        tool = cls()
+        info = tool.get_info()
+        assert isinstance(info, dict)
+        assert info["name"] == tool.name
+        assert info["provider"] == "dashscope"
+        assert info["runtime"] == "api"
+        assert info["agent_skills"] == ["dashscope"]
+
+    def test_execution_mode(self, cls):
+        tool = cls()
+        assert tool.execution_mode == EXPECTED_EXECUTION_MODE[cls]
+
+    def test_status_unavailable_without_key(self, cls, monkeypatch):
+        monkeypatch.delenv("DASHSCOPE_API_KEY", raising=False)
+        tool = cls()
+        assert tool.get_status() == ToolStatus.UNAVAILABLE
+
+    def test_status_available_with_key(self, cls, monkeypatch):
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing")
+        tool = cls()
+        assert tool.get_status() == ToolStatus.AVAILABLE
+
+    def test_idempotency_key_fields(self, cls):
+        tool = cls()
+        assert len(tool.idempotency_key_fields) > 0
+
+    def test_has_resource_profile(self, cls):
+        tool = cls()
+        assert tool.resource_profile.network_required is True
+        assert tool.resource_profile.vram_mb == 0
+
+    def test_has_retry_policy(self, cls):
+        tool = cls()
+        assert tool.retry_policy.max_retries >= 0
+
+    def test_has_side_effects(self, cls):
+        tool = cls()
+        assert len(tool.side_effects) > 0
+        # Must mention it calls the API
+        assert any("API" in s for s in tool.side_effects)
+
+    def test_has_user_visible_verification(self, cls):
+        tool = cls()
+        assert len(tool.user_visible_verification) > 0
+
+    def test_lazy_imports_requests(self, cls):
+        """Tool module must not import requests at top level (registry
+        discovery must stay fast)."""
+        import importlib
+        import sys
+        # Remove requests from cache to simulate fresh import
+        mod_name = cls.__module__
+        if "requests" in sys.modules:
+            del sys.modules["requests"]
+        # Re-import the tool module — should not pull in requests
+        # (requests is imported inside execute(), not at module level)
+        importlib.reload(sys.modules[mod_name])
+        # The tool module itself should not have imported requests
+        # (it's inside execute, so module-level reload shouldn't trigger it)
+        # This is a smoke test — the real proof is that registry.discover()
+        # works without requests installed, but requests IS installed here.
+
+    def test_estimate_cost_returns_float(self, cls):
+        tool = cls()
+        # Use tool-specific minimal inputs
+        if cls is DashscopeImage:
+            cost = tool.estimate_cost({"prompt": "test", "n": 1})
+        elif cls is DashscopeTTS:
+            cost = tool.estimate_cost({"text": "test"})
+        else:
+            cost = tool.estimate_cost({"audio_url": "https://x.com/a.mp3"})
+        assert isinstance(cost, float)
+        assert cost >= 0.0
+
+    def test_dry_run_returns_dict(self, cls):
+        tool = cls()
+        if cls is DashscopeImage:
+            result = tool.dry_run({"prompt": "test"})
+        elif cls is DashscopeTTS:
+            result = tool.dry_run({"text": "test"})
+        else:
+            result = tool.dry_run({"audio_url": "https://x.com/a.mp3"})
+        assert isinstance(result, dict)
+        assert "tool" in result
+        assert result["tool"] == tool.name
+
+
+# ------------------------------------------------------------------
+# Image-specific tests
+# ------------------------------------------------------------------
+
+class TestDashscopeImageSpecific:
+
+    def test_default_model_is_qwen_image_2_pro(self):
+        tool = DashscopeImage()
+        assert tool.input_schema["properties"]["model"]["default"] == "qwen-image-2.0-pro"
+
+    def test_default_size_uses_asterisk_format(self):
+        """CRITICAL: DashScope uses W*H (asterisk), not WxH."""
+        tool = DashscopeImage()
+        size_default = tool.input_schema["properties"]["size"]["default"]
+        assert "*" in size_default
+        assert "x" not in size_default.lower()
+
+    def test_cost_positive_for_image(self):
+        tool = DashscopeImage()
+        assert tool.estimate_cost({"prompt": "test", "n": 1}) > 0.0
+
+    def test_cost_scales_with_n(self):
+        tool = DashscopeImage()
+        cost1 = tool.estimate_cost({"prompt": "test", "n": 1})
+        cost3 = tool.estimate_cost({"prompt": "test", "n": 3})
+        assert cost3 > cost1
+
+    def test_build_payload_uses_asterisk_size(self):
+        tool = DashscopeImage()
+        payload = tool._build_payload({"prompt": "test"})
+        assert "*" in payload["parameters"]["size"]
+
+    def test_build_payload_includes_messages_structure(self):
+        tool = DashscopeImage()
+        payload = tool._build_payload({"prompt": "a cat"})
+        assert "input" in payload
+        assert "messages" in payload["input"]
+        assert payload["input"]["messages"][0]["content"][0]["text"] == "a cat"
+
+    def test_build_payload_optional_negative_prompt(self):
+        tool = DashscopeImage()
+        payload = tool._build_payload({
+            "prompt": "test",
+            "negative_prompt": "blurry",
+        })
+        assert payload["parameters"]["negative_prompt"] == "blurry"
+
+    def test_build_payload_omits_negative_prompt_when_absent(self):
+        tool = DashscopeImage()
+        payload = tool._build_payload({"prompt": "test"})
+        assert "negative_prompt" not in payload["parameters"]
+
+    def test_safe_error_redacts_key(self, monkeypatch):
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345")
+        redacted = DashscopeImage._safe_error(
+            Exception("failed with key secret-key-12345")
+        )
+        assert "secret-key-12345" not in redacted
+        assert "[redacted]" in redacted
+
+
+# ------------------------------------------------------------------
+# PR review regressions: multi-image download + idempotency keys
+# ------------------------------------------------------------------
+
+class TestDashscopeImageMultiOutput:
+    """Regression tests for PR #240 review: the tool advertised
+    multiple_outputs and accepted n>1 but only downloaded the first image.
+    Verify every returned URL is saved and returned as an artifact."""
+
+    def test_extract_image_urls_across_choices(self):
+        data = {
+            "output": {
+                "choices": [
+                    {"finish_reason": "stop", "message": {"content": [{"image": "https://x/1.png"}]}},
+                    {"finish_reason": "stop", "message": {"content": [{"image": "https://x/2.png"}]}},
+                    {"finish_reason": "stop", "message": {"content": [{"image": "https://x/3.png"}]}},
+                ]
+            }
+        }
+        assert DashscopeImage._extract_image_urls(data) == [
+            "https://x/1.png",
+            "https://x/2.png",
+            "https://x/3.png",
+        ]
+
+    def test_extract_image_urls_within_single_choice(self):
+        data = {
+            "output": {
+                "choices": [
+                    {"finish_reason": "stop", "message": {"content": [
+                        {"image": "https://x/1.png"},
+                        {"image": "https://x/2.png"},
+                    ]}}
+                ]
+            }
+        }
+        assert DashscopeImage._extract_image_urls(data) == [
+            "https://x/1.png",
+            "https://x/2.png",
+        ]
+
+    def test_extract_image_urls_empty_when_no_images(self):
+        assert DashscopeImage._extract_image_urls({}) == []
+        assert DashscopeImage._extract_image_urls(
+            {"output": {"choices": []}}
+        ) == []
+        assert DashscopeImage._extract_image_urls(
+            {"output": {"choices": [{"message": {"content": [{"text": "x"}]}}]}}
+        ) == []
+
+    def test_extract_image_urls_skips_failed_choices(self):
+        """Per Qwen Cloud docs, a multi-output task can be SUCCEEDED with
+        partial failures. Choices with finish_reason != "stop" must be
+        skipped so we don't download partial/empty results. The failed
+        choice here carries a non-empty URL to prove it is the
+        finish_reason filter (not the truthy-url check) that skips it."""
+        data = {
+            "output": {
+                "choices": [
+                    {
+                        "finish_reason": "stop",
+                        "message": {"content": [{"image": "https://x/ok.png"}]},
+                    },
+                    {
+                        "finish_reason": "content_filter",
+                        "message": {"content": [{"image": "https://x/blocked.png"}]},
+                    },
+                ]
+            }
+        }
+        assert DashscopeImage._extract_image_urls(data) == ["https://x/ok.png"]
+
+    def test_resolve_output_paths_single_unchanged(self):
+        paths = DashscopeImage._resolve_output_paths("foo.png", 1)
+        assert paths == [Path("foo.png")]
+
+    def test_resolve_output_paths_multiple_inserts_index(self):
+        paths = DashscopeImage._resolve_output_paths("foo.png", 3)
+        assert paths == [
+            Path("foo_1.png"),
+            Path("foo_2.png"),
+            Path("foo_3.png"),
+        ]
+
+    def test_resolve_output_paths_multiple_without_extension(self):
+        paths = DashscopeImage._resolve_output_paths("foo", 2)
+        assert paths == [Path("foo_1"), Path("foo_2")]
+
+    def test_execute_downloads_all_images(self, monkeypatch, tmp_path):
+        """The bug: n=3 returned images_generated=3 but downloaded 1 file.
+        Mock the DashScope response with 3 URLs and verify all 3 are saved."""
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key")
+
+        class FakeResp:
+            def __init__(self, payload, content=b""):
+                self._payload = payload
+                self.content = content
+
+            def raise_for_status(self):
+                pass
+
+            def json(self):
+                return self._payload
+
+        api_response = {
+            "output": {
+                "choices": [
+                    {"finish_reason": "stop", "message": {"content": [{"image": f"https://x/{i}.png"}]}}
+                    for i in range(1, 4)
+                ]
+            },
+            "usage": {"image_count": 3},
+        }
+
+        import requests
+
+        monkeypatch.setattr(
+            requests, "post", lambda *a, **kw: FakeResp(api_response)
+        )
+        monkeypatch.setattr(
+            requests,
+            "get",
+            lambda url, **kw: FakeResp({}, content=f"img-{url}".encode()),
+        )
+
+        out = tmp_path / "shot.png"
+        result = DashscopeImage().execute({
+            "prompt": "test", "n": 3, "output_path": str(out),
+        })
+
+        assert result.success is True
+        assert result.data["images_generated"] == 3
+        assert len(result.artifacts) == 3
+        assert (tmp_path / "shot_1.png").exists()
+        assert (tmp_path / "shot_2.png").exists()
+        assert (tmp_path / "shot_3.png").exists()
+
+    def test_execute_single_image_uses_base_path(self, monkeypatch, tmp_path):
+        """n=1 must keep the legacy single-path behavior (no _1 suffix)."""
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key")
+
+        class FakeResp:
+            def __init__(self, payload, content=b""):
+                self._payload = payload
+                self.content = content
+
+            def raise_for_status(self):
+                pass
+
+            def json(self):
+                return self._payload
+
+        api_response = {
+            "output": {
+                "choices": [
+                    {"finish_reason": "stop", "message": {"content": [{"image": "https://x/1.png"}]}}
+                ]
+            },
+            "usage": {"image_count": 1},
+        }
+
+        import requests
+
+        monkeypatch.setattr(
+            requests, "post", lambda *a, **kw: FakeResp(api_response)
+        )
+        monkeypatch.setattr(
+            requests,
+            "get",
+            lambda url, **kw: FakeResp({}, content=b"img-bytes"),
+        )
+
+        out = tmp_path / "shot.png"
+        result = DashscopeImage().execute({
+            "prompt": "test", "n": 1, "output_path": str(out),
+        })
+
+        assert result.success is True
+        assert result.data["images_generated"] == 1
+        assert result.artifacts == [str(out)]
+        assert out.exists()
+        assert not (tmp_path / "shot_1.png").exists()
+
+
+class TestDashscopeIdempotencyKeys:
+    """Regression tests for PR #240 review: idempotency keys must include
+    all output-affecting fields so different requests don't collide and
+    reuse stale artifacts."""
+
+    def test_image_idempotency_includes_all_output_fields(self):
+        fields = DashscopeImage().idempotency_key_fields
+        for field in (
+            "prompt", "model", "size", "n",
+            "negative_prompt", "seed", "prompt_extend", "watermark",
+        ):
+            assert field in fields, f"image idempotency missing {field}"
+
+    def test_image_idempotency_differs_on_negative_prompt(self):
+        tool = DashscopeImage()
+        base = {"prompt": "x", "model": "m", "size": "1024*1024", "n": 1}
+        assert tool.idempotency_key(base) != tool.idempotency_key(
+            {**base, "negative_prompt": "blurry"}
+        )
+
+    def test_image_idempotency_differs_on_seed(self):
+        tool = DashscopeImage()
+        base = {"prompt": "x", "model": "m", "size": "1024*1024", "n": 1}
+        assert tool.idempotency_key(base) != tool.idempotency_key(
+            {**base, "seed": 42}
+        )
+
+    def test_image_idempotency_differs_on_prompt_extend(self):
+        tool = DashscopeImage()
+        base = {"prompt": "x", "model": "m", "size": "1024*1024", "n": 1}
+        assert tool.idempotency_key(
+            {**base, "prompt_extend": True}
+        ) != tool.idempotency_key({**base, "prompt_extend": False})
+
+    def test_image_idempotency_differs_on_watermark(self):
+        tool = DashscopeImage()
+        base = {"prompt": "x", "model": "m", "size": "1024*1024", "n": 1}
+        assert tool.idempotency_key(
+            {**base, "watermark": False}
+        ) != tool.idempotency_key({**base, "watermark": True})
+
+    def test_tts_idempotency_includes_instructions(self):
+        assert "instructions" in DashscopeTTS().idempotency_key_fields
+
+    def test_tts_idempotency_differs_on_instructions(self):
+        tool = DashscopeTTS()
+        base = {
+            "text": "hi", "voice": "Cherry",
+            "model": "qwen3-tts-flash", "language_type": "Auto",
+        }
+        assert tool.idempotency_key(base) != tool.idempotency_key(
+            {**base, "instructions": "speak softly"}
+        )
+
+    def test_asr_idempotency_includes_enable_words_and_language_hints(self):
+        fields = DashscopeAsr().idempotency_key_fields
+        assert "enable_words" in fields
+        assert "language_hints" in fields
+
+    def test_asr_idempotency_differs_on_enable_words(self):
+        tool = DashscopeAsr()
+        base = {"audio_url": "https://x/a.mp3", "model": "qwen3-asr-flash-filetrans"}
+        assert tool.idempotency_key(
+            {**base, "enable_words": True}
+        ) != tool.idempotency_key({**base, "enable_words": False})
+
+    def test_asr_idempotency_differs_on_language_hints(self):
+        tool = DashscopeAsr()
+        base = {"audio_url": "https://x/a.mp3", "model": "qwen3-asr-flash-filetrans"}
+        assert tool.idempotency_key(
+            {**base, "language_hints": ["zh"]}
+        ) != tool.idempotency_key(
+            {**base, "language_hints": ["zh", "en"]}
+        )
+
+
+# ------------------------------------------------------------------
+# TTS-specific tests
+# ------------------------------------------------------------------
+
+class TestDashscopeTtsSpecific:
+
+    def test_default_model_is_qwen3_tts_flash(self):
+        tool = DashscopeTTS()
+        assert tool.input_schema["properties"]["model"]["default"] == "qwen3-tts-flash"
+
+    def test_default_voice_is_cherry(self):
+        tool = DashscopeTTS()
+        assert tool.input_schema["properties"]["voice"]["default"] == "Cherry"
+
+    def test_default_language_is_auto(self):
+        tool = DashscopeTTS()
+        assert tool.input_schema["properties"]["language_type"]["default"] == "Auto"
+
+    def test_cost_scales_with_text_length(self):
+        tool = DashscopeTTS()
+        cost_short = tool.estimate_cost({"text": "hi"})
+        cost_long = tool.estimate_cost({"text": "hi " * 100})
+        assert cost_long > cost_short
+
+    def test_build_payload_includes_input_text_voice(self):
+        tool = DashscopeTTS()
+        payload = tool._build_payload({"text": "hello", "voice": "Ethan"})
+        assert payload["input"]["text"] == "hello"
+        assert payload["input"]["voice"] == "Ethan"
+
+    def test_build_payload_adds_instructions_for_instruct_model(self):
+        tool = DashscopeTTS()
+        payload = tool._build_payload({
+            "text": "hello",
+            "instructions": "speak softly",
+        })
+        assert payload["input"]["instructions"] == "speak softly"
+        assert payload["input"]["optimize_instructions"] is True
+
+    def test_fallback_includes_piper(self):
+        """Piper is the free offline fallback — must be in fallback list."""
+        tool = DashscopeTTS()
+        assert "piper_tts" in tool.fallback_tools
+
+    def test_safe_error_redacts_key(self, monkeypatch):
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345")
+        redacted = DashscopeTTS._safe_error(
+            Exception("failed with key secret-key-12345")
+        )
+        assert "secret-key-12345" not in redacted
+        assert "[redacted]" in redacted
+
+
+# ------------------------------------------------------------------
+# ASR-specific tests
+# ------------------------------------------------------------------
+
+class TestDashscopeAsrSpecific:
+
+    def test_default_model_is_filetrans(self):
+        """CRITICAL: must use qwen3-asr-flash-filetrans, NOT qwen3-asr-flash.
+        The sync version does not support word-level timestamps."""
+        tool = DashscopeAsr()
+        assert tool.input_schema["properties"]["model"]["default"] == "qwen3-asr-flash-filetrans"
+
+    def test_execution_mode_is_async(self):
+        tool = DashscopeAsr()
+        assert tool.execution_mode == ExecutionMode.ASYNC
+
+    def test_default_enable_words_is_true(self):
+        """Word-level timestamps must be enabled by default."""
+        tool = DashscopeAsr()
+        assert tool.input_schema["properties"]["enable_words"]["default"] is True
+
+    def test_default_language_hints_includes_zh_en(self):
+        tool = DashscopeAsr()
+        hints = tool.input_schema["properties"]["language_hints"]["default"]
+        assert "zh" in hints
+        assert "en" in hints
+
+    def test_rejects_local_file_path(self, monkeypatch):
+        """audio_url must be a public URL — local paths are rejected."""
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing")
+        tool = DashscopeAsr()
+        result = tool.execute({"audio_url": "/local/path/audio.mp3"})
+        assert result.success is False
+        assert "publicly accessible URL" in result.error
+
+    def test_rejects_relative_path(self, monkeypatch):
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing")
+        tool = DashscopeAsr()
+        result = tool.execute({"audio_url": "audio.mp3"})
+        assert result.success is False
+        assert "publicly accessible URL" in result.error
+
+    def test_rejects_empty_url(self, monkeypatch):
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "fake-key-for-testing")
+        tool = DashscopeAsr()
+        result = tool.execute({"audio_url": ""})
+        assert result.success is False
+        assert "required" in result.error.lower()
+
+    def test_rejects_no_key(self, monkeypatch):
+        monkeypatch.delenv("DASHSCOPE_API_KEY", raising=False)
+        tool = DashscopeAsr()
+        result = tool.execute({"audio_url": "https://example.com/audio.mp3"})
+        assert result.success is False
+        assert "DASHSCOPE_API_KEY" in result.error
+
+    def test_build_payload_enables_words(self):
+        tool = DashscopeAsr()
+        payload = tool._build_payload({"audio_url": "https://x.com/a.mp3"})
+        assert payload["parameters"]["enable_words"] is True
+
+    def test_build_payload_includes_file_url(self):
+        """qwen3-asr-flash-filetrans uses file_url (singular string),
+        NOT file_urls (plural array) like paraformer-v2."""
+        tool = DashscopeAsr()
+        payload = tool._build_payload({"audio_url": "https://x.com/a.mp3"})
+        assert payload["input"]["file_url"] == "https://x.com/a.mp3"
+
+    def test_extract_words_normalizes_ms_to_seconds(self):
+        """Word timestamps from DashScope are in milliseconds; the tool
+        must normalize to seconds for downstream subtitle building."""
+        fake_transcription = {
+            "transcripts": [
+                {
+                    "sentences": [
+                        {
+                            "words": [
+                                {"text": "hello", "begin_time": 1000, "end_time": 1500},
+                                {"text": "world", "begin_time": 1500, "end_time": 2000},
+                            ]
+                        }
+                    ]
+                }
+            ]
+        }
+        words = DashscopeAsr._extract_words(fake_transcription)
+        assert len(words) == 2
+        assert words[0]["text"] == "hello"
+        assert words[0]["begin_time_seconds"] == 1.0
+        assert words[0]["end_time_seconds"] == 1.5
+        assert words[1]["begin_time_seconds"] == 1.5
+        assert words[1]["end_time_seconds"] == 2.0
+
+    def test_extract_words_handles_empty_transcription(self):
+        words = DashscopeAsr._extract_words({})
+        assert words == []
+
+    def test_is_public_url_accepts_https(self):
+        assert DashscopeAsr._is_public_url("https://example.com/audio.mp3") is True
+
+    def test_is_public_url_rejects_local(self):
+        assert DashscopeAsr._is_public_url("/local/path/audio.mp3") is False
+        assert DashscopeAsr._is_public_url("audio.mp3") is False
+        assert DashscopeAsr._is_public_url("ftp://example.com/audio.mp3") is False
+
+    def test_safe_error_redacts_key(self, monkeypatch):
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "secret-key-12345")
+        redacted = DashscopeAsr._safe_error(
+            Exception("failed with key secret-key-12345")
+        )
+        assert "secret-key-12345" not in redacted
+        assert "[redacted]" in redacted
+
+
+# ------------------------------------------------------------------
+# Registry discovery
+# ------------------------------------------------------------------
+
+class TestDashscopeRegistryDiscovery:
+
+    def test_all_three_tools_discoverable(self):
+        from tools.tool_registry import ToolRegistry
+        registry = ToolRegistry()
+        registry.discover()
+        dashscope_tools = [
+            t for t in registry._tools.values()
+            if t.provider == "dashscope"
+        ]
+        names = {t.name for t in dashscope_tools}
+        assert names == {"dashscope_image", "dashscope_tts", "dashscope_asr"}
+
+    def test_image_selector_finds_dashscope(self):
+        """image_selector should auto-discover dashscope_image by capability."""
+        from tools.graphics.image_selector import ImageSelector
+        selector = ImageSelector()
+        # Selector discovers providers by capability="image_generation"
+        # dashscope_image has that capability, so it should be routable
+        assert DashscopeImage().capability == "image_generation"
+
+    def test_tts_selector_finds_dashscope(self):
+        """tts_selector should auto-discover dashscope_tts by capability."""
+        from tools.audio.tts_selector import TTSSelector
+        selector = TTSSelector()
+        assert DashscopeTTS().capability == "tts"

tests/contracts/test_phase3_contracts.py

@@ -161,7 +161,14 @@ class TestCapabilityMetadata:
         catalog = reg.capability_catalog()
         assert "tts" in catalog
         providers = {item["provider"] for item in catalog["tts"] if item["provider"] != "selector"}
-        assert providers == {"doubao", "elevenlabs", "google_tts", "openai", "piper"}
+        assert providers == {
+            "dashscope",
+            "doubao",
+            "elevenlabs",
+            "google_tts",
+            "openai",
+            "piper",
+        }
 
 
 # ---- Animated Explainer Pipeline ----

tools/analysis/dashscope_asr.py

@@ -0,0 +1,386 @@
+"""DashScope (Alibaba Cloud Bailian) ASR with word-level timestamps.
+
+Uses the DashScope-native async transcription endpoint with
+X-DashScope-Async: enable header. The model qwen3-asr-flash-filetrans is the
+ONLY DashScope path that returns word-level timestamps (the sync
+qwen3-asr-flash via /chat/completions does not).
+
+Pattern: submit (POST) -> poll (GET /tasks/{task_id}) -> download
+transcription_url -> parse transcripts[].sentences[].words[].
+
+This tool replaces the broken `whisperx` slot for subtitle-aligned
+transcription. Word timestamps are normalized from milliseconds to seconds.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+from tools.base_tool import (
+    BaseTool,
+    Determinism,
+    ExecutionMode,
+    ResourceProfile,
+    RetryPolicy,
+    ToolResult,
+    ToolRuntime,
+    ToolStability,
+    ToolStatus,
+    ToolTier,
+)
+
+
+class DashscopeAsr(BaseTool):
+    name = "dashscope_asr"
+    version = "0.1.0"
+    tier = ToolTier.ANALYZE
+    capability = "analysis"
+    provider = "dashscope"
+    stability = ToolStability.EXPERIMENTAL
+    execution_mode = ExecutionMode.ASYNC
+    determinism = Determinism.DETERMINISTIC
+    runtime = ToolRuntime.API
+
+    dependencies = []
+    install_instructions = (
+        "Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n"
+        "  Get one at https://dashscope.aliyun.com/"
+    )
+    fallback = "transcriber"
+    fallback_tools = ["transcriber"]
+    agent_skills = ["dashscope"]
+
+    capabilities = [
+        "speech_to_text",
+        "word_timestamps",
+        "multilingual",
+    ]
+    supports = {
+        "word_timestamps": True,
+        "multilingual": True,
+        "offline": False,
+    }
+    best_for = [
+        "word-level timestamp transcription for subtitle alignment",
+        "Mandarin and English speech recognition",
+        "replacing whisperx when word-level granularity is needed",
+    ]
+    not_good_for = [
+        "real-time transcription",
+        "local/offline transcription",
+    ]
+
+    input_schema = {
+        "type": "object",
+        "required": ["audio_url"],
+        "properties": {
+            "audio_url": {
+                "type": "string",
+                "description": (
+                    "Publicly accessible URL of the audio file to transcribe. "
+                    "Must be reachable by DashScope servers — local paths "
+                    "are not supported."
+                ),
+            },
+            "model": {
+                "type": "string",
+                "enum": ["qwen3-asr-flash-filetrans"],
+                "default": "qwen3-asr-flash-filetrans",
+            },
+            "language_hints": {
+                "type": "array",
+                "items": {"type": "string"},
+                "default": ["zh", "en"],
+                "description": (
+                    "Language hints to improve accuracy. "
+                    'Examples: ["zh", "en", "ja"].'
+                ),
+            },
+            "enable_words": {
+                "type": "boolean",
+                "default": True,
+                "description": (
+                    "Enable word-level timestamps. Required for subtitle "
+                    "alignment."
+                ),
+            },
+            "output_path": {"type": "string"},
+            "poll_interval_seconds": {
+                "type": "number",
+                "default": 5.0,
+                "minimum": 1.0,
+            },
+            "timeout_seconds": {
+                "type": "integer",
+                "default": 300,
+                "minimum": 30,
+            },
+        },
+    }
+
+    resource_profile = ResourceProfile(
+        cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=20, network_required=True
+    )
+    retry_policy = RetryPolicy(
+        max_retries=2,
+        backoff_seconds=2.0,
+        retryable_errors=["timeout", "rate_limit"],
+    )
+    idempotency_key_fields = ["audio_url", "model", "enable_words", "language_hints"]
+    side_effects = [
+        "writes transcription JSON to output_path",
+        "calls DashScope (Alibaba Cloud) ASR API (async submit + poll)",
+    ]
+    user_visible_verification = [
+        "Check transcription text for accuracy",
+        "Verify word-level timestamps before building subtitles",
+    ]
+
+    SUBMIT_URL = (
+        "https://dashscope.aliyuncs.com/api/v1/services/audio/asr/"
+        "transcription"
+    )
+    POLL_URL_TEMPLATE = (
+        "https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}"
+    )
+
+    def get_status(self) -> ToolStatus:
+        if os.environ.get("DASHSCOPE_API_KEY"):
+            return ToolStatus.AVAILABLE
+        return ToolStatus.UNAVAILABLE
+
+    def estimate_cost(self, inputs: dict[str, Any]) -> float:
+        # DashScope ASR pricing is per-minute; check console for actual cost.
+        return 0.0
+
+    def execute(self, inputs: dict[str, Any]) -> ToolResult:
+        api_key = os.environ.get("DASHSCOPE_API_KEY")
+        if not api_key:
+            return ToolResult(
+                success=False,
+                error="DASHSCOPE_API_KEY not set. " + self.install_instructions,
+            )
+
+        audio_url = inputs.get("audio_url", "").strip()
+        if not audio_url:
+            return ToolResult(
+                success=False, error="audio_url is required."
+            )
+        if not self._is_public_url(audio_url):
+            return ToolResult(
+                success=False,
+                error=(
+                    "audio_url must be a publicly accessible URL (http/https). "
+                    "DashScope servers fetch the file; local paths are not "
+                    "supported. Upload the audio to a public location first."
+                ),
+            )
+        # DashScope ASR rejects http:// URLs with InvalidParameter.MalformedURL;
+        # upgrade to https:// before submitting. Note: signed OSS URLs with
+        # query params (Expires, Signature) may also be rejected — prefer clean
+        # public file URLs when possible.
+        if audio_url.startswith("http://"):
+            audio_url = "https://" + audio_url[len("http://"):]
+            inputs = {**inputs, "audio_url": audio_url}
+
+        start = time.time()
+        try:
+            result = self._transcribe(inputs, api_key=api_key)
+        except Exception as exc:
+            return ToolResult(
+                success=False,
+                error=f"DashScope ASR failed: {self._safe_error(exc)}",
+            )
+
+        result.duration_seconds = round(time.time() - start, 2)
+        return result
+
+    def _transcribe(
+        self, inputs: dict[str, Any], *, api_key: str
+    ) -> ToolResult:
+        import json
+        import requests
+
+        payload = self._build_payload(inputs)
+        headers = {
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+            "X-DashScope-Async": "enable",
+        }
+
+        # Submit
+        submit_resp = requests.post(
+            self.SUBMIT_URL, headers=headers, json=payload, timeout=(10, 60)
+        )
+        submit_data = self._json_or_raise(submit_resp)
+        self._raise_for_error(submit_resp.status_code, submit_data)
+
+        task_id = submit_data.get("output", {}).get("task_id")
+        if not task_id:
+            raise RuntimeError(
+                "DashScope ASR submit succeeded but did not return "
+                "output.task_id"
+            )
+
+        # Poll
+        poll_data = self._poll_task(
+            requests_module=requests,
+            api_key=api_key,
+            task_id=task_id,
+            poll_interval=float(inputs.get("poll_interval_seconds", 5.0)),
+            timeout_seconds=int(inputs.get("timeout_seconds", 300)),
+        )
+
+        # qwen3-asr-flash-filetrans returns output.result.transcription_url
+        # (singular "result", NOT "results" array like paraformer-v2)
+        result = poll_data.get("output", {}).get("result", {})
+        transcription_url = result.get("transcription_url")
+        if not transcription_url:
+            raise RuntimeError(
+                "DashScope ASR task succeeded but "
+                "result.transcription_url missing"
+            )
+
+        # Download transcription JSON
+        trans_resp = requests.get(transcription_url, timeout=120)
+        trans_resp.raise_for_status()
+        transcription = trans_resp.json()
+
+        # Save full transcription
+        output_path = Path(
+            inputs.get("output_path", "dashscope_asr.json")
+        )
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(
+            json.dumps(transcription, ensure_ascii=False, indent=2) + "\n",
+            encoding="utf-8",
+        )
+
+        # Parse word-level timestamps (normalize ms -> seconds)
+        words = self._extract_words(transcription)
+        transcripts = transcription.get("transcripts", [])
+
+        return ToolResult(
+            success=True,
+            data={
+                "provider": "dashscope",
+                "model": payload["model"],
+                "audio_url": inputs["audio_url"],
+                "task_id": task_id,
+                "transcripts": transcripts,
+                "words": words,
+                "word_count": len(words),
+                "output": str(output_path),
+            },
+            artifacts=[str(output_path)],
+            cost_usd=self.estimate_cost(inputs),
+            model=payload["model"],
+        )
+
+    def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "model": inputs.get(
+                "model", "qwen3-asr-flash-filetrans"
+            ),
+            "input": {
+                "file_url": inputs["audio_url"],
+            },
+            "parameters": {
+                "enable_words": bool(inputs.get("enable_words", True)),
+                "language_hints": inputs.get(
+                    "language_hints", ["zh", "en"]
+                ),
+            },
+        }
+
+    def _poll_task(
+        self,
+        *,
+        requests_module: Any,
+        api_key: str,
+        task_id: str,
+        poll_interval: float,
+        timeout_seconds: int,
+    ) -> dict[str, Any]:
+        deadline = time.time() + timeout_seconds
+        headers = {"Authorization": f"Bearer {api_key}"}
+        while time.time() < deadline:
+            time.sleep(poll_interval)
+            resp = requests_module.get(
+                self.POLL_URL_TEMPLATE.format(task_id=task_id),
+                headers=headers,
+                timeout=(10, 60),
+            )
+            data = self._json_or_raise(resp)
+            self._raise_for_error(resp.status_code, data)
+            status = data.get("output", {}).get("task_status")
+            if status == "SUCCEEDED":
+                return data
+            if status == "FAILED":
+                msg = data.get("output", {}).get(
+                    "message", "unknown error"
+                )
+                raise RuntimeError(
+                    f"DashScope ASR task failed: {msg}"
+                )
+        raise TimeoutError(
+            f"DashScope ASR task {task_id} did not finish within "
+            f"{timeout_seconds}s"
+        )
+
+    @staticmethod
+    def _is_public_url(url: str) -> bool:
+        return url.startswith("http://") or url.startswith("https://")
+
+    @staticmethod
+    def _extract_words(
+        transcription: dict[str, Any]
+    ) -> list[dict[str, Any]]:
+        """Extract flat word list with timestamps normalized to seconds."""
+        words: list[dict[str, Any]] = []
+        for transcript in transcription.get("transcripts", []):
+            for sentence in transcript.get("sentences", []):
+                for word in sentence.get("words", []):
+                    words.append(
+                        {
+                            "text": word.get("text", ""),
+                            "begin_time_seconds": round(
+                                word.get("begin_time", 0) / 1000.0, 3
+                            ),
+                            "end_time_seconds": round(
+                                word.get("end_time", 0) / 1000.0, 3
+                            ),
+                        }
+                    )
+        return words
+
+    @staticmethod
+    def _json_or_raise(response: Any) -> dict[str, Any]:
+        try:
+            return response.json()
+        except ValueError as exc:
+            raise RuntimeError(
+                f"Non-JSON response from DashScope API: "
+                f"HTTP {response.status_code}"
+            ) from exc
+
+    def _raise_for_error(
+        self, http_status: int, payload: dict[str, Any]
+    ) -> None:
+        if http_status < 400:
+            return
+        code = payload.get("code")
+        message = payload.get("message", "unknown error")
+        raise RuntimeError(
+            f"DashScope API error: HTTP {http_status}, "
+            f"code {code}: {message}"
+        )
+
+    @staticmethod
+    def _safe_error(exc: Exception) -> str:
+        return str(exc).replace(
+            os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]"
+        )

tools/audio/dashscope_tts.py

@@ -0,0 +1,243 @@
+"""DashScope (Alibaba Cloud Bailian) text-to-speech via Qwen-TTS models.
+
+Uses the DashScope-native multimodal-generation endpoint (same as image gen).
+The response contains a temporary audio URL (WAV, valid ~24h) that must be
+downloaded separately — unlike OpenAI TTS which returns raw audio bytes.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+from tools.base_tool import (
+    BaseTool,
+    Determinism,
+    ExecutionMode,
+    ResourceProfile,
+    RetryPolicy,
+    ToolResult,
+    ToolRuntime,
+    ToolStability,
+    ToolStatus,
+    ToolTier,
+)
+
+
+class DashscopeTTS(BaseTool):
+    name = "dashscope_tts"
+    version = "0.1.0"
+    tier = ToolTier.VOICE
+    capability = "tts"
+    provider = "dashscope"
+    stability = ToolStability.EXPERIMENTAL
+    execution_mode = ExecutionMode.SYNC
+    determinism = Determinism.STOCHASTIC
+    runtime = ToolRuntime.API
+
+    dependencies = []
+    install_instructions = (
+        "Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n"
+        "  Get one at https://dashscope.aliyun.com/"
+    )
+    fallback = "piper_tts"
+    fallback_tools = [
+        "doubao_tts",
+        "elevenlabs_tts",
+        "openai_tts",
+        "piper_tts",
+    ]
+    agent_skills = ["dashscope"]
+
+    capabilities = [
+        "text_to_speech",
+        "voice_selection",
+        "multilingual",
+    ]
+    supports = {
+        "voice_cloning": False,
+        "multilingual": True,
+        "offline": False,
+        "native_audio": True,
+    }
+    best_for = [
+        "natural Mandarin and multilingual narration via Qwen-TTS",
+        "cost-effective TTS via Alibaba Cloud",
+        "Chinese-language voiceover production",
+    ]
+    not_good_for = [
+        "fully offline production",
+        "voice clone matching",
+    ]
+
+    input_schema = {
+        "type": "object",
+        "required": ["text"],
+        "properties": {
+            "text": {
+                "type": "string",
+                "description": (
+                    "Text to convert to speech "
+                    "(max 600 chars for qwen3-tts-flash)."
+                ),
+            },
+            "model": {
+                "type": "string",
+                "enum": [
+                    "qwen3-tts-flash",
+                    "qwen3-tts-instruct-flash",
+                    "qwen-tts-2025-05-22",
+                ],
+                "default": "qwen3-tts-flash",
+            },
+            "voice": {
+                "type": "string",
+                "default": "Cherry",
+                "description": (
+                    'DashScope voice name. Examples: "Cherry", "Ethan", '
+                    '"Chelsie".'
+                ),
+            },
+            "language_type": {
+                "type": "string",
+                "default": "Auto",
+                "enum": ["Auto", "Chinese", "English", "Japanese", "Korean"],
+                "description": "Language hint for the TTS model.",
+            },
+            "instructions": {
+                "type": "string",
+                "description": (
+                    "Natural language delivery instructions "
+                    "(only for qwen3-tts-instruct-flash)."
+                ),
+            },
+            "output_path": {"type": "string"},
+        },
+    }
+
+    resource_profile = ResourceProfile(
+        cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=50, network_required=True
+    )
+    retry_policy = RetryPolicy(
+        max_retries=2, retryable_errors=["rate_limit", "timeout"]
+    )
+    idempotency_key_fields = ["text", "voice", "model", "language_type", "instructions"]
+    side_effects = [
+        "writes audio file to output_path",
+        "calls DashScope (Alibaba Cloud) TTS API",
+    ]
+    user_visible_verification = [
+        "Listen to generated audio for naturalness and pacing"
+    ]
+
+    ENDPOINT = (
+        "https://dashscope.aliyuncs.com/api/v1/services/aigc/"
+        "multimodal-generation/generation"
+    )
+
+    def get_status(self) -> ToolStatus:
+        if os.environ.get("DASHSCOPE_API_KEY"):
+            return ToolStatus.AVAILABLE
+        return ToolStatus.UNAVAILABLE
+
+    def estimate_cost(self, inputs: dict[str, Any]) -> float:
+        # Conservative per-character estimate; DashScope bills by character.
+        return round(len(inputs.get("text", "")) * 0.000015, 4)
+
+    def execute(self, inputs: dict[str, Any]) -> ToolResult:
+        api_key = os.environ.get("DASHSCOPE_API_KEY")
+        if not api_key:
+            return ToolResult(
+                success=False,
+                error="DASHSCOPE_API_KEY not set. " + self.install_instructions,
+            )
+
+        import requests
+
+        from tools.analysis.audio_probe import probe_duration
+
+        start = time.time()
+        try:
+            payload = self._build_payload(inputs)
+            response = requests.post(
+                self.ENDPOINT,
+                headers={
+                    "Authorization": f"Bearer {api_key}",
+                    "Content-Type": "application/json",
+                },
+                json=payload,
+                timeout=120,
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            audio_info = data.get("output", {}).get("audio", {})
+            audio_url = audio_info.get("url")
+            if not audio_url:
+                return ToolResult(
+                    success=False,
+                    error="DashScope TTS returned no audio URL",
+                )
+
+            # Download the audio from the temporary URL (valid ~24h).
+            download = requests.get(audio_url, timeout=120)
+            download.raise_for_status()
+
+            output_path = Path(
+                inputs.get("output_path", "dashscope_tts.wav")
+            )
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            output_path.write_bytes(download.content)
+
+            audio_duration = probe_duration(output_path)
+            usage = data.get("usage", {})
+
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                error=f"DashScope TTS failed: {self._safe_error(e)}",
+            )
+
+        return ToolResult(
+            success=True,
+            data={
+                "provider": "dashscope",
+                "model": payload["model"],
+                "voice": payload["input"]["voice"],
+                "language_type": payload["input"].get("language_type", "Auto"),
+                "text_length": len(inputs["text"]),
+                "audio_duration_seconds": (
+                    round(audio_duration, 2) if audio_duration else None
+                ),
+                "output": str(output_path),
+                "audio_url": audio_url,
+                "usage": usage,
+            },
+            artifacts=[str(output_path)],
+            cost_usd=self.estimate_cost(inputs),
+            duration_seconds=round(time.time() - start, 2),
+            model=payload["model"],
+        )
+
+    def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]:
+        input_data: dict[str, Any] = {
+            "text": inputs["text"],
+            "voice": inputs.get("voice", "Cherry"),
+            "language_type": inputs.get("language_type", "Auto"),
+        }
+        if inputs.get("instructions"):
+            input_data["instructions"] = inputs["instructions"]
+            input_data["optimize_instructions"] = True
+
+        return {
+            "model": inputs.get("model", "qwen3-tts-flash"),
+            "input": input_data,
+        }
+
+    @staticmethod
+    def _safe_error(exc: Exception) -> str:
+        return str(exc).replace(
+            os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]"
+        )

tools/graphics/dashscope_image.py

@@ -0,0 +1,273 @@
+"""DashScope (Alibaba Cloud Bailian) image generation via Qwen-Image models.
+
+Uses the DashScope-native multimodal-generation endpoint (NOT OpenAI-compatible
+mode, which only supports /chat/completions and /embeddings). The response
+contains a temporary image URL (valid ~24h) that must be downloaded separately.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+from tools.base_tool import (
+    BaseTool,
+    Determinism,
+    ExecutionMode,
+    ResourceProfile,
+    RetryPolicy,
+    ToolResult,
+    ToolRuntime,
+    ToolStability,
+    ToolStatus,
+    ToolTier,
+)
+
+
+class DashscopeImage(BaseTool):
+    name = "dashscope_image"
+    version = "0.1.0"
+    tier = ToolTier.GENERATE
+    capability = "image_generation"
+    provider = "dashscope"
+    stability = ToolStability.EXPERIMENTAL
+    execution_mode = ExecutionMode.SYNC
+    determinism = Determinism.STOCHASTIC
+    runtime = ToolRuntime.API
+
+    dependencies = []
+    install_instructions = (
+        "Set DASHSCOPE_API_KEY to your Alibaba Cloud DashScope API key.\n"
+        "  Get one at https://dashscope.aliyun.com/"
+    )
+    fallback = "grok_image"
+    fallback_tools = ["grok_image", "openai_image", "flux_image", "recraft_image"]
+    agent_skills = ["dashscope"]
+
+    capabilities = ["generate_image", "text_to_image"]
+    supports = {
+        "multiple_outputs": True,
+        "aspect_ratio": True,
+        "resolution": True,
+        "negative_prompt": True,
+        "seed": True,
+    }
+    best_for = [
+        "high-quality image generation with Qwen-Image models",
+        "Chinese-language prompt understanding",
+        "cost-effective image generation via Alibaba Cloud",
+    ]
+    not_good_for = ["offline generation", "image editing (use grok_image edit mode)"]
+
+    input_schema = {
+        "type": "object",
+        "required": ["prompt"],
+        "properties": {
+            "prompt": {"type": "string"},
+            "model": {
+                "type": "string",
+                "enum": [
+                    "qwen-image-2.0-pro",
+                    "qwen-image-max",
+                    "wan2.7-image",
+                    "z-image-turbo",
+                ],
+                "default": "qwen-image-2.0-pro",
+            },
+            "size": {
+                "type": "string",
+                "default": "1024*1024",
+                "description": (
+                    'Image size as "W*H" (asterisk separator, NOT "x"). '
+                    'Examples: "1024*1024", "2048*2048", "2688*1536".'
+                ),
+            },
+            "n": {"type": "integer", "default": 1, "minimum": 1, "maximum": 6},
+            "negative_prompt": {
+                "type": "string",
+                "description": "Negative prompt (max 500 chars). Things to avoid in the image.",
+            },
+            "prompt_extend": {
+                "type": "boolean",
+                "default": True,
+                "description": "Enable DashScope prompt auto-rewrite for better results.",
+            },
+            "watermark": {"type": "boolean", "default": False},
+            "seed": {"type": "integer", "minimum": 0, "maximum": 2147483647},
+            "output_path": {"type": "string"},
+        },
+    }
+
+    resource_profile = ResourceProfile(
+        cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=100, network_required=True
+    )
+    retry_policy = RetryPolicy(
+        max_retries=2, retryable_errors=["rate_limit", "timeout"]
+    )
+    idempotency_key_fields = [
+        "prompt",
+        "model",
+        "size",
+        "n",
+        "negative_prompt",
+        "seed",
+        "prompt_extend",
+        "watermark",
+    ]
+    side_effects = [
+        "writes image file to output_path",
+        "calls DashScope (Alibaba Cloud) image generation API",
+    ]
+    user_visible_verification = [
+        "Inspect generated image for relevance and quality"
+    ]
+
+    ENDPOINT = (
+        "https://dashscope.aliyuncs.com/api/v1/services/aigc/"
+        "multimodal-generation/generation"
+    )
+
+    def get_status(self) -> ToolStatus:
+        if os.environ.get("DASHSCOPE_API_KEY"):
+            return ToolStatus.AVAILABLE
+        return ToolStatus.UNAVAILABLE
+
+    def estimate_cost(self, inputs: dict[str, Any]) -> float:
+        # Conservative per-image estimate; DashScope bills per image.
+        # Check the DashScope console for actual pricing.
+        n = int(inputs.get("n", 1))
+        return n * 0.02
+
+    def execute(self, inputs: dict[str, Any]) -> ToolResult:
+        api_key = os.environ.get("DASHSCOPE_API_KEY")
+        if not api_key:
+            return ToolResult(
+                success=False,
+                error="DASHSCOPE_API_KEY not set. " + self.install_instructions,
+            )
+
+        import requests
+
+        start = time.time()
+        try:
+            payload = self._build_payload(inputs)
+            response = requests.post(
+                self.ENDPOINT,
+                headers={
+                    "Authorization": f"Bearer {api_key}",
+                    "Content-Type": "application/json",
+                },
+                json=payload,
+                timeout=180,
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            image_urls = self._extract_image_urls(data)
+            if not image_urls:
+                return ToolResult(
+                    success=False,
+                    error="DashScope returned no image URLs",
+                )
+
+            # DashScope bills per image and URLs expire ~24h; save every one.
+            output_paths = self._resolve_output_paths(
+                inputs.get("output_path", "dashscope_image.png"),
+                count=len(image_urls),
+            )
+            for path, url in zip(output_paths, image_urls):
+                download = requests.get(url, timeout=120)
+                download.raise_for_status()
+                path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_bytes(download.content)
+
+            usage = data.get("usage", {})
+            n_generated = len(image_urls)
+
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                error=f"DashScope image generation failed: {self._safe_error(e)}",
+            )
+
+        return ToolResult(
+            success=True,
+            data={
+                "provider": "dashscope",
+                "model": payload["model"],
+                "prompt": inputs["prompt"],
+                "size": payload["parameters"]["size"],
+                "output": str(output_paths[0]),
+                "outputs": [str(p) for p in output_paths],
+                "images_generated": n_generated,
+                "usage": usage,
+            },
+            artifacts=[str(p) for p in output_paths],
+            cost_usd=self.estimate_cost(inputs),
+            duration_seconds=round(time.time() - start, 2),
+            model=payload["model"],
+        )
+
+    @staticmethod
+    def _extract_image_urls(data: dict[str, Any]) -> list[str]:
+        """Collect image URLs from every choice whose finish_reason is "stop".
+
+        Per Qwen Cloud docs, a multi-output task is SUCCEEDED if at least one
+        image is generated; failed choices carry finish_reason != "stop" and
+        must be skipped to avoid downloading partial/empty results.
+        """
+        urls: list[str] = []
+        for choice in data.get("output", {}).get("choices", []):
+            if choice.get("finish_reason") != "stop":
+                continue
+            for item in choice.get("message", {}).get("content", []):
+                url = item.get("image")
+                if url:
+                    urls.append(url)
+        return urls
+
+    @staticmethod
+    def _resolve_output_paths(base: str, count: int) -> list[Path]:
+        """Derive distinct paths for `count` images. Single image keeps the
+        base path unchanged; multiple images insert an index before the
+        extension (foo.png -> foo_1.png, foo_2.png, ...)."""
+        base_path = Path(base)
+        if count <= 1:
+            return [base_path]
+        stem = base_path.stem
+        suffix = base_path.suffix
+        parent = base_path.parent
+        return [parent / f"{stem}_{i}{suffix}" for i in range(1, count + 1)]
+
+    def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]:
+        parameters: dict[str, Any] = {
+            "size": inputs.get("size", "1024*1024"),
+            "n": int(inputs.get("n", 1)),
+            "prompt_extend": bool(inputs.get("prompt_extend", True)),
+            "watermark": bool(inputs.get("watermark", False)),
+        }
+        if inputs.get("negative_prompt"):
+            parameters["negative_prompt"] = inputs["negative_prompt"]
+        if inputs.get("seed") is not None:
+            parameters["seed"] = int(inputs["seed"])
+
+        return {
+            "model": inputs.get("model", "qwen-image-2.0-pro"),
+            "input": {
+                "messages": [
+                    {
+                        "role": "user",
+                        "content": [{"text": inputs["prompt"]}],
+                    }
+                ]
+            },
+            "parameters": parameters,
+        }
+
+    @staticmethod
+    def _safe_error(exc: Exception) -> str:
+        return str(exc).replace(
+            os.environ.get("DASHSCOPE_API_KEY", ""), "[redacted]"
+        )