diff --git a/gateway/platforms/yuanbao.py b/gateway/platforms/yuanbao.py index 23c0d4d45f6..4657afd5219 100644 --- a/gateway/platforms/yuanbao.py +++ b/gateway/platforms/yuanbao.py @@ -179,6 +179,16 @@ OBSERVED_MEDIA_BACKFILL_LOOKBACK = 50 # Max number of resource references to resolve per inbound turn OBSERVED_MEDIA_BACKFILL_MAX_RESOLVE_PER_TURN = 12 +# Bounded concurrency for inbound media resolve/download. +# - 1 = sequential (legacy behavior, safe rollback knob) +# - 6 = default; aligns with the per-origin HTTP/1.1 ceiling browsers use, +# balances first-token latency vs. backend pressure +# - 12 = upper clamp; matches OBSERVED_MEDIA_BACKFILL_MAX_RESOLVE_PER_TURN +# Configured via config.yaml: platforms.yuanbao.extra.media_resolve_concurrency. +_DEFAULT_RESOLVE_CONCURRENCY = 6 +_MIN_RESOLVE_CONCURRENCY = 1 +_MAX_RESOLVE_CONCURRENCY = 12 + class MarkdownProcessor: """Encapsulates all Markdown-related utilities for the Yuanbao platform. @@ -2813,45 +2823,94 @@ class MediaResolveMiddleware(InboundMiddleware): Yuanbao COS hostnames resolve to private IPs, tripping the SSRF guard in vision_tools. We download ourselves and return local cache paths. - """ - media_urls: List[str] = [] - media_types: List[str] = [] + Resolution runs with bounded concurrency + (``adapter.media_resolve_concurrency``); see :meth:`_resolve_ybres_refs` + for the same order-preserving / exception-isolated contract. + """ + # Pre-filter resolvable refs, preserving input order. + active: List[Tuple[str, str, str, str]] = [] for ref in media_refs: kind = str(ref.get("kind") or "").strip().lower() url = str(ref.get("url") or "").strip() filename = str(ref.get("name") or "").strip() if kind not in _RESOLVABLE_MEDIA_KINDS or not url: continue - - # Extract resourceId from the placeholder URL for cache dedup. rid = ExtractContentMiddleware._parse_resource_id(url) - if rid and cls._append_cached_resource(adapter, rid, media_urls, media_types): - continue + active.append((kind, url, filename, rid)) - try: - fetch_url = await cls._resolve_download_url(adapter, url) - except Exception as exc: - logger.warning( - "[%s] inbound media resolve failed: kind=%s url=%s err=%s", - adapter.name, kind, url, exc, + if not active: + return [], [] + + semaphore = asyncio.Semaphore(adapter.media_resolve_concurrency) + + async def _resolve_one( + kind: str, url: str, filename: str, rid: str, + ) -> Optional[Tuple[str, str]]: + # Cache dedup first — avoids the RPC + download entirely on a hit. + if rid: + hit = cls._get_cached_resource(rid) + if hit is not None: + logger.debug( + "[%s] resource cache hit: rid=%s path=%s", + adapter.name, rid, hit[0], + ) + return hit + async with semaphore: + try: + fetch_url = await cls._resolve_download_url(adapter, url) + except Exception as exc: + logger.warning( + "[%s] inbound media resolve failed: kind=%s url=%s err=%s", + adapter.name, kind, url, exc, + ) + return None + return await cls._download_and_cache( + adapter, + fetch_url=fetch_url, + kind=kind, + file_name=filename or None, + log_tag=f"placeholder_url={url[:80]}", + resource_id=rid, ) - continue - cached = await cls._download_and_cache( - adapter, - fetch_url=fetch_url, - kind=kind, - file_name=filename or None, - log_tag=f"placeholder_url={url[:80]}", - resource_id=rid, - ) - if cached is None: + _t0 = time.monotonic() + results = await asyncio.gather( + *(_resolve_one(kind, url, filename, rid) + for kind, url, filename, rid in active), + return_exceptions=True, + ) + _elapsed_ms = int((time.monotonic() - _t0) * 1000) + + media_urls: List[str] = [] + media_types: List[str] = [] + _failed = 0 + for (kind, url, _filename, _rid), result in zip(active, results): + if isinstance(result, BaseException): + logger.warning( + "[%s] inbound media resolve crashed: kind=%s url=%s err=%s", + adapter.name, kind, url[:80], result, + ) + _failed += 1 continue - local_path, mime = cached + if result is None: + _failed += 1 + continue + local_path, mime = result media_urls.append(local_path) media_types.append(mime) + # Batch summary: keep fields stable for offline aggregation + # (concurrency vs elapsed_ms is the core knob-tuning view). + logger.info( + "[%s] media resolve batch: scope=media concurrency=%d total=%d ok=%d failed=%d elapsed_ms=%d", + adapter.name, + adapter.media_resolve_concurrency, + len(active), + len(media_urls), + _failed, + _elapsed_ms, + ) return media_urls, media_types @classmethod @@ -2862,36 +2921,86 @@ class MediaResolveMiddleware(InboundMiddleware): *, log_prefix: str, ) -> Tuple[List[str], List[str]]: - """Resolve a list of ``(rid, kind, filename)`` ybres tuples to local paths. + """Resolve ``(rid, kind, filename)`` ybres tuples to local paths. + + Runs with bounded concurrency (``adapter.media_resolve_concurrency``) + so cold-start turns with many anchors don't pay ``N × (RPC + download)`` + sequentially. Output order matches input; per-rid failures are isolated. """ + # Pre-filter resolvable kinds, preserving input order. + active: List[Tuple[str, str, str]] = [ + (rid, kind, filename) + for rid, kind, filename in refs + if kind in _RESOLVABLE_MEDIA_KINDS + ] + if not active: + return [], [] + + semaphore = asyncio.Semaphore(adapter.media_resolve_concurrency) + + async def _resolve_one(rid: str, kind: str, filename: str) -> Optional[Tuple[str, str]]: + # Cache dedup first — avoids the RPC + download entirely on a hit. + hit = cls._get_cached_resource(rid) + if hit is not None: + logger.debug( + "[%s] resource cache hit: rid=%s path=%s", + adapter.name, rid, hit[0], + ) + return hit + async with semaphore: + try: + fresh_url = await cls._fetch_resource_url(adapter, rid) + except Exception as exc: + logger.warning( + "[%s] %s resolve failed: rid=%s kind=%s err=%s", + adapter.name, log_prefix, rid, kind, exc, + ) + return None + return await cls._download_and_cache( + adapter, + fetch_url=fresh_url, + kind=kind, + file_name=filename or None, + log_tag=f"{log_prefix} rid={rid}", + resource_id=rid, + ) + + # return_exceptions=True isolates per-coroutine failures. + _t0 = time.monotonic() + results = await asyncio.gather( + *(_resolve_one(rid, kind, filename) for rid, kind, filename in active), + return_exceptions=True, + ) + _elapsed_ms = int((time.monotonic() - _t0) * 1000) + media_paths: List[str] = [] mimes: List[str] = [] - for rid, kind, filename in refs: - if kind not in _RESOLVABLE_MEDIA_KINDS: - continue - if cls._append_cached_resource(adapter, rid, media_paths, mimes): - continue - try: - fresh_url = await cls._fetch_resource_url(adapter, rid) - except Exception as exc: + _failed = 0 + for (rid, kind, _filename), result in zip(active, results): + if isinstance(result, BaseException): logger.warning( - "[%s] %s resolve failed: rid=%s kind=%s err=%s", - adapter.name, log_prefix, rid, kind, exc, + "[%s] %s resolve crashed: rid=%s kind=%s err=%s", + adapter.name, log_prefix, rid, kind, result, ) + _failed += 1 continue - cached = await cls._download_and_cache( - adapter, - fetch_url=fresh_url, - kind=kind, - file_name=filename or None, - log_tag=f"{log_prefix} rid={rid}", - resource_id=rid, - ) - if cached is None: + if result is None: + _failed += 1 continue - path, mime = cached + path, mime = result media_paths.append(path) mimes.append(mime) + + # Batch summary: stable fields for offline aggregation. + logger.info( + "[%s] media resolve batch: scope=ybres concurrency=%d total=%d ok=%d failed=%d elapsed_ms=%d", + adapter.name, + adapter.media_resolve_concurrency, + len(active), + len(media_paths), + _failed, + _elapsed_ms, + ) return media_paths, mimes @classmethod @@ -5071,6 +5180,19 @@ class YuanbaoAdapter(BasePlatformAdapter): self._api_domain: str = (_extra.get("api_domain") or DEFAULT_API_DOMAIN).rstrip("/") self._route_env: str = (_extra.get("route_env") or "").strip() + # Bounded concurrency for inbound media resolve/download. + # See _DEFAULT_RESOLVE_CONCURRENCY for rationale; clamped to [min, max] + # so a misconfigured value cannot hammer the resource backend nor + # accidentally drop below sequential behavior. + try: + _raw_concurrency = int(_extra.get("media_resolve_concurrency", _DEFAULT_RESOLVE_CONCURRENCY)) + except (TypeError, ValueError): + _raw_concurrency = _DEFAULT_RESOLVE_CONCURRENCY + self.media_resolve_concurrency: int = max( + _MIN_RESOLVE_CONCURRENCY, + min(_MAX_RESOLVE_CONCURRENCY, _raw_concurrency), + ) + # Core managers (UML composition) self._connection: ConnectionManager = ConnectionManager(self) self._outbound: OutboundManager = OutboundManager(self) diff --git a/tests/test_yuanbao_pipeline.py b/tests/test_yuanbao_pipeline.py index d55dad7e007..fa19f5ec6bc 100644 --- a/tests/test_yuanbao_pipeline.py +++ b/tests/test_yuanbao_pipeline.py @@ -10,6 +10,7 @@ Tests cover: 6. OOP middleware ABC and class tests """ +import asyncio import sys import os import json @@ -45,6 +46,8 @@ from gateway.platforms.yuanbao import ( DispatchMiddleware, InboundPipelineBuilder, YuanbaoAdapter, + _MIN_RESOLVE_CONCURRENCY, + _MAX_RESOLVE_CONCURRENCY, ) from gateway.config import PlatformConfig @@ -1600,6 +1603,245 @@ class TestResolveMediaUrlsCacheHit: MediaResolveMiddleware._resource_cache.clear() +class TestResolveYbresRefsConcurrency: + """Bounded-concurrency contracts for ``_resolve_ybres_refs``.""" + + # ------------------------------------------------------------------ + # Bounded-concurrency contracts (issue 3 in + # yuanbao-media-pipeline-optimizations.md). These are behavior + # contracts, not implementation snapshots — they assert the + # invariants the new gather()-based path must hold, not how it's + # wired internally. + # ------------------------------------------------------------------ + + @pytest.mark.asyncio + async def test_concurrent_resolve_preserves_input_order(self): + """Order of returned (paths, mimes) must match input ``refs`` order + even when later refs finish downloading first. + """ + adapter = make_adapter(extra={"media_resolve_concurrency": 4}) + refs = [ + ("rid-A", "image", ""), + ("rid-B", "image", ""), + ("rid-C", "image", ""), + ] + + # _fetch is fast and uniform; the interesting variation is in + # _download_and_cache, where rid-A is the slowest. If results + # were assembled by completion order, rid-A would land last. + async def slow_fetch(_adapter, rid): + return f"https://fresh/{rid}" + + delays = {"rid-A": 0.06, "rid-B": 0.02, "rid-C": 0.0} + results_by_rid = { + "rid-A": ("/cache/A.jpg", "image/jpeg"), + "rid-B": ("/cache/B.jpg", "image/jpeg"), + "rid-C": ("/cache/C.jpg", "image/jpeg"), + } + + async def slow_download(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id): + await asyncio.sleep(delays[resource_id]) + return results_by_rid[resource_id] + + with patch.object( + MediaResolveMiddleware, "_fetch_resource_url", + new=AsyncMock(side_effect=slow_fetch), + ), patch.object( + MediaResolveMiddleware, "_download_and_cache", + new=AsyncMock(side_effect=slow_download), + ): + paths, mimes = await MediaResolveMiddleware._resolve_ybres_refs( + adapter, refs, log_prefix="test", + ) + + assert paths == ["/cache/A.jpg", "/cache/B.jpg", "/cache/C.jpg"] + assert mimes == ["image/jpeg", "image/jpeg", "image/jpeg"] + + @pytest.mark.asyncio + async def test_concurrency_one_equivalent_to_sequential(self): + """``media_resolve_concurrency = 1`` must behave like the legacy + sequential path — at any moment at most one ``_download_and_cache`` + is in flight. + """ + adapter = make_adapter(extra={"media_resolve_concurrency": 1}) + refs = [("rid-A", "image", ""), ("rid-B", "image", ""), ("rid-C", "image", "")] + + in_flight = 0 + max_in_flight = 0 + + async def tracked_download(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id): + nonlocal in_flight, max_in_flight + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + try: + # Yield to the event loop so any concurrent coroutine + # would have a chance to also enter the critical section. + await asyncio.sleep(0.01) + return (f"/cache/{resource_id}.jpg", "image/jpeg") + finally: + in_flight -= 1 + + with patch.object( + MediaResolveMiddleware, "_fetch_resource_url", + new=AsyncMock(side_effect=lambda _a, rid: f"https://fresh/{rid}"), + ), patch.object( + MediaResolveMiddleware, "_download_and_cache", + new=AsyncMock(side_effect=tracked_download), + ): + paths, _ = await MediaResolveMiddleware._resolve_ybres_refs( + adapter, refs, log_prefix="test", + ) + + assert max_in_flight == 1 + assert paths == ["/cache/rid-A.jpg", "/cache/rid-B.jpg", "/cache/rid-C.jpg"] + + @pytest.mark.asyncio + async def test_concurrency_caps_inflight_downloads(self): + """Configured concurrency bounds the number of in-flight downloads.""" + adapter = make_adapter(extra={"media_resolve_concurrency": 2}) + refs = [(f"rid-{i}", "image", "") for i in range(6)] + + in_flight = 0 + max_in_flight = 0 + + async def tracked_download(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id): + nonlocal in_flight, max_in_flight + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + try: + await asyncio.sleep(0.01) + return (f"/cache/{resource_id}.jpg", "image/jpeg") + finally: + in_flight -= 1 + + with patch.object( + MediaResolveMiddleware, "_fetch_resource_url", + new=AsyncMock(side_effect=lambda _a, rid: f"https://fresh/{rid}"), + ), patch.object( + MediaResolveMiddleware, "_download_and_cache", + new=AsyncMock(side_effect=tracked_download), + ): + paths, _ = await MediaResolveMiddleware._resolve_ybres_refs( + adapter, refs, log_prefix="test", + ) + + assert max_in_flight == 2 + assert paths == [f"/cache/rid-{i}.jpg" for i in range(6)] + + @pytest.mark.asyncio + async def test_download_exception_isolated_to_single_ref(self): + """An exception raised inside ``_download_and_cache`` for one rid + must not poison the whole batch; surviving refs still resolve. + """ + adapter = make_adapter(extra={"media_resolve_concurrency": 4}) + refs = [ + ("rid-ok-1", "image", ""), + ("rid-boom", "image", ""), + ("rid-ok-2", "image", ""), + ] + + async def maybe_boom(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id): + if resource_id == "rid-boom": + raise RuntimeError("download crashed") + return (f"/cache/{resource_id}.jpg", "image/jpeg") + + with patch.object( + MediaResolveMiddleware, "_fetch_resource_url", + new=AsyncMock(side_effect=lambda _a, rid: f"https://fresh/{rid}"), + ), patch.object( + MediaResolveMiddleware, "_download_and_cache", + new=AsyncMock(side_effect=maybe_boom), + ): + paths, mimes = await MediaResolveMiddleware._resolve_ybres_refs( + adapter, refs, log_prefix="test", + ) + + assert paths == ["/cache/rid-ok-1.jpg", "/cache/rid-ok-2.jpg"] + assert mimes == ["image/jpeg", "image/jpeg"] + + @pytest.mark.asyncio + async def test_misconfigured_concurrency_clamped(self): + """Out-of-range or non-int concurrency values are clamped, not crashed.""" + # Negative -> clamped up to MIN + adapter_low = make_adapter(extra={"media_resolve_concurrency": -3}) + assert adapter_low.media_resolve_concurrency >= _MIN_RESOLVE_CONCURRENCY + + # Huge -> clamped down to MAX + adapter_high = make_adapter(extra={"media_resolve_concurrency": 9999}) + assert adapter_high.media_resolve_concurrency <= _MAX_RESOLVE_CONCURRENCY + + # Non-int garbage -> falls back to default, doesn't raise + adapter_garbage = make_adapter(extra={"media_resolve_concurrency": "fast"}) + assert ( + _MIN_RESOLVE_CONCURRENCY + <= adapter_garbage.media_resolve_concurrency + <= _MAX_RESOLVE_CONCURRENCY + ) + + +class TestResolveMediaUrlsConcurrency: + """Bounded-concurrency contracts for ``_resolve_media_urls`` (own-turn + media). Same invariants as ``_resolve_ybres_refs`` — order preserved, + failures isolated, concurrency clamped. + """ + + @pytest.mark.asyncio + async def test_preserves_input_order(self): + adapter = make_adapter(extra={"media_resolve_concurrency": 4}) + media_refs = [ + {"kind": "image", "url": "https://hunyuan.tencent.com/api/resource/download?resourceId=rid-A", "name": ""}, + {"kind": "image", "url": "https://hunyuan.tencent.com/api/resource/download?resourceId=rid-B", "name": ""}, + {"kind": "image", "url": "https://hunyuan.tencent.com/api/resource/download?resourceId=rid-C", "name": ""}, + ] + + delays = {"rid-A": 0.05, "rid-B": 0.02, "rid-C": 0.0} + + async def slow_download(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id): + await asyncio.sleep(delays[resource_id]) + return (f"/cache/{resource_id}.jpg", "image/jpeg") + + with patch.object( + MediaResolveMiddleware, "_resolve_download_url", + new=AsyncMock(side_effect=lambda _a, url: url), + ), patch.object( + MediaResolveMiddleware, "_download_and_cache", + new=AsyncMock(side_effect=slow_download), + ): + paths, mimes = await MediaResolveMiddleware._resolve_media_urls( + adapter, media_refs, + ) + + assert paths == ["/cache/rid-A.jpg", "/cache/rid-B.jpg", "/cache/rid-C.jpg"] + assert mimes == ["image/jpeg"] * 3 + + @pytest.mark.asyncio + async def test_failure_isolated(self): + adapter = make_adapter(extra={"media_resolve_concurrency": 4}) + media_refs = [ + {"kind": "image", "url": "https://x/api/resource/download?resourceId=ok", "name": ""}, + {"kind": "image", "url": "https://x/api/resource/download?resourceId=boom", "name": ""}, + ] + + async def maybe_boom(_adapter, *, fetch_url, kind, file_name, log_tag, resource_id): + if resource_id == "boom": + raise RuntimeError("download crashed") + return ("/cache/ok.jpg", "image/jpeg") + + with patch.object( + MediaResolveMiddleware, "_resolve_download_url", + new=AsyncMock(side_effect=lambda _a, url: url), + ), patch.object( + MediaResolveMiddleware, "_download_and_cache", + new=AsyncMock(side_effect=maybe_boom), + ): + paths, mimes = await MediaResolveMiddleware._resolve_media_urls( + adapter, media_refs, + ) + + assert paths == ["/cache/ok.jpg"] + assert mimes == ["image/jpeg"] + + class TestMediaResolveMiddlewareRouting: """Branch-routing tests for MediaResolveMiddleware.handle()."""