From a440d81d8f9ff565871c45780f4b9e680b4c8ec8 Mon Sep 17 00:00:00 2001 From: 1012839419a-alt <1012839419a-alt@users.noreply.github.com> Date: Fri, 7 Aug 2026 19:56:45 +0800 Subject: [PATCH] feat(channels): migrate doubao/douyin to thick fetch() with bounded retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both session-affinity channels declared capabilities (default_rate, auth_kind) but never overrode fetch(), so run_channel treated them as unmigrated: no RateLimitedClient, no runner protocol, and CDP race failures (CDP connection is not open, Inspected target navigated or closed, cloneNode) had no retry semantics — the outer worker had to brute-force retry, and a captcha wall was the only classified error. - doubao_research_channel: classify transient CDP races as ConnectionError (retryable in error_taxonomy) alongside the existing captcha_challenge classification; add fetch() override with exponential-backoff bounded retry (max_retries default 3, retry_base_delay default 2s). Captcha is NEVER auto-retried — the pipeline's captcha branch (PR #65) pauses the source for a human. - douyin_detail_channel: same transient classification on open/eval failures + same fetch() retry loop; permanent failures fail immediately. - collect() stays the single source of truth in both; ctx.http is not threaded through (subprocess transport, same trade-off as opencli_channel.fetch()). Tests: 19 doubao (7 new), 6 douyin (3 new) — retry-then-success, give-up after max_retries, captcha/permanent never retried, metadata passthrough, CDP transient classification. channel contract + runner suites green. --- backend/channels/doubao_research_channel.py | 92 ++++++++++++- backend/channels/douyin_detail_channel.py | 71 +++++++++- .../channels/test_doubao_research_channel.py | 127 ++++++++++++++++++ .../channels/test_douyin_detail_channel.py | 66 +++++++++ 4 files changed, 349 insertions(+), 7 deletions(-) diff --git a/backend/channels/doubao_research_channel.py b/backend/channels/doubao_research_channel.py index 6a9f430a..8fe59c29 100644 --- a/backend/channels/doubao_research_channel.py +++ b/backend/channels/doubao_research_channel.py @@ -4,7 +4,14 @@ import re from typing import Any -from backend.channels.base import AbstractChannel, Capabilities, ChannelResult +from backend.channels.base import ( + AbstractChannel, + Capabilities, + ChannelFetchError, + ChannelResult, + FetchContext, + FetchResult, +) from backend.channels.registry import register_channel _URL_RE = re.compile(r"https?://[^\s<>\[\](){}'\"]+", re.IGNORECASE) @@ -17,6 +24,18 @@ "人机验证", "验证码", ) +#: Transient CDP/browser race conditions — retrying the same question in the +#: same session usually succeeds once the navigation settles. Classified as +#: ``ConnectionError`` (retryable in error_taxonomy) so the thick ``fetch()`` +#: retry loop picks them up, while a captcha wall (above) is NOT retried. +_TRANSIENT_CDP_MARKERS = ( + "CDP connection is not open", + "Inspected target navigated or closed", + "Execution context was destroyed", + "Target closed", + "Session closed", + "cloneNode", +) def _citations(text: str) -> list[dict[str, str]]: @@ -66,6 +85,13 @@ def _is_captcha_block(stderr: str, stdout: str) -> bool: return any(marker in text for marker in _CAPTCHA_MARKERS) +def _is_transient_cdp_fault(stderr: str, stdout: str) -> bool: + """True when the adapter hit a CDP/browser race (navigation, closed tab, + session teardown) — retrying the same question usually succeeds.""" + text = f"{stderr} {stdout}" + return any(marker in text for marker in _TRANSIENT_CDP_MARKERS) + + async def _run_doubao_command(command: list[str]) -> tuple[int, str, str]: """Late import avoids the channel registry's legacy OpenCLI import cycle.""" from backend.channels.opencli_channel import _run_opencli @@ -131,9 +157,16 @@ async def collect(self, config: dict[str, Any], parameters: dict[str, Any]) -> C ) if returncode: - # Classify captcha walls so the runner can apply a human-in-the-loop - # or cooldown-retry policy instead of treating it as a permanent failure. - error_type = "captcha_challenge" if _is_captcha_block(stderr, stdout) else None + # Classify: captcha walls (human-cleared — never auto-retry) and + # transient CDP races (retryable) get structured error_types so + # the runner / thick fetch() can act on them; anything else stays + # a generic failure instead of a permanent misclassification. + if _is_captcha_block(stderr, stdout): + error_type = "captcha_challenge" + elif _is_transient_cdp_fault(stderr, stdout): + error_type = "ConnectionError" + else: + error_type = None return ChannelResult.fail( f"opencli doubao ask exited with code {returncode}: {stderr[:500]}", error_type=error_type, @@ -195,3 +228,54 @@ async def validate_config(self, config: dict[str, Any]) -> list[str]: if str(config.get("question") or "").strip() else ["'question' is required for doubao_research channel"] ) + + async def fetch(self, ctx: FetchContext) -> FetchResult: + """Thick-contract entry point: migrate doubao onto the runner protocol + (``type(chan).fetch is not AbstractChannel.fetch``) and own a bounded + retry on TRANSIENT faults only. + + ``collect()`` stays the single source of truth for prompt construction + and evidence output; this override adds the retry loop around it: + - captcha walls (``captcha_challenge``) are never auto-retried — the + pipeline's captcha branch (PR #65) pauses the source for a human + instead; + - non-retryable failures (``error_taxonomy.is_retryable`` False) fail + immediately; + - transient faults (CDP races classified as ``ConnectionError``, + timeouts) retry with exponential backoff up to ``max_retries`` + (default 3, config key ``max_retries``; ``retry_base_delay`` + seconds, default 2). + + ``ctx.http`` is deliberately not threaded into ``collect()`` — the + transport is a local opencli subprocess, not an HTTP request the + runner's RateLimitedClient was built for (same trade-off documented in + ``opencli_channel.fetch()``). + """ + import asyncio + + from backend.pipeline.error_taxonomy import is_retryable + + # Local captcha marker (equals error_taxonomy.CAPTCHA_CHALLENGE, which + # PR #65 exposes as is_captcha()) — kept inline so this PR merges + # independently of #65. + _CAPTCHA = "captcha_challenge" + + max_retries = int(ctx.config.get("max_retries", 3)) + base_delay = float(ctx.config.get("retry_base_delay", 2.0)) + last: ChannelResult | None = None + for attempt in range(max_retries + 1): + result = await self.collect(ctx.config, ctx.params) + if result.success: + return FetchResult(items=result.items, metadata=result.metadata) + if result.error_type == _CAPTCHA or not is_retryable(result.error_type): + raise ChannelFetchError( + result.error or "doubao collect failed", + error_type=result.error_type, + ) + last = result + if attempt < max_retries: + await asyncio.sleep(base_delay * (2**attempt)) + raise ChannelFetchError( + last.error or "doubao collect failed", + error_type=last.error_type, + ) diff --git a/backend/channels/douyin_detail_channel.py b/backend/channels/douyin_detail_channel.py index 1f0f82bc..c068e116 100644 --- a/backend/channels/douyin_detail_channel.py +++ b/backend/channels/douyin_detail_channel.py @@ -5,9 +5,28 @@ from typing import Any from urllib.parse import urlparse -from backend.channels.base import AbstractChannel, Capabilities, ChannelResult +from backend.channels.base import ( + AbstractChannel, + Capabilities, + ChannelFetchError, + ChannelResult, + FetchContext, + FetchResult, +) from backend.channels.registry import register_channel +#: Transient CDP/browser race conditions — retrying the same video lookup +#: usually succeeds once navigation settles. Classified as ``ConnectionError`` +#: (retryable in error_taxonomy) so the thick ``fetch()`` retry loop picks +#: them up. +_TRANSIENT_CDP_MARKERS = ( + "CDP connection is not open", + "Inspected target navigated or closed", + "Execution context was destroyed", + "Target closed", + "Session closed", +) + _AWEME_ID_RE = re.compile(r"(?:/video/|/share/video/|[?&]aweme_id=)(\d{15,25})(?:[/?&#]|$)") @@ -96,6 +115,13 @@ def _parse_detail(stdout: str) -> dict[str, Any]: return detail +def _is_transient_cdp_fault(stderr: str, stdout: str) -> bool: + """True when the adapter hit a CDP/browser race (navigation, closed tab, + session teardown) — retrying the same lookup usually succeeds.""" + text = f"{stderr} {stdout}" + return any(marker in text for marker in _TRANSIENT_CDP_MARKERS) + + async def _run_douyin_command(command: list[str]) -> tuple[int, str, str]: """Use OpenCLI's bounded subprocess helper rather than spawning a shell.""" import os @@ -139,7 +165,10 @@ async def collect(self, config: dict[str, Any], parameters: dict[str, Any]) -> C ) if open_code: return ChannelResult.fail( - f"OpenCLI browser open exited with code {open_code}: {open_stderr[:500]}" + f"OpenCLI browser open exited with code {open_code}: {open_stderr[:500]}", + error_type=( + "ConnectionError" if _is_transient_cdp_fault(open_stderr, "") else None + ), ) eval_code, stdout, stderr = await _run_douyin_command( [ @@ -165,7 +194,10 @@ async def collect(self, config: dict[str, Any], parameters: dict[str, Any]) -> C if eval_code: return ChannelResult.fail( - f"OpenCLI browser eval exited with code {eval_code}: {stderr[:500]}" + f"OpenCLI browser eval exited with code {eval_code}: {stderr[:500]}", + error_type=( + "ConnectionError" if _is_transient_cdp_fault(stderr, stdout) else None + ), ) try: item = _detail_item(_parse_detail(stdout), aweme_id) @@ -185,3 +217,36 @@ async def validate_config(self, config: dict[str, Any]) -> list[str]: def identity(self, item: dict[str, Any]) -> str | None: value = item.get("aweme_id") return str(value) if value else None + + async def fetch(self, ctx: FetchContext) -> FetchResult: + """Thick-contract entry point: migrate douyin onto the runner protocol + and own a bounded retry on transient CDP faults (same pattern as + ``doubao_research_channel.fetch()`` — see its docstring for the + rationale). ``collect()`` stays the single source of truth; this + override only adds the retry loop for ``error_taxonomy``-retryable + failures (``max_retries`` default 3, ``retry_base_delay`` default 2s, + exponential backoff). Permanent failures fail immediately. + """ + import asyncio + + from backend.pipeline.error_taxonomy import is_retryable + + max_retries = int(ctx.config.get("max_retries", 3)) + base_delay = float(ctx.config.get("retry_base_delay", 2.0)) + last: ChannelResult | None = None + for attempt in range(max_retries + 1): + result = await self.collect(ctx.config, ctx.params) + if result.success: + return FetchResult(items=result.items, metadata=result.metadata) + if not is_retryable(result.error_type): + raise ChannelFetchError( + result.error or "douyin detail collect failed", + error_type=result.error_type, + ) + last = result + if attempt < max_retries: + await asyncio.sleep(base_delay * (2**attempt)) + raise ChannelFetchError( + last.error or "douyin detail collect failed", + error_type=last.error_type, + ) diff --git a/tests/unit/channels/test_doubao_research_channel.py b/tests/unit/channels/test_doubao_research_channel.py index d2475c92..50a4e9f5 100644 --- a/tests/unit/channels/test_doubao_research_channel.py +++ b/tests/unit/channels/test_doubao_research_channel.py @@ -1,5 +1,6 @@ import pytest +from backend.channels.base import ChannelFetchError, ChannelResult, FetchContext from backend.channels.doubao_research_channel import ( DoubaoResearchChannel, _citations, @@ -148,3 +149,129 @@ def test_source_schema_accepts_doubao_research_channel(): ) assert source.channel_type == "doubao_research" + +# ── thick fetch(): bounded retry on transient faults (PR-thick-fetch) ──── + + +def _ctx(**config) -> FetchContext: + cfg = {"question": "x", "max_retries": 3, "retry_base_delay": 0.001} + cfg.update(config) + return FetchContext(config=cfg, params={}) + + +@pytest.mark.asyncio +async def test_fetch_retries_transient_then_succeeds(monkeypatch): + channel = DoubaoResearchChannel() + results = [ + ChannelResult.fail("CDP connection is not open", error_type="ConnectionError"), + ChannelResult.fail("CDP connection is not open", error_type="ConnectionError"), + ChannelResult.ok([{"title": "回答"}]), + ] + calls: list[int] = [] + + async def fake_collect(config, parameters): + calls.append(1) + return results.pop(0) + + monkeypatch.setattr(channel, "collect", fake_collect) + out = await channel.fetch(_ctx()) + + assert out.items == [{"title": "回答"}] + assert len(calls) == 3 # 2 failures + 1 success + + +@pytest.mark.asyncio +async def test_fetch_gives_up_after_max_retries(monkeypatch): + channel = DoubaoResearchChannel() + calls: list[int] = [] + + async def fake_collect(config, parameters): + calls.append(1) + return ChannelResult.fail("CDP connection is not open", error_type="ConnectionError") + + monkeypatch.setattr(channel, "collect", fake_collect) + + with pytest.raises(ChannelFetchError) as ei: + await channel.fetch(_ctx(max_retries=3)) + + assert ei.value.error_type == "ConnectionError" + assert len(calls) == 4 # initial + 3 retries + + +@pytest.mark.asyncio +async def test_fetch_does_not_retry_captcha(monkeypatch): + channel = DoubaoResearchChannel() + calls: list[int] = [] + + async def fake_collect(config, parameters): + calls.append(1) + return ChannelResult.fail("verification challenge", error_type="captcha_challenge") + + monkeypatch.setattr(channel, "collect", fake_collect) + + with pytest.raises(ChannelFetchError) as ei: + await channel.fetch(_ctx()) + + assert ei.value.error_type == "captcha_challenge" + assert len(calls) == 1 # captcha must never auto-retry + + +@pytest.mark.asyncio +async def test_fetch_does_not_retry_permanent_errors(monkeypatch): + channel = DoubaoResearchChannel() + calls: list[int] = [] + + async def fake_collect(config, parameters): + calls.append(1) + return ChannelResult.fail("bad config", error_type="ValueError") + + monkeypatch.setattr(channel, "collect", fake_collect) + + with pytest.raises(ChannelFetchError) as ei: + await channel.fetch(_ctx()) + + assert ei.value.error_type == "ValueError" + assert len(calls) == 1 + + +@pytest.mark.asyncio +async def test_fetch_success_passes_metadata_through(monkeypatch): + channel = DoubaoResearchChannel() + async def fake_collect(config, parameters): + return ChannelResult.ok( + [{"title": "x"}], citation_count=3, citation_capture="answer_url_extraction" + ) + + monkeypatch.setattr(channel, "collect", fake_collect) + out = await channel.fetch(_ctx()) + + assert out.metadata["citation_count"] == 3 + assert out.metadata["citation_capture"] == "answer_url_extraction" + + +# ── transient CDP classification in collect (PR-thick-fetch) ───────────── + + +@pytest.mark.asyncio +async def test_collect_classifies_cdp_transient_as_connection_error(monkeypatch): + async def fake_run(command): + return 1, "", "error: CDP connection is not open" + + monkeypatch.setattr("backend.channels.doubao_research_channel._run_doubao_command", fake_run) + result = await DoubaoResearchChannel().collect({"question": "测试"}, {}) + + assert not result.success + assert result.error_type == "ConnectionError" # retryable per error_taxonomy + + +@pytest.mark.asyncio +async def test_collect_classifies_target_navigated_as_connection_error(monkeypatch): + async def fake_run(command): + return 1, "", "Inspected target navigated or closed" + + monkeypatch.setattr("backend.channels.doubao_research_channel._run_doubao_command", fake_run) + result = await DoubaoResearchChannel().collect({"question": "测试"}, {}) + + assert not result.success + assert result.error_type == "ConnectionError" + diff --git a/tests/unit/channels/test_douyin_detail_channel.py b/tests/unit/channels/test_douyin_detail_channel.py index 338179f2..74b0d29e 100644 --- a/tests/unit/channels/test_douyin_detail_channel.py +++ b/tests/unit/channels/test_douyin_detail_channel.py @@ -2,6 +2,7 @@ import pytest +from backend.channels.base import ChannelFetchError, ChannelResult, FetchContext from backend.channels.douyin_detail_channel import DouyinDetailChannel, _aweme_id @@ -61,3 +62,68 @@ async def should_not_run(command): assert not result.success assert "url" in (result.error or "") + + +# ── transient classification + thick fetch() (PR-thick-fetch) ───────────── + + +def _ctx(**config) -> FetchContext: + cfg = { + "url": "https://www.douyin.com/video/7664819289043537167", + "max_retries": 3, + "retry_base_delay": 0.001, + } + cfg.update(config) + return FetchContext(config=cfg, params={}) + + +@pytest.mark.asyncio +async def test_collect_classifies_cdp_transient_as_connection_error(monkeypatch): + async def fake_run(command): + return 1, "", "CDP connection is not open" + + monkeypatch.setattr("backend.channels.douyin_detail_channel._run_douyin_command", fake_run) + result = await DouyinDetailChannel().collect( + {"url": "https://www.douyin.com/video/7664819289043537167"}, {} + ) + + assert not result.success + assert result.error_type == "ConnectionError" + + +@pytest.mark.asyncio +async def test_fetch_retries_transient_then_succeeds(monkeypatch): + channel = DouyinDetailChannel() + results = [ + ChannelResult.fail("CDP connection is not open", error_type="ConnectionError"), + ChannelResult.ok([{"title": "ok", "aweme_id": "7664819289043537167"}]), + ] + calls: list[int] = [] + + async def fake_collect(config, parameters): + calls.append(1) + return results.pop(0) + + monkeypatch.setattr(channel, "collect", fake_collect) + out = await channel.fetch(_ctx()) + + assert out.items[0]["aweme_id"] == "7664819289043537167" + assert len(calls) == 2 + + +@pytest.mark.asyncio +async def test_fetch_gives_up_after_max_retries(monkeypatch): + channel = DouyinDetailChannel() + calls: list[int] = [] + + async def fake_collect(config, parameters): + calls.append(1) + return ChannelResult.fail("CDP connection is not open", error_type="ConnectionError") + + monkeypatch.setattr(channel, "collect", fake_collect) + + with pytest.raises(ChannelFetchError) as ei: + await channel.fetch(_ctx(max_retries=2)) + + assert ei.value.error_type == "ConnectionError" + assert len(calls) == 3