Merge pull request #448 from clarkhjc/feat/hunyuan_cloud_video

feat: add Tencent Hunyuan cloud video provider via TokenHub API

.env.example

@@ -68,8 +68,10 @@ DOUBAO_SPEECH_VOICE_TYPE=
 # Get one at https://dashscope.aliyun.com/
 DASHSCOPE_API_KEY=
 
-# --- Tencent Hunyuan TokenHub ---
-# Hunyuan Image 3.0 via Tencent TokenHub.
+# --- Tencent Hunyuan TokenHub API ---
+# Tencent Hunyuan (腾讯混元) image and cloud video generation via TokenHub
+# (Bearer token).
+# Get it at https://console.cloud.tencent.com/tokenhub.
 TENCENT_TOKENHUB_API_KEY=
 
 # --- Music ---

docs/PROVIDERS.md

@@ -20,10 +20,11 @@ Everything you need to know about every provider in OpenMontage — setup instru
 | 8 | **pay-as-you-go** | Kling Official | Official direct Kling video, image, TTS, avatar, and lip-sync API, separate from fal.ai Kling |
 | 9 | **pay-as-you-go** | Volcengine Ark | Official direct Seedance 2.0 Standard/Fast/Mini API |
 | 10 | **$12/month** | Runway | Gen-4 video — highest quality AI video |
-| 11 | **pay-as-you-go** | HeyGen | Avatar videos, multi-model video gateway |
-| 12 | **pay-as-you-go** | Suno | Full song generation with vocals and lyrics |
-| 13 | **$0 + GPU** | Local video gen | WAN 2.1, Hunyuan, CogVideo, LTX — free, offline |
-| 14 | **$0 + GPU** | Local Diffusion | Stable Diffusion images — free, offline |
+| 11 | **pay-as-you-go** | Hunyuan cloud video | Chinese-friendly T2V + I2V |
+| 12 | **pay-as-you-go** | HeyGen | Avatar videos, multi-model video gateway |
+| 13 | **pay-as-you-go** | Suno | Full song generation with vocals and lyrics |
+| 14 | **$0 + GPU** | Local video gen | WAN 2.1, Hunyuan, CogVideo, LTX — free, offline |
+| 15 | **$0 + GPU** | Local Diffusion | Stable Diffusion images — free, offline |
 
 ### Environment Variable Summary
 
@@ -65,6 +66,9 @@ HEYGEN_API_KEY=              # HeyGen avatar video gateway
 RUNWAY_API_KEY=              # Runway Gen-4 video (direct)
 SUNO_API_KEY=                # Suno music generation
 
+# TENCLOUD HUNYUAN VIDEO
+TENCENT_TOKENHUB_API_KEY=    # Tencent Hunyuan cloud video via TokenHub API
+
 # LOCAL (no keys needed — just GPU + install)
 VIDEO_GEN_LOCAL_ENABLED=     # Set to "true" for local video gen
 VIDEO_GEN_LOCAL_MODEL=       # wan2.1-1.3b, wan2.1-14b, hunyuan-1.5, ltx2-local, cogvideo-5b
@@ -464,6 +468,79 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon
 
 ---
 
+### Tencent Hunyuan Cloud — Video Generation
+
+> **Tencent Hunyuan (腾讯混元) cloud video generation via TokenHub API.** Generates
+> videos from text or images using Tencent's Hunyuan models through the Tencent
+> TokenHub API — an OpenAI-compatible gateway (tokenhub.tencentmaas.com) with
+> simple Bearer-token authentication. No TC3-HMAC-SHA256 signing required.
+
+**Tools unlocked:** `hunyuan_cloud_video`
+**Env var:** `TENCENT_TOKENHUB_API_KEY`
+
+#### Setup
+
+1. Go to the [Tencent Cloud TokenHub console](https://console.cloud.tencent.com/tokenhub).
+2. Create an application or navigate to the **API Key** section.
+3. Generate an API key and copy its value.
+4. Add to `.env`:
+   ```bash
+   TENCENT_TOKENHUB_API_KEY=your-tokenhub-api-key
+   ```
+
+#### What It's Best For
+
+- **Chinese-friendly prompt understanding** — Hunyuan models natively understand Chinese prompts better than most Western APIs
+- **Simple auth** — Bearer token, no complex signing (just an HTTP Authorization header)
+- **Direct Tencent Cloud quota** — uses your own Tencent Cloud credits, not a third-party gateway mark-up
+- **Both T2V and I2V** — one API key unlocks text-to-video and image-to-video
+
+#### API Notes
+
+TokenHub uses a **submit-then-poll** pattern:
+
+```text
+# Submit a generation task
+POST https://tokenhub.tencentmaas.com/v1/api/video/submit
+Authorization: Bearer ${TENCENT_TOKENHUB_API_KEY}
+
+# Poll for results
+POST https://tokenhub.tencentmaas.com/v1/api/video/query
+Authorization: Bearer ${TENCENT_TOKENHUB_API_KEY}
+```
+
+| Model | Type | Pricing |
+|-------|------|---------|
+| `hy-video-1.5` | Text-to-video | 1.5 credits (~$0.25) |
+| `yt-video-2.0` | Image-to-video | 2–5 credits (~$0.33–0.83) |
+
+Resolution options: **720p** (default) or **1080p**.
+
+A watermark (`logo_add`) is added by default. Set `logo_add: 0` to disable it (requires console approval from Tencent).
+
+**Schema constraints:**
+- **Prompt:** max 200 UTF-8 characters
+- **Image:** max 10MB, 50–5000 px per side, aspect ratio 1:4 to 4:1
+- **Formats:** jpg, png, jpeg, webp, bmp, tiff
+
+#### Fallback Tools
+
+If `hunyuan_cloud_video` returns an error, the agent may retry with: `jimeng_video`, `kling_official_video`, `minimax_video`
+
+#### Pricing
+
+Tencent TokenHub uses a credit-based pricing system (1 credit = 1.2 RMB ≈ $0.167 USD):
+
+| Model | Resolution | Credits | Estimated USD |
+|-------|-----------|---------|---------------|
+| HY-Video-1.5 | any | 1.5 | ~$0.25 |
+| YT-Video-2.0 | 480p | 2 | ~$0.33 |
+| YT-Video-2.0 | 720p / 1080p | 5 | ~$0.83 |
+
+> **Free tier:** Tencent occasionally offers new-user credits for TokenHub. Check the [TokenHub console](https://console.cloud.tencent.com/tokenhub) for current promotions.
+
+---
+
 ### Azure AI Speech — Speech-to-Text
 
 > **Cloud transcription.** Azure AI Speech Fast Transcription turns local audio into text with word-level timestamps, speaker diarization, and multi-language identification — no GPU required. Optional: the local faster-whisper `transcriber` remains the default offline STT path. When `AZURE_SPEECH_KEY` is set, the agent prefers `azure_stt` for cloud transcription.
@@ -1064,6 +1141,7 @@ These tools require only FFmpeg or Python packages — no GPU, no API key.
 | **Higgsfield** | `HIGGSFIELD_API_KEY` + `HIGGSFIELD_API_SECRET` | `higgsfield_video` | Subscription ($15-84/mo) |
 | **HeyGen** | `HEYGEN_API_KEY` | `heygen_video` | Pay-as-you-go |
 | **Suno** | `SUNO_API_KEY` | `suno_music` | Pay-as-you-go |
+| **Tencent Hunyuan** | `TENCENT_TOKENHUB_API_KEY` | `hunyuan_cloud_video` | Pay-as-you-go (~$0.25–0.83/gen) |
 | **Local GPU** | `VIDEO_GEN_LOCAL_ENABLED` | `wan_video`, `hunyuan_video`, `cogvideo_video`, `ltx_video_local` | Free (GPU required) |
 | **Local Diffusion** | — (install only) | `local_diffusion` | Free (GPU required) |
 | **Modal** | `MODAL_LTX2_ENDPOINT_URL` | `ltx_video_modal` | Self-hosted cloud |
@@ -1077,7 +1155,7 @@ How many providers cover each capability:
 | Capability | Cloud Providers | Local Providers | Free Options |
 |-----------|----------------|-----------------|--------------|
 | **Image Generation** | FLUX, Kling Official, Grok, Google Imagen, GPT Image 2, Recraft | Local Diffusion | Pexels, Pixabay (stock) |
-| **Video Generation** | Grok, Kling Official, Kling via fal.ai, Seedance via Volcengine Ark, Runway, Veo, Gemini Omni, Higgsfield, MiniMax, HeyGen | WAN, Hunyuan, CogVideo, LTX | Pexels, Pixabay (stock) |
+| **Video Generation** | Grok, Kling Official, Kling via fal.ai, Seedance via Volcengine Ark, Runway, Veo, Gemini Omni, Higgsfield, MiniMax, HeyGen, Tencent Hunyuan | WAN, Hunyuan, CogVideo, LTX | Pexels, Pixabay (stock) |
 | **Text-to-Speech** | ElevenLabs, Google TTS, Kling Official, OpenAI | Piper | Piper, Google free tier, ElevenLabs free tier |
 | **Music Generation** | ElevenLabs, Suno, Google Lyria | — | ElevenLabs free tier |
 | **Post-Production** | — | FFmpeg (compose, stitch, trim, mix, enhance, grade) | All free |

tests/contracts/test_hunyuan_cloud_video.py

@@ -0,0 +1,843 @@
+"""Contract tests for the Tencent Hunyuan cloud video provider tool (TokenHub API).
+
+These tests verify that the tool satisfies the BaseTool contract without
+requiring real Tencent Cloud credentials or making any API calls.
+
+Run: pytest tests/contracts/test_hunyuan_cloud_video.py -v
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+from tools.base_tool import (
+    BaseTool,
+    ExecutionMode,
+    ToolRuntime,
+    ToolStability,
+    ToolStatus,
+    ToolTier,
+)
+from tools.video.hunyuan_cloud_video import HunyuanCloudVideo
+
+
+# ------------------------------------------------------------------
+# Fake HTTP infrastructure (used by execute-path tests)
+# ------------------------------------------------------------------
+
+class FakeResponse:
+    def __init__(self, json_data=None, content=b"", ok=True, status_code=200, headers=None, text=""):
+        self._json = json_data
+        self.content = content
+        self.ok = ok
+        self.status_code = status_code
+        self.headers = headers or {}
+        self.text = text or (json.dumps(json_data) if json_data is not None else "")
+
+    def json(self):
+        return self._json
+
+    def raise_for_status(self):
+        if not self.ok:
+            raise RuntimeError(f"HTTP {self.status_code}")
+
+
+def _install_fake_requests(monkeypatch, post_responses, get_responses):
+    """Inject a fake requests module; returns the recorded calls."""
+    calls = {"post": [], "get": []}
+    fake = types.ModuleType("requests")
+
+    def fake_post(url, headers=None, json=None, data=None, timeout=None, params=None):
+        calls["post"].append({"url": url, "headers": headers, "json": json, "data": data})
+        return post_responses.pop(0)
+
+    def fake_get(url, headers=None, timeout=None, params=None):
+        calls["get"].append({"url": url, "headers": headers, "params": params})
+        return get_responses.pop(0)
+
+    fake.post = fake_post
+    fake.get = fake_get
+    monkeypatch.setitem(sys.modules, "requests", fake)
+    return calls
+
+
+# ------------------------------------------------------------------
+# Fixtures
+# ------------------------------------------------------------------
+
+@pytest.fixture()
+def hunyuan_env(monkeypatch):
+    """Set fake TokenHub API credentials."""
+    monkeypatch.setenv("TENCENT_TOKENHUB_API_KEY", "thub-fake-test-key")
+
+
+@pytest.fixture()
+def no_hunyuan_env(monkeypatch):
+    """Ensure no TokenHub credentials are set."""
+    monkeypatch.delenv("TENCENT_TOKENHUB_API_KEY", raising=False)
+
+
+# ------------------------------------------------------------------
+# Contract compliance
+# ------------------------------------------------------------------
+
+class TestContract:
+
+    def test_inherits_base_tool(self):
+        assert issubclass(HunyuanCloudVideo, BaseTool)
+
+    def test_has_required_identity(self):
+        tool = HunyuanCloudVideo()
+        assert tool.name == "hunyuan_cloud_video"
+        assert tool.version == "0.1.0"
+        assert tool.provider == "hunyuan_cloud"
+        assert tool.capability == "video_generation"
+        assert tool.tier == ToolTier.GENERATE
+        assert tool.stability == ToolStability.EXPERIMENTAL
+        assert tool.runtime == ToolRuntime.API
+
+    def test_execution_mode_is_async(self):
+        assert HunyuanCloudVideo().execution_mode == ExecutionMode.ASYNC
+
+    def test_has_input_schema(self):
+        schema = HunyuanCloudVideo().input_schema
+        assert schema.get("type") == "object"
+        props = schema.get("properties", {})
+        required = schema.get("required", [])
+        assert required == ["prompt"]
+        for field in required:
+            assert field in props
+
+    def test_has_capabilities(self):
+        tool = HunyuanCloudVideo()
+        assert "text_to_video" in tool.capabilities
+        assert "image_to_video" in tool.capabilities
+
+    def test_has_agent_skills(self):
+        assert "ai-video-gen" in HunyuanCloudVideo().agent_skills
+
+    def test_has_fallbacks(self):
+        tool = HunyuanCloudVideo()
+        assert "jimeng_video" in tool.fallback_tools
+        assert "kling_official_video" in tool.fallback_tools
+        assert "minimax_video" in tool.fallback_tools
+
+    def test_has_install_instructions(self):
+        tool = HunyuanCloudVideo()
+        assert "TENCENT_TOKENHUB_API_KEY" in tool.install_instructions
+
+    def test_get_info_returns_dict(self):
+        info = HunyuanCloudVideo().get_info()
+        assert isinstance(info, dict)
+        assert info["name"] == "hunyuan_cloud_video"
+        assert info["provider"] == "hunyuan_cloud"
+        assert info["runtime"] == "api"
+        assert info["capability"] == "video_generation"
+
+    def test_status_unavailable_without_keys(self, no_hunyuan_env):
+        assert HunyuanCloudVideo().get_status() == ToolStatus.UNAVAILABLE
+
+    def test_status_available_with_keys(self, hunyuan_env):
+        assert HunyuanCloudVideo().get_status() == ToolStatus.AVAILABLE
+
+    def test_has_resource_profile(self):
+        rp = HunyuanCloudVideo().resource_profile
+        assert rp.network_required is True
+        assert rp.vram_mb == 0
+
+    def test_has_retry_policy(self):
+        assert HunyuanCloudVideo().retry_policy.max_retries >= 0
+
+    def test_has_side_effects(self):
+        side = HunyuanCloudVideo().side_effects
+        assert len(side) > 0
+        assert any("API" in s for s in side) or any("TokenHub" in s for s in side)
+
+    def test_has_user_visible_verification(self):
+        assert len(HunyuanCloudVideo().user_visible_verification) > 0
+
+    def test_estimate_cost_returns_float(self):
+        cost = HunyuanCloudVideo().estimate_cost({"prompt": "x"})
+        assert isinstance(cost, float)
+        assert cost > 0.0
+
+    def test_dry_run_returns_dict(self):
+        result = HunyuanCloudVideo().dry_run({"prompt": "test"})
+        assert isinstance(result, dict)
+        assert result["tool"] == "hunyuan_cloud_video"
+
+
+# ------------------------------------------------------------------
+# Supports flags
+# ------------------------------------------------------------------
+
+class TestSupports:
+
+    def test_text_to_video_supported(self):
+        assert HunyuanCloudVideo().supports["text_to_video"] is True
+
+    def test_image_to_video_supported(self):
+        assert HunyuanCloudVideo().supports["image_to_video"] is True
+
+    def test_native_audio_not_supported(self):
+        assert HunyuanCloudVideo().supports["native_audio"] is False
+
+    def test_seed_not_supported(self):
+        assert HunyuanCloudVideo().supports["seed"] is False
+
+
+# ------------------------------------------------------------------
+# Idempotency keys
+# ------------------------------------------------------------------
+
+class TestIdempotencyKeys:
+
+    def test_includes_all_output_affecting_fields(self):
+        fields = HunyuanCloudVideo().idempotency_key_fields
+        for field in (
+            "prompt", "operation", "model", "image_url", "image_path",
+            "resolution", "logo_add",
+        ):
+            assert field in fields, f"missing idempotency field: {field}"
+
+    def test_excludes_execution_only_fields(self):
+        fields = HunyuanCloudVideo().idempotency_key_fields
+        for field in ("output_path", "poll_interval_seconds", "timeout_seconds"):
+            assert field not in fields
+
+    def test_differs_on_operation(self):
+        tool = HunyuanCloudVideo()
+        base = {"prompt": "x"}
+        assert tool.idempotency_key(base) != tool.idempotency_key(
+            {**base, "operation": "image_to_video"}
+        )
+
+    def test_differs_on_model(self):
+        tool = HunyuanCloudVideo()
+        base = {"prompt": "x"}
+        assert tool.idempotency_key(base) != tool.idempotency_key(
+            {**base, "model": "yt-video-2.0"}
+        )
+
+    def test_differs_on_image_url(self):
+        tool = HunyuanCloudVideo()
+        base = {"prompt": "x", "operation": "image_to_video"}
+        assert tool.idempotency_key(base) != tool.idempotency_key(
+            {**base, "image_url": "https://example.com/img.png"}
+        )
+
+    def test_ignores_execution_params(self):
+        tool = HunyuanCloudVideo()
+        base = {"prompt": "x"}
+        assert tool.idempotency_key(base) == tool.idempotency_key(
+            {**base, "output_path": "/tmp/out.mp4", "timeout_seconds": 999}
+        )
+
+
+# ------------------------------------------------------------------
+# Tool-specific behavior
+# ------------------------------------------------------------------
+
+class TestToolSpecific:
+
+    def test_default_operation_is_text_to_video(self):
+        tool = HunyuanCloudVideo()
+        assert tool.input_schema["properties"]["operation"]["default"] == "text_to_video"
+
+    def test_default_resolution_is_720p(self):
+        tool = HunyuanCloudVideo()
+        assert tool.input_schema["properties"]["resolution"]["default"] == "720p"
+
+    def test_default_logo_add_is_1(self):
+        tool = HunyuanCloudVideo()
+        assert tool.input_schema["properties"]["logo_add"]["default"] == 1
+
+    def test_default_poll_interval_is_5(self):
+        tool = HunyuanCloudVideo()
+        assert tool.input_schema["properties"]["poll_interval_seconds"]["default"] == 5.0
+
+    def test_default_timeout_is_600(self):
+        tool = HunyuanCloudVideo()
+        assert tool.input_schema["properties"]["timeout_seconds"]["default"] == 600
+
+    # -- _resolve_model --
+
+    def test_resolve_model_defaults_t2v(self):
+        model = HunyuanCloudVideo._resolve_model({"prompt": "test"})
+        assert model == "hy-video-1.5"
+
+    def test_resolve_model_defaults_i2v(self):
+        model = HunyuanCloudVideo._resolve_model({
+            "prompt": "test", "operation": "image_to_video",
+        })
+        assert model == "yt-video-2.0"
+
+    def test_resolve_model_explicit_input(self):
+        model = HunyuanCloudVideo._resolve_model({
+            "prompt": "test", "model": "yt-video-2.0",
+        })
+        assert model == "yt-video-2.0"
+
+    # -- _build_payload --
+
+    def test_build_payload_t2v_minimal(self):
+        """Minimal T2V payload — only prompt is required by TokenHub."""
+        tool = HunyuanCloudVideo()
+        payload = tool._build_payload({"prompt": "一只猫"})
+        assert payload["prompt"] == "一只猫"
+        # resolution and logo_add are optional in TokenHub; only included when
+        # explicitly passed in inputs
+        assert "resolution" not in payload
+        assert "logo_add" not in payload
+
+    def test_build_payload_t2v_with_resolution_and_logo(self):
+        tool = HunyuanCloudVideo()
+        payload = tool._build_payload({
+            "prompt": "test", "resolution": "720p", "logo_add": 0,
+        })
+        assert payload["prompt"] == "test"
+        assert payload["resolution"] == "720p"
+        assert payload["logo_add"] == 0
+
+    def test_build_payload_t2v_custom_logo_add(self):
+        tool = HunyuanCloudVideo()
+        payload = tool._build_payload({"prompt": "test", "logo_add": 0})
+        assert payload["logo_add"] == 0
+
+    def test_build_payload_i2v_includes_image_url(self):
+        tool = HunyuanCloudVideo()
+        payload = tool._build_payload({
+            "prompt": "motion",
+            "operation": "image_to_video",
+            "image_url": "https://example.com/frame.png",
+        })
+        assert payload["image_url"] == "https://example.com/frame.png"
+
+    def test_build_payload_t2v_omits_image(self):
+        tool = HunyuanCloudVideo()
+        payload = tool._build_payload({"prompt": "a cat", "operation": "text_to_video"})
+        assert "image_url" not in payload
+        assert "image" not in payload
+
+    def test_build_payload_i2v_base64_from_path(self, tmp_path):
+        """image_path should be base64-encoded into the image field."""
+        img = tmp_path / "frame.jpg"
+        img.write_bytes(b"\xff\xd8\xff\xe0test-jpeg-data")
+        tool = HunyuanCloudVideo()
+        payload = tool._build_payload({
+            "prompt": "test",
+            "operation": "image_to_video",
+            "image_path": str(img),
+        })
+        assert payload["image"].startswith("/9j/")
+
+    def test_encode_image_returns_base64(self, tmp_path):
+        img = tmp_path / "ref.jpg"
+        img.write_bytes(b"\xff\xd8\xff\xe0\x00\x10JFIF")
+        encoded = HunyuanCloudVideo._encode_image(str(img))
+        assert isinstance(encoded, str)
+        assert encoded.startswith("/9j/")
+
+    def test_encode_image_raises_on_missing(self):
+        with pytest.raises(FileNotFoundError):
+            HunyuanCloudVideo._encode_image("/nonexistent/file.jpg")
+
+    # -- Error paths --
+
+    def test_no_keys_returns_error(self, no_hunyuan_env):
+        result = HunyuanCloudVideo().execute({"prompt": "test"})
+        assert result.success is False
+        assert "TENCENT_TOKENHUB_API_KEY" in result.error
+
+    def test_i2v_without_image_fails(self, hunyuan_env):
+        result = HunyuanCloudVideo().execute(
+            {"prompt": "test", "operation": "image_to_video"}
+        )
+        assert result.success is False
+        assert "image_url" in result.error or "image_path" in result.error
+
+    @pytest.mark.parametrize(
+        ("operation", "model", "expected"),
+        [
+            ("text_to_video", "yt-video-2.0", "hy-video-1.5"),
+            ("image_to_video", "hy-video-1.5", "yt-video-2.0"),
+        ],
+    )
+    def test_incompatible_model_operation_fails_before_submit(
+        self, hunyuan_env, monkeypatch, operation, model, expected
+    ):
+        monkeypatch.setattr(
+            HunyuanCloudVideo,
+            "_generate",
+            lambda *args, **kwargs: pytest.fail("must not submit a paid task"),
+        )
+        inputs = {"prompt": "test", "operation": operation, "model": model}
+        if operation == "image_to_video":
+            inputs["image_url"] = "https://example.com/frame.png"
+        result = HunyuanCloudVideo().execute(inputs)
+        assert not result.success
+        assert expected in (result.error or "")
+
+    def test_i2v_both_url_and_path_fails(self, hunyuan_env, tmp_path):
+        img = tmp_path / "ref.jpg"
+        img.write_bytes(b"fake-jpeg")
+        result = HunyuanCloudVideo().execute({
+            "prompt": "test",
+            "operation": "image_to_video",
+            "image_url": "https://example.com/img.jpg",
+            "image_path": str(img),
+        })
+        assert result.success is False
+        assert "not both" in result.error.lower()
+
+    def test_safe_error_redacts_keys(self, monkeypatch):
+        monkeypatch.setenv("TENCENT_TOKENHUB_API_KEY", "thub-secret-key")
+        redacted = HunyuanCloudVideo._safe_error(
+            Exception("failed with thub-secret-key in message")
+        )
+        assert "thub-secret-key" not in redacted
+        assert "[redacted]" in redacted
+
+    def test_safe_error_no_empty_string_bug(self, no_hunyuan_env):
+        msg = HunyuanCloudVideo._safe_error(Exception("abc"))
+        assert msg == "abc"
+
+
+# ------------------------------------------------------------------
+# TokenHub auth headers
+# ------------------------------------------------------------------
+
+class TestAuthHeaders:
+
+    def test_auth_headers_includes_bearer(self):
+        headers = HunyuanCloudVideo._auth_headers("test-api-key")
+        assert headers["Authorization"] == "Bearer test-api-key"
+        assert headers["Content-Type"] == "application/json"
+
+
+# ------------------------------------------------------------------
+# Error handling helpers
+# ------------------------------------------------------------------
+
+class TestErrorHelpers:
+
+    def test_json_or_raise_returns_dict(self):
+        class FakeResp:
+            status_code = 200
+            def json(self):
+                return {"id": "task-123", "status": "queued"}
+        result = HunyuanCloudVideo._json_or_raise(FakeResp())
+        assert result == {"id": "task-123", "status": "queued"}
+
+    def test_json_or_raise_raises_on_non_json(self):
+        class FakeResp:
+            status_code = 500
+            def json(self):
+                raise ValueError("not JSON")
+        with pytest.raises(RuntimeError, match="Non-JSON"):
+            HunyuanCloudVideo._json_or_raise(FakeResp())
+
+    def test_check_response_passes_on_success(self):
+        HunyuanCloudVideo._check_response(
+            {"id": "task-123", "status": "completed"}
+        )
+
+    def test_check_response_passes_without_error_field(self):
+        HunyuanCloudVideo._check_response(
+            {"id": "task-456", "status": "running", "progress": 50}
+        )
+
+    def test_check_response_raises_on_api_error(self):
+        with pytest.raises(RuntimeError, match="Prompt too long"):
+            HunyuanCloudVideo._check_response({
+                "error": {"code": "invalid_parameter", "message": "Prompt too long"},
+            })
+
+    def test_check_response_raises_on_auth_failure(self):
+        with pytest.raises(RuntimeError, match="Invalid API key"):
+            HunyuanCloudVideo._check_response({
+                "error": {
+                    "type": "authentication_error",
+                    "message": "Invalid API key",
+                },
+            })
+
+
+# ------------------------------------------------------------------
+# Execute with mocked HTTP
+# ------------------------------------------------------------------
+
+class TestExecuteWithMocks:
+
+    def test_text_to_video_success(self, hunyuan_env, tmp_path, monkeypatch):
+        """Full T2V flow: submit -> poll -> download -> write output."""
+        task_id = "143-test-task-12345"
+        calls = _install_fake_requests(
+            monkeypatch,
+            post_responses=[
+                # Submit response
+                FakeResponse({
+                    "id": task_id,
+                    "request_id": "req-sub-001",
+                    "object": "video",
+                    "created_at": 1700000000,
+                    "status": "queued",
+                }),
+                # Poll response (completed)
+                FakeResponse({
+                    "request_id": "req-poll-001",
+                    "object": "video",
+                    "created_at": 1700000000,
+                    "completed_at": 1700000120,
+                    "status": "completed",
+                    "progress": 100,
+                    "data": {"url": "https://example.com/output.mp4"},
+                }),
+            ],
+            get_responses=[
+                # Download response
+                FakeResponse(content=b"fake-hunyuan-mp4-data"),
+            ],
+        )
+
+        output_path = tmp_path / "hunyuan_out.mp4"
+        result = HunyuanCloudVideo().execute({
+            "prompt": "一只猫在草原上奔跑",
+            "output_path": str(output_path),
+        })
+
+        assert result.success, result.error
+        assert output_path.read_bytes() == b"fake-hunyuan-mp4-data"
+        assert result.data["provider"] == "hunyuan_cloud"
+        assert result.data["route"] == "tokenhub"
+        assert result.data["model"] == "hy-video-1.5"
+        assert result.data["task_id"] == task_id
+        assert result.data["operation"] == "text_to_video"
+        assert result.cost_usd == pytest.approx(0.25)
+        assert len(result.artifacts) == 1
+        assert result.artifacts[0] == str(output_path)
+
+        # Verify submit was called to the correct TokenHub endpoint
+        submit_call = calls["post"][0]
+        assert "tokenhub.tencentmaas.com" in submit_call["url"]
+        assert submit_call["url"].endswith("/v1/api/video/submit")
+        assert submit_call["headers"]["Authorization"] == "Bearer thub-fake-test-key"
+        assert submit_call["json"]["model"] == "hy-video-1.5"
+        assert submit_call["json"]["prompt"] == "一只猫在草原上奔跑"
+
+        # Verify poll was called
+        poll_call = calls["post"][1]
+        assert poll_call["url"].endswith("/v1/api/video/query")
+        assert poll_call["json"]["model"] == "hy-video-1.5"
+        assert poll_call["json"]["id"] == task_id
+
+        # Verify download was called
+        assert len(calls["get"]) == 1
+        assert calls["get"][0]["url"] == "https://example.com/output.mp4"
+
+    def test_image_to_video_with_url_success(self, hunyuan_env, tmp_path, monkeypatch):
+        """Full I2V flow with an image URL."""
+        task_id = "i2v-task-999"
+        _install_fake_requests(
+            monkeypatch,
+            post_responses=[
+                FakeResponse({
+                    "id": task_id, "request_id": "req-sub", "object": "video",
+                    "created_at": 1700000000, "status": "queued",
+                }),
+                FakeResponse({
+                    "request_id": "req-poll", "object": "video",
+                    "created_at": 1700000000, "completed_at": 1700000120,
+                    "status": "completed", "progress": 100,
+                    "data": {"url": "https://example.com/i2v_out.mp4"},
+                }),
+            ],
+            get_responses=[
+                FakeResponse(content=b"fake-i2v-video"),
+            ],
+        )
+
+        output_path = tmp_path / "i2v_out.mp4"
+        result = HunyuanCloudVideo().execute({
+            "prompt": "让画面动起来",
+            "operation": "image_to_video",
+            "image_url": "https://example.com/frame.jpg",
+            "output_path": str(output_path),
+        })
+
+        assert result.success, result.error
+        assert output_path.read_bytes() == b"fake-i2v-video"
+        assert result.data["operation"] == "image_to_video"
+        assert result.data["model"] == "yt-video-2.0"
+        assert result.data["task_id"] == task_id
+
+    def test_i2v_with_local_image_path(self, hunyuan_env, tmp_path, monkeypatch):
+        """Full I2V flow with a local image path -> base64 encoding."""
+        img = tmp_path / "frame.jpg"
+        img.write_bytes(b"\xff\xd8\xff\xe0test-jpeg-image-data")
+
+        task_id = "i2v-local-task"
+        _install_fake_requests(
+            monkeypatch,
+            post_responses=[
+                FakeResponse({
+                    "id": task_id, "request_id": "req-sub", "object": "video",
+                    "created_at": 1700000000, "status": "queued",
+                }),
+                FakeResponse({
+                    "request_id": "req-poll", "object": "video",
+                    "created_at": 1700000000, "completed_at": 1700000120,
+                    "status": "completed", "progress": 100,
+                    "data": {"url": "https://example.com/i2v_local.mp4"},
+                }),
+            ],
+            get_responses=[
+                FakeResponse(content=b"fake-i2v-local-video"),
+            ],
+        )
+
+        output_path = tmp_path / "i2v_local.mp4"
+        result = HunyuanCloudVideo().execute({
+            "prompt": "animate this frame",
+            "operation": "image_to_video",
+            "image_path": str(img),
+            "output_path": str(output_path),
+        })
+
+        assert result.success, result.error
+        assert output_path.read_bytes() == b"fake-i2v-local-video"
+
+    def test_explicit_incompatible_model_is_rejected(self, hunyuan_env):
+        result = HunyuanCloudVideo().execute({
+            "prompt": "test",
+            "operation": "image_to_video",
+            "model": "hy-video-1.5",
+            "image_url": "https://example.com/frame.jpg",
+        })
+
+        assert not result.success
+        assert "yt-video-2.0" in (result.error or "")
+
+    def test_polling_retries_until_success(self, hunyuan_env, tmp_path, monkeypatch):
+        """Polling should retry when status is queued/running, then succeed."""
+        task_id = "poll-retry-task"
+        _install_fake_requests(
+            monkeypatch,
+            post_responses=[
+                FakeResponse({
+                    "id": task_id, "request_id": "req-sub", "object": "video",
+                    "created_at": 1700000000, "status": "queued",
+                }),
+                FakeResponse({
+                    "request_id": "r1", "object": "video",
+                    "status": "queued", "progress": 0,
+                }),
+                FakeResponse({
+                    "request_id": "r2", "object": "video",
+                    "status": "running", "progress": 45,
+                }),
+                FakeResponse({
+                    "request_id": "r3", "object": "video",
+                    "status": "completed", "progress": 100,
+                    "data": {"url": "https://example.com/final.mp4"},
+                }),
+            ],
+            get_responses=[
+                FakeResponse(content=b"final-video-data"),
+            ],
+        )
+
+        result = HunyuanCloudVideo().execute({
+            "prompt": "test polling",
+            "poll_interval_seconds": 0.1,
+            "output_path": str(tmp_path / "polled.mp4"),
+        })
+
+        assert result.success, result.error
+        assert result.data["task_id"] == task_id
+
+    def test_polling_fails_on_task_failed(self, hunyuan_env, tmp_path, monkeypatch):
+        """When the API returns status=failed, execute should return error."""
+        task_id = "failed-task"
+        _install_fake_requests(
+            monkeypatch,
+            post_responses=[
+                FakeResponse({
+                    "id": task_id, "request_id": "req-sub", "object": "video",
+                    "created_at": 1700000000, "status": "queued",
+                }),
+                FakeResponse({
+                    "request_id": "req-fail", "object": "video",
+                    "status": "failed",
+                    "error": {"code": "internal_error", "message": "Service unavailable"},
+                }),
+            ],
+            get_responses=[],
+        )
+
+        result = HunyuanCloudVideo().execute({
+            "prompt": "this will fail",
+            "poll_interval_seconds": 0.1,
+            "output_path": str(tmp_path / "failed.mp4"),
+        })
+
+        assert result.success is False
+        assert "failed" in result.error.lower()
+        assert "Service unavailable" in result.error
+
+    def test_submit_error_returns_failure(self, hunyuan_env, tmp_path, monkeypatch):
+        """API-level error on submit should be returned as ToolResult error."""
+        _install_fake_requests(
+            monkeypatch,
+            post_responses=[
+                FakeResponse({
+                    "error": {"code": "invalid_parameter", "message": "Prompt exceeds limit"},
+                }),
+            ],
+            get_responses=[],
+        )
+
+        result = HunyuanCloudVideo().execute({
+            "prompt": "test",
+            "output_path": str(tmp_path / "err.mp4"),
+        })
+
+        assert result.success is False
+        assert "Prompt exceeds limit" in result.error
+
+
+# ------------------------------------------------------------------
+# Registry discovery
+# ------------------------------------------------------------------
+
+class TestRegistryDiscovery:
+
+    def test_discoverable(self, isolated_tool_registry):
+        isolated_tool_registry.discover()
+        tool = isolated_tool_registry.get("hunyuan_cloud_video")
+        assert tool is not None
+        assert tool.provider == "hunyuan_cloud"
+        assert tool.capability == "video_generation"
+
+    def test_distinct_from_local_hunyuan_tool(self, isolated_tool_registry):
+        isolated_tool_registry.discover()
+        cloud = isolated_tool_registry.get("hunyuan_cloud_video")
+        local = isolated_tool_registry.get("hunyuan_video")
+        assert cloud is not None
+        assert local is not None
+        assert cloud.provider == "hunyuan_cloud"
+        assert local.provider == "hunyuan"
+        assert cloud.runtime == ToolRuntime.API
+        assert local.runtime == ToolRuntime.LOCAL_GPU
+
+    def test_video_selector_routes_to_hunyuan_cloud(self, hunyuan_env, monkeypatch):
+        from tools.base_tool import ToolResult
+        from tools.video.video_selector import VideoSelector
+
+        tool = HunyuanCloudVideo()
+        selector = VideoSelector()
+        monkeypatch.setattr(selector, "_providers", lambda: [tool])
+        monkeypatch.setattr(
+            selector,
+            "_select_best_tool",
+            lambda _inputs, _candidates, _context: (tool, None),
+        )
+        monkeypatch.setattr(
+            tool,
+            "execute",
+            lambda inputs: ToolResult(
+                success=True,
+                data={"received": inputs},
+                artifacts=[inputs["output_path"]],
+            ),
+        )
+
+        result = selector.execute(
+            {
+                "prompt": "test",
+                "preferred_provider": "hunyuan_cloud",
+                "output_path": "out.mp4",
+            }
+        )
+        assert result.success
+        assert result.data["selected_tool"] == "hunyuan_cloud_video"
+        assert result.data["selected_provider"] == "hunyuan_cloud"
+
+
+# ------------------------------------------------------------------
+# Schema validation
+# ------------------------------------------------------------------
+
+class TestSchemaValidation:
+
+    def test_prompt_max_length_200(self):
+        schema = HunyuanCloudVideo().input_schema
+        assert schema["properties"]["prompt"]["maxLength"] == 200
+
+    def test_prompt_rejects_over_200_chars(self):
+        import jsonschema
+        schema = HunyuanCloudVideo().input_schema
+        instance = {"prompt": "x" * 201}
+        with pytest.raises(jsonschema.ValidationError):
+            jsonschema.validate(instance, schema)
+
+    def test_prompt_accepts_200_chars(self):
+        import jsonschema
+        schema = HunyuanCloudVideo().input_schema
+        instance = {"prompt": "x" * 200}
+        jsonschema.validate(instance, schema)
+
+    def test_operation_accepts_valid_values(self):
+        import jsonschema
+        schema = HunyuanCloudVideo().input_schema
+        for op in ["text_to_video", "image_to_video"]:
+            jsonschema.validate({"prompt": "test", "operation": op}, schema)
+
+    def test_operation_rejects_invalid_values(self):
+        import jsonschema
+        schema = HunyuanCloudVideo().input_schema
+        for invalid in ["video_to_video", "", "TEXT_TO_VIDEO"]:
+            with pytest.raises(jsonschema.ValidationError):
+                jsonschema.validate({"prompt": "test", "operation": invalid}, schema)
+
+    def test_model_accepts_valid_values(self):
+        import jsonschema
+        schema = HunyuanCloudVideo().input_schema
+        for m in ["hy-video-1.5", "yt-video-2.0"]:
+            jsonschema.validate({"prompt": "test", "model": m}, schema)
+
+    def test_model_rejects_invalid_values(self):
+        import jsonschema
+        schema = HunyuanCloudVideo().input_schema
+        with pytest.raises(jsonschema.ValidationError):
+            jsonschema.validate({"prompt": "test", "model": "invalid-model"}, schema)
+
+    def test_logo_add_accepts_0_and_1(self):
+        import jsonschema
+        schema = HunyuanCloudVideo().input_schema
+        for val in [0, 1]:
+            jsonschema.validate({"prompt": "test", "logo_add": val}, schema)
+
+    def test_logo_add_rejects_other_values(self):
+        import jsonschema
+        schema = HunyuanCloudVideo().input_schema
+        for invalid in [2, -1, 99]:
+            with pytest.raises(jsonschema.ValidationError):
+                jsonschema.validate({"prompt": "test", "logo_add": invalid}, schema)
+
+    def test_poll_interval_minimum_2(self):
+        schema = HunyuanCloudVideo().input_schema
+        assert schema["properties"]["poll_interval_seconds"]["minimum"] == 2
+
+    def test_timeout_minimum_60(self):
+        schema = HunyuanCloudVideo().input_schema
+        assert schema["properties"]["timeout_seconds"]["minimum"] == 60
+
+    def test_resolution_only_720p(self):
+        schema = HunyuanCloudVideo().input_schema
+        assert schema["properties"]["resolution"]["enum"] == ["720p", "1080p"]

tools/video/hunyuan_cloud_video.py

@@ -0,0 +1,530 @@
+"""Tencent Hunyuan (腾讯混元) cloud video generation via TokenHub API.
+
+Calls the Tencent TokenHub API (tokenhub.tencentmaas.com) using simple Bearer
+token authentication. This is the OpenAI-compatible API gateway for Tencent
+Hunyuan video models — no TC3-HMAC-SHA256 signing required.
+
+API flow: POST /v1/api/video/submit -> poll /v1/api/video/query ->
+download data.url.
+
+Authentication uses a TokenHub API key obtained from the Tencent Cloud
+TokenHub console (https://console.cloud.tencent.com/tokenhub).
+"""
+
+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,
+)
+
+_HOST = "tokenhub.tencentmaas.com"
+_SUBMIT_PATH = "/v1/api/video/submit"
+_QUERY_PATH = "/v1/api/video/query"
+
+# TokenHub model identifiers
+_MODEL_T2V = "hy-video-1.5"       # text-to-video
+_MODEL_I2V = "yt-video-2.0"       # image-to-video
+
+
+class HunyuanCloudVideo(BaseTool):
+    """Tencent Hunyuan cloud video generation via TokenHub API."""
+
+    name = "hunyuan_cloud_video"
+    version = "0.1.0"
+    tier = ToolTier.GENERATE
+    capability = "video_generation"
+    provider = "hunyuan_cloud"
+    stability = ToolStability.EXPERIMENTAL
+    execution_mode = ExecutionMode.ASYNC
+    determinism = Determinism.STOCHASTIC
+    runtime = ToolRuntime.API
+
+    dependencies = ["env:TENCENT_TOKENHUB_API_KEY"]
+    install_instructions = (
+        "Set TENCENT_TOKENHUB_API_KEY to your Tencent Cloud TokenHub API key.\n"
+        "  Get it at https://console.cloud.tencent.com/tokenhub"
+    )
+    agent_skills = ["ai-video-gen"]
+
+    capabilities = ["text_to_video", "image_to_video"]
+    supports = {
+        "text_to_video": True,
+        "image_to_video": True,
+        "native_audio": False,
+        "seed": False,
+    }
+    best_for = [
+        "Hunyuan text-to-video and image-to-video via Tencent TokenHub API",
+        "simple Bearer-token auth (no TC3 signing required)",
+        "direct Tencent Cloud quota usage (not through a third-party gateway)",
+        "Chinese-language prompt understanding"
+    ]
+    not_good_for = [
+        "offline generation or air-gapped environments",
+        "users without Tencent Cloud account and real-name verification"
+    ]
+    fallback_tools = ["jimeng_video", "kling_official_video", "minimax_video"]
+
+    input_schema = {
+        "type": "object",
+        "required": ["prompt"],
+        "properties": {
+            "prompt": {
+                "type": "string",
+                "maxLength": 200,
+                "description": (
+                    "Video description. Max 200 UTF-8 characters. "
+                    "Supports Chinese and English. Be specific about subject, action, "
+                    "setting, and style."
+                ),
+            },
+            "operation": {
+                "type": "string",
+                "enum": ["text_to_video", "image_to_video"],
+                "default": "text_to_video",
+                "description": "Generation mode.",
+            },
+            "model": {
+                "type": "string",
+                "enum": ["hy-video-1.5", "yt-video-2.0"],
+                "description": (
+                    "TokenHub model ID. hy-video-1.5 for text-to-video, "
+                    "yt-video-2.0 for image-to-video. Defaults to the recommended "
+                    "model for the chosen operation."
+                ),
+            },
+            "image_url": {
+                "type": "string",
+                "description": (
+                    "Reference image URL for image-to-video. "
+                    "Must be publicly accessible. Max 10MB. "
+                    "Formats: jpg/png/jpeg/webp/bmp/tiff. "
+                    "Resolution: 50-5000 pixels per side, aspect ratio 1:4 to 4:1."
+                ),
+            },
+            "image_path": {
+                "type": "string",
+                "description": (
+                    "Local path to a reference image for image-to-video. "
+                    "Will be base64-encoded and sent inline. "
+                    "Mutually exclusive with image_url."
+                ),
+            },
+            "resolution": {
+                "type": "string",
+                "enum": ["720p", "1080p"],
+                "default": "720p",
+                "description": "Output resolution.",
+            },
+            "logo_add": {
+                "type": "integer",
+                "enum": [0, 1],
+                "default": 1,
+                "description": (
+                    "Add 'AI-generated' watermark. 1 = add watermark (default), "
+                    "0 = no watermark (requires console approval from Tencent)."
+                ),
+            },
+            "output_path": {
+                "type": "string",
+                "description": "Output file path for the generated video (MP4).",
+            },
+            "poll_interval_seconds": {
+                "type": "number",
+                "minimum": 2,
+                "default": 5.0,
+                "description": "Seconds between status polls.",
+            },
+            "timeout_seconds": {
+                "type": "integer",
+                "minimum": 60,
+                "default": 600,
+                "description": "Maximum seconds to wait for generation.",
+            },
+        },
+    }
+
+    resource_profile = ResourceProfile(
+        cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True,
+    )
+    retry_policy = RetryPolicy(
+        max_retries=2,
+        backoff_seconds=2.0,
+        retryable_errors=["rate_limit", "timeout"],
+    )
+    idempotency_key_fields = [
+        "prompt",
+        "operation",
+        "model",
+        "image_url",
+        "image_path",
+        "resolution",
+        "logo_add",
+    ]
+    side_effects = [
+        "writes video file to output_path",
+        "calls Tencent TokenHub API (Bearer-token submit + poll + download)",
+    ]
+    user_visible_verification = [
+        "Watch generated clip for motion coherence and prompt adherence",
+        "Check for watermark if logo_add=0 was requested",
+    ]
+
+    # ------------------------------------------------------------------
+    # Credential helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _api_key() -> str | None:
+        val = os.environ.get("TENCENT_TOKENHUB_API_KEY", "")
+        if val and not val.strip().startswith("#"):
+            return val.strip()
+        return None
+
+    # ------------------------------------------------------------------
+    # Tool contract methods
+    # ------------------------------------------------------------------
+
+    def get_status(self) -> ToolStatus:
+        if self._api_key():
+            return ToolStatus.AVAILABLE
+        return ToolStatus.UNAVAILABLE
+
+    def estimate_cost(self, inputs: dict[str, Any]) -> float:
+        """Estimate cost in USD based on model and resolution.
+
+        Tencent TokenHub credit-based pricing (1 credit = 1.2 RMB ≈ $0.167 USD):
+        - HY-Video-1.5: 1.5 credits/generation → $0.25
+        - YT-Video-2.0 480p:       2   credits/generation → $0.33
+        - YT-Video-2.0 720p/1080p: 5   credits/generation → $0.83
+
+        Source: https://cloud.tencent.com.cn/document/product/1823/130054
+        """
+        model = self._resolve_model(inputs)
+        resolution = inputs.get("resolution", "720p")
+
+        _CREDIT_TO_USD = 1.2 / 7.2  # 1 credit = 1.2 RMB, ~7.2 RMB/USD
+
+        if model == _MODEL_I2V:
+            # YT-Video-2.0 has resolution-tiered pricing
+            if resolution in ("720p", "1080p"):
+                credits = 5.0
+            else:
+                credits = 2.0  # 480p and below
+        else:
+            # HY-Video-1.5 (and fallback for unknown models)
+            credits = 1.5
+
+        return round(credits * _CREDIT_TO_USD, 2)
+
+    def estimate_runtime(self, inputs: dict[str, Any]) -> float:
+        """Estimate wall-clock time in seconds.
+
+        Empirical estimate: most cloud video generation APIs (Kling, Pika, etc.)
+        queue + generate in 60-180s. No official latency published by Tencent.
+        180s is a safe upper-bound for timeout planning.
+        """
+        return 180.0
+
+    # ------------------------------------------------------------------
+    # Main execution
+    # ------------------------------------------------------------------
+
+    def execute(self, inputs: dict[str, Any]) -> ToolResult:
+        api_key = self._api_key()
+        if not api_key:
+            return ToolResult(
+                success=False,
+                error="TENCENT_TOKENHUB_API_KEY not set. " + self.install_instructions,
+            )
+
+        operation = inputs.get("operation", "text_to_video")
+        model = self._resolve_model(inputs)
+        expected_model = _MODEL_I2V if operation == "image_to_video" else _MODEL_T2V
+        if model != expected_model:
+            return ToolResult(
+                success=False,
+                error=(
+                    f"Model '{model}' is not compatible with operation '{operation}'. "
+                    f"Use '{expected_model}'."
+                ),
+            )
+        if operation == "image_to_video" and not inputs.get("image_url") and not inputs.get("image_path"):
+            return ToolResult(
+                success=False,
+                error="image_to_video requires image_url (public URL) or image_path (local file).",
+            )
+
+        if inputs.get("image_url") and inputs.get("image_path"):
+            return ToolResult(
+                success=False,
+                error="Provide only one of image_url or image_path, not both.",
+            )
+
+        start = time.time()
+        try:
+            result = self._generate(inputs, api_key=api_key)
+        except Exception as exc:
+            return ToolResult(
+                success=False,
+                error=f"Hunyuan TokenHub video generation failed: {self._safe_error(exc)}",
+            )
+
+        result.duration_seconds = round(time.time() - start, 2)
+        return result
+
+    # ------------------------------------------------------------------
+    # Generation pipeline
+    # ------------------------------------------------------------------
+
+    def _generate(
+        self, inputs: dict[str, Any], *, api_key: str,
+    ) -> ToolResult:
+        import requests
+
+        from tools.video._shared import probe_output
+
+        payload = self._build_payload(inputs)
+        model = self._resolve_model(inputs)
+        task_id = self._submit_task(payload, model=model, api_key=api_key)
+        video_url = self._poll_task(
+            task_id,
+            model=model,
+            api_key=api_key,
+            poll_interval=float(inputs.get("poll_interval_seconds", 5.0)),
+            timeout_seconds=int(inputs.get("timeout_seconds", 600)),
+        )
+
+        download = requests.get(video_url, timeout=120)
+        download.raise_for_status()
+
+        output_path = Path(
+            inputs.get("output_path", f"hunyuan_cloud_{task_id}.mp4")
+        )
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_bytes(download.content)
+
+        probed = probe_output(output_path)
+        return ToolResult(
+            success=True,
+            data={
+                "provider": "hunyuan_cloud",
+                "route": "tokenhub",
+                "model": model,
+                "prompt": inputs["prompt"],
+                "operation": inputs.get("operation", "text_to_video"),
+                "resolution": inputs.get("resolution", "720p"),
+                "logo_add": payload.get("logo_add", 1),
+                "task_id": task_id,
+                "video_url": video_url,
+                "output": str(output_path),
+                "format": "mp4",
+                **probed,
+            },
+            artifacts=[str(output_path)],
+            cost_usd=self.estimate_cost(inputs),
+            model=model,
+        )
+
+    # ------------------------------------------------------------------
+    # Payload construction
+    # ------------------------------------------------------------------
+
+    def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]:
+        """Build the request body for TokenHub video submit."""
+        operation = inputs.get("operation", "text_to_video")
+        payload: dict[str, Any] = {
+            "prompt": inputs["prompt"],
+        }
+
+        # Optional parameters (TokenHub uses lowercase_with_underscores)
+        if inputs.get("resolution"):
+            payload["resolution"] = inputs["resolution"]
+        if "logo_add" in inputs:
+            payload["logo_add"] = int(inputs["logo_add"])
+
+        # Image for image-to-video
+        if operation == "image_to_video":
+            if inputs.get("image_url"):
+                payload["image_url"] = inputs["image_url"]
+            elif inputs.get("image_path"):
+                payload["image"] = self._encode_image(inputs["image_path"])
+
+        return payload
+
+    @staticmethod
+    def _resolve_model(inputs: dict[str, Any]) -> str:
+        """Resolve the TokenHub model ID.
+
+        Order of precedence:
+        1. Explicit ``model`` input
+        2. Default based on operation (hy-video-1.5 for T2V, yt-video-2.0 for I2V)
+        """
+        if inputs.get("model"):
+            return inputs["model"]
+        operation = inputs.get("operation", "text_to_video")
+        return _MODEL_I2V if operation == "image_to_video" else _MODEL_T2V
+
+    @staticmethod
+    def _encode_image(path: str) -> str:
+        """Read a local image file and return a base64-encoded string."""
+        import base64
+
+        image_path = Path(path)
+        if not image_path.is_file():
+            raise FileNotFoundError(f"Image not found: {path}")
+
+        raw = image_path.read_bytes()
+        max_raw = 6 * 1024 * 1024  # 6MB raw ≈ 8MB base64
+        if len(raw) > max_raw:
+            raise ValueError(
+                f"Image too large ({len(raw)} bytes). Max ~6MB raw (8MB base64-encoded)."
+            )
+
+        return base64.b64encode(raw).decode("ascii")
+
+    # ------------------------------------------------------------------
+    # API communication (TokenHub OpenAI-compatible)
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _auth_headers(api_key: str) -> dict[str, str]:
+        """Build common request headers for TokenHub API calls."""
+        return {
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+        }
+
+    def _submit_task(
+        self, payload: dict[str, Any], *, model: str, api_key: str,
+    ) -> str:
+        """Submit a video generation task and return the task ID."""
+        import requests
+
+        body = {
+            "model": model,
+            **payload,
+        }
+        url = f"https://{_HOST}{_SUBMIT_PATH}"
+        resp = requests.post(
+            url,
+            json=body,
+            headers=self._auth_headers(api_key),
+            timeout=30,
+        )
+        data = self._json_or_raise(resp)
+        self._check_response(data)
+
+        task_id = data.get("id")
+        if not task_id:
+            raise RuntimeError(
+                f"TokenHub submit returned no task id: {data}"
+            )
+        return task_id
+
+    def _poll_task(
+        self,
+        task_id: str,
+        *,
+        model: str,
+        api_key: str,
+        poll_interval: float,
+        timeout_seconds: int,
+    ) -> str:
+        """Poll /v1/api/video/query until completion, return video download URL."""
+        import requests
+
+        url = f"https://{_HOST}{_QUERY_PATH}"
+
+        deadline = time.time() + timeout_seconds
+        while time.time() < deadline:
+            time.sleep(poll_interval)
+
+            resp = requests.post(
+                url,
+                json={"model": model, "id": task_id},
+                headers=self._auth_headers(api_key),
+                timeout=30,
+            )
+            data = self._json_or_raise(resp)
+            self._check_response(data)
+
+            status = data.get("status", "")
+
+            if status == "completed":
+                result_data = data.get("data") or {}
+                video_url = result_data.get("url")
+                if not video_url:
+                    raise RuntimeError(
+                        f"TokenHub task {task_id} completed but no data.url: {data}"
+                    )
+                return video_url
+
+            if status == "failed":
+                error_info = data.get("error") or {}
+                error_msg = error_info.get("message", "unknown error")
+                raise RuntimeError(
+                    f"TokenHub task {task_id} failed: {error_msg}"
+                )
+
+            # queued / running / in_progress — continue polling
+            if status not in ("queued", "running", "in_progress"):
+                raise RuntimeError(
+                    f"TokenHub task {task_id} returned unknown status: {status}"
+                )
+
+        raise TimeoutError(
+            f"TokenHub task {task_id} did not finish within {timeout_seconds}s"
+        )
+
+    # ------------------------------------------------------------------
+    # Error handling helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _safe_error(exc: Exception) -> str:
+        """Redact secret values from exception messages."""
+        msg = str(exc)
+        for var in ("TENCENT_TOKENHUB_API_KEY",):
+            val = os.environ.get(var, "")
+            if val:
+                msg = msg.replace(val, "[redacted]")
+        return msg
+
+    @staticmethod
+    def _json_or_raise(response: Any) -> dict[str, Any]:
+        """Parse JSON response body or raise with HTTP status."""
+        try:
+            return response.json()
+        except ValueError as exc:
+            raise RuntimeError(
+                f"Non-JSON response from TokenHub API: HTTP {response.status_code}"
+            ) from exc
+
+    @staticmethod
+    def _check_response(payload: dict[str, Any]) -> None:
+        """Check the TokenHub API response for errors.
+
+        TokenHub returns errors at the top level with an ``error`` field.
+        """
+        error = payload.get("error")
+        if error:
+            message = error.get("message", "unknown error")
+            code = error.get("code", error.get("type", "unknown"))
+            raise RuntimeError(
+                f"TokenHub API error: code={code}, message={message}"
+            )