feat(channels): migrate doubao/douyin to thick fetch() with bounded retry (#66)

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.

Co-authored-by: 1012839419a-alt <1012839419a-alt@users.noreply.github.com>

Author: 1012839419a-alt

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,
+        )

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,
+        )

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"
+