dashscope: fix multi-image download and complete idempotency keys

Address PR #240 review feedback from @calesthio:

1. dashscope_image: save EVERY returned image URL, not just the first.
   The tool advertised multiple_outputs and accepted n>1 but only read
   content[0], silently dropping paid outputs. Now collects all image
   URLs across choices/content and downloads each to a distinct indexed
   path (foo.png -> foo_1.png, foo_2.png, ...). images_generated now
   reflects the actual count downloaded.

   Per Qwen Cloud docs, a multi-output task is SUCCEEDED if at least one
   image is generated; choices with finish_reason != "stop" are skipped
   to avoid downloading partial/failed results.

2. Complete idempotency_key_fields so different requests no longer
   collide and reuse stale artifacts:
   - image: + negative_prompt, seed, prompt_extend, watermark
   - tts:   + instructions
   - asr:   + enable_words, language_hints

Adds 19 regression tests (114 total, all pass, no API keys needed):
- TestDashscopeImageMultiOutput: URL extraction across choices / within
  one choice / failed-choice skipping, path resolution for
  single/multi/no-extension, end-to-end multi-image download with a
  mocked 3-URL DashScope response verifying all 3 files land on disk,
  single-image legacy path behavior
- TestDashscopeIdempotencyKeys: field presence + key-differs-on-value
  for every newly added field across all three tools

tests/contracts/test_dashscope_tools.py

@@ -250,6 +250,266 @@ class TestDashscopeImageSpecific:
         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
 # ------------------------------------------------------------------

tools/analysis/dashscope_asr.py

@@ -129,7 +129,7 @@ class DashscopeAsr(BaseTool):
         backoff_seconds=2.0,
         retryable_errors=["timeout", "rate_limit"],
     )
-    idempotency_key_fields = ["audio_url", "model"]
+    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)",

tools/audio/dashscope_tts.py

@@ -123,7 +123,7 @@ class DashscopeTTS(BaseTool):
     retry_policy = RetryPolicy(
         max_retries=2, retryable_errors=["rate_limit", "timeout"]
     )
-    idempotency_key_fields = ["text", "voice", "model", "language_type"]
+    idempotency_key_fields = ["text", "voice", "model", "language_type", "instructions"]
     side_effects = [
         "writes audio file to output_path",
         "calls DashScope (Alibaba Cloud) TTS API",

tools/graphics/dashscope_image.py

@@ -106,7 +106,16 @@ class DashscopeImage(BaseTool):
     retry_policy = RetryPolicy(
         max_retries=2, retryable_errors=["rate_limit", "timeout"]
     )
-    idempotency_key_fields = ["prompt", "model", "size", "n"]
+    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",
@@ -156,39 +165,26 @@ class DashscopeImage(BaseTool):
             response.raise_for_status()
             data = response.json()
 
-            choices = data.get("output", {}).get("choices", [])
-            if not choices:
+            image_urls = self._extract_image_urls(data)
+            if not image_urls:
                 return ToolResult(
                     success=False,
-                    error="DashScope returned no image choices",
+                    error="DashScope returned no image URLs",
                 )
 
-            content = choices[0].get("message", {}).get("content", [])
-            if not content:
-                return ToolResult(
-                    success=False,
-                    error="DashScope returned no image content",
-                )
-
-            image_url = content[0].get("image")
-            if not image_url:
-                return ToolResult(
-                    success=False,
-                    error="DashScope image output missing URL",
-                )
-
-            # Download the image from the temporary URL (valid ~24h).
-            download = requests.get(image_url, timeout=120)
-            download.raise_for_status()
-
-            output_path = Path(
-                inputs.get("output_path", "dashscope_image.png")
+            # 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),
             )
-            output_path.parent.mkdir(parents=True, exist_ok=True)
-            output_path.write_bytes(download.content)
+            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 = int(usage.get("image_count", 1))
+            n_generated = len(image_urls)
 
         except Exception as e:
             return ToolResult(
@@ -203,17 +199,48 @@ class DashscopeImage(BaseTool):
                 "model": payload["model"],
                 "prompt": inputs["prompt"],
                 "size": payload["parameters"]["size"],
-                "output": str(output_path),
-                "outputs": [str(output_path)],
+                "output": str(output_paths[0]),
+                "outputs": [str(p) for p in output_paths],
                 "images_generated": n_generated,
                 "usage": usage,
             },
-            artifacts=[str(output_path)],
+            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"),