fix(video_selector): dedup race + preferred-provider gap + motion-aware fallback

Three routing defects in video_selector, none previously covered by
tests (REVIEW §8 #3, #5, #7); plus the routing-test coverage itself (#10).

#3 Seedance dedup race
  tool_by_provider keyed by provider STRING, so two tools legitimately
  sharing provider="seedance" (seedance_video=fal, seedance_replicate)
  collided — only the first-registered was ever selectable; the other
  was invisible to the selector regardless of rank. Key selectable tools
  by NAME instead; ranking picks the best of the shared-provider backends.

#5 preferred_provider had no score-gap gate
  The selector returned the preferred provider on the first ranking match
  no matter how far below the top it scored (the comment claimed "unless
  drastically worse" but nothing enforced it). Add a configurable
  preferred_provider_gap (default 0.15): honor the preference only when
  its best ranked tool is within the gap of the overall top, else yield
  to the top-ranked provider.

#7 fallback_tools appended image_selector unconditionally
  The motion-required prohibition lived only in director skills, so a
  direct caller could silently fall back to an image-only tool for an
  image_to_video / reference_to_video brief. Add input-aware
  fallback_tools_for(inputs) that drops image_selector for
  motion-required operations; keep the static fallback_tools property
  (with image_selector) for external consumers / contracts.

#10 routing coverage
  First routing tests for video_selector: dedup reachability, the gap
  gate (honored / ignored / configurable), motion-aware fallback, and
  estimate_cost / estimate_runtime delegation. 13 tests, scoring patched
  for determinism so they test routing logic, not the scorer.

Full tools + contracts suite green (638 passed, 6 skipped).

Refs: docs/REVIEW-image-to-video-voice.md §8 #3, #5, #7, #10

Co-Authored-By: Claude <noreply@anthropic.com>

tests/tools/test_video_selector_routing.py

@@ -0,0 +1,281 @@
+"""Provider-routing regression coverage for VideoSelector (REVIEW §8 #3, #5, #7, #10).
+
+The selector had NO routing tests (``ls tests | grep video`` turned up only
+provider-specific suites), so several routing defects shipped:
+
+- #3 Seedance dedup race: two tools sharing provider="seedance" (the fal and
+  Replicate backends) were keyed by provider string in tool_by_provider, so
+  only the first-registered was ever selectable. The other was invisible.
+- #5 preferred_provider had no score-gap gate: it returned the preferred
+  provider on the first ranking match regardless of how far below the top it
+  scored (the comment claimed "unless drastically worse" but nothing enforced it).
+- #7 fallback_tools appended image_selector unconditionally — a motion-required
+  brief could fall back to an image-only tool.
+
+These tests exercise _select_best_tool / estimate_cost / estimate_runtime /
+fallback_tools_for directly with stub providers, patching lib.scoring.rank_providers
+for deterministic rankings so we test ROUTING logic, not the scorer.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from tools.base_tool import ToolStatus
+from tools.video.video_selector import VideoSelector
+
+
+class _StubTool:
+    """Minimal stand-in satisfying what _select_best_tool / _filter_candidates touch."""
+
+    capability = "video_generation"
+
+    def __init__(
+        self,
+        name: str,
+        provider: str,
+        *,
+        supports_image_to_video: bool = True,
+        status: ToolStatus = ToolStatus.AVAILABLE,
+        cost: float = 0.10,
+        runtime: float = 60.0,
+    ) -> None:
+        self.name = name
+        self.provider = provider
+        self.quality_score: float | None = None
+        self.best_for = [name]
+        self.supports = {
+            "text_to_video": True,
+            "image_to_video": supports_image_to_video,
+        }
+        self.input_schema = {"properties": {"prompt": {}}}
+        self._status = status
+        self._cost = cost
+        self._runtime = runtime
+
+    # --- BaseTool surface used by the selector -------------------------------
+    def get_status(self) -> ToolStatus:
+        return self._status
+
+    def is_operation_available(self, operation: str) -> bool:
+        return self.supports.get(operation, False)
+
+    def get_info(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "provider": self.provider,
+            "agent_skills": [],
+            "best_for": self.best_for,
+            "supports": self.supports,
+            "quality_score": self.quality_score,
+        }
+
+    def estimate_cost(self, inputs: dict[str, Any]) -> float:
+        return self._cost
+
+    def estimate_runtime(self, inputs: dict[str, Any]) -> float:
+        return self._runtime
+
+
+# ProviderScore.weighted_score is a read-only computed property, so we can't
+# override it per-instance. Instead _ScoreStub exposes the same attribute surface
+# (provider / tool_name / weighted_score) the selector reads via getattr.
+class _ScoreStub:
+    def __init__(self, tool_name: str, provider: str, weighted: float) -> None:
+        self.tool_name = tool_name
+        self.provider = provider
+        self.weighted_score = weighted
+
+    def explain(self) -> str:  # noqa: D401 - selector may call this
+        return f"{self.tool_name} ({self.provider}): {self.weighted_score:.2f}"
+
+    def to_dict(self) -> dict[str, Any]:
+        return {"tool_name": self.tool_name, "provider": self.provider, "weighted_score": self.weighted_score}
+
+
+@pytest.fixture()
+def rankings(monkeypatch):
+    """Set the ranking table the patched rank_providers returns."""
+    table: list[_ScoreStub] = []
+
+    def fake_rank(candidates, task_context):  # noqa: ANN001
+        return list(table)
+
+    monkeypatch.setattr("lib.scoring.rank_providers", fake_rank)
+    return table
+
+
+# ---------------------------------------------------------------------------
+# #3 — Seedance dedup race: two tools, same provider, must both be selectable
+# ---------------------------------------------------------------------------
+
+def test_two_tools_sharing_provider_are_both_selectable(rankings):
+    """The higher-RANKED of two same-provider tools wins; the other isn't shadowed.
+
+    Pre-fix, tool_by_provider keyed by provider string, so whichever of
+    seedance_video / seedance_replicate registered second was unreachable
+    even if it ranked higher.
+    """
+    fal = _StubTool("seedance_video", "seedance")
+    rep = _StubTool("seedance_replicate", "seedance")
+    rankings.extend([
+        _ScoreStub("seedance_replicate", "seedance", 0.90),  # ranked higher
+        _ScoreStub("seedance_video", "seedance", 0.80),
+    ])
+
+    tool, score = VideoSelector()._select_best_tool(
+        {"preferred_provider": "auto"}, [fal, rep], {}
+    )
+    assert tool is not None
+    assert tool.name == "seedance_replicate", "higher-ranked same-provider tool must win"
+
+
+def test_lower_ranked_same_provider_still_reachable_when_higher_unavailable(rankings):
+    """If the top-ranked same-provider tool is unavailable, the other is selected.
+
+    Pre-fix the unavailable one could shadow the available one in tool_by_provider
+    depending on registration order.
+    """
+    fal = _StubTool("seedance_video", "seedance", status=ToolStatus.UNAVAILABLE)
+    rep = _StubTool("seedance_replicate", "seedance")
+    rankings.extend([
+        _ScoreStub("seedance_video", "seedance", 0.95),     # ranked higher but unavailable
+        _ScoreStub("seedance_replicate", "seedance", 0.80),
+    ])
+
+    tool, score = VideoSelector()._select_best_tool(
+        {"preferred_provider": "auto"}, [fal, rep], {}
+    )
+    assert tool is not None
+    assert tool.name == "seedance_replicate"
+
+
+# ---------------------------------------------------------------------------
+# #5 — preferred_provider score-gap gate
+# ---------------------------------------------------------------------------
+
+def test_preferred_provider_honored_when_within_gap(rankings):
+    """Preferred provider ranked #2 but within the gap → selected."""
+    veo = _StubTool("veo_video", "veo")
+    kling = _StubTool("kling_video", "kling")
+    rankings.extend([
+        _ScoreStub("veo_video", "veo", 0.90),
+        _ScoreStub("kling_video", "kling", 0.80),  # 0.10 below top, within default 0.15 gap
+    ])
+
+    tool, score = VideoSelector()._select_best_tool(
+        {"preferred_provider": "kling"}, [veo, kling], {}
+    )
+    assert tool.name == "kling_video"
+
+
+def test_preferred_provider_ignored_when_drastically_worse(rankings):
+    """Preferred provider far below top → top-ranked provider wins instead.
+
+    Pre-fix the preferred provider was returned on the first ranking match
+    regardless of the gap (no gate), silently dragging selection to a worse tool.
+    """
+    veo = _StubTool("veo_video", "veo")
+    kling = _StubTool("kling_video", "kling")
+    rankings.extend([
+        _ScoreStub("veo_video", "veo", 0.95),
+        _ScoreStub("kling_video", "kling", 0.50),  # 0.45 below top, outside 0.15 gap
+    ])
+
+    tool, score = VideoSelector()._select_best_tool(
+        {"preferred_provider": "kling"}, [veo, kling], {}
+    )
+    assert tool.name == "veo_video", "preference must yield to a drastically better top"
+
+
+def test_preferred_provider_gap_is_configurable(rankings):
+    """A wider gap lets an otherwise-too-low preferred provider win."""
+    veo = _StubTool("veo_video", "veo")
+    kling = _StubTool("kling_video", "kling")
+    rankings.extend([
+        _ScoreStub("veo_video", "veo", 0.95),
+        _ScoreStub("kling_video", "kling", 0.70),  # 0.25 below top
+    ])
+
+    # default gap (0.15) → veo wins
+    tool_default, _ = VideoSelector()._select_best_tool(
+        {"preferred_provider": "kling"}, [veo, kling], {}
+    )
+    assert tool_default.name == "veo_video"
+
+    # widened gap (0.30) → kling wins
+    tool_wide, _ = VideoSelector()._select_best_tool(
+        {"preferred_provider": "kling", "preferred_provider_gap": 0.30}, [veo, kling], {}
+    )
+    assert tool_wide.name == "kling_video"
+
+
+def test_preferred_provider_not_in_rankings_falls_through(rankings):
+    """An unknown/preferred provider that doesn't rank yields the top provider."""
+    veo = _StubTool("veo_video", "veo")
+    rankings.append(_ScoreStub("veo_video", "veo", 0.90))
+
+    tool, _ = VideoSelector()._select_best_tool(
+        {"preferred_provider": "nonexistent"}, [veo], {}
+    )
+    assert tool.name == "veo_video"
+
+
+# ---------------------------------------------------------------------------
+# #7 — fallback_tools gate for motion-required briefs
+# ---------------------------------------------------------------------------
+
+def test_fallback_excludes_image_selector_for_image_to_video():
+    sel = VideoSelector()
+    fallback = sel.fallback_tools_for({"operation": "image_to_video"})
+    assert "image_selector" not in fallback
+
+
+def test_fallback_excludes_image_selector_for_reference_to_video():
+    sel = VideoSelector()
+    fallback = sel.fallback_tools_for({"operation": "reference_to_video"})
+    assert "image_selector" not in fallback
+
+
+def test_fallback_keeps_image_selector_for_text_to_video():
+    """A still-image degraded fallback is acceptable for a non-motion brief."""
+    sel = VideoSelector()
+    fallback = sel.fallback_tools_for({"operation": "text_to_video"})
+    assert "image_selector" in fallback
+
+
+def test_static_fallback_tools_property_still_lists_image_selector():
+    """The input-agnostic property preserves the old shape for external consumers."""
+    assert "image_selector" in VideoSelector().fallback_tools
+
+
+# ---------------------------------------------------------------------------
+# #10 — estimate_cost / estimate_runtime delegate to the selected provider
+# ---------------------------------------------------------------------------
+
+def test_estimate_cost_uses_selected_provider(rankings):
+    veo = _StubTool("veo_video", "veo", cost=0.42)
+    kling = _StubTool("kling_video", "kling", cost=0.99)
+    rankings.append(_ScoreStub("veo_video", "veo", 0.90))
+    rankings.append(_ScoreStub("kling_video", "kling", 0.50))
+
+    sel = VideoSelector()
+    sel._providers = lambda: [veo, kling]  # type: ignore[assignment]
+    assert sel.estimate_cost({"prompt": "x"}) == pytest.approx(0.42)
+
+
+def test_estimate_runtime_uses_selected_provider(rankings):
+    veo = _StubTool("veo_video", "veo", runtime=123.0)
+    rankings.append(_ScoreStub("veo_video", "veo", 0.90))
+
+    sel = VideoSelector()
+    sel._providers = lambda: [veo]  # type: ignore[assignment]
+    assert sel.estimate_runtime({"prompt": "x"}) == pytest.approx(123.0)
+
+
+def test_estimate_cost_zero_when_no_providers():
+    sel = VideoSelector()
+    sel._providers = lambda: []  # type: ignore[assignment]
+    assert sel.estimate_cost({"prompt": "x"}) == 0.0

tools/video/video_selector.py

@@ -14,7 +14,7 @@ from tools.base_tool import BaseTool, ToolResult, ToolRuntime, ToolStability, To
 
 class VideoSelector(BaseTool):
     name = "video_selector"
-    version = "0.3.0"
+    version = "0.3.1"
     tier = ToolTier.GENERATE
     capability = "video_generation"
     provider = "selector"
@@ -22,6 +22,12 @@ class VideoSelector(BaseTool):
     runtime = ToolRuntime.HYBRID
     agent_skills = ["ai-video-gen", "create-video", "ltx2"]
 
+    # Operations that REQUIRE motion: an image-only tool (image_selector) is not
+    # an acceptable last-resort fallback for these, so fallback_tools_for() drops it.
+    MOTION_REQUIRED_OPERATIONS = frozenset({"image_to_video", "reference_to_video"})
+    # Default score gap for the preferred_provider override (see input_schema).
+    PREFERRED_PROVIDER_GAP = 0.15
+
     capabilities = [
         "text_to_video", "image_to_video", "stock_video",
         "provider_selection", "search_video", "download_video",
@@ -48,6 +54,19 @@ class VideoSelector(BaseTool):
                 "description": "Provider name or 'auto'. Valid values are discovered at runtime from the registry.",
                 "default": "auto",
             },
+            "preferred_provider_gap": {
+                "type": "number",
+                "minimum": 0,
+                "maximum": 1,
+                "default": 0.15,
+                "description": (
+                    "Max weighted-score gap (0-1) within which an explicit preferred_provider "
+                    "overrides the top-ranked provider. If the preferred provider's best score "
+                    "falls more than this far below the overall top, the preference is ignored "
+                    "and the top-ranked provider wins. Default 0.15 — honors a preference unless "
+                    "it would drag selection to a drastically worse provider."
+                ),
+            },
             "allowed_providers": {"type": "array", "items": {"type": "string"}},
             "operation": {
                 "type": "string",
@@ -141,9 +160,29 @@ class VideoSelector(BaseTool):
 
     @property
     def fallback_tools(self) -> list[str]:
-        """Dynamically built from discovered providers + image_selector as last resort."""
+        """Static (input-agnostic) fallback list for external consumers / contracts.
+
+        See :meth:`fallback_tools_for` for the input-aware form used during
+        routing, which drops ``image_selector`` for motion-required briefs.
+        """
         return [t.name for t in self._providers()] + ["image_selector"]
 
+    def fallback_tools_for(self, inputs: dict[str, object]) -> list[str]:
+        """Input-aware fallback list used during routing.
+
+        ``image_selector`` is a legitimate degraded last-resort for a still-image
+        brief (text_to_video with no motion requirement), but for motion-required
+        operations (image_to_video / reference_to_video) an image-only fallback
+        silently defeats the brief. Gate it here at the selector layer so a direct
+        caller — with no director skill enforcing the prohibition — still cannot
+        fall back to an image tool when motion was requested.
+        """
+        tools = [t.name for t in self._providers()]
+        operation = inputs.get("operation", "text_to_video")
+        if operation in self.MOTION_REQUIRED_OPERATIONS:
+            return tools
+        return tools + ["image_selector"]
+
     @property
     def provider_matrix(self) -> dict[str, dict[str, str]]:
         """Built at runtime from each provider's best_for field."""
@@ -228,6 +267,8 @@ class VideoSelector(BaseTool):
                 t.name for t in candidates
                 if t.name != tool.name and t.get_status().value == "available"
             ]
+            # Input-aware fallback list (drops image_selector for motion-required briefs).
+            result.data.setdefault("fallback_tools", self.fallback_tools_for(inputs))
         return result
 
     def _select_best_tool(
@@ -263,23 +304,41 @@ class VideoSelector(BaseTool):
 
         rankings = rank_providers(candidates, task_context)
 
-        # Build tool lookup: provider → tool (first selectable per provider)
-        tool_by_provider: dict[str, BaseTool] = {}
-        for tool in candidates:
-            if tool.provider not in tool_by_provider and self._tool_selectable(tool, inputs):
-                tool_by_provider[tool.provider] = tool
+        # Selectable tools, keyed by NAME (not provider). Keying by provider
+        # string shadowed one of two tools that legitimately share a provider —
+        # e.g. seedance_video (fal) and seedance_replicate both have
+        # provider="seedance", so only the first-registered was ever reachable.
+        # Keying by name keeps every backend selectable; ranking picks the best.
+        selectable_by_name: dict[str, BaseTool] = {
+            tool.name: tool for tool in candidates if self._tool_selectable(tool, inputs)
+        }
 
-        # If a preferred provider is explicitly requested and available,
-        # boost it to the top unless its score is drastically worse.
-        if preferred != "auto":
-            for score in rankings:
-                if score.provider == preferred and score.provider in tool_by_provider:
-                    return tool_by_provider[score.provider], score
+        def _tool_for(score: object) -> BaseTool | None:
+            return selectable_by_name.get(getattr(score, "tool_name", None))
 
-        # Return the highest-scored available provider
+        # If a preferred provider is explicitly requested, honor it ONLY when its
+        # best ranked tool is within a configurable score gap of the overall top.
+        # The prior code returned the preferred provider on the first ranking
+        # match regardless of how far below the top it scored (the comment
+        # claimed "unless drastically worse" but no gate enforced it).
+        if preferred != "auto" and rankings:
+            try:
+                gap = float(inputs.get("preferred_provider_gap", self.PREFERRED_PROVIDER_GAP))
+            except (TypeError, ValueError):
+                gap = self.PREFERRED_PROVIDER_GAP
+            top_score = rankings[0].weighted_score
+            preferred_score = next(
+                (s for s in rankings if s.provider == preferred and _tool_for(s) is not None),
+                None,
+            )
+            if preferred_score is not None and preferred_score.weighted_score >= top_score - gap:
+                return _tool_for(preferred_score), preferred_score
+
+        # Return the highest-scored selectable provider
         for score in rankings:
-            if score.provider in tool_by_provider:
-                return tool_by_provider[score.provider], score
+            tool = _tool_for(score)
+            if tool is not None:
+                return tool, score
 
         return None, None