Merge pull request #341 from yiyabo/feat/jimeng-video

Add Volcengine Jimeng (即梦 AI) video provider with V4 signing

.env.example

@@ -50,6 +50,8 @@ SUNO_API_KEY=                # Suno AI music generation (full songs, instrumenta
 # --- Video Generation ---
 HEYGEN_API_KEY=              # HeyGen API (VEO, Sora, Runway, Kling, Seedance via single key)
 RUNWAY_API_KEY=              # Runway Gen-4 (direct API, alternative to fal.ai routing)
+VOLC_ACCESSKEY=              # Volcengine Jimeng (即梦 AI) video generation via official API (HMAC-SHA256 V4 signing)
+VOLC_SECRETKEY=             # Secret Access Key paired with VOLC_ACCESSKEY. Get both at https://console.volcengine.com/iam/keymanage
 VIDEO_GEN_LOCAL_ENABLED=     # Set to "true" for local video gen (needs GPU + diffusers)
 VIDEO_GEN_LOCAL_MODEL=       # Local model: wan2.1-1.3b, wan2.1-14b, hunyuan-1.5, ltx2-local, cogvideo-5b
 MODAL_LTX2_ENDPOINT_URL=    # Modal self-hosted LTX-2 endpoint (optional)

docs/PROVIDERS.md

@@ -104,6 +104,53 @@ OpenMontage now uses those published rates in the Grok tool estimators.
 
 ---
 
+### Volcengine Jimeng — 即梦 AI Video Generation
+
+> **Direct ByteDance API via V4 signing.** Calls the Volcengine visual API (visual.volcengineapi.com) with HMAC-SHA256 request signing using IAM AK/SK credentials. Supports text-to-video and image-to-video via Jimeng 3.0 Pro.
+
+**Tools unlocked:** `jimeng_video`
+**Env vars:** `VOLC_ACCESSKEY` (Access Key ID) + `VOLC_SECRETKEY` (Secret Access Key)
+
+#### Setup
+
+1. Go to [console.volcengine.com/iam/keymanage](https://console.volcengine.com/iam/keymanage)
+2. Create a Volcengine account if you don't have one
+3. Create an Access Key pair (AK + SK)
+4. Ensure your account has access to Jimeng AI (即梦) video generation service
+5. Add to `.env`: `VOLC_ACCESSKEY=...` and `VOLC_SECRETKEY=...`
+
+#### What it's best for
+
+- Direct ByteDance/Volcengine API quota usage
+- Jimeng 3.0 Pro text-to-video and image-to-video
+- Chinese-language prompt understanding
+- Configurable frame count (121=5s, 241=10s) and aspect ratio
+
+#### API notes
+
+Authentication uses Volcengine IAM V4 signing (HMAC-SHA256), not a Bearer token. The signing process builds a canonical request, derives a signing key from SK → date → region → service, and signs the request.
+
+API flow: `POST ?Action=CVSync2AsyncSubmitTask` → poll `POST ?Action=CVSync2AsyncGetResult` → download `video_url`.
+
+The implementation uses the compatible generic `CVSync2Async*` route (API version `2022-08-31`) rather than the model-specific `2024-06-06` actions presented in the public API explorer. This is intentional — the generic route supports the same Jimeng 3.0 Pro model via `req_key` while remaining stable across model updates.
+
+The `req_key` for video is `jimeng_ti2v_v30_pro`. Success code is `10000`. Task statuses: `in_queue`, `generating`, `done`, `not_found`, `expired`.
+
+**Authoritative API reference:** [Jimeng TI2V V30 Pro SubmitTask](https://api.volcengine.com/api-docs/view?action=JimengTI2VV30PROSubmitTask&serviceCode=cv&version=2024-06-06)
+
+**Schema constraints** (enforced by `input_schema` to prevent paid-call failures):
+- `prompt`: max 800 characters
+- `frames`: must be exactly `121` (5s) or `241` (10s) at 24fps
+- `seed`: `-1` for random, or any non-negative integer
+
+#### Pricing
+
+| Model | Price |
+|------|-------|
+| Jimeng 3.0 Pro (video) | ~$0.05/sec (check Volcengine console for actual rate) |
+
+---
+
 ### 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.

tests/contracts/test_jimeng_video.py

@@ -0,0 +1,403 @@
+"""Contract tests for the Volcengine Jimeng video provider tool.
+
+These tests verify that the tool satisfies the BaseTool contract without
+requiring real Volcengine AK/SK credentials or making any API calls.
+
+Run: pytest tests/contracts/test_jimeng_video.py -v
+"""
+
+import pytest
+
+from tools.base_tool import (
+    BaseTool,
+    Determinism,
+    ExecutionMode,
+    ToolRuntime,
+    ToolStability,
+    ToolStatus,
+    ToolTier,
+)
+from tools.video.jimeng_video import JimengVideo
+
+
+# ------------------------------------------------------------------
+# Contract compliance
+# ------------------------------------------------------------------
+
+class TestContract:
+
+    def test_inherits_base_tool(self):
+        assert issubclass(JimengVideo, BaseTool)
+
+    def test_has_required_identity(self):
+        tool = JimengVideo()
+        assert tool.name == "jimeng_video"
+        assert tool.version
+        assert tool.provider == "volcengine"
+        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 JimengVideo().execution_mode == ExecutionMode.ASYNC
+
+    def test_has_input_schema(self):
+        schema = JimengVideo().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 = JimengVideo()
+        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 JimengVideo().agent_skills
+
+    def test_has_fallbacks(self):
+        tool = JimengVideo()
+        assert "minimax_video" in tool.fallback_tools
+        assert "kling_video" in tool.fallback_tools
+
+    def test_has_install_instructions(self):
+        tool = JimengVideo()
+        assert "VOLC_ACCESSKEY" in tool.install_instructions
+        assert "VOLC_SECRETKEY" in tool.install_instructions
+
+    def test_get_info_returns_dict(self):
+        info = JimengVideo().get_info()
+        assert isinstance(info, dict)
+        assert info["name"] == "jimeng_video"
+        assert info["provider"] == "volcengine"
+        assert info["runtime"] == "api"
+
+    def test_status_unavailable_without_keys(self, monkeypatch):
+        monkeypatch.delenv("VOLC_ACCESSKEY", raising=False)
+        monkeypatch.delenv("VOLC_SECRETKEY", raising=False)
+        assert JimengVideo().get_status() == ToolStatus.UNAVAILABLE
+
+    def test_status_available_with_keys(self, monkeypatch):
+        monkeypatch.setenv("VOLC_ACCESSKEY", "fake-ak")
+        monkeypatch.setenv("VOLC_SECRETKEY", "fake-sk")
+        assert JimengVideo().get_status() == ToolStatus.AVAILABLE
+
+    def test_status_unavailable_with_only_ak(self, monkeypatch):
+        monkeypatch.setenv("VOLC_ACCESSKEY", "fake-ak")
+        monkeypatch.delenv("VOLC_SECRETKEY", raising=False)
+        assert JimengVideo().get_status() == ToolStatus.UNAVAILABLE
+
+    def test_has_resource_profile(self):
+        rp = JimengVideo().resource_profile
+        assert rp.network_required is True
+        assert rp.vram_mb == 0
+
+    def test_has_retry_policy(self):
+        assert JimengVideo().retry_policy.max_retries >= 0
+
+    def test_has_side_effects(self):
+        side = JimengVideo().side_effects
+        assert len(side) > 0
+        assert any("API" in s for s in side)
+
+    def test_has_user_visible_verification(self):
+        assert len(JimengVideo().user_visible_verification) > 0
+
+    def test_lazy_imports_requests(self, monkeypatch):
+        import importlib
+        import sys
+        mod_name = "tools.video.jimeng_video"
+        if "requests" in sys.modules:
+            monkeypatch.delitem(sys.modules, "requests")
+        importlib.reload(sys.modules[mod_name])
+
+    def test_estimate_cost_returns_float(self):
+        cost = JimengVideo().estimate_cost({"prompt": "x", "frames": 121})
+        assert isinstance(cost, float)
+        assert cost > 0.0
+
+    def test_dry_run_returns_dict(self):
+        result = JimengVideo().dry_run({"prompt": "test"})
+        assert isinstance(result, dict)
+        assert result["tool"] == "jimeng_video"
+
+
+# ------------------------------------------------------------------
+# Idempotency keys
+# ------------------------------------------------------------------
+
+class TestIdempotencyKeys:
+
+    def test_includes_all_output_affecting_fields(self):
+        fields = JimengVideo().idempotency_key_fields
+        for field in ("prompt", "operation", "image_url", "frames", "aspect_ratio", "seed"):
+            assert field in fields, f"missing idempotency field: {field}"
+
+    def test_excludes_execution_only_fields(self):
+        fields = JimengVideo().idempotency_key_fields
+        for field in ("output_path", "poll_interval_seconds", "timeout_seconds"):
+            assert field not in fields
+
+    def test_differs_on_frames(self):
+        tool = JimengVideo()
+        base = {"prompt": "x"}
+        assert tool.idempotency_key(base) != tool.idempotency_key({**base, "frames": 241})
+
+    def test_differs_on_aspect_ratio(self):
+        tool = JimengVideo()
+        base = {"prompt": "x"}
+        assert tool.idempotency_key({**base, "aspect_ratio": "16:9"}) != tool.idempotency_key(
+            {**base, "aspect_ratio": "9:16"}
+        )
+
+    def test_differs_on_seed(self):
+        tool = JimengVideo()
+        base = {"prompt": "x"}
+        assert tool.idempotency_key({**base, "seed": -1}) != tool.idempotency_key(
+            {**base, "seed": 42}
+        )
+
+    def test_differs_on_image_url(self):
+        tool = JimengVideo()
+        base = {"prompt": "x", "operation": "image_to_video"}
+        assert tool.idempotency_key(base) != tool.idempotency_key(
+            {**base, "image_url": "https://example.com/img.png"}
+        )
+
+
+# ------------------------------------------------------------------
+# Tool-specific behavior
+# ------------------------------------------------------------------
+
+class TestToolSpecific:
+
+    def test_default_frames_is_121(self):
+        tool = JimengVideo()
+        assert tool.input_schema["properties"]["frames"]["default"] == 121
+
+    def test_default_aspect_ratio_is_16_9(self):
+        tool = JimengVideo()
+        assert tool.input_schema["properties"]["aspect_ratio"]["default"] == "16:9"
+
+    def test_default_seed_is_negative_one(self):
+        tool = JimengVideo()
+        assert tool.input_schema["properties"]["seed"]["default"] == -1
+
+    def test_cost_scales_with_frames(self):
+        tool = JimengVideo()
+        cost_5s = tool.estimate_cost({"prompt": "x", "frames": 121})
+        cost_10s = tool.estimate_cost({"prompt": "x", "frames": 241})
+        assert cost_10s > cost_5s
+
+    def test_build_payload_t2v(self):
+        tool = JimengVideo()
+        payload = tool._build_payload({"prompt": "a cat"})
+        assert payload["req_key"] == "jimeng_ti2v_v30_pro"
+        assert payload["prompt"] == "a cat"
+        assert payload["frames"] == 121
+        assert payload["aspect_ratio"] == "16:9"
+        assert payload["seed"] == -1
+        assert "image_urls" not in payload
+
+    def test_build_payload_i2v_includes_image(self):
+        tool = JimengVideo()
+        payload = tool._build_payload({
+            "prompt": "motion",
+            "operation": "image_to_video",
+            "image_url": "https://example.com/img.png",
+        })
+        assert payload["image_urls"] == ["https://example.com/img.png"]
+
+    def test_build_payload_t2v_omits_image(self):
+        tool = JimengVideo()
+        payload = tool._build_payload({"prompt": "a cat", "operation": "text_to_video"})
+        assert "image_urls" not in payload
+
+    def test_i2v_without_image_fails(self, monkeypatch):
+        monkeypatch.setenv("VOLC_ACCESSKEY", "fake-ak")
+        monkeypatch.setenv("VOLC_SECRETKEY", "fake-sk")
+        result = JimengVideo().execute({"prompt": "test", "operation": "image_to_video"})
+        assert result.success is False
+        assert "image_url" in result.error
+
+    def test_no_keys_returns_error(self, monkeypatch):
+        monkeypatch.delenv("VOLC_ACCESSKEY", raising=False)
+        monkeypatch.delenv("VOLC_SECRETKEY", raising=False)
+        result = JimengVideo().execute({"prompt": "test"})
+        assert result.success is False
+        assert "VOLC_ACCESSKEY" in result.error
+        assert "VOLC_SECRETKEY" in result.error
+
+    def test_safe_error_redacts_keys(self, monkeypatch):
+        monkeypatch.setenv("VOLC_ACCESSKEY", "my-ak-secret")
+        monkeypatch.setenv("VOLC_SECRETKEY", "my-sk-secret")
+        redacted = JimengVideo._safe_error(
+            Exception("failed with ak=my-ak-secret sk=my-sk-secret")
+        )
+        assert "my-ak-secret" not in redacted
+        assert "my-sk-secret" not in redacted
+        assert "[redacted]" in redacted
+
+    def test_safe_error_no_empty_string_bug(self, monkeypatch):
+        """Regression: when no keys are set, _safe_error must not mangle."""
+        monkeypatch.delenv("VOLC_ACCESSKEY", raising=False)
+        monkeypatch.delenv("VOLC_SECRETKEY", raising=False)
+        msg = JimengVideo._safe_error(Exception("abc"))
+        assert msg == "abc"
+
+    def test_sign_returns_authorization_header(self):
+        headers = JimengVideo._sign(
+            "POST", "/",
+            {"Action": "CVSync2AsyncSubmitTask", "Version": "2022-08-31"},
+            {}, b'{"prompt":"test"}',
+            "fake-ak", "fake-sk",
+        )
+        assert "Authorization" in headers
+        assert "HMAC-SHA256" in headers["Authorization"]
+        assert "fake-ak" in headers["Authorization"]
+        assert "Host" in headers
+        assert "X-Date" in headers
+        assert "X-Content-Sha256" in headers
+
+    def test_sign_includes_content_type(self):
+        headers = JimengVideo._sign("POST", "/", {}, {}, b"{}", "ak", "sk")
+        assert headers["Content-Type"] == "application/json"
+
+    def test_json_or_raise_returns_dict(self):
+        class FakeResp:
+            status_code = 200
+            def json(self):
+                return {"code": 10000, "data": {"task_id": "123"}}
+        assert JimengVideo._json_or_raise(FakeResp()) == {"code": 10000, "data": {"task_id": "123"}}
+
+    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"):
+            JimengVideo._json_or_raise(FakeResp())
+
+    def test_check_code_passes_on_success(self):
+        JimengVideo._check_code(200, {"code": 10000, "message": "Success"})
+
+    def test_check_code_raises_on_api_error(self):
+        with pytest.raises(RuntimeError, match="code=10008"):
+            JimengVideo._check_code(200, {"code": 10008, "message": "Insufficient balance"})
+
+    def test_check_code_raises_on_http_error(self):
+        with pytest.raises(RuntimeError, match="HTTP 401"):
+            JimengVideo._check_code(401, {"code": 10004, "message": "Auth failed"})
+
+    def test_check_code_defaults_to_success_when_code_missing(self):
+        """If code field is absent on HTTP 2xx, default to 10000 (success)."""
+        JimengVideo._check_code(200, {"data": {"task_id": "123"}})
+
+
+# ------------------------------------------------------------------
+# Registry discovery
+# ------------------------------------------------------------------
+
+class TestRegistryDiscovery:
+
+    def test_discoverable(self):
+        from tools.tool_registry import ToolRegistry
+        registry = ToolRegistry()
+        registry.discover()
+        names = {t.name for t in registry._tools.values()}
+        assert "jimeng_video" in names
+
+    def test_distinct_from_other_minimax_tools(self):
+        from tools.tool_registry import ToolRegistry
+        registry = ToolRegistry()
+        registry.discover()
+        jimeng = [t for t in registry._tools.values() if t.name == "jimeng_video"]
+        assert len(jimeng) == 1
+        assert jimeng[0].provider == "volcengine"
+
+
+# ------------------------------------------------------------------
+# Schema validation — reject invalid inputs before paid API call
+# ------------------------------------------------------------------
+
+class TestSchemaValidation:
+
+    def test_frames_accepts_121(self):
+        schema = JimengVideo().input_schema
+        valid = schema["properties"]["frames"]
+        assert valid["enum"] == [121, 241]
+
+    def test_frames_rejects_non_enum(self):
+        import jsonschema
+        schema = JimengVideo().input_schema
+        for invalid in [1, 100, 200, 500, 0, -1]:
+            instance = {"prompt": "test", "frames": invalid}
+            with pytest.raises(jsonschema.ValidationError):
+                jsonschema.validate(instance, schema)
+
+    def test_prompt_max_length_800(self):
+        schema = JimengVideo().input_schema
+        assert schema["properties"]["prompt"]["maxLength"] == 800
+
+    def test_prompt_rejects_over_800_chars(self):
+        import jsonschema
+        schema = JimengVideo().input_schema
+        instance = {"prompt": "x" * 801}
+        with pytest.raises(jsonschema.ValidationError):
+            jsonschema.validate(instance, schema)
+
+    def test_prompt_accepts_800_chars(self):
+        import jsonschema
+        schema = JimengVideo().input_schema
+        instance = {"prompt": "x" * 800}
+        jsonschema.validate(instance, schema)
+
+    def test_seed_minimum_is_negative_one(self):
+        schema = JimengVideo().input_schema
+        assert schema["properties"]["seed"]["minimum"] == -1
+
+    def test_seed_rejects_below_negative_one(self):
+        import jsonschema
+        schema = JimengVideo().input_schema
+        for invalid in [-2, -10, -100]:
+            instance = {"prompt": "test", "seed": invalid}
+            with pytest.raises(jsonschema.ValidationError):
+                jsonschema.validate(instance, schema)
+
+    def test_seed_accepts_negative_one(self):
+        import jsonschema
+        schema = JimengVideo().input_schema
+        jsonschema.validate({"prompt": "test", "seed": -1}, schema)
+
+    def test_seed_accepts_zero_and_positive(self):
+        import jsonschema
+        schema = JimengVideo().input_schema
+        for valid in [0, 1, 42, 999999]:
+            jsonschema.validate({"prompt": "test", "seed": valid}, schema)
+
+
+# ------------------------------------------------------------------
+# Selector duration → frames mapping
+# ------------------------------------------------------------------
+
+class TestSelectorDurationMapping:
+
+    def test_duration_5_maps_to_121_frames(self):
+        payload = JimengVideo._build_payload({"prompt": "x", "duration": 5})
+        assert payload["frames"] == 121
+
+    def test_duration_10_maps_to_241_frames(self):
+        payload = JimengVideo._build_payload({"prompt": "x", "duration": 10})
+        assert payload["frames"] == 241
+
+    def test_duration_defaults_to_5_when_absent(self):
+        payload = JimengVideo._build_payload({"prompt": "x"})
+        assert payload["frames"] == 121
+
+    def test_frames_takes_priority_over_duration(self):
+        payload = JimengVideo._build_payload({"prompt": "x", "frames": 241, "duration": 5})
+        assert payload["frames"] == 241

tools/video/jimeng_video.py

@@ -0,0 +1,405 @@
+"""Volcengine Jimeng (即梦 AI) video generation via the official API.
+
+Calls the Volcengine visual API directly (visual.volcengineapi.com) using
+HMAC-SHA256 V4 request signing with AK/SK credentials. Supports text-to-video
+and image-to-video via the Hailuo/Jimeng 3.0 Pro model.
+
+API flow: POST CVSync2AsyncSubmitTask -> poll CVSync2AsyncGetResult ->
+download video_url.
+
+Authentication uses Volcengine IAM V4 signing (not Bearer token), which
+requires an Access Key ID (AK) and Secret Access Key (SK) pair from
+console.volcengine.com/iam/keymanage.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import time
+import urllib.parse
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from tools.base_tool import (
+    BaseTool,
+    Determinism,
+    ExecutionMode,
+    ResourceProfile,
+    RetryPolicy,
+    ToolResult,
+    ToolRuntime,
+    ToolStability,
+    ToolStatus,
+    ToolTier,
+)
+
+
+_HOST = "visual.volcengineapi.com"
+_REGION = "cn-north-1"
+_SERVICE = "cv"
+_ALGORITHM = "HMAC-SHA256"
+_API_VERSION = "2022-08-31"
+_REQ_KEY_VIDEO = "jimeng_ti2v_v30_pro"
+
+
+class JimengVideo(BaseTool):
+    name = "jimeng_video"
+    version = "0.1.0"
+    tier = ToolTier.GENERATE
+    capability = "video_generation"
+    provider = "volcengine"
+    stability = ToolStability.EXPERIMENTAL
+    execution_mode = ExecutionMode.ASYNC
+    determinism = Determinism.STOCHASTIC
+    runtime = ToolRuntime.API
+
+    dependencies = ["env:VOLC_ACCESSKEY", "env:VOLC_SECRETKEY"]
+    install_instructions = (
+        "Set VOLC_ACCESSKEY and VOLC_SECRETKEY to your Volcengine IAM credentials.\n"
+        "  Get them at https://console.volcengine.com/iam/keymanage\n"
+        "  Ensure your account has access to Jimeng AI (即梦) video generation."
+    )
+    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": True,
+    }
+    best_for = [
+        "Jimeng 3.0 Pro text-to-video and image-to-video via Volcengine",
+        "direct ByteDance API quota usage (not through a gateway)",
+        "Chinese-language prompt understanding",
+    ]
+    not_good_for = ["offline generation", "users without Volcengine AK/SK"]
+    fallback_tools = ["minimax_video", "kling_video", "veo_video"]
+
+    input_schema = {
+        "type": "object",
+        "required": ["prompt"],
+        "properties": {
+            "prompt": {
+                "type": "string",
+                "maxLength": 800,
+                "description": "Video description. Max 800 chars. Supports Chinese.",
+            },
+            "operation": {
+                "type": "string",
+                "enum": ["text_to_video", "image_to_video"],
+                "default": "text_to_video",
+            },
+            "image_url": {
+                "type": "string",
+                "description": (
+                    "First frame image URL for image-to-video. "
+                    "Must be publicly accessible."
+                ),
+            },
+            "frames": {
+                "type": "integer",
+                "enum": [121, 241],
+                "default": 121,
+                "description": "Total frames. 121=5s, 241=10s at 24fps.",
+            },
+            "aspect_ratio": {
+                "type": "string",
+                "enum": ["16:9", "9:16", "1:1", "4:3", "3:4", "21:9"],
+                "default": "16:9",
+            },
+            "seed": {
+                "type": "integer",
+                "minimum": -1,
+                "default": -1,
+                "description": "Random seed. -1 for random.",
+            },
+            "output_path": {"type": "string"},
+            "poll_interval_seconds": {
+                "type": "number",
+                "minimum": 2,
+                "default": 5.0,
+            },
+            "timeout_seconds": {
+                "type": "integer",
+                "minimum": 60,
+                "default": 600,
+            },
+        },
+    }
+
+    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",
+        "image_url",
+        "frames",
+        "aspect_ratio",
+        "seed",
+    ]
+    side_effects = [
+        "writes video file to output_path",
+        "calls Volcengine Jimeng API (V4-signed submit + poll + download)",
+    ]
+    user_visible_verification = [
+        "Watch generated clip for motion coherence and prompt adherence",
+    ]
+
+    def _ak(self) -> str | None:
+        val = os.environ.get("VOLC_ACCESSKEY", "")
+        if val and not val.strip().startswith("#"):
+            return val.strip()
+        return None
+
+    def _sk(self) -> str | None:
+        val = os.environ.get("VOLC_SECRETKEY", "")
+        if val and not val.strip().startswith("#"):
+            return val.strip()
+        return None
+
+    def get_status(self) -> ToolStatus:
+        if self._ak() and self._sk():
+            return ToolStatus.AVAILABLE
+        return ToolStatus.UNAVAILABLE
+
+    def estimate_cost(self, inputs: dict[str, Any]) -> float:
+        frames = int(inputs.get("frames", 121))
+        seconds = frames / 24.0
+        return round(0.05 * seconds, 2)
+
+    def estimate_runtime(self, inputs: dict[str, Any]) -> float:
+        return 120.0 + int(inputs.get("frames", 121)) * 1.0
+
+    def execute(self, inputs: dict[str, Any]) -> ToolResult:
+        ak = self._ak()
+        sk = self._sk()
+        if not ak or not sk:
+            return ToolResult(
+                success=False,
+                error="VOLC_ACCESSKEY or VOLC_SECRETKEY not set. " + self.install_instructions,
+            )
+
+        operation = inputs.get("operation", "text_to_video")
+        if operation == "image_to_video" and not inputs.get("image_url"):
+            return ToolResult(
+                success=False,
+                error="image_to_video requires image_url (public URL).",
+            )
+
+        start = time.time()
+        try:
+            result = self._generate(inputs, ak=ak, sk=sk)
+        except Exception as exc:
+            return ToolResult(
+                success=False,
+                error=f"Jimeng video generation failed: {self._safe_error(exc)}",
+            )
+
+        result.duration_seconds = round(time.time() - start, 2)
+        return result
+
+    def _generate(self, inputs: dict[str, Any], *, ak: str, sk: str) -> ToolResult:
+        import requests
+
+        from tools.video._shared import probe_output
+
+        payload = self._build_payload(inputs)
+        task_id = self._submit_task(payload, ak=ak, sk=sk)
+        video_url = self._poll_task(
+            task_id, ak=ak, sk=sk,
+            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", "jimeng_video.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": "volcengine",
+                "route": "jimeng_direct",
+                "model": _REQ_KEY_VIDEO,
+                "prompt": inputs["prompt"],
+                "operation": inputs.get("operation", "text_to_video"),
+                "frames": payload.get("frames", 121),
+                "aspect_ratio": payload.get("aspect_ratio", "16:9"),
+                "seed": payload.get("seed", -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=_REQ_KEY_VIDEO,
+        )
+
+    @staticmethod
+    def _duration_to_frames(duration: int) -> int:
+        if duration >= 10:
+            return 241
+        return 121
+
+    @staticmethod
+    def _build_payload(inputs: dict[str, Any]) -> dict[str, Any]:
+        operation = inputs.get("operation", "text_to_video")
+        frames = inputs.get("frames")
+        if frames is None:
+            frames = JimengVideo._duration_to_frames(int(inputs.get("duration", 5)))
+        payload: dict[str, Any] = {
+            "req_key": _REQ_KEY_VIDEO,
+            "prompt": inputs["prompt"],
+            "frames": int(frames),
+            "aspect_ratio": inputs.get("aspect_ratio", "16:9"),
+            "seed": int(inputs.get("seed", -1)),
+        }
+        if operation == "image_to_video" and inputs.get("image_url"):
+            payload["image_urls"] = [inputs["image_url"]]
+        return payload
+
+    def _submit_task(self, payload: dict[str, Any], *, ak: str, sk: str) -> str:
+        import requests
+
+        query = {"Action": "CVSync2AsyncSubmitTask", "Version": _API_VERSION}
+        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+        headers = self._sign("POST", "/", query, {}, body, ak, sk)
+        url = f"https://{_HOST}/?{urllib.parse.urlencode(sorted(query.items()))}"
+        resp = requests.post(url, data=body, headers=headers, timeout=30)
+        data = self._json_or_raise(resp)
+        self._check_code(resp.status_code, data)
+        task_id = data.get("data", {}).get("task_id")
+        if not task_id:
+            raise RuntimeError(f"Jimeng submit returned no task_id: {data}")
+        return task_id
+
+    def _poll_task(
+        self, task_id: str, *, ak: str, sk: str,
+        poll_interval: float, timeout_seconds: int,
+    ) -> str:
+        import requests
+
+        query = {"Action": "CVSync2AsyncGetResult", "Version": _API_VERSION}
+        body = json.dumps({
+            "req_key": _REQ_KEY_VIDEO,
+            "task_id": task_id,
+            "req_json": json.dumps({"return_url": True}),
+        }, ensure_ascii=False).encode("utf-8")
+
+        deadline = time.time() + timeout_seconds
+        while time.time() < deadline:
+            time.sleep(poll_interval)
+            headers = self._sign("POST", "/", query, {}, body, ak, sk)
+            url = f"https://{_HOST}/?{urllib.parse.urlencode(sorted(query.items()))}"
+            resp = requests.post(url, data=body, headers=headers, timeout=30)
+            data = self._json_or_raise(resp)
+            self._check_code(resp.status_code, data)
+            status = (data.get("data") or {}).get("status", "")
+            if status == "done":
+                video_url = (data.get("data") or {}).get("video_url")
+                if not video_url:
+                    raise RuntimeError(f"Jimeng task done but no video_url: {data}")
+                return video_url
+            if status in ("not_found", "expired"):
+                raise RuntimeError(f"Jimeng task invalid: status={status}")
+        raise TimeoutError(f"Jimeng task {task_id} did not finish within {timeout_seconds}s")
+
+    @staticmethod
+    def _sign(
+        method: str, path: str, query_params: dict,
+        headers: dict, body: bytes, ak: str, sk: str,
+    ) -> dict:
+        now = datetime.now(timezone.utc)
+        x_date = now.strftime("%Y%m%dT%H%M%SZ")
+        short_date = x_date[:8]
+
+        body_hash = hashlib.sha256(body).hexdigest()
+        headers = dict(headers)
+        headers["Host"] = _HOST
+        headers["X-Date"] = x_date
+        headers["X-Content-Sha256"] = body_hash
+        headers["Content-Type"] = "application/json"
+
+        lower_headers = {k.lower(): v.strip() for k, v in headers.items()}
+        signed_names = sorted(lower_headers)
+        canonical_headers = "".join(
+            f"{k}:{lower_headers[k]}\n"
+            for k in signed_names
+        )
+        signed_str = ";".join(signed_names)
+
+        canonical_query = "&".join(
+            f"{urllib.parse.quote(str(k), safe='')}={urllib.parse.quote(str(v), safe='')}"
+            for k, v in sorted(query_params.items())
+        )
+
+        canonical_request = "\n".join([
+            method.upper(), path, canonical_query,
+            canonical_headers, signed_str, body_hash,
+        ])
+
+        credential_scope = f"{short_date}/{_REGION}/{_SERVICE}/request"
+        string_to_sign = "\n".join([
+            _ALGORITHM, x_date, credential_scope,
+            hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
+        ])
+
+        k_date = hmac.new(sk.encode("utf-8"), short_date.encode("utf-8"), hashlib.sha256).digest()
+        k_region = hmac.new(k_date, _REGION.encode("utf-8"), hashlib.sha256).digest()
+        k_service = hmac.new(k_region, _SERVICE.encode("utf-8"), hashlib.sha256).digest()
+        k_signing = hmac.new(k_service, b"request", hashlib.sha256).digest()
+
+        signature = hmac.new(k_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
+
+        headers["Authorization"] = (
+            f"{_ALGORITHM} Credential={ak}/{credential_scope}, "
+            f"SignedHeaders={signed_str}, Signature={signature}"
+        )
+        return headers
+
+    @staticmethod
+    def _safe_error(exc: Exception) -> str:
+        msg = str(exc)
+        for var in ("VOLC_SECRETKEY", "VOLC_ACCESSKEY"):
+            val = os.environ.get(var, "")
+            if val:
+                msg = msg.replace(val, "[redacted]")
+        return msg
+
+    @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 Jimeng API: HTTP {response.status_code}"
+            ) from exc
+
+    @staticmethod
+    def _check_code(http_status: int, payload: dict[str, Any]) -> None:
+        if http_status < 400:
+            code = payload.get("code", 10000)
+            if code == 10000:
+                return
+            msg = payload.get("message", "unknown error")
+            raise RuntimeError(f"Jimeng API error: code={code}, msg={msg}")
+        code = payload.get("code", "unknown")
+        msg = payload.get("message", "unknown error")
+        raise RuntimeError(f"Jimeng API error: HTTP {http_status}, code={code}, msg={msg}")