diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index 85a62aeea70..a777db1f1e3 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -102,6 +102,7 @@ CONFIGURABLE_TOOLSETS = [ ("video", "🎬 Video Analysis", "video_analyze (requires video-capable model)"), ("image_gen", "🎨 Image Generation", "image_generate"), ("video_gen", "🎬 Video Generation", "video_generate (text/image/reference)"), + ("bfl", "🎬 BFL FLUX 3 Video", "bfl_flux3_*"), ("x_search", "🐦 X (Twitter) Search", "x_search (requires xAI OAuth or XAI_API_KEY)"), ("tts", "πŸ”Š Text-to-Speech", "text_to_speech"), ("stt", "πŸŽ™οΈ Speech-to-Text", "voice transcription (gateway voice messages + voice mode)"), diff --git a/tests/tools/test_flux3_video_tool.py b/tests/tools/test_flux3_video_tool.py new file mode 100644 index 00000000000..caab52aa28e --- /dev/null +++ b/tests/tools/test_flux3_video_tool.py @@ -0,0 +1,568 @@ +"""Native BFL FLUX 3 tools: gating, transport, media delivery, redaction.""" + +import asyncio +import base64 +import json +from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from tools import flux3_video_tool as flux3 + +GATEWAY = "https://tool-gateway.example.com" +BASE_URL = f"{GATEWAY}/api/bfl" +UPLOAD_PATH = "/api/uploads/bfl" + +_PNG = base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" +) + + +@pytest.fixture(autouse=True) +def _endpoints(): + """Every test runs as if the mount is reachable unless it says otherwise.""" + with patch.object( + flux3, + "managed_vendor_endpoints", + return_value={"origin": GATEWAY, "base_url": BASE_URL, "upload_path": UPLOAD_PATH}, + ): + yield + + +class _FakeResponse: + def __init__(self, status_code=200, payload=None, text=""): + self.status_code = status_code + self._payload = payload + self.text = text or (json.dumps(payload) if payload is not None else "") + + def json(self): + if self._payload is None: + raise ValueError("no json") + return self._payload + + +class _FakeClient: + """Captures the one request each handler makes.""" + + def __init__(self, response, sink): + self._response = response + self._sink = sink + + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc): + return False + + async def request(self, method, url, headers=None, json=None): + self._sink.append({"method": method, "url": url, "headers": headers or {}, "json": json}) + if isinstance(self._response, Exception): + raise self._response + return self._response + + +class _FakeStream: + """A streaming GET that yields `body` in one chunk.""" + + def __init__(self, body, status_code=200): + self._body = body + self.status_code = status_code + + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc): + return False + + def raise_for_status(self): + if self.status_code >= 400: + raise RuntimeError(f"HTTP {self.status_code}") + + async def aiter_bytes(self): + yield self._body + + +@contextmanager +def _fake_download(body, status_code=200): + """Stub the clip download; yields the list of URLs that were fetched. + + Patched at `create_ssrf_safe_async_client` rather than at httpx, which both + stubs the transport and asserts the download goes through the SSRF-guarded + client β€” the URL is vendor-supplied and fetched from the user's machine. + """ + from tools import url_safety + + fetched = [] + + class _Client: + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc): + return False + + def stream(self, _method, url): + fetched.append(url) + return _FakeStream(body, status_code) + + with patch.object(url_safety, "create_ssrf_safe_async_client", lambda **_kw: _Client()): + yield fetched + + +def _run(coro): + return asyncio.run(coro) + + +def _call(handler, args, response, headers=None): + """Invoke a handler with the transport stubbed; returns (parsed, requests).""" + sink = [] + import httpx + + with patch.object( + flux3, + "managed_gateway_auth_headers", + return_value=headers if headers is not None else {"Authorization": "Bearer nous-token"}, + ), patch.object(httpx, "AsyncClient", lambda **_kw: _FakeClient(response, sink)): + raw = _run(handler(args)) + return json.loads(raw), sink + + +class TestGating: + def test_hidden_without_a_reachable_mount(self): + with patch.object(flux3, "managed_vendor_endpoints", return_value=None): + assert flux3.check_bfl_requirements() is False + + def test_hidden_without_paid_service_access(self): + # The free tool pool does not fund BFL, so a pool-only user must never + # see the tools rather than see them and be refused. + account = SimpleNamespace(logged_in=True, paid_service_access=False, tool_gateway_entitled=True) + with patch("hermes_cli.nous_account.get_nous_portal_account_info", return_value=account): + assert flux3.check_bfl_requirements() is False + + def test_hidden_when_logged_out(self): + account = SimpleNamespace(logged_in=False, paid_service_access=False) + with patch("hermes_cli.nous_account.get_nous_portal_account_info", return_value=account): + assert flux3.check_bfl_requirements() is False + + def test_visible_for_a_paid_portal_account(self): + account = SimpleNamespace(logged_in=True, paid_service_access=True) + with patch("hermes_cli.nous_account.get_nous_portal_account_info", return_value=account): + assert flux3.check_bfl_requirements() is True + + def test_fails_closed_when_the_account_probe_raises(self): + with patch("hermes_cli.nous_account.get_nous_portal_account_info", side_effect=RuntimeError("portal down")): + assert flux3.check_bfl_requirements() is False + + +class TestSubmitTransport: + def test_text_to_video_posts_the_mode_and_arguments(self): + response = _FakeResponse(200, {"id": "bfl_job_1", "status": "submitted", "guidance": "Poll bfl_flux3_get_result with id=bfl_job_1"}) + + parsed, requests = _call( + flux3._handle_text_to_video, + {"prompt": "a lake", "aspect_ratio": "16:9", "duration": 5}, + response, + ) + + assert requests[0]["method"] == "POST" + assert requests[0]["url"] == f"{BASE_URL}/generations" + assert requests[0]["json"] == { + "prompt": "a lake", + "aspect_ratio": "16:9", + "duration": 5, + "mode": "text_to_video", + } + assert requests[0]["headers"]["Authorization"] == "Bearer nous-token" + # The gateway's guidance is the model-facing text, verbatim. + assert parsed["result"] == "Poll bfl_flux3_get_result with id=bfl_job_1" + assert parsed["details"]["id"] == "bfl_job_1" + + def test_each_generate_tool_sends_its_own_mode(self): + for handler, args, mode in [ + (flux3._handle_text_to_video, {"prompt": "a"}, "text_to_video"), + (flux3._handle_image_to_video, {"prompt": "a", "input_image": "https://x/a.png"}, "image_to_video"), + ( + flux3._handle_keyframes_to_video, + {"prompt": "a", "input_images": ["https://x/a.png"], "keyframe_indices": [0]}, + "keyframes_to_video", + ), + (flux3._handle_video_continuation, {"prompt": "a", "input_video": "https://x/c.mp4"}, "video_continuation"), + ]: + _parsed, requests = _call(handler, args, _FakeResponse(200, {"id": "j", "status": "submitted", "guidance": "ok"})) + assert requests[0]["json"]["mode"] == mode + + def test_urls_pass_through_without_an_upload(self): + # Forwarding a URL is cheaper than downloading and re-uploading it. + _parsed, requests = _call( + flux3._handle_image_to_video, + {"prompt": "a", "input_image": "https://example.com/a.png"}, + _FakeResponse(200, {"id": "j", "status": "submitted", "guidance": "ok"}), + ) + assert requests[0]["json"]["input_image"] == "https://example.com/a.png" + + def test_a_refusal_is_surfaced_as_the_tools_result_text(self): + # Throttles are designed to be hit: the message is written for the + # model and must reach it intact, with the machine detail alongside. + response = _FakeResponse( + 429, + { + "error": { + "code": "BFL_GENERATION_COOLDOWN", + "message": "A new BFL video generation may be started once every 5 minutes. Wait 210 seconds.", + "details": {"retryAfterSeconds": 210}, + } + }, + ) + + parsed, _requests = _call(flux3._handle_text_to_video, {"prompt": "a"}, response) + + assert parsed["error"] == "A new BFL video generation may be started once every 5 minutes. Wait 210 seconds." + assert parsed["details"] == {"retryAfterSeconds": 210} + + def test_a_401_asks_for_a_nous_sign_in(self): + parsed, _requests = _call(flux3._handle_text_to_video, {"prompt": "a"}, _FakeResponse(401, {"error": {"code": "AUTH_ERROR"}})) + + assert parsed["needs_reauth"] is True + assert "sign in" in parsed["error"].lower() + + def test_missing_credentials_ask_for_a_sign_in_without_calling_out(self): + parsed, requests = _call(flux3._handle_text_to_video, {"prompt": "a"}, _FakeResponse(200, {}), headers={}) + + assert requests == [] + assert "sign in" in parsed["error"].lower() + + def test_a_transport_failure_reports_the_cause(self): + parsed, _requests = _call( + flux3._handle_text_to_video, + {"prompt": "a"}, + RuntimeError("connect failed"), + ) + + assert "Could not reach the video-generation gateway" in parsed["error"] + assert "connect failed" in parsed["error"] + + def test_an_unreadable_body_does_not_masquerade_as_success(self): + parsed, _requests = _call(flux3._handle_text_to_video, {"prompt": "a"}, _FakeResponse(502, None, text="upstream exploded")) + + assert "error" in parsed + + +class TestPollTransport: + def test_poll_gets_the_job_and_returns_guidance(self): + response = _FakeResponse(200, {"id": "bfl_job_1", "status": "Generating", "guidance": "Still going."}) + + parsed, requests = _call(flux3._handle_get_result, {"id": "bfl_job_1"}, response) + + assert requests[0]["method"] == "GET" + assert requests[0]["url"] == f"{BASE_URL}/generations/bfl_job_1" + assert requests[0]["json"] is None + assert parsed["result"] == "Still going." + assert parsed["details"]["status"] == "Generating" + + def test_ready_saves_the_clip_and_never_returns_the_signed_url(self, tmp_path): + # The signed URL is a bearer credential for the clip and it used to be + # re-keyed into a shell command by hand, dropping characters. Neither + # can happen if the model never sees it. + signed = "https://cdn.example/container/flux3-clip.mp4?sig=abc%2Bdef%3D&se=2026" + response = _FakeResponse(200, {"id": "bfl_job_1", "status": "Ready", "result": {"sample": signed}, "guidance": "Deliver the saved file."}) + + with _fake_download(b"x" * (128 * 1024)) as fetched: + parsed, _requests = _call( + flux3._handle_get_result, + {"id": "bfl_job_1", "save_to": str(tmp_path)}, + response, + ) + + saved = tmp_path / "flux3-clip.mp4" + assert saved.read_bytes() == b"x" * (128 * 1024) + assert parsed["details"]["saved_path"] == str(saved) + assert parsed["details"]["result"].get("sample") is None + assert signed not in json.dumps(parsed) + # The gateway still owns the delivery wording; the client only supplies + # the path it cannot know. + assert parsed["result"].startswith(f"Saved to {saved}.") + assert "Deliver the saved file." in parsed["result"] + assert fetched == [signed] + + def test_ready_never_overwrites_an_existing_file(self, tmp_path): + (tmp_path / "flux3-clip.mp4").write_bytes(b"an earlier clip") + response = _FakeResponse(200, {"id": "bfl_job_1", "status": "Ready", "result": {"sample": "https://cdn.example/x/flux3-clip.mp4?sig=a"}, "guidance": "g"}) + + with _fake_download(b"y" * (128 * 1024)): + parsed, _requests = _call(flux3._handle_get_result, {"id": "bfl_job_1", "save_to": str(tmp_path)}, response) + + assert parsed["details"]["saved_path"] == str(tmp_path / "flux3-clip-2.mp4") + assert (tmp_path / "flux3-clip.mp4").read_bytes() == b"an earlier clip" + + def test_a_rejected_download_fails_loudly_and_leaves_no_file(self, tmp_path): + # The original bug: a bad signature returns an XML error body, curl + # writes it to the .mp4 and exits 0, and it reads as success. A short + # body is not a video whatever the status code said. + response = _FakeResponse(200, {"id": "bfl_job_1", "status": "Ready", "result": {"sample": "https://cdn.example/x/flux3-clip.mp4?sig=bad"}, "guidance": "g"}) + + with _fake_download(b"AuthenticationFailed"): + parsed, _requests = _call(flux3._handle_get_result, {"id": "bfl_job_1", "save_to": str(tmp_path)}, response) + + assert "saving it failed" in parsed["result"] + assert "Poll this job again" in parsed["result"] + # Neither a half-written .part nor a plausible-looking .mp4 survives. + assert [p.name for p in tmp_path.glob("*.mp4*")] == [] + assert "saved_path" not in parsed["details"] + + def test_poll_requires_an_id(self): + parsed, requests = _call(flux3._handle_get_result, {}, _FakeResponse(200, {})) + + assert "id is required" in parsed["error"] + assert requests == [] + + def test_poll_url_encodes_the_job_id(self): + _parsed, requests = _call( + flux3._handle_get_result, + {"id": "weird/../id"}, + _FakeResponse(200, {"id": "x", "guidance": "ok"}), + ) + assert requests[0]["url"] == f"{BASE_URL}/generations/weird%2F..%2Fid" + + +class TestMediaDelivery: + def _resolved(self, mime="image/png", data=_PNG): + return SimpleNamespace(data=data, mime=mime) + + def test_a_local_path_is_uploaded_and_replaced_with_a_reference(self): + async def fake_uploader(data, mime): + assert data == _PNG + assert mime == "image/png" + return "nous-upload:token-1" + + with patch.object(flux3, "build_managed_media_uploader", return_value=fake_uploader), patch( + "tools.image_source.resolve_image_source", return_value=self._resolved() + ) as resolve: + _parsed, requests = _call( + flux3._handle_image_to_video, + {"prompt": "a", "input_image": "/tmp/frame.png"}, + _FakeResponse(200, {"id": "j", "status": "submitted", "guidance": "ok"}), + ) + + assert requests[0]["json"]["input_image"] == "nous-upload:token-1" + # Images and video ride the same safety pipeline; only the permitted + # type differs, and an image field must not accept a video. + assert resolve.call_args.kwargs["permitted"] == ("image",) + + def test_video_fields_permit_video_only(self): + async def fake_uploader(data, mime): + return "nous-upload:token-v" + + with patch.object(flux3, "build_managed_media_uploader", return_value=fake_uploader), patch( + "tools.image_source.resolve_image_source", return_value=self._resolved("video/mp4", b"\x00\x00\x00\x18ftypmp42") + ) as resolve: + _parsed, requests = _call( + flux3._handle_video_continuation, + {"prompt": "a", "input_video": "/tmp/clip.mp4"}, + _FakeResponse(200, {"id": "j", "status": "submitted", "guidance": "ok"}), + ) + + assert requests[0]["json"]["input_video"] == "nous-upload:token-v" + assert resolve.call_args.kwargs["permitted"] == ("video",) + + def test_every_keyframe_path_is_uploaded(self): + uploads = [] + + async def fake_uploader(data, mime): + uploads.append(mime) + return f"nous-upload:token-{len(uploads)}" + + with patch.object(flux3, "build_managed_media_uploader", return_value=fake_uploader), patch( + "tools.image_source.resolve_image_source", return_value=self._resolved() + ): + _parsed, requests = _call( + flux3._handle_keyframes_to_video, + {"prompt": "a", "input_images": ["/tmp/a.png", "https://x/b.png", "/tmp/c.png"], "keyframe_indices": [0, 24, 48]}, + _FakeResponse(200, {"id": "j", "status": "submitted", "guidance": "ok"}), + ) + + # The URL in the middle is forwarded untouched. + assert requests[0]["json"]["input_images"] == [ + "nous-upload:token-1", + "https://x/b.png", + "nous-upload:token-2", + ] + + def test_a_list_valued_input_image_is_still_uploaded(self): + # The gateway accepts input_image as a string OR a list, so a list of + # local paths must not slip past unsanitized β€” that would send raw + # filesystem paths to the vendor and disclose the user's directories. + async def fake_uploader(data, mime): + return "nous-upload:token-1" + + with patch.object(flux3, "build_managed_media_uploader", return_value=fake_uploader), patch( + "tools.image_source.resolve_image_source", return_value=self._resolved() + ): + _parsed, requests = _call( + flux3._handle_image_to_video, + {"prompt": "a", "input_image": ["/tmp/frame.png"]}, + _FakeResponse(200, {"id": "j", "status": "submitted", "guidance": "ok"}), + ) + + assert requests[0]["json"]["input_image"] == ["nous-upload:token-1"] + assert "/tmp/frame.png" not in json.dumps(requests[0]["json"]) + + def test_media_fields_are_sanitized_whatever_the_mode_expects(self): + # The gateway prefers input_image over input_images, so sanitizing only + # the field this mode documents would let the other one through. + uploads = [] + + async def fake_uploader(data, mime): + uploads.append(mime) + return f"nous-upload:token-{len(uploads)}" + + with patch.object(flux3, "build_managed_media_uploader", return_value=fake_uploader), patch( + "tools.image_source.resolve_image_source", return_value=self._resolved() + ): + _parsed, requests = _call( + flux3._handle_keyframes_to_video, + { + "prompt": "a", + "input_images": ["https://x/b.png"], + "input_image": "/tmp/sneaky.png", + "keyframe_indices": [0], + }, + _FakeResponse(200, {"id": "j", "status": "submitted", "guidance": "ok"}), + ) + + body = json.dumps(requests[0]["json"]) + assert "/tmp/sneaky.png" not in body + assert requests[0]["json"]["input_image"] == "nous-upload:token-1" + + def test_text_to_video_strips_media_fields_instead_of_uploading_them(self): + # The mode takes no media, so an upload would spend the caller's quota + # on a value the gateway ignores. + def _must_not_upload(*_args, **_kwargs): + raise AssertionError("text-to-video must not upload anything") + + with patch.object(flux3, "build_managed_media_uploader", _must_not_upload): + _parsed, requests = _call( + flux3._handle_text_to_video, + {"prompt": "a", "input_image": "/tmp/frame.png"}, + _FakeResponse(200, {"id": "j", "status": "submitted", "guidance": "ok"}), + ) + + assert "input_image" not in requests[0]["json"] + assert requests[0]["json"]["mode"] == "text_to_video" + + def test_an_over_long_image_list_is_refused_before_any_upload(self): + def _must_not_upload(*_args, **_kwargs): + raise AssertionError("an over-long list must be refused before uploading") + + with patch.object(flux3, "build_managed_media_uploader", _must_not_upload): + parsed, requests = _call( + flux3._handle_keyframes_to_video, + {"prompt": "a", "input_images": [f"/tmp/{i}.png" for i in range(11)], "keyframe_indices": [0]}, + _FakeResponse(200, {}), + ) + + assert "at most 10" in parsed["error"] + assert requests == [] + + def test_an_upload_refusal_becomes_the_tools_error(self): + async def failing_uploader(data, mime): + raise RuntimeError("the daily upload budget for this account is exhausted") + + with patch.object(flux3, "build_managed_media_uploader", return_value=failing_uploader), patch( + "tools.image_source.resolve_image_source", return_value=self._resolved() + ): + parsed, requests = _call( + flux3._handle_image_to_video, + {"prompt": "a", "input_image": "/tmp/frame.png"}, + _FakeResponse(200, {}), + ) + + assert "daily upload budget" in parsed["error"] + # A failed upload must not reach the gateway as a bare local path. + assert requests == [] + + def test_an_unreadable_file_is_reported_without_dumping_the_value(self): + from tools.image_source import SourceNotFound + + with patch.object(flux3, "build_managed_media_uploader", return_value=lambda *a: None), patch( + "tools.image_source.resolve_image_source", side_effect=SourceNotFound("media file not found", src="/tmp/x.png") + ): + parsed, requests = _call( + flux3._handle_image_to_video, + # Long, but unmistakably a path (dots and dashes are outside + # the base64 alphabet, so the payload guard leaves it alone). + {"prompt": "a", "input_image": "/tmp/" + "a-b." * 2000 + "frame.png"}, + _FakeResponse(200, {}), + ) + + assert "error" in parsed + # The offending value is truncated: echoing it whole would blow up the + # model's context on the way to reporting a bad path. + assert len(parsed["error"]) < 500 + assert requests == [] + + +class TestLocalPathDetection: + @pytest.mark.parametrize( + "value", + ["/tmp/frame.png", "~/Pictures/f.png", "./f.png", "../f.png", "file:///tmp/f.png", r"C:\Users\me\f.png", r"\\nas\share\f.png"], + ) + def test_rooted_paths_are_read_off_disk(self, value): + assert flux3._looks_like_local_path(value) is True + + @pytest.mark.parametrize( + "value", + [ + "frame.png", + "https://example.com/f.png", + "nous-upload:eyJhbGciOiJIUzI1NiJ9.e30.sig", + "C:frame.png", + # Inline base64 of a JPEG always starts "/9j/" (first byte 0xFF), + # which must not read as an absolute POSIX path. + "/9j/4AAQSkZJRgABAQAAAQ" + "A" * 300 + "==", + ], + ) + def test_ambiguous_and_remote_values_are_forwarded(self, value): + assert flux3._looks_like_local_path(value) is False + + def test_a_short_base64_lookalike_path_is_still_a_path(self): + assert flux3._looks_like_local_path("/tmp/frames/a1") is True + + +class TestSchemas: + def test_every_tool_is_registered_under_the_bfl_toolset(self): + from tools.registry import registry + + for name in [ + "bfl_flux3_text_to_video", + "bfl_flux3_image_to_video", + "bfl_flux3_keyframes_to_video", + "bfl_flux3_video_continuation", + "bfl_flux3_get_result", + "bfl_flux3_prompting_guide", + ]: + entry = registry.get_entry(name) + assert entry is not None, f"{name} is not registered" + assert entry.toolset == "bfl" + assert entry.check_fn is flux3.check_bfl_requirements + + def test_generate_tools_point_at_the_guide_and_the_poll_tool(self): + # Descriptions are the only text guaranteed to be in context when a + # model picks a tool, so the pointers live there. + for schema in [flux3.TEXT_TO_VIDEO_SCHEMA, flux3.IMAGE_TO_VIDEO_SCHEMA, flux3.KEYFRAMES_TO_VIDEO_SCHEMA, flux3.VIDEO_CONTINUATION_SCHEMA]: + assert "bfl_flux3_prompting_guide" in schema["description"] + assert "bfl_flux3_get_result" in schema["description"] + + def test_the_guide_covers_the_methodology_without_pinning_server_policy(self): + guide = flux3.FLUX3_PROMPTING_GUIDE + assert "grounding" in guide.lower() + assert "bfl_flux3_get_result" in guide + # Waits and limits ship live in the gateway's responses; pinning them + # here would let the client lie about what the server enforces. + assert "5 minutes" not in guide + assert "per minute" not in guide + + def test_the_guide_tool_takes_no_arguments_and_calls_nothing(self): + assert flux3.PROMPTING_GUIDE_SCHEMA["parameters"]["properties"] == {} + assert _run(flux3._handle_prompting_guide({})) == flux3.FLUX3_PROMPTING_GUIDE diff --git a/tests/tools/test_managed_tool_gateway.py b/tests/tools/test_managed_tool_gateway.py index a1aac581aeb..1757d5141fa 100644 --- a/tests/tools/test_managed_tool_gateway.py +++ b/tests/tools/test_managed_tool_gateway.py @@ -6,6 +6,8 @@ from pathlib import Path import sys from unittest.mock import patch +import pytest + MODULE_PATH = Path(__file__).resolve().parents[2] / "tools" / "managed_tool_gateway.py" MODULE_SPEC = spec_from_file_location("managed_tool_gateway_test_module", MODULE_PATH) assert MODULE_SPEC and MODULE_SPEC.loader @@ -35,6 +37,147 @@ def test_resolve_managed_tool_gateway_derives_vendor_origin_from_shared_domain() assert result.managed_mode is True +def test_resolve_managed_tool_gateway_uses_vendor_specific_override(): + with patch.dict( + os.environ, + { + "BROWSER_USE_GATEWAY_URL": "http://browser-use-gateway.localhost:3009/", + }, + clear=False, + ), patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + result = resolve_managed_tool_gateway( + "browser-use", + token_reader=lambda: "nous-token", + ) + + assert result is not None + assert result.gateway_origin == "http://browser-use-gateway.localhost:3009" + + +def test_resolve_managed_tool_gateway_is_inactive_without_nous_token(): + with patch.dict( + os.environ, + { + "TOOL_GATEWAY_DOMAIN": "nousresearch.com", + }, + clear=False, + ), patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + result = resolve_managed_tool_gateway( + "firecrawl", + token_reader=lambda: None, + ) + + assert result is None + + +def test_resolve_managed_tool_gateway_is_disabled_without_subscription(): + with patch.dict(os.environ, {"TOOL_GATEWAY_DOMAIN": "nousresearch.com"}, clear=False), \ + patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=False): + result = resolve_managed_tool_gateway( + "firecrawl", + token_reader=lambda: "nous-token", + ) + + assert result is None + + +def test_read_nous_access_token_refreshes_expiring_cached_token(tmp_path, monkeypatch): + monkeypatch.delenv("TOOL_GATEWAY_USER_TOKEN", raising=False) + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + expires_at = (datetime.now(timezone.utc) + timedelta(seconds=30)).isoformat() + (tmp_path / "auth.json").write_text(json.dumps({ + "providers": { + "nous": { + "access_token": "stale-token", + "refresh_token": "refresh-token", + "expires_at": expires_at, + } + } + })) + monkeypatch.setattr( + "hermes_cli.auth.resolve_nous_access_token", + lambda refresh_skew_seconds=120: "fresh-token", + ) + + assert managed_tool_gateway.read_nous_access_token() == "fresh-token" + + +def test_managed_vendor_endpoints_pin_the_deployed_gateway_url(): + """The exact URL an agent may connect to is a code fact, not a lookup. + + Exercises the real ``build_vendor_gateway_url`` (which once resolved a + typo'd pseudo-vendor to a non-existent host while every other test stubbed + it): default builder, real deployed host, pinned vendor path. + """ + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True), \ + patch.dict( + os.environ, + {"TOOL_GATEWAY_DOMAIN": "nousresearch.com", "TOOL_GATEWAY_SCHEME": "https"}, + clear=False, + ): + os.environ.pop("TOOL_GATEWAY_URL", None) + endpoints = managed_tool_gateway.managed_vendor_endpoints("bfl") + + assert endpoints == { + "origin": "https://tool-gateway.nousresearch.com", + "base_url": "https://tool-gateway.nousresearch.com/api/bfl", + "upload_path": "/api/uploads/bfl", + } + + +def test_managed_vendor_endpoints_unreachable_when_managed_tools_disabled(): + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=False): + assert managed_tool_gateway.managed_vendor_endpoints("bfl") is None + + +def test_managed_gateway_auth_headers_carry_the_bearer(): + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + headers = managed_tool_gateway.managed_gateway_auth_headers( + "https://tool-gateway.example.com/api/bfl/generations", + gateway_builder=lambda vendor: f"https://{vendor}-gateway.example.com", + token_reader=lambda: "nous-token", + ) + + assert headers == {"Authorization": "Bearer nous-token"} + + +def test_managed_gateway_auth_headers_reflect_a_rotated_token(): + # Read fresh on every call: a Nous access token expires within the hour, + # and a long session must not keep presenting a dead bearer. + tokens = iter(["first-token", "second-token"]) + builder = lambda vendor: f"https://{vendor}-gateway.example.com" + url = "https://tool-gateway.example.com/api/bfl/generations" + + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + first = managed_tool_gateway.managed_gateway_auth_headers(url, builder, lambda: next(tokens)) + second = managed_tool_gateway.managed_gateway_auth_headers(url, builder, lambda: next(tokens)) + + assert first["Authorization"] == "Bearer first-token" + assert second["Authorization"] == "Bearer second-token" + + +def test_managed_gateway_auth_headers_refuse_a_url_off_the_gateway_origin(): + # Gated on the URL, never a name: our bearer must never be handed to a + # host that merely looks managed. + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + assert managed_tool_gateway.managed_gateway_auth_headers( + "https://attacker.example/api/bfl/generations", + gateway_builder=lambda vendor: f"https://{vendor}-gateway.example.com", + token_reader=lambda: "nous-token", + ) == {} + + +def test_managed_gateway_auth_headers_empty_without_a_token(): + # Empty rather than raising, so a caller can say "sign in" instead of + # sending an unauthenticated request. + with patch.object(managed_tool_gateway, "managed_nous_tools_enabled", return_value=True): + assert managed_tool_gateway.managed_gateway_auth_headers( + "https://tool-gateway.example.com/api/bfl/generations", + gateway_builder=lambda vendor: f"https://{vendor}-gateway.example.com", + token_reader=lambda: None, + ) == {} + + def test_is_managed_tool_gateway_ready_skips_refresh_for_expired_cached_token(tmp_path, monkeypatch): monkeypatch.delenv("TOOL_GATEWAY_USER_TOKEN", raising=False) monkeypatch.setenv("HERMES_HOME", str(tmp_path)) diff --git a/tools/flux3_video_tool.py b/tools/flux3_video_tool.py new file mode 100644 index 00000000000..49dda1c86e9 --- /dev/null +++ b/tools/flux3_video_tool.py @@ -0,0 +1,914 @@ +"""Native BFL FLUX 3 video generation tools, backed by the Nous tool gateway. + +These are service-gated native tools in the ``image_generate`` mold: schemas +and descriptions are pinned here as build-time facts, the handlers speak the +gateway's own REST contract, and ``check_fn`` hides the whole toolset unless +the user is signed in to Nous Portal with paid service access. No runtime +discovery, and no server-supplied schema is ever consulted β€” that is the point +of the design. + +The wire is two calls against the gateway's managed mount, and it names the +vendor but not the vendor's API: + +- ``POST {base}/generations`` with ``{mode, prompt, ...}`` -> ``{id, status, + guidance}`` +- ``GET {base}/generations/`` -> the job state plus ``guidance`` + +``guidance`` is the gateway's live policy channel: exact waits, what to do +next, and how to deliver a finished clip ship from the server so they cannot +drift from what it actually enforces. Handlers surface it verbatim as the +tool's result text, and surface ``error.message`` the same way on a refusal. + +Media inputs: handlers know their own media fields explicitly. A local file +path is resolved through :func:`tools.image_source.resolve_image_source` +(sandbox confinement, credential guard) and delivered via the Nous upload +protocol (presign, direct PUT to storage, ``nous-upload:`` reference). +URLs pass through untouched. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import re +from typing import Optional + +from tools.registry import registry +from tools.managed_tool_gateway import ( + build_managed_media_uploader, + managed_gateway_auth_headers, + managed_vendor_endpoints, + read_nous_access_token, +) + +logger = logging.getLogger(__name__) + +_TOOLSET = "bfl" +_VENDOR = "bfl" + +# Submit sits behind the gateway's upstream call plus upload-reference +# resolution, and the gateway bounds a poll server-side. One generous read +# timeout covers both without ever approaching the agent's per-tool budget. +_TRANSPORT_READ_TIMEOUT_SECONDS = 180.0 +_TRANSPORT_CONNECT_TIMEOUT_SECONDS = 10.0 + +_SIGN_IN_MESSAGE = ( + "BFL video generation needs a Nous Portal sign-in with an active paid plan. " + "Ask the user to run `hermes model` and sign in to Nous, then retry." +) + +# --------------------------------------------------------------------------- +# Local-path detection (moved from the retired MCP media-argument walker) +# --------------------------------------------------------------------------- + +# Drive-absolute Windows paths: ``C:\Users\...`` / ``C:/Users/...``. +_WINDOWS_DRIVE_PATH = re.compile(r"^[A-Za-z]:[\\/]") + +# A whole string of base64 alphabet (optionally line-wrapped, optionally +# padded). Base64's alphabet includes "/", so an inline JPEG payload always +# starts with "/9j/" β€” which reads as an absolute path unless caught first. +_BASE64_PAYLOAD = re.compile(r"^[A-Za-z0-9+/\r\n]+={0,2}[\r\n]*$") +# Real filesystem paths are short; base64 of even a thumbnail runs to +# thousands of characters. Anything this long and alphabet-pure is a payload. +_MIN_BASE64_PAYLOAD_LENGTH = 256 + + +def _looks_like_local_path(value: str) -> bool: + """True for things we should read off disk rather than forward as-is. + + Deliberately narrow: only explicitly rooted paths and ``file://`` qualify. + Bare names are ambiguous with opaque ids, URLs pass through, and base64 + payloads (checked first β€” a JPEG's base64 starts ``/9j/``) fall through + untouched. + """ + if len(value) >= _MIN_BASE64_PAYLOAD_LENGTH and _BASE64_PAYLOAD.match(value): + return False + if value.startswith("file://"): + return True + if value == "~" or value.startswith(("~/", "~\\")): + return True + if value.startswith(("/", "./", "../", ".\\", "..\\")): + return True + if _WINDOWS_DRIVE_PATH.match(value) or value.startswith("\\\\"): + return True + return False + + +def _display_path(path: str) -> str: + """A value safe to embed in an error message shown to the model.""" + return path if len(path) <= 200 else f"{path[:200]}… ({len(path)} characters)" + + +# --------------------------------------------------------------------------- +# Gateway transport +# --------------------------------------------------------------------------- + +def _endpoints() -> Optional[dict]: + """Absolute URLs for the managed BFL routes, or None when unreachable.""" + return managed_vendor_endpoints(_VENDOR) + + +async def _call_gateway(method: str, url: str, json_body: Optional[dict] = None) -> str: + """One REST round trip, rendered as this tool's result. + + The gateway's ``guidance`` (on success) and ``error.message`` (on a + refusal) are both written for a model to act on, so they are surfaced + verbatim. A refusal is a normal outcome the model can respond to β€” being + throttled is not a broken tool β€” so only genuinely unreadable responses + become ``error``. + """ + import httpx + + headers = managed_gateway_auth_headers(url) + if not headers: + return json.dumps({"error": _SIGN_IN_MESSAGE}) + headers["Content-Type"] = "application/json" + + timeout = httpx.Timeout(_TRANSPORT_CONNECT_TIMEOUT_SECONDS, read=_TRANSPORT_READ_TIMEOUT_SECONDS) + try: + async with httpx.AsyncClient(timeout=timeout) as client: + response = await client.request(method, url, headers=headers, json=json_body) + except Exception as exc: + return json.dumps({ + "error": f"Could not reach the video-generation gateway: {type(exc).__name__}: {exc}", + }) + + if response.status_code == 401: + return json.dumps({"error": _SIGN_IN_MESSAGE, "needs_reauth": True}) + + try: + payload = response.json() + except Exception: + payload = None + + if not isinstance(payload, dict): + return json.dumps({ + "error": f"The video-generation gateway answered HTTP {response.status_code} with an unreadable body.", + }) + + if response.status_code >= 400: + error = payload.get("error") if isinstance(payload.get("error"), dict) else {} + message = error.get("message") or f"the gateway refused the request (HTTP {response.status_code})" + out: dict = {"error": str(message)} + if isinstance(error.get("details"), dict): + out["details"] = error["details"] + return json.dumps(out, ensure_ascii=False) + + guidance = payload.pop("guidance", None) + return json.dumps( + {"result": guidance or "Request accepted.", "details": payload}, + ensure_ascii=False, + ) + + +# --------------------------------------------------------------------------- +# Media delivery +# --------------------------------------------------------------------------- + +# Every media field the gateway understands, and the media kind each accepts. +# `input_image` and `input_images` are interchangeable server-side, so both +# must be sanitized whatever the mode. +_MEDIA_FIELDS = { + "input_image": ("image",), + "input_images": ("image",), + "input_video": ("video",), +} +# The vendor takes at most ten keyframes. Checked here as well as server-side +# so an over-long list is refused before it spends the caller's upload quota. +_MAX_IMAGES = 10 + + +def _warm_nous_token() -> None: + """Refresh the Nous token once, before any parallel upload needs it. + + ``read_nous_access_token`` takes no lock and, when a refresh fails, falls + back to returning the stale cached token. Uploading in parallel therefore + had every request discover the token was expiring at the same instant and + fire its own refresh; the rotating refresh token means the first wins and + the rest quietly send the stale bearer, which the gateway answers with 401. + One warm-up call first puts a fresh token in the cache, so the fan-out all + reads it instead of racing for it. + """ + try: + read_nous_access_token() + except Exception as exc: # pragma: no cover β€” the real read retries below + logger.debug("Nous token warm-up failed before parallel uploads: %s", exc) + + +async def _prepare_media(args: dict, task_id: Optional[str]) -> dict: + """Replace local paths with upload references in every media field. + + Deliberately covers all media fields rather than the one this mode + expects. The gateway accepts `input_image` and `input_images` + interchangeably, so a value left unsanitized in the "wrong" field still + reaches the vendor β€” as a raw local path, which both fails the generation + and discloses the user's directory layout to a third party. + """ + prepared = dict(args or {}) + _warm_nous_token() + for field, permitted in _MEDIA_FIELDS.items(): + value = prepared.get(field) + if value is None: + continue + if isinstance(value, list): + if len(value) > _MAX_IMAGES: + raise ValueError(f"{field} takes at most {_MAX_IMAGES} images; you passed {len(value)}.") + # Uploaded together rather than one after another: each entry is a + # presign round trip plus a full-file PUT, and ten keyframes in + # sequence took long enough that a submit looked hung. + prepared[field] = list( + await asyncio.gather(*(_deliver_media(entry, permitted, task_id) for entry in value)) + ) + else: + prepared[field] = await _deliver_media(value, permitted, task_id) + return prepared + + +def _without_media(args: dict) -> dict: + """Drop media fields entirely β€” for the mode that takes none. + + The gateway ignores them for text-to-video, so stripping here matches its + behaviour while avoiding an upload the caller never needed. + """ + return {k: v for k, v in dict(args or {}).items() if k not in _MEDIA_FIELDS} + + +async def _deliver_media(value, permitted: tuple, task_id: Optional[str]): + """Replace a local path with a ``nous-upload:`` reference; pass URLs through. + + Raises ``ValueError`` with a model-readable sentence when the file cannot + be read or uploaded β€” the caller turns that into the tool's error payload. + """ + if not isinstance(value, str) or not _looks_like_local_path(value): + return value + + from tools.image_source import ImageResolutionError, ResolveContext, resolve_image_source + + try: + resolved = await resolve_image_source(value, ResolveContext(task_id=task_id), permitted=permitted) + except ImageResolutionError as exc: + raise ValueError(f"Could not read {_display_path(value)}: {exc}") from exc + + endpoints = _endpoints() + uploader = None + if endpoints is not None: + uploader = build_managed_media_uploader(endpoints["base_url"], endpoints["upload_path"]) + if uploader is None: + raise ValueError("Local file uploads are not available in this build; pass a URL instead.") + + try: + return await uploader(resolved.data, resolved.mime) + except Exception as exc: + raise ValueError(f"Could not upload {_display_path(value)}: {exc}") from exc + + +# --------------------------------------------------------------------------- +# Saving the finished clip +# --------------------------------------------------------------------------- + +# Generous: a 50MB clip over a slow link, still well inside the agent's budget. +_DOWNLOAD_READ_TIMEOUT_SECONDS = 300.0 +_DOWNLOAD_CONNECT_TIMEOUT_SECONDS = 15.0 +# A rejection page is a few hundred bytes of XML; a clip is megabytes. Anything +# smaller than this is not the video, whatever the HTTP status said. +_MIN_PLAUSIBLE_VIDEO_BYTES = 64 * 1024 +# Enough collision suffixes to be useful, few enough to fail fast if something +# is generating files in a loop. +_MAX_FILENAME_ATTEMPTS = 50 + + +async def _save_if_ready(raw: str, save_to) -> str: + """Download a finished clip and swap the signed URL for a local path. + + The URL is handled here rather than by the model on purpose. It is long and + percent-encoded, and re-keying it into a shell command dropped characters + often enough to be the main failure mode of this tool β€” a corrupted + signature is rejected, but `curl` still writes the rejection body to the + output file and exits 0, so it read as success. Passing the exact string + from this response removes the transcription step entirely. + + It also keeps a bearer credential out of the transcript: the signed URL + grants the clip to anyone holding it for the next 15-60 minutes, and + returning it put it in the model's context and the saved conversation. The + gateway already scrubs presigned URLs for *input* media for exactly this + reason; this makes the output side match. + """ + try: + payload = json.loads(raw) + except Exception: + return raw + if not isinstance(payload, dict): + return raw + + details = payload.get("details") + if not isinstance(details, dict) or details.get("status") != "Ready": + return raw + + result = details.get("result") + url = result.get("sample") if isinstance(result, dict) else None + if not isinstance(url, str) or not url.strip(): + return raw + + # Dropped whether or not the save succeeds. A retry re-polls, which mints a + # fresh URL, so nothing is lost by not handing this one to the model. + result.pop("sample", None) + + try: + target, size = await _download_video(url.strip(), save_to) + except Exception as exc: + payload["result"] = ( + f"The clip finished but saving it failed: {type(exc).__name__}: {exc}. " + "Poll this job again to retry the download; the job itself is unaffected." + ) + return json.dumps(payload, ensure_ascii=False) + + details["saved_path"] = str(target) + details["saved_bytes"] = size + payload["result"] = f"Saved to {target}. " + str(payload.get("result") or "") + return json.dumps(payload, ensure_ascii=False) + + +async def _download_video(url: str, save_to) -> tuple: + """Stream the clip to disk, returning (path, bytes). + + SSRF-guarded, for the same reason the upload PUT is: this URL comes from + the vendor by way of the gateway, and it is fetched from the user's own + machine. Real result URLs are public CDN objects, which the guard allows. + """ + import httpx + + from tools.url_safety import create_ssrf_safe_async_client + + target = _resolve_destination(save_to, _filename_from_url(url)) + # Written under a .part name and renamed only once it is complete and + # plausible, so a failed download can never leave something that looks like + # a playable file behind. + partial = target.with_name(target.name + ".part") + timeout = httpx.Timeout(_DOWNLOAD_CONNECT_TIMEOUT_SECONDS, read=_DOWNLOAD_READ_TIMEOUT_SECONDS) + + try: + async with create_ssrf_safe_async_client(timeout=timeout, follow_redirects=True) as client: + async with client.stream("GET", url) as response: + response.raise_for_status() + with partial.open("wb") as handle: + async for chunk in response.aiter_bytes(): + handle.write(chunk) + + size = partial.stat().st_size + if size < _MIN_PLAUSIBLE_VIDEO_BYTES: + raise ValueError(f"the download returned only {size} bytes, which is not a video") + partial.replace(target) + except BaseException: + partial.unlink(missing_ok=True) + raise + + return target, size + + +def _filename_from_url(url: str) -> str: + from pathlib import PurePosixPath + from urllib.parse import unquote, urlsplit + + name = PurePosixPath(unquote(urlsplit(url).path)).name + # The path segment is vendor-controlled, so keep only a plain filename. + name = re.sub(r"[^A-Za-z0-9._-]", "_", name).lstrip(".")[:120] + return name or "flux3-video.mp4" + + +def _resolve_destination(save_to, filename: str): + """Where to write, honouring an explicit request and never overwriting.""" + from pathlib import Path + + if isinstance(save_to, str) and save_to.strip(): + requested = Path(save_to.strip()).expanduser() + if requested.is_dir() or save_to.rstrip().endswith(("/", "\\")): + directory, name = requested, filename + else: + directory, name = requested.parent, requested.name + else: + downloads = Path.home() / "Downloads" + directory, name = (downloads if downloads.is_dir() else Path.cwd()), filename + + directory.mkdir(parents=True, exist_ok=True) + return _free_path(directory / name) + + +def _free_path(candidate): + """`name.mp4` -> `name-2.mp4` -> `name-3.mp4` … so nothing is clobbered.""" + if not candidate.exists(): + return candidate + for suffix in range(2, _MAX_FILENAME_ATTEMPTS + 2): + sibling = candidate.with_name(f"{candidate.stem}-{suffix}{candidate.suffix}") + if not sibling.exists(): + return sibling + raise ValueError(f"could not find a free filename next to {candidate}") + + +# --------------------------------------------------------------------------- +# Handlers +# --------------------------------------------------------------------------- + +def _error(message: str) -> str: + return json.dumps({"error": message}, ensure_ascii=False) + + +def _submit_args(mode: str, args: dict) -> dict: + """The wire body: the model's arguments, minus Nones, plus our mode. + + Arguments pass through as the model gave them; the gateway owns validation + and the translation onto the vendor's own fields. + """ + body = {k: v for k, v in dict(args or {}).items() if v is not None} + body["mode"] = mode + return body + + +async def _submit(mode: str, args: dict) -> str: + endpoints = _endpoints() + if endpoints is None: + return _error("BFL video generation is not available in this build.") + return await _call_gateway("POST", f"{endpoints['base_url']}/generations", _submit_args(mode, args)) + + +async def _handle_text_to_video(args: dict, **kwargs) -> str: + return await _submit("text_to_video", _without_media(args)) + + +async def _handle_image_to_video(args: dict, **kwargs) -> str: + try: + prepared = await _prepare_media(args, kwargs.get("task_id")) + except ValueError as exc: + return _error(str(exc)) + return await _submit("image_to_video", prepared) + + +async def _handle_keyframes_to_video(args: dict, **kwargs) -> str: + images = (args or {}).get("input_images") + if not isinstance(images, list) or not images: + return _error("input_images must be a non-empty list of 1-10 images (local paths or URLs).") + try: + prepared = await _prepare_media(args, kwargs.get("task_id")) + except ValueError as exc: + return _error(str(exc)) + return await _submit("keyframes_to_video", prepared) + + +async def _handle_video_continuation(args: dict, **kwargs) -> str: + try: + prepared = await _prepare_media(args, kwargs.get("task_id")) + except ValueError as exc: + return _error(str(exc)) + return await _submit("video_continuation", prepared) + + +async def _handle_get_result(args: dict, **kwargs) -> str: + job_id = (args or {}).get("id") + if not isinstance(job_id, str) or not job_id.strip(): + return _error("id is required: the job id returned by the generate tool.") + endpoints = _endpoints() + if endpoints is None: + return _error("BFL video generation is not available in this build.") + from urllib.parse import quote + + raw = await _call_gateway("GET", f"{endpoints['base_url']}/generations/{quote(job_id.strip(), safe='')}") + return await _save_if_ready(raw, (args or {}).get("save_to")) + + +async def _handle_prompting_guide(args: dict, **kwargs) -> str: + return FLUX3_PROMPTING_GUIDE + + +# --------------------------------------------------------------------------- +# Gating +# --------------------------------------------------------------------------- + +def check_bfl_requirements() -> bool: + try: + if _endpoints() is None: + return False + from hermes_cli.nous_account import get_nous_portal_account_info + + info = get_nous_portal_account_info() + return bool(getattr(info, "logged_in", False) and getattr(info, "paid_service_access", False)) + except Exception: + return False + + +# --------------------------------------------------------------------------- +# Pinned schemas +# --------------------------------------------------------------------------- + +_ASPECT_RATIOS = ["auto", "21:9", "16:9", "4:3", "1:1", "3:4", "9:16", "9:21"] +_RESOLUTIONS = ["720p"] + +_GUIDE_POINTER = "Read bfl_flux3_prompting_guide before your first generation. " +_MEDIA_SENTENCE = ( + "Media fields accept a local file path (uploaded automatically to Nous-managed temporary " + "storage and deleted when the generation finishes) or a URL. " +) +_OVERRIDE_SENTENCE = "All guidance is defaults: explicit user instructions override it." + + +def _shared_submit_properties() -> dict: + return { + "prompt": { + "type": "string", + "minLength": 1, + "description": ( + "Generation brief in plain prose (it is interpreted and expanded by a reasoning " + "harness). Order it subject, distinguishing visual specifics, action, camera, " + "lighting, environment, audio, style. Audio is generated by default β€” say " + '"no music" if unwanted.' + ), + }, + "aspect_ratio": { + "type": "string", + "enum": list(_ASPECT_RATIOS), + "default": "auto", + "description": 'Output aspect ratio. "auto" lets the model choose.', + }, + "duration": { + "oneOf": [ + {"type": "integer", "minimum": 5, "maximum": 20}, + {"type": "string", "const": "auto"}, + ], + "default": "auto", + "description": 'Clip duration in whole seconds (5-20), or "auto".', + }, + "resolution": { + "type": "string", + "enum": list(_RESOLUTIONS), + "default": "720p", + "description": "Output resolution bin.", + }, + "generate_audio": { + "type": "boolean", + "default": True, + "description": "Generate synchronized audio.", + }, + "grounding": { + "type": "boolean", + "default": True, + "description": "Allow a short research pass before generation.", + }, + "seed": { + "type": "integer", + "minimum": 0, + "maximum": 4294967295, + "description": "Optional reproducibility seed.", + }, + "version": { + "type": "string", + "description": 'Model version pin. Defaults to "latest".', + }, + } + + +TEXT_TO_VIDEO_SCHEMA = { + "name": "bfl_flux3_text_to_video", + "description": ( + _GUIDE_POINTER + + "FLUX 3 text-to-video: generates a clip (with audio) from the prompt alone. Nothing but " + "the prompt anchors the subject here, so research anything with a real, checkable " + "appearance before writing it β€” whatever you leave unspecified is filled in for you. " + "Generation takes several minutes: this returns a job id immediately; poll " + "bfl_flux3_get_result. " + + _OVERRIDE_SENTENCE + ), + "parameters": { + "type": "object", + "properties": _shared_submit_properties(), + "required": ["prompt"], + "additionalProperties": False, + }, +} + +IMAGE_TO_VIDEO_SCHEMA = { + "name": "bfl_flux3_image_to_video", + "description": ( + _GUIDE_POINTER + + "FLUX 3 image-to-video: animates one image as the literal opening frame β€” those pixels " + "are frame 0 and the clip moves from there. " + + _MEDIA_SENTENCE + + "Returns a job id; poll bfl_flux3_get_result. " + + _OVERRIDE_SENTENCE + ), + "parameters": { + "type": "object", + "properties": { + **_shared_submit_properties(), + "input_image": { + "type": "string", + "minLength": 1, + "description": "Exactly one opening-frame image: a local file path or a URL (PNG/JPEG/WebP, up to 10MB).", + }, + }, + "required": ["prompt", "input_image"], + "additionalProperties": False, + }, +} + +KEYFRAMES_TO_VIDEO_SCHEMA = { + "name": "bfl_flux3_keyframes_to_video", + "description": ( + _GUIDE_POINTER + + "FLUX 3 keyframe video: a storyboard of 1-10 images pinned at chosen frame positions " + "(24fps). Name the subject and describe the motion that carries it between the pins, and " + "keep the subject consistent across every pinned image. " + + _MEDIA_SENTENCE + + "Returns a job id; poll bfl_flux3_get_result. " + + _OVERRIDE_SENTENCE + ), + "parameters": { + "type": "object", + "properties": { + **_shared_submit_properties(), + "input_images": { + "type": "array", + "items": {"type": "string", "minLength": 1}, + "minItems": 1, + "maxItems": 10, + "description": "1-10 keyframe images: local file paths or URLs (PNG/JPEG/WebP, up to 10MB each).", + }, + "keyframe_indices": { + "type": "array", + "items": {"type": "integer", "minimum": 0, "maximum": 480}, + "minItems": 1, + "maxItems": 10, + "description": ( + "One unique non-negative frame index per image (24fps). Each must be at most " + 'durationΓ—24, so set an explicit duration rather than "auto" whenever you pin ' + 'indices β€” "auto" resolves to 5, 10, 15 or 20 seconds and an index past the ' + "length it picks is rejected." + ), + }, + }, + "required": ["prompt", "input_images", "keyframe_indices"], + "additionalProperties": False, + }, +} + +VIDEO_CONTINUATION_SCHEMA = { + "name": "bfl_flux3_video_continuation", + "description": ( + _GUIDE_POINTER + + "FLUX 3 video continuation: the new generation picks up from the input clip's final " + 'frames. Open the prompt with "Continue this video from its final frames:", re-establish ' + "the subject and the moment it ended on, then describe what happens next. input_video " + "must be an mp4 of at most 50MB and 15 seconds, and the generated segment tops out at 15s " + "too β€” chain a second continuation for a longer sequence. duration is the " + "new segment only. " + + _MEDIA_SENTENCE + + "Returns a job id; poll bfl_flux3_get_result. " + + _OVERRIDE_SENTENCE + ), + "parameters": { + "type": "object", + "properties": { + **_shared_submit_properties(), + "input_video": { + "type": "string", + "minLength": 1, + "description": "The clip to continue: a local file path or a URL. mp4 only, at most 50MB and 15 seconds.", + }, + }, + "required": ["prompt", "input_video"], + "additionalProperties": False, + }, +} + +GET_RESULT_SCHEMA = { + "name": "bfl_flux3_get_result", + "description": ( + "Poll a FLUX 3 video job by the job id a generate tool returned. Generation takes minutes " + "and a long Generating phase is normal. Every response states the job's status, how long " + "to wait before polling again, and what you may do next β€” follow it rather than guessing. " + "On Ready the clip is downloaded for you and the response gives its local path; your only " + "remaining step is to deliver that file as the response describes." + ), + "parameters": { + "type": "object", + "properties": { + "id": { + "type": "string", + "minLength": 1, + "description": "Job id from a previous bfl_flux3_* generate call.", + }, + "save_to": { + "type": "string", + "description": ( + "Where to save the finished clip: a directory or a full file path. Set this " + "only when the user asked for a particular location; the default is " + "~/Downloads. An existing file is never overwritten." + ), + }, + }, + "required": ["id"], + "additionalProperties": False, + }, +} + +PROMPTING_GUIDE_SCHEMA = { + "name": "bfl_flux3_prompting_guide", + "description": ( + "Read this before your first FLUX 3 generation. The prompting and grounding guide: how " + "to research a subject so it renders as itself, how to assemble a prompt, which generate " + "tool fits, and how to save and deliver the finished clip. Takes no arguments and spends " + "no generation budget." + ), + "parameters": {"type": "object", "properties": {}, "additionalProperties": False}, +} + + +# --------------------------------------------------------------------------- +# Pinned prompting guide (methodology only β€” policy numbers such as waits and +# limits arrive live in the gateway's tool responses, so they cannot drift) +# --------------------------------------------------------------------------- + +FLUX3_PROMPTING_GUIDE = """# FLUX 3 video generation β€” how to get the best results + +Everything here is guidance, not policy: the user's explicit instructions +always win. If they say skip the research, save somewhere specific, or deliver +differently β€” do it their way without arguing. Only the server's own validation +and limits are non-negotiable. + +Start from the user's own wording. The prompt is rewritten by a reasoning +harness before anything is generated, so restyling it yourself just stacks a +second rewrite on top and gives their intent another chance to drift. There are +two reasons to add anything: grounding facts for a subject with a real, +checkable appearance, and the behaviours below, which they had no way to know +about. Absent those, send what they wrote. + +## Grounding + +Grounding is your job, and it is the highest-leverage step in the workflow. +What you specify is preserved; what you leave out is filled in for you. So +when a subject has a real, checkable appearance, research it first and put +what you found into the prompt β€” that is what makes the result yours rather +than an approximation of it. + +Ground whenever someone who knows the subject could watch the clip and say +"that's not what that is": a named person or place, a landmark, a particular +vehicle or machine, anything technical, anything culturally specific, or a +period setting. Skip it for generics ("a dog on a beach") β€” there is no fact +to get wrong. + +To ground: search for VISUAL references, where three good photographs beat any +amount of prose. If you can analyze images, analyze what you find. Then put +only what a camera could see into the prompt: silhouette and proportion, +materials and finish, specific colours, distinctive details, era-correct +context. + +## How this model behaves + +The prompt is read by a reasoning harness rather than a tag encoder, so keyword +tricks and word order do nothing. Write plain prose at whatever length the +brief deserves; a single line is a valid prompt. + +Audio is generated by default, whether or not you mention it, so leaving it out +gets you invented sound rather than silence. Name the ambient sound, the music +and any speech separately β€” each lands as its own layer β€” and say "no music" +when you do not want it. + +A quoted line becomes speech only if a speaker is visible on camera. Without +one it tends to render as burned-in text instead, so quote the line, describe +the speaker, and add "no on-screen text, no subtitles". + +Multi-shot sequences work inside a single generation: "SHOT ONE ... HARD CUT. +SHOT TWO ..." produces real cuts, and "one continuous unbroken shot" gets an +uncut take. Consecutive shots have to contrast in scale, location or colour or +the cut will not read as one β€” near-identical coverage blends back into a +continuous take. + +## Choosing a tool + +- No input media -> bfl_flux3_text_to_video. +- Animate one image as the literal opening frame -> + bfl_flux3_image_to_video (those pixels are frame 0, and the clip moves from + there). Name the subject and its distinguishing specifics here as you would + anywhere else: only frame 0 is pinned, every frame after it is generated, + and what you leave out is the model's call rather than yours. +- Several images pinned at chosen moments -> bfl_flux3_keyframes_to_video + (name the subject and describe the motion that carries it between the pins; + keep it consistent across every pin, since a mismatch becomes a visible + morph mid-clip). Pin exact indices and you want an explicit duration too β€” + every index must fall within durationΓ—24. +- Keep going from where a clip ends, or chain segments -> + bfl_flux3_video_continuation. Open with "Continue this video from its final + frames:", re-establish the subject and the moment the clip ended on, then + describe what happens next; duration is the new segment only, so plan + segments of 15 seconds or less to feed outputs back in. + +## Media inputs + +Pass local file paths directly β€” never hand-encode file contents into an argument. +URLs also work. Limits per file: +images 10MB (PNG/JPEG/WebP), video one mp4 of at most 50MB and 15 seconds. +Do not pre-shrink files to fit imagined caps; oversized pixel dimensions are +auto-downscaled and output tops out at 720p. + +## Workflow + +Submit returns a job id immediately β€” the video does not exist yet. Poll +bfl_flux3_get_result with that id; generation takes several minutes and a long +Generating phase is normal, not a stall. Nothing reaches disk before the job is +Ready, so checking folders mid-run tells you nothing. Every response +states the job status, how long to wait before polling again, and what you may +do next β€” follow it literally rather than guessing an interval. A job survives +client restarts: re-poll the same id rather than resubmitting, which would only +spend your budgets on duplicate work. + +## Save and deliver + +bfl_flux3_get_result saves the clip itself and returns saved_path. The download +is not yours to do and no URL is handed to you to fetch. Pass save_to only when +the user named a location; otherwise it lands in ~/Downloads, and an existing +file is never overwritten. If saving fails the response says so β€” poll the same +job again to retry, which is safe and spends no generation budget. + +Then deliver that file so the clip plays inline. Which markup plays inline is the +host's decision, so check your system prompt or platform instructions and use +exactly the form they give; the common ones are a MEDIA: tag alone on its own +line and a markdown embed. Two things break it. Write the real absolute path, +with ~ expanded. And keep the markup plain: wrapping it in bold, backticks or a +code fence, or rewriting it as a [link](path), turns an inline player into +literal text or a click-target. That is the most common way this step fails, +and it reads as success because the filename is on screen. Where the +instructions call for no delivery markup, or the host has no such mechanism, +follow them and state the absolute path in plain text. + +Report what you did rather than what the job says it did: the echoed prompt +field describes intent and often overstates what the render preserved. +""" + + +# --------------------------------------------------------------------------- +# Registration (auto-discovered: top-level registry.register calls) +# --------------------------------------------------------------------------- + +registry.register( + name="bfl_flux3_text_to_video", + toolset=_TOOLSET, + schema=TEXT_TO_VIDEO_SCHEMA, + handler=_handle_text_to_video, + check_fn=check_bfl_requirements, + requires_env=[], + is_async=True, + emoji="🎬", +) + +registry.register( + name="bfl_flux3_image_to_video", + toolset=_TOOLSET, + schema=IMAGE_TO_VIDEO_SCHEMA, + handler=_handle_image_to_video, + check_fn=check_bfl_requirements, + requires_env=[], + is_async=True, + emoji="🎬", +) + +registry.register( + name="bfl_flux3_keyframes_to_video", + toolset=_TOOLSET, + schema=KEYFRAMES_TO_VIDEO_SCHEMA, + handler=_handle_keyframes_to_video, + check_fn=check_bfl_requirements, + requires_env=[], + is_async=True, + emoji="🎬", +) + +registry.register( + name="bfl_flux3_video_continuation", + toolset=_TOOLSET, + schema=VIDEO_CONTINUATION_SCHEMA, + handler=_handle_video_continuation, + check_fn=check_bfl_requirements, + requires_env=[], + is_async=True, + emoji="🎬", +) + +registry.register( + name="bfl_flux3_get_result", + toolset=_TOOLSET, + schema=GET_RESULT_SCHEMA, + handler=_handle_get_result, + check_fn=check_bfl_requirements, + requires_env=[], + is_async=True, + emoji="🎬", +) + +registry.register( + name="bfl_flux3_prompting_guide", + toolset=_TOOLSET, + schema=PROMPTING_GUIDE_SCHEMA, + handler=_handle_prompting_guide, + check_fn=check_bfl_requirements, + requires_env=[], + is_async=True, + emoji="πŸ“–", +) diff --git a/tools/image_source.py b/tools/image_source.py index e8d55cebabb..e46651e4821 100644 --- a/tools/image_source.py +++ b/tools/image_source.py @@ -1,10 +1,16 @@ -"""Single resolver for every vision_analyze image source -> bytes + mime. +"""Single resolver for every media source -> bytes + mime. All source handling (data:/http(s)/file/local/container) funnels through :func:`resolve_image_source` so size and magic-byte checks are enforced exactly once. Returns raw bytes (not a path): the downstream step is base64 -> data URL (RFC 2397) and provider base64 content blocks. +Images are the default and the historical purpose. Callers whose argument +takes video opt in via ``permitted=("video",)`` β€” the same confinement and +credential-guard pipeline applies, and only the type check at the end differs +(extension-table typing plus an mp4 magic sniff, rather than image magic +bytes). Every existing call site keeps the image-only default unchanged. + Security (terminal-backend confinement, GHSA-gpxw-6wxv-w3qq): under a non-local terminal backend the file tools are confined to the sandbox (SECURITY.md 2.2), but vision read images host-side. This resolver enforces the same boundary: @@ -86,18 +92,23 @@ class ResolvedImage: _SCHEME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.\-]*://") -async def resolve_image_source(src: str, ctx: ResolveContext) -> ResolvedImage: +async def resolve_image_source( + src: str, + ctx: ResolveContext, + *, + permitted: tuple = ("image",), +) -> ResolvedImage: if not isinstance(src, str) or not src.strip(): raise SourceNotFound("image_url is required", src=str(src)) s = src.strip() if s.startswith("data:"): data, mime = _resolve_data_url(s) - return _finalize(data, mime, "data", s) + return _finalize(data, mime, "data", s, permitted) if s.startswith(("http://", "https://")): reason = _http_block_reason(s) if reason: raise SourceUnsafe(reason, src=s) - return _finalize(await _download_to_bytes(s), "", "http", s) + return _finalize(await _download_to_bytes(s), "", "http", s, permitted) if _SCHEME_RE.match(s) and not s.lower().startswith("file://"): raise UnsupportedScheme( @@ -134,15 +145,15 @@ async def resolve_image_source(src: str, ctx: ResolveContext) -> ResolvedImage: except ValueError as exc: raise SourceUnsafe(str(exc), src=s, origin="file") data = await asyncio.to_thread(host_target.read_bytes) - return _finalize(data, "", "file", s) + return _finalize(data, "", "file", s, permitted) if _is_local_terminal_backend(): # Local backend: any path was host-readable, so a miss simply means # the file doesn't exist β€” no sandbox to fall back to. - raise SourceNotFound(f"image file not found: '{p}'", src=s, origin="file") + raise SourceNotFound(f"media file not found: '{p}'", src=s, origin="file") # Not a permitted host read (or the host file is absent) -> read the # bytes inside the sandbox. Under a sandbox this reads the container's # filesystem, never the host's. - return await _resolve_container_fallback(p, ctx, s) + return await _resolve_container_fallback(p, ctx, s, permitted) def _resolve_data_url(s: str) -> tuple[bytes, str]: @@ -270,7 +281,9 @@ def _get_active_env(task_id: Optional[str]): return None -async def _resolve_container_fallback(p: Path, ctx: ResolveContext, src: str) -> ResolvedImage: +async def _resolve_container_fallback( + p: Path, ctx: ResolveContext, src: str, permitted: tuple = ("image",) +) -> ResolvedImage: """Read the image bytes inside the sandbox (fail-closed when none exists). Reached when a host read is not permitted or the host file is absent. The @@ -312,27 +325,66 @@ async def _resolve_container_fallback(p: Path, ctx: ResolveContext, src: str) -> except Exception as exc: raise NotAnImage(f"sandbox returned non-image data for '{p}': {exc}", src=src) if len(data) > _MAX_INGEST_BYTES: - raise SourceTooLarge("image exceeds size limit", src=src, origin="container") - return _finalize(data, "", "container", src) + raise SourceTooLarge("media exceeds size limit", src=src, origin="container") + return _finalize(data, "", "container", src, permitted) -def _finalize(data: bytes, declared_mime: str, origin: str, src: str) -> ResolvedImage: - """Intrinsic-correctness chokepoint: ingest byte cap + magic-byte sniff. +def _finalize( + data: bytes, declared_mime: str, origin: str, src: str, permitted: tuple = ("image",) +) -> ResolvedImage: + """Intrinsic-correctness chokepoint: ingest byte cap + type check. The cap here is the generous 50MB *ingest* budget, not the 20MB provider payload cap β€” a 20-50MB image must survive this step so the call site can resize it under the payload cap. See ``_MAX_INGEST_BYTES``. + + Images are typed by magic bytes. Video (opt-in via ``permitted``) is typed + by the extension table plus an mp4 container sniff: extension typing is + sufficient because every downstream consumer re-validates β€” the upload + gateway signs the content type into its presigned URL and the vendor + rejects undecodable input β€” so a wrong guess is a clean rejection there + rather than a hole here. """ from tools.vision_tools import _detect_image_mime_type_from_bytes if len(data) > _MAX_INGEST_BYTES: - raise SourceTooLarge("image exceeds size limit", src=src, origin=origin) + raise SourceTooLarge("media exceeds size limit", src=src, origin=origin) + sniffed = _detect_image_mime_type_from_bytes(data) - if sniffed is None: - if b" Optional[str]: + """Video MIME from the extension table, else the mp4/mov container magic. + + The magic fallback covers extensionless sources (data: URLs, URLs with + query strings): ISO base-media files carry ``ftyp`` at offset 4. + """ + from urllib.parse import urlsplit + + from tools.vision_tools import _detect_video_mime_type + + path_part = urlsplit(src).path if _SCHEME_RE.match(src) else src + by_extension = _detect_video_mime_type(Path(path_part)) + if by_extension is not None: + return by_extension + if len(data) > 12 and data[4:8] == b"ftyp": + return "video/mp4" + return None diff --git a/tools/managed_tool_gateway.py b/tools/managed_tool_gateway.py index 2bbc9dcff16..73910813662 100644 --- a/tools/managed_tool_gateway.py +++ b/tools/managed_tool_gateway.py @@ -8,6 +8,7 @@ import os from datetime import datetime, timezone from dataclasses import dataclass from typing import Callable, Optional +from urllib.parse import urlsplit logger = logging.getLogger(__name__) @@ -190,3 +191,232 @@ def is_managed_tool_gateway_ready( gateway_builder=gateway_builder, token_reader=token_reader or peek_nous_access_token, ) is not None + + +# --------------------------------------------------------------------------- +# Managed vendor endpoints +# --------------------------------------------------------------------------- +# +# Vendors the gateway serves on its own origin (rather than on a +# `{vendor}-gateway` host) are pinned HERE, in code, the same way every other +# managed vendor's gateway URL is pinned: adding one is a Hermes release, and +# the exact URL a user's agent may connect to is reviewable in this file. A +# runtime discovery catalog was tried and deliberately removed β€” a remote +# endpoint that can add tools to every entitled install is a bigger trust +# surface than a code diff. +# +# The gateway exposes a Nous-owned REST contract per vendor; it names the +# vendor but not the vendor's own API, so nothing here needs to know the +# upstream's endpoint or field names. + +# Pseudo-vendor used only to resolve the shared tool-gateway origin via +# build_vendor_gateway_url (honors TOOL_GATEWAY_URL / TOOL_GATEWAY_DOMAIN). +_MANAGED_GATEWAY_VENDOR = "tool" + +def managed_vendor_base_path(vendor: str) -> str: + """Base path for a managed vendor's REST routes on the gateway host.""" + return f"/api/{vendor}" + + +def managed_vendor_upload_path(vendor: str) -> str: + """Media upload endpoint for a managed vendor, on the same host.""" + return f"/api/uploads/{vendor}" + + +def managed_vendor_endpoints( + vendor: str, + gateway_builder: Optional[Callable[[str], str]] = None, +) -> Optional[dict]: + """Absolute URLs for a managed vendor, or ``None`` when unreachable. + + ``None`` means managed Nous tools are disabled for this build, which is + what keeps a user who could never use the vendor from being shown its + tools. + """ + if not managed_nous_tools_enabled(): + return None + + builder = gateway_builder or build_vendor_gateway_url + origin = builder(_MANAGED_GATEWAY_VENDOR).rstrip("/") + return { + "origin": origin, + "base_url": f"{origin}{managed_vendor_base_path(vendor)}", + "upload_path": managed_vendor_upload_path(vendor), + } + + +def is_managed_nous_gateway_url( + url: object, + gateway_builder: Optional[Callable[[str], str]] = None, +) -> bool: + """True when ``url`` is on the Nous tool-gateway origin this client builds. + + Anything granting a URL extra trust β€” our bearer, reading files off disk to + upload β€” must gate on this rather than on a name, so an arbitrary URL can + never inherit that trust. + """ + if not isinstance(url, str) or not url.strip(): + return False + + builder = gateway_builder or build_vendor_gateway_url + try: + expected = urlsplit(builder(_MANAGED_GATEWAY_VENDOR)) + actual = urlsplit(url.strip()) + except ValueError: + return False + + return bool(actual.scheme) and (actual.scheme, actual.netloc) == (expected.scheme, expected.netloc) + + +def managed_gateway_auth_headers( + url: object, + gateway_builder: Optional[Callable[[str], str]] = None, + token_reader: Optional[Callable[[], Optional[str]]] = None, +) -> dict: + """Live auth headers for a managed gateway URL, or ``{}`` when not managed. + + Read fresh on every call rather than cached: a Nous access token expires + within the hour, and a long session would otherwise keep presenting a dead + bearer. Returns ``{}`` rather than raising when no token is available, so a + caller can report "sign in" instead of sending an unauthenticated request. + """ + if not is_managed_nous_gateway_url(url, gateway_builder): + return {} + + resolved_token_reader = token_reader or read_nous_access_token + try: + token = resolved_token_reader() + except Exception as exc: # pragma: no cover β€” defensive + logger.debug("Managed gateway token read failed for %s: %s", url, exc) + return {} + if not isinstance(token, str) or not token.strip(): + return {} + + return {"Authorization": f"Bearer {token.strip()}"} + + +# --------------------------------------------------------------------------- +# Managed media uploads +# --------------------------------------------------------------------------- +# +# Media arguments used to be inlined as base64, which capped a whole tool call +# at ~2MB of real bytes under the gateway's request ceiling and ruled out video +# entirely. Each pinned managed server carries an upload endpoint +# (`upload_path`); the bytes go straight to storage via a presigned URL, and +# the tool argument carries an opaque `nous-upload:` reference instead. +# +# The protocol lives HERE rather than in a vendor tool module: the presign +# request shape, the response contract, and the `nous-upload:` scheme are Nous +# gateway specifics shared by every managed vendor that takes media. + +_MEDIA_UPLOAD_PRESIGN_TIMEOUT_SECONDS = 15.0 +# The PUT carries up to 50MB of video; a flat 60s would fail a legitimate +# clip on an ordinary residential uplink, so only the write phase is long. +_MEDIA_UPLOAD_PUT_READ_TIMEOUT_SECONDS = 60.0 +_MEDIA_UPLOAD_PUT_WRITE_TIMEOUT_SECONDS = 300.0 + + +def _describe_media_upload_refusal(response) -> str: + """A model-actionable reason from a gateway refusal, or a generic one. + + The gateway's 4xx bodies carry deliberate guidance (rate-limit waits, size + caps, "you could not submit anyway"), so surface `error.message` verbatim + rather than a bare status code. + """ + try: + payload = response.json() + message = payload.get("error", {}).get("message") + if isinstance(message, str) and message.strip(): + return message.strip() + except Exception: + pass + return f"the gateway refused the upload (HTTP {response.status_code})" + + +def build_managed_media_uploader( + server_url: object, + upload_path: object, + gateway_builder: Optional[Callable[[str], str]] = None, + token_reader: Optional[Callable[[], Optional[str]]] = None, +) -> Optional[Callable]: + """Async ``(data, mime) -> argument value`` uploader for one managed vendor. + + Returns ``None`` when there is no usable upload endpoint (not a managed + Nous URL, or no ``upload_path``); callers then refuse local paths with a + clear message instead of silently forwarding them. + + The three steps of the protocol: + + 1. POST ``origin + upload_path`` with the declared content type and exact + byte length, using the same live auth headers as the vendor calls. + The gateway answers with a presigned single-object PUT URL (short + expiry; type and length are signed into it) and an upload token. + 2. PUT the bytes to that URL. This goes directly to storage β€” never + through the gateway β€” which is what removes the request-size ceiling. + 3. Return ``nous-upload:`` for the tool argument. The token is + bound to this Nous principal and is redeemable only through the + gateway, so it is inert anywhere else it might end up. + """ + if not is_managed_nous_gateway_url(server_url, gateway_builder): + return None + if not isinstance(upload_path, str) or not upload_path.startswith("/"): + return None + + parts = urlsplit(str(server_url).strip()) + origin = f"{parts.scheme}://{parts.netloc}" + presign_url = f"{origin}{upload_path}" + + async def upload(data: bytes, mime: str) -> str: + import httpx + + from tools.url_safety import create_ssrf_safe_async_client + + headers = managed_gateway_auth_headers(server_url, gateway_builder, token_reader) + if not headers: + raise RuntimeError("no Nous credential is available for the upload") + + # Two clients on purpose, split by whose address we are trusting. + # + # The presign POST goes to `presign_url`, which is entirely determined + # by the managed gateway origin (already validated by + # is_managed_nous_gateway_url) plus the pinned upload_path β€” the same + # first-party host the vendor calls go to freely. SSRF-guarding it + # protects against nothing and would reject a local gateway on + # 127.0.0.1, so it uses a plain client. The PUT target, by contrast, is + # a URL the gateway *returned*, so it keeps the SSRF-safe client as + # defense in depth (real presigned URLs are public R2, which it allows). + presign_timeout = httpx.Timeout(_MEDIA_UPLOAD_PRESIGN_TIMEOUT_SECONDS) + async with httpx.AsyncClient(timeout=presign_timeout) as client: + presign = await client.post( + presign_url, + headers=headers, + json={"contentType": mime, "contentLength": len(data)}, + ) + if presign.status_code != 200: + raise RuntimeError(_describe_media_upload_refusal(presign)) + + try: + payload = presign.json() + except Exception: + payload = None + upload_url = payload.get("uploadUrl") if isinstance(payload, dict) else None + token = payload.get("token") if isinstance(payload, dict) else None + if not (isinstance(upload_url, str) and upload_url and isinstance(token, str) and token): + raise RuntimeError("the gateway's upload response was malformed") + + put_timeout = httpx.Timeout( + _MEDIA_UPLOAD_PRESIGN_TIMEOUT_SECONDS, + read=_MEDIA_UPLOAD_PUT_READ_TIMEOUT_SECONDS, + write=_MEDIA_UPLOAD_PUT_WRITE_TIMEOUT_SECONDS, + ) + async with create_ssrf_safe_async_client(timeout=put_timeout) as client: + # The presigned URL signs the exact Content-Type and Content-Length, + # so this PUT must send precisely what was declared above. + put = await client.put(upload_url, content=data, headers={"Content-Type": mime}) + if put.status_code != 200: + raise RuntimeError(f"storage refused the upload (HTTP {put.status_code})") + + return f"nous-upload:{token}" + + return upload + diff --git a/toolsets.py b/toolsets.py index 4c7588b0abd..f4bb3c3343a 100644 --- a/toolsets.py +++ b/toolsets.py @@ -42,6 +42,10 @@ _HERMES_CORE_TOOLS = [ "read_file", "write_file", "patch", "search_files", # Vision + image generation "vision_analyze", "image_generate", + # BFL FLUX 3 video generation + "bfl_flux3_text_to_video", "bfl_flux3_image_to_video", + "bfl_flux3_keyframes_to_video", "bfl_flux3_video_continuation", + "bfl_flux3_get_result", "bfl_flux3_prompting_guide", # Skills "skills_list", "skill_view", "skill_manage", # Browser automation @@ -151,6 +155,25 @@ TOOLSETS = { "includes": [] }, + "bfl": { + "description": ( + "Black Forest Labs FLUX 3 video generation through the Nous tool " + "gateway: per-mode submit tools (text, image, keyframes, " + "continuation), a poll tool, and a prompting guide. Generations " + "take minutes, so submit returns a job id and the model polls for " + "the result." + ), + "tools": [ + "bfl_flux3_text_to_video", + "bfl_flux3_image_to_video", + "bfl_flux3_keyframes_to_video", + "bfl_flux3_video_continuation", + "bfl_flux3_get_result", + "bfl_flux3_prompting_guide", + ], + "includes": [] + }, + "computer_use": { "description": ( "Background desktop control via cua-driver (macOS/Windows/Linux) β€” " @@ -410,6 +433,10 @@ TOOLSETS = { "read_file", "write_file", "patch", "search_files", # Vision + image generation "vision_analyze", "image_generate", + # BFL FLUX 3 video generation + "bfl_flux3_text_to_video", "bfl_flux3_image_to_video", + "bfl_flux3_keyframes_to_video", "bfl_flux3_video_continuation", + "bfl_flux3_get_result", "bfl_flux3_prompting_guide", # Skills "skills_list", "skill_view", "skill_manage", # Browser automation