diff --git a/tests/tools/test_video_selector_routing.py b/tests/tools/test_video_selector_routing.py new file mode 100644 index 00000000..162e907d --- /dev/null +++ b/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 diff --git a/tools/video/video_selector.py b/tools/video/video_selector.py index 8e9a9412..a651cd82 100644 --- a/tools/video/video_selector.py +++ b/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