From 0a190bc8315e3a5c7b74b9ec21a9919d96ae8f96 Mon Sep 17 00:00:00 2001 From: Curry Date: Sat, 25 Jul 2026 16:28:05 +0800 Subject: [PATCH 1/4] feat(channels): migrate opencli channel to fetch() contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit What: OpenCLIChannel now overrides fetch() (narrow delegation to the existing collect(), mirroring BrowserActChannel's pattern) instead of relying on AbstractChannel's default adapter. capabilities.default_rate is now spelled out explicitly (60/min, unchanged value); incremental/ paginated stay False — opencli's site/command catalog is an external binary probed via --help at runtime, with no cursor or page-token contract anywhere in this codebase to drive the runner's pagination loop against. Why: flips channel_runner.run_channel's `type(chan).fetch is not AbstractChannel.fetch` check so opencli is treated as migrated: the runner now builds it a RateLimitedClient from the declared rate (same accepted no-op cost BrowserActChannel documents, since opencli's transport is local subprocess / LAN-agent HTTP-WS, not ctx.http's public-API shape) and a raised failure's error_type reaches error_taxonomy.effective_error_type directly rather than through the identical inherited default. collect() is untouched — same site-routing/auth logic, same observable behavior; fetch() is pure addition. Tests: 6 new unit tests covering fetch() success via mocked _run_opencli, retryable (TimeoutError) vs permanent (FileNotFoundError) error-taxonomy classification through fetch()'s ChannelFetchError, collect() not routing through fetch(), the migration identity check, and run_channel actually building a RateLimitedClient for the real channel. Full tests/unit: 1418 passed (was 1412), same pre-existing 2 GBK-encoding failures in test_nodes_install_script.py (unrelated, backend/api/v1/nodes.py:435 read_text() with no explicit encoding). --- backend/channels/opencli_channel.py | 66 +++++++- tests/unit/channels/test_opencli_channel.py | 173 +++++++++++++++++++- 2 files changed, 235 insertions(+), 4 deletions(-) diff --git a/backend/channels/opencli_channel.py b/backend/channels/opencli_channel.py index c654734..3b83b36 100644 --- a/backend/channels/opencli_channel.py +++ b/backend/channels/opencli_channel.py @@ -14,7 +14,14 @@ import yaml -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 from backend.opencli_runtime import configured_opencli_bin, resolve_opencli_bin @@ -610,12 +617,27 @@ async def _collect_with_opencli_subprocess( @register_channel class OpenCLIChannel(AbstractChannel): - """Collect data by running the opencli CLI tool.""" + """Collect data by running the opencli CLI tool. + + Migrated onto the thick ``fetch()`` contract (see ``fetch()`` below) via the + same narrow-override pattern as ``BrowserActChannel`` (backend.channels. + browser_act_channel): ``collect()`` stays the single source of truth for + site/command routing, and ``fetch()`` only exists to flip channel_runner's + ``channel_migrated`` check. + """ channel_type = "opencli" # Drives a real Chrome from the shared pool → must run on the node holding the # live session; the pipeline resolves a site-keyed browser binding for it. - capabilities = Capabilities(session_affinity=True) + # incremental/paginated stay False: opencli's site/command catalog is an + # external binary discovered at runtime via `--help` (see _get_named_options / + # _command_requires_browser) — there is no cursor or page-token contract for + # it anywhere in this codebase to drive a runner-owned pagination loop against. + # default_rate is spelled out (rather than left to the dataclass default) to + # document it's a deliberate choice, not an oversight: same 60/min every other + # browser-driving channel (BrowserActChannel, SkillChannel) accepts, since + # there's no empirical number specific to opencli to justify a different one. + capabilities = Capabilities(session_affinity=True, default_rate="60/min") async def collect( self, config: dict[str, Any], parameters: dict[str, Any] @@ -787,6 +809,44 @@ async def collect( return result + async def fetch(self, ctx: FetchContext) -> FetchResult: + """Thick-contract entry point, overridden narrowly (same pattern as + ``BrowserActChannel.fetch()``, backend.channels.browser_act_channel:454) so + that ``type(chan).fetch is not AbstractChannel.fetch`` and + ``channel_runner.run_channel`` (backend.pipeline.channel_runner) treats + opencli as migrated: it builds a ``RateLimitedClient`` from + ``capabilities.default_rate`` for the run instead of skipping it, and any + failure's ``error_type`` (already set by ``_collect_with_opencli_subprocess`` + for TimeoutError/FileNotFoundError/OSError/JSON-parse errors — see that + function) reaches ``error_taxonomy.effective_error_type`` directly instead + of relying on the inherited default doing the identical thing implicitly. + + ``collect()`` remains the single source of truth for site/command routing + (direct subprocess vs. LAN-agent HTTP vs. WS-agent dispatch, browser-pool + acquisition, CDP tab snapshot/cleanup) — this body is intentionally just + the inherited default's, because unlike ``BrowserActChannel`` there is no + ``ctx.source_id``-driven credential lookup to thread through: opencli has + no per-source encrypted-credential path (``capabilities.auth_kind`` stays + "none", so ``run_channel`` resolves ``AuthContext`` without a DB hit). + + ``ctx.http`` is deliberately NOT threaded into ``collect()``: opencli's + transport is a local subprocess (direct/cdp/bridge modes) or a LAN-agent + HTTP/WS dispatch to an internal node (``_collect_via_agent`` / + ``_collect_via_ws_agent``) authenticated with the fleet's own bearer + token — neither is the public-API/SSRF-guarded shape ``ctx.http``'s + rate-limited client is built for. Same accepted trade-off + ``BrowserActChannel.fetch()`` documents: the ``RateLimitedClient`` the + runner builds but this channel never reads is one Python object for the + run's duration, not an open socket. + """ + result = await self.collect(ctx.config, ctx.params) + if not result.success: + raise ChannelFetchError( + result.error or f"{self.channel_type} collect failed", + error_type=result.error_type, + ) + return FetchResult(items=result.items, metadata=result.metadata) + async def validate_config(self, config: dict[str, Any]) -> list[str]: errors: list[str] = [] if not config.get("site"): diff --git a/tests/unit/channels/test_opencli_channel.py b/tests/unit/channels/test_opencli_channel.py index e9576aa..af5455d 100644 --- a/tests/unit/channels/test_opencli_channel.py +++ b/tests/unit/channels/test_opencli_channel.py @@ -6,7 +6,12 @@ import pytest from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker -from backend.channels.base import ChannelResult +from backend.channels.base import ( + AbstractChannel, + ChannelFetchError, + ChannelResult, + FetchContext, +) from backend.channels.opencli_channel import ( OpenCLIChannel, _collect_via_agent, @@ -19,6 +24,7 @@ _parse_yaml, _run_opencli, ) +from backend.pipeline.error_taxonomy import effective_error_type, is_retryable def _sessionmaker(db_engine): @@ -1108,3 +1114,168 @@ async def test_collect_subprocess_exception(channel): assert result.success is False assert "Failed to run" in result.error + + +# ── fetch(): thick-contract migration ─────────────────────────────────────── +# +# OpenCLIChannel migrates onto fetch() via the same narrow-override pattern as +# BrowserActChannel (backend.channels.browser_act_channel): fetch() delegates +# straight to collect() — the internal transport (subprocess / LAN-agent HTTP / +# WS-agent dispatch) is unchanged and unmocked-transport-wise identical to the +# collect() tests above; only the entry point differs. + + +def test_opencli_channel_fetch_is_migrated(): + """channel_runner.run_channel's migration check + (`type(chan).fetch is not AbstractChannel.fetch`) must recognize + OpenCLIChannel as migrated now that fetch() is overridden.""" + assert OpenCLIChannel.fetch is not AbstractChannel.fetch + + +@pytest.mark.asyncio +async def test_fetch_returns_items_via_mocked_subprocess(channel): + """fetch() — the entry point run_channel actually calls in production — + returns items from opencli's real collection machinery, mocked at the same + seam (_run_opencli) the collect() tests above use. Never a real subprocess.""" + mock_pool = _make_mock_pool(mode="cdp") + mock_settings = _make_mock_settings(collection_mode="local") + + with ( + patch("backend.browser_pool.get_pool", return_value=mock_pool), + patch("backend.config.get_settings", return_value=mock_settings), + patch( + "backend.channels.opencli_channel._run_opencli", + new=AsyncMock(return_value=(0, '[{"title": "test"}]', "")), + ), + ): + ctx = FetchContext( + config={"site": "example.com", "command": "list", "format": "json"}, + params={}, + ) + result = await channel.fetch(ctx) + + assert result.items == [{"title": "test"}] + + +@pytest.mark.asyncio +async def test_fetch_transport_timeout_classifies_as_retryable(channel): + """A subprocess timeout's error_type ("TimeoutError", set by + _collect_with_opencli_subprocess) must reach error_taxonomy as retryable — + proves fetch()'s ChannelFetchError(error_type=...) plumbing carries the + classification through, not just that collect() sets it right internally.""" + mock_pool = _make_mock_pool(mode="cdp") + mock_settings = _make_mock_settings(collection_mode="local") + + with ( + patch("backend.browser_pool.get_pool", return_value=mock_pool), + patch("backend.config.get_settings", return_value=mock_settings), + patch( + "backend.channels.opencli_channel._run_opencli", + new=AsyncMock(side_effect=TimeoutError()), + ), + ): + ctx = FetchContext(config={"site": "example.com", "command": "list"}, params={}) + with pytest.raises(ChannelFetchError) as exc_info: + await channel.fetch(ctx) + + assert exc_info.value.error_type == "TimeoutError" + assert is_retryable(effective_error_type(exc_info.value)) is True + + +@pytest.mark.asyncio +async def test_fetch_binary_not_found_classifies_as_permanent(channel): + """FileNotFoundError's error_type ("FileNotFoundError") must classify as + permanent through the same fetch()->ChannelFetchError->error_taxonomy path — + retrying a missing binary can't ever succeed.""" + mock_pool = _make_mock_pool(mode="cdp") + mock_settings = _make_mock_settings(collection_mode="local") + + with ( + patch("backend.browser_pool.get_pool", return_value=mock_pool), + patch("backend.config.get_settings", return_value=mock_settings), + patch( + "backend.channels.opencli_channel._run_opencli", + new=AsyncMock(side_effect=FileNotFoundError("binary not found")), + ), + ): + ctx = FetchContext(config={"site": "example.com", "command": "list"}, params={}) + with pytest.raises(ChannelFetchError) as exc_info: + await channel.fetch(ctx) + + assert exc_info.value.error_type == "FileNotFoundError" + assert is_retryable(effective_error_type(exc_info.value)) is False + + +@pytest.mark.asyncio +async def test_collect_does_not_route_through_fetch(channel): + """Legacy path guard: collect() must stay the single source of truth for + site-routing (unlike api_channel/crawl4ai_channel's inverted + collect()-calls-fetch() pattern) — migrating fetch() must not silently + become a rewrite of collect()'s dispatch logic. Same mocked seam and config + as test_collect_local_cdp_success, plus a spy proving fetch() is never + called from inside collect().""" + mock_pool = _make_mock_pool(mode="cdp") + mock_settings = _make_mock_settings(collection_mode="local") + + with ( + patch("backend.browser_pool.get_pool", return_value=mock_pool), + patch("backend.config.get_settings", return_value=mock_settings), + patch( + "backend.channels.opencli_channel._run_opencli", + new=AsyncMock(return_value=(0, '[{"title": "test"}]', "")), + ), + patch.object( + channel, + "fetch", + new=AsyncMock(side_effect=AssertionError("collect() must not call fetch()")), + ) as mock_fetch, + ): + result = await channel.collect( + {"site": "example.com", "command": "list", "format": "json"}, {} + ) + + assert result.success is True + assert result.items == [{"title": "test"}] + mock_fetch.assert_not_called() + + +@pytest.mark.asyncio +async def test_run_channel_builds_rate_limited_client_for_opencli( + opencli_manifest_mocks, +): + """End-to-end proof of the migration through the real runner entry point + (mirrors test_channel_runner.py's + test_migrated_channel_still_builds_rate_limited_client_when_none_injected, + and test_rss_fetch.py's test_run_channel_drives_rss_and_persists_cursor, but + with the real OpenCLIChannel): a migrated channel gets a RateLimitedClient + built from its declared default_rate when the caller injects no http of its + own — even though opencli's fetch() never reads ctx.http (documented + accepted trade-off, same as BrowserActChannel).""" + from types import SimpleNamespace + + from backend.pipeline.channel_runner import run_channel + from backend.pipeline.cursor_store import InMemoryCursorStore + + opencli_manifest_mocks["requires_browser"] = False + mock_settings = _make_mock_settings(collection_mode="local") + source = SimpleNamespace( + id="src-opencli-1", + channel_type="opencli", + channel_config={"site": "bbc", "command": "news", "format": "json"}, + ) + + with ( + patch("backend.config.get_settings", return_value=mock_settings), + patch( + "backend.channels.opencli_channel._run_opencli", + new=AsyncMock(return_value=(0, '[{"title": "news"}]', "")), + ), + patch("backend.pipeline.channel_runner.RateLimitedClient") as mock_rlc, + ): + mock_rlc.return_value.aclose = AsyncMock() + result = await run_channel( + source, {}, channel=OpenCLIChannel(), cursor_store=InMemoryCursorStore() + ) + + assert result.items == [{"title": "news"}] + mock_rlc.assert_called_once() From 4cb6c94ab76192228155f680fe9d8d0d4a84e808 Mon Sep 17 00:00:00 2001 From: Curry Date: Sat, 25 Jul 2026 16:44:57 +0800 Subject: [PATCH 2/4] feat(workflow): demand assembler matches needs against opencli adapter catalog _source_slots_for_need() now tries the ~205-adapter opencli catalog (via list_opencli_adapter_nodes) before falling back to the hardcoded Xiaohongshu/Bilibili keyword floor. Matching is deterministic: exact site id/name token > Chinese alias > description/domain/strategy keyword, capped at 3 slots, read-access adapters only. Catalog access is wrapped in try/except so a load failure (missing binary, subprocess/decode error -- including the known GBK crash in _load_opencli_catalog on this box) is treated as "catalog unavailable" and falls back to the legacy keyword matcher, which is unchanged. New tests mock the catalog boundary and never invoke the real opencli CLI. --- backend/workflow/demand_assembler.py | 110 +++++++++++++ tests/unit/test_demand_assembler.py | 230 +++++++++++++++++++++++++++ 2 files changed, 340 insertions(+) create mode 100644 tests/unit/test_demand_assembler.py diff --git a/backend/workflow/demand_assembler.py b/backend/workflow/demand_assembler.py index 429063e..0e6a3a3 100644 --- a/backend/workflow/demand_assembler.py +++ b/backend/workflow/demand_assembler.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging import re from typing import Any @@ -14,8 +15,11 @@ WorkflowProjectEdge, WorkflowProjectNode, ) +from backend.workflow.opencli_adapter_nodes import list_opencli_adapter_nodes from backend.workflow.patcher import preview_workflow_patch +logger = logging.getLogger(__name__) + def draft_workflow_demand(body: WorkflowDemandDraftRequest) -> WorkflowPatchResponse: """Translate a user collection need into reviewable native-node patches. @@ -416,6 +420,19 @@ def _data_operators_for_need(text: str) -> list[dict[str, Any]]: def _source_slots_for_need(text: str) -> list[dict[str, Any]]: + """Resolve a collection need to native OpenCLI source slots. + + Consults the OpenCLI adapter catalog first (~205 site adapters) so any + catalog-known site can be matched, not just the two hardcoded below. + Catalog access is best-effort and never raises: if it can't be reached + (binary missing, subprocess/decode failure, anything at all) this falls + back to the legacy keyword floor, which is byte-for-byte what this + function did before catalog matching existed. + """ + return _catalog_slots_for_need(text) or _legacy_keyword_slots_for_need(text) + + +def _legacy_keyword_slots_for_need(text: str) -> list[dict[str, Any]]: normalized = text.lower() slots: list[dict[str, Any]] = [] keyword = _keyword_from_need(text) @@ -459,6 +476,99 @@ def _keyword_from_need(text: str) -> str: return value or "热门" +# --- OpenCLI adapter catalog matching -------------------------------------- +# +# Chinese aliases for catalog sites that are commonly typed in Chinese rather +# than by their OpenCLI site slug. An alias for a site that isn't present in +# the loaded catalog is harmless -- it simply never matches anything. +_CATALOG_SITE_ALIASES: dict[str, tuple[str, ...]] = { + "xiaohongshu": ("小红书", "xiaohongshu", "xhs"), + "bilibili": ("哔哩", "bilibili", "b站", "bili"), +} + +_CATALOG_SLOT_CAP = 3 +_CATALOG_TOKEN_PATTERN = re.compile(r"[a-zA-Z一-鿿]+") + + +def _catalog_slots_for_need(text: str) -> list[dict[str, Any]]: + """Match a need against the OpenCLI adapter catalog. + + Deterministic, no LLM. Scores each read-access catalog site against the + need text by (a) exact site id/name token, (b) Chinese alias substring, + (c) description/domain/strategy keyword -- in that priority -- and keeps + only the best-scoring tier per site. Never raises: any catalog access + failure (missing binary, subprocess error, decode error, ...) is caught + and treated as "no catalog available", returning an empty list so the + caller can fall back to the legacy keyword floor. + """ + try: + nodes = list_opencli_adapter_nodes(include_write=False).nodes + except Exception: + logger.debug( + "opencli adapter catalog unavailable for demand matching", exc_info=True + ) + return [] + + normalized = text.strip().lower() + if not normalized or not nodes: + return [] + + keyword = _keyword_from_need(text) + best: dict[str, tuple[int, int, str]] = {} + for index, node in enumerate(nodes): + tier = _catalog_match_tier(normalized, node) + if tier is None: + continue + current = best.get(node.site) + if current is None or tier < current[0]: + best[node.site] = (tier, index, node.command) + + ordered_sites = sorted(best.items(), key=lambda item: (item[1][0], item[1][1])) + slots: list[dict[str, Any]] = [] + for site, (_tier, _index, command) in ordered_sites[:_CATALOG_SLOT_CAP]: + slots.append( + { + "id": _catalog_slot_id(site), + "label": f"{site.title()} {command.title()}".strip(), + "sourceGroup": "opencli", + "site": site, + "command": command, + "args": {"keyword": keyword}, + } + ) + return slots + + +def _catalog_match_tier(normalized: str, node: Any) -> int | None: + site = node.site.lower() + command = (node.command or "").lower() + if (len(site) >= 2 and site in normalized) or ( + len(command) >= 2 and command in normalized + ): + return 0 + for alias in _CATALOG_SITE_ALIASES.get(site, ()): + if alias.lower() in normalized: + return 1 + if any(token in normalized for token in _catalog_description_tokens(node)): + return 2 + return None + + +def _catalog_description_tokens(node: Any) -> list[str]: + haystack = " ".join( + value for value in (node.description, node.domain, node.strategy) if value + ) + return [ + token.lower() + for token in _CATALOG_TOKEN_PATTERN.findall(haystack) + if len(token) >= 2 + ] + + +def _catalog_slot_id(site: str) -> str: + return re.sub(r"[^a-z0-9]+", "-", site.strip().lower()).strip("-") or "source" + + def _unique_node_id(project: WorkflowProject, base: str) -> str: return _unique_id({node.id for node in project.nodes}, base) diff --git a/tests/unit/test_demand_assembler.py b/tests/unit/test_demand_assembler.py new file mode 100644 index 0000000..7f2d1af --- /dev/null +++ b/tests/unit/test_demand_assembler.py @@ -0,0 +1,230 @@ +"""Unit tests for OpenCLI-catalog-aware source-slot matching in demand_assembler. + +These tests mock the catalog boundary (``_load_opencli_catalog``, exactly the +seam the rest of the test suite already patches for opencli adapter tests) +and never invoke the real ``opencli`` CLI. +""" + +from __future__ import annotations + +from typing import Any + +from backend.workflow.demand_assembler import ( + _catalog_slots_for_need, + _legacy_keyword_slots_for_need, + _source_slots_for_need, +) + +_CATALOG_PATCH_TARGET = "backend.workflow.opencli_adapter_nodes._load_opencli_catalog" + + +def _catalog_entry( + site: str, + name: str = "search", + *, + description: str = "", + access: str = "read", +) -> dict[str, Any]: + return { + "site": site, + "name": name, + "description": description, + "access": access, + "browser": False, + "args": [], + "columns": [], + } + + +def _patch_catalog(monkeypatch, entries: list[dict[str, Any]]) -> None: + monkeypatch.setattr(_CATALOG_PATCH_TARGET, lambda: tuple(entries)) + + +def _patch_catalog_raises(monkeypatch, exc: Exception) -> None: + def _raise() -> tuple[dict[str, Any], ...]: + raise exc + + monkeypatch.setattr(_CATALOG_PATCH_TARGET, _raise) + + +def test_catalog_slots_hit_by_exact_site_name_token(monkeypatch): + _patch_catalog( + monkeypatch, + [_catalog_entry("acmenews", "headlines", description="Breaking news wire")], + ) + + slots = _catalog_slots_for_need("抓 acmenews 热门内容") + + assert slots == [ + { + "id": "acmenews", + "label": "Acmenews Headlines", + "sourceGroup": "opencli", + "site": "acmenews", + "command": "headlines", + "args": {"keyword": "acmenews"}, + } + ] + + +def test_catalog_slots_hit_by_chinese_alias(monkeypatch): + _patch_catalog( + monkeypatch, + [_catalog_entry("xiaohongshu", "search", description="Xiaohongshu search endpoint")], + ) + + slots = _catalog_slots_for_need("抓 小红书 热门内容") + + assert slots == [ + { + "id": "xiaohongshu", + "label": "Xiaohongshu Search", + "sourceGroup": "opencli", + "site": "xiaohongshu", + "command": "search", + "args": {"keyword": "热门"}, + } + ] + + +def test_catalog_slots_hit_by_description_keyword(monkeypatch): + _patch_catalog( + monkeypatch, + [_catalog_entry("widgetco", "fetch", description="gadget marketplace listings")], + ) + + slots = _catalog_slots_for_need("抓 gadget 热门内容") + + assert slots == [ + { + "id": "widgetco", + "label": "Widgetco Fetch", + "sourceGroup": "opencli", + "site": "widgetco", + "command": "fetch", + "args": {"keyword": "gadget"}, + } + ] + + +def test_catalog_slots_excludes_write_only_adapters(monkeypatch): + _patch_catalog( + monkeypatch, + [_catalog_entry("acmenews", "publish", description="acmenews publish", access="write")], + ) + + assert _catalog_slots_for_need("抓 acmenews 热门内容") == [] + + +def test_no_match_returns_empty_from_catalog_and_legacy(monkeypatch): + _patch_catalog( + monkeypatch, + [_catalog_entry("widgetco", "fetch", description="gadget marketplace listings")], + ) + + text = "帮我盯着未知平台的更新" + + assert _catalog_slots_for_need(text) == [] + assert _legacy_keyword_slots_for_need(text) == [] + assert _source_slots_for_need(text) == [] + + +def test_catalog_load_raises_falls_back_to_legacy_keywords(monkeypatch): + _patch_catalog_raises(monkeypatch, TypeError("'NoneType' object is not iterable")) + + text = "抓小红书热帖" + + assert _catalog_slots_for_need(text) == [] + assert _source_slots_for_need(text) == _legacy_keyword_slots_for_need(text) + assert _source_slots_for_need(text) == [ + { + "id": "xiaohongshu", + "label": "Xiaohongshu Search", + "sourceGroup": "social", + "site": "xiaohongshu", + "command": "search", + "args": {"keyword": "热门"}, + } + ] + + +def test_catalog_load_empty_falls_back_to_legacy_keywords_for_bilibili(monkeypatch): + _patch_catalog(monkeypatch, []) + + text = "看下B站AI相关的热门帖子" + + assert _source_slots_for_need(text) == _legacy_keyword_slots_for_need(text) + assert [slot["site"] for slot in _source_slots_for_need(text)] == ["bilibili"] + + +def test_catalog_slot_cap_enforced_at_three(monkeypatch): + _patch_catalog( + monkeypatch, + [ + _catalog_entry("sitea", "search", description="alpha site"), + _catalog_entry("siteb", "search", description="beta site"), + _catalog_entry("sitec", "search", description="gamma site"), + _catalog_entry("sited", "search", description="delta site"), + ], + ) + + slots = _catalog_slots_for_need("抓 sitea siteb sitec sited 热门内容") + + assert len(slots) == 3 + assert [slot["site"] for slot in slots] == ["sitea", "siteb", "sitec"] + + +def test_catalog_match_tier_priority_beats_catalog_order(monkeypatch): + # "alpha" sorts before "zylo" (list_opencli_adapter_nodes sorts by site), + # but "zylo" is an exact-token (tier 0) hit while "alpha" only matches on + # a description keyword (tier 2). Tier priority must win over both catalog + # order and alphabetical order. + _patch_catalog( + monkeypatch, + [ + _catalog_entry("alpha", "search", description="gadget reviews"), + _catalog_entry("zylo", "search", description="unrelated stuff"), + ], + ) + + slots = _catalog_slots_for_need("抓 zylo gadget 热门内容") + + assert [slot["site"] for slot in slots] == ["zylo", "alpha"] + + +def test_source_slots_for_need_prefers_catalog_match_when_available(monkeypatch): + _patch_catalog( + monkeypatch, + [_catalog_entry("acmenews", "headlines", description="Breaking news wire")], + ) + + assert _source_slots_for_need("抓 acmenews 热门内容") == _catalog_slots_for_need( + "抓 acmenews 热门内容" + ) + + +def test_legacy_keyword_slots_unchanged_for_both_known_sites(monkeypatch): + # No monkeypatch needed: this exercises the pure legacy function directly, + # which never touches the catalog. + text = "抓小红书和B站AI热帖" + + slots = _legacy_keyword_slots_for_need(text) + + assert slots == [ + { + "id": "xiaohongshu", + "label": "Xiaohongshu Search", + "sourceGroup": "social", + "site": "xiaohongshu", + "command": "search", + "args": {"keyword": "和 AI"}, + }, + { + "id": "bilibili", + "label": "Bilibili Search", + "sourceGroup": "video", + "site": "bilibili", + "command": "search", + "args": {"keyword": "和 AI"}, + }, + ] From c0d991528a2b2397b583e3e00e9414b39935af22 Mon Sep 17 00:00:00 2001 From: Curry Date: Sat, 25 Jul 2026 17:03:53 +0800 Subject: [PATCH 3/4] feat(pipeline): prove per-source dedup is a storage-layer guarantee MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Investigated per task: content_hash (normalizer, unconditional) + DB unique constraint uq_source_content(source_id, content_hash) + pre-insert/in-batch checks + IntegrityError retry already fully enforce per-source dedup at ingest for every write_strategy that owns a DB row (legacy, odp_shadow/dual/primary), independent of the workflow-graph text.deduplicate operator. Verified every call site (LocalExecutor, Celery tasks, plan_ir executor) funnels through the same run_pipeline -> storer path. alembic check: no model/migration drift, no schema change needed. No code gap found, so no new machinery added. Closed the one real test gap (in-batch content_hash duplicate with no identity() — previously only the cross-run and identity-based cases were covered) plus a distinct-items sanity test, and documented the guarantee (including the previously undocumented identity_key column) in docs/schema.md. --- docs/schema.md | 11 ++- tests/unit/pipeline/test_storer.py | 134 +++++++++++++++++++++++++++-- 2 files changed, 139 insertions(+), 6 deletions(-) diff --git a/docs/schema.md b/docs/schema.md index a6e0f16..806b84c 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -17,12 +17,21 @@ SQLAlchemy 模型：[`backend/models/record.py`](../backend/models/record.py) | `normalized_data` | JSON | **下游应消费的标准字段**，详见下表 | | `ai_enrichment` | JSON / NULL | AI processor 写回的 enrichment；schema 由 processor + prompt 决定，无强制契约 | | `content_hash` | TEXT (sha256) | `(source_id, content_hash)` 唯一约束用于去重 | +| `identity_key` | TEXT / NULL | 渠道 `identity()` 提供的稳定原生 id（RSS entry id、tweet id...）；NULL 表示渠道未实现 `identity()`。补充去重键，非 `content_hash` 的替代——同一 identity 命中且内容变化时就地更新该行，而不是插入新行 | | `status` | TEXT | `raw` → `normalized` → `ai_processed` → `notified`；失败为 `error` | | `error_message` | TEXT / NULL | `status='error'` 时填错误描述 | | `created_at` | DATETIME | 入库时间 | | `updated_at` | DATETIME | 末次更新（含 AI 回填） | -唯一约束：`uq_source_content (source_id, content_hash)`。 +唯一约束：`uq_source_content (source_id, content_hash)`。非唯一索引：`ix_collected_records_source_identity (source_id, identity_key)`。 + +### 去重保证 + +按数据源去重是**入库层的强制保证**，不是可选的工作流节点：任何 source（不论通过 `POST /sources` 直接创建，还是工作流编排）在写入这张表时都会经过同一条路径——`backend/pipeline/normalizer.py` 对每条 item 无条件计算 `content_hash`，`backend/pipeline/storer.py` 在写入前按 `source_id` 查重（含同一批次内部去重），并由上面的唯一约束在 DB 层兜底并发竞争。跳过的条数经由 `SinkResult.duplicates` 一路传导到 `SourceMeasurement.duplicates`/`duplicate_rate`（见 [`CONTROL_THEORY_ARCHITECTURE.md`](CONTROL_THEORY_ARCHITECTURE.md)）。 + +工作流图上的 `text.deduplicate` 算子（`backend/workflow/demand_assembler.py`，仅在需求文本命中质量类关键词时才挂载）是完全独立的另一层：面向已入库内容的 AI 语料整理（exact/SimHash 近似去重可配置字段），不是入库去重的前提条件——没有它，入库去重依然生效。 + +写入路径由 `write_strategy` 决定（`backend/pipeline/sinks/strategy.py`）：默认 `legacy` 和 `odp_shadow`/`odp_dual_required`/`odp_primary`（`DualSink`，legacy leg 为准）都会经过上述去重路径；只有 `odp_only`（不落这张表）把去重交给独立的 Rust ODP ingest 服务，语义由该服务自行保证，不在本文档范围内。 ## `normalized_data` JSON 内的标准字段 diff --git a/tests/unit/pipeline/test_storer.py b/tests/unit/pipeline/test_storer.py index 5ebb519..4e651c5 100644 --- a/tests/unit/pipeline/test_storer.py +++ b/tests/unit/pipeline/test_storer.py @@ -26,12 +26,18 @@ async def test_store_new_records(db_session): triples = [ ( {"title": "Article 1"}, - {"title": "Article 1", "url": "", "content": "", "author": "", "published_at": "", "source_id": source.id}, + { + "title": "Article 1", "url": "", "content": "", "author": "", + "published_at": "", "source_id": source.id, + }, "hash_abc123_1", ), ( {"title": "Article 2"}, - {"title": "Article 2", "url": "", "content": "", "author": "", "published_at": "", "source_id": source.id}, + { + "title": "Article 2", "url": "", "content": "", "author": "", + "published_at": "", "source_id": source.id, + }, "hash_abc123_2", ), ] @@ -60,7 +66,10 @@ async def test_store_deduplication(db_session): triple = ( {"title": "Same Article"}, - {"title": "Same", "url": "", "content": "", "author": "", "published_at": "", "source_id": source.id}, + { + "title": "Same", "url": "", "content": "", "author": "", + "published_at": "", "source_id": source.id, + }, "same_hash_xyz", ) @@ -75,6 +84,115 @@ async def test_store_deduplication(db_session): assert skipped2 == 1 +@pytest.mark.asyncio +async def test_store_dedup_within_single_batch(db_session): + """Two triples in the SAME store_records() call sharing a content_hash, + neither previously stored (no identity involved): only one row lands, + same net result as test_store_deduplication's cross-run case above but + within one call — e.g. a channel returning the same item twice in one + fetch (pagination overlap, a feed listing an entry twice). + + Two layers cooperate to guarantee this, both already covered elsewhere: + the in-memory ``seen_in_batch`` fast path (this test's primary target) + skips the second triple before it ever reaches flush(); if that guard + were bypassed, the (source_id, content_hash) unique constraint plus the + per-record retry-on-IntegrityError path (test_store_survives_concurrent_ + race_on_flush) would still catch it at flush() — confirmed empirically + by temporarily disabling seen_in_batch and observing this test still + pass via that path. This test pins the end-to-end guarantee; it does not + by itself distinguish which layer fired.""" + from backend.models.source import DataSource + from backend.models.task import CollectionTask + + source = DataSource( + name="Batch Dedup Source", + channel_type="rss", + channel_config={"feed_url": "https://example.com/feed.xml"}, + ) + db_session.add(source) + await db_session.flush() + + task = CollectionTask(source_id=source.id, trigger_type="manual", parameters={}) + db_session.add(task) + await db_session.flush() + + triples = [ + ( + {"title": "Same Article"}, + { + "title": "Same", "url": "", "content": "", "author": "", + "published_at": "", "source_id": source.id, + }, + "batch_dup_hash", + ), + ( + {"title": "Same Article (repeat)"}, + { + "title": "Same", "url": "", "content": "", "author": "", + "published_at": "", "source_id": source.id, + }, + "batch_dup_hash", + ), + ] + + new_records, skipped = await store_records(db_session, task.id, source.id, triples) + assert len(new_records) == 1 + assert skipped == 1 + + from sqlalchemy import select as sa_select + + from backend.models.record import CollectedRecord + + rows = ( + await db_session.execute( + sa_select(CollectedRecord).where(CollectedRecord.source_id == source.id) + ) + ).scalars().all() + assert len(rows) == 1 # DB agrees: one row, not two + assert rows[0].content_hash == "batch_dup_hash" + + +@pytest.mark.asyncio +async def test_store_distinct_items_in_one_batch_all_land(db_session): + """Sanity complement to the dedup tests above: several genuinely distinct + items in one batch are all stored, none mistaken for duplicates of each + other (proves the guards above key on content_hash equality, not on + batch position or count).""" + from backend.models.source import DataSource + from backend.models.task import CollectionTask + + source = DataSource( + name="Distinct Items Source", + channel_type="rss", + channel_config={"feed_url": "https://example.com/feed.xml"}, + ) + db_session.add(source) + await db_session.flush() + + task = CollectionTask(source_id=source.id, trigger_type="manual", parameters={}) + db_session.add(task) + await db_session.flush() + + triples = [ + ( + {"title": f"Article {i}"}, + { + "title": f"Article {i}", "url": "", "content": "", "author": "", + "published_at": "", "source_id": source.id, + }, + f"distinct_hash_{i}", + ) + for i in range(3) + ] + + new_records, skipped = await store_records(db_session, task.id, source.id, triples) + assert skipped == 0 + assert len(new_records) == 3 + assert {r.content_hash for r in new_records} == { + "distinct_hash_0", "distinct_hash_1", "distinct_hash_2", + } + + @pytest.mark.asyncio async def test_store_empty_input(db_session): new_records, skipped = await store_records(db_session, "task-id", "src-id", []) @@ -118,12 +236,18 @@ async def test_store_survives_concurrent_race_on_flush(db_session): triples = [ ( {"title": "Loser"}, - {"title": "Loser", "url": "", "content": "", "author": "", "published_at": "", "source_id": source.id}, + { + "title": "Loser", "url": "", "content": "", "author": "", + "published_at": "", "source_id": source.id, + }, "race_hash", ), ( {"title": "Clean"}, - {"title": "Clean", "url": "", "content": "", "author": "", "published_at": "", "source_id": source.id}, + { + "title": "Clean", "url": "", "content": "", "author": "", + "published_at": "", "source_id": source.id, + }, "clean_hash", ), ] From 7408814ce41fa2bb5199516eb0a710f4ddeda571 Mon Sep 17 00:00:00 2001 From: Curry Date: Sat, 25 Jul 2026 17:56:22 +0800 Subject: [PATCH 4/4] fix: address CodeRabbit review on acquisition hardening MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - demand_assembler: drop command-only matches from tier 0 — command names repeat across unrelated sites and must not outrank exact site matches - opencli_channel.fetch(): delegate to AbstractChannel.fetch instead of duplicating its body (override kept for the migration identity check) - tests: assert RateLimitedClient token bucket is built from the declared 60/min default_rate; assert fetch() metadata passthrough (chrome_mode) --- backend/channels/opencli_channel.py | 13 +++---------- backend/workflow/demand_assembler.py | 7 +++---- tests/unit/channels/test_opencli_channel.py | 9 +++++++++ 3 files changed, 15 insertions(+), 14 deletions(-) diff --git a/backend/channels/opencli_channel.py b/backend/channels/opencli_channel.py index 3b83b36..03c1b4f 100644 --- a/backend/channels/opencli_channel.py +++ b/backend/channels/opencli_channel.py @@ -17,7 +17,6 @@ from backend.channels.base import ( AbstractChannel, Capabilities, - ChannelFetchError, ChannelResult, FetchContext, FetchResult, @@ -823,8 +822,8 @@ async def fetch(self, ctx: FetchContext) -> FetchResult: ``collect()`` remains the single source of truth for site/command routing (direct subprocess vs. LAN-agent HTTP vs. WS-agent dispatch, browser-pool - acquisition, CDP tab snapshot/cleanup) — this body is intentionally just - the inherited default's, because unlike ``BrowserActChannel`` there is no + acquisition, CDP tab snapshot/cleanup) — this override delegates straight + to the inherited default, because unlike ``BrowserActChannel`` there is no ``ctx.source_id``-driven credential lookup to thread through: opencli has no per-source encrypted-credential path (``capabilities.auth_kind`` stays "none", so ``run_channel`` resolves ``AuthContext`` without a DB hit). @@ -839,13 +838,7 @@ async def fetch(self, ctx: FetchContext) -> FetchResult: runner builds but this channel never reads is one Python object for the run's duration, not an open socket. """ - result = await self.collect(ctx.config, ctx.params) - if not result.success: - raise ChannelFetchError( - result.error or f"{self.channel_type} collect failed", - error_type=result.error_type, - ) - return FetchResult(items=result.items, metadata=result.metadata) + return await AbstractChannel.fetch(self, ctx) async def validate_config(self, config: dict[str, Any]) -> list[str]: errors: list[str] = [] diff --git a/backend/workflow/demand_assembler.py b/backend/workflow/demand_assembler.py index 0e6a3a3..fe13092 100644 --- a/backend/workflow/demand_assembler.py +++ b/backend/workflow/demand_assembler.py @@ -540,11 +540,10 @@ def _catalog_slots_for_need(text: str) -> list[dict[str, Any]]: def _catalog_match_tier(normalized: str, node: Any) -> int | None: + # Command names (node.command) repeat across unrelated sites, so a + # command-only hit must never claim the exact-site tier. site = node.site.lower() - command = (node.command or "").lower() - if (len(site) >= 2 and site in normalized) or ( - len(command) >= 2 and command in normalized - ): + if len(site) >= 2 and site in normalized: return 0 for alias in _CATALOG_SITE_ALIASES.get(site, ()): if alias.lower() in normalized: diff --git a/tests/unit/channels/test_opencli_channel.py b/tests/unit/channels/test_opencli_channel.py index af5455d..0670e3c 100644 --- a/tests/unit/channels/test_opencli_channel.py +++ b/tests/unit/channels/test_opencli_channel.py @@ -1155,6 +1155,9 @@ async def test_fetch_returns_items_via_mocked_subprocess(channel): result = await channel.fetch(ctx) assert result.items == [{"title": "test"}] + # Metadata passthrough matters for opencli specifically (node_url / + # chrome_mode reach the runner through FetchResult.metadata). + assert result.metadata.get("chrome_mode") == "cdp" @pytest.mark.asyncio @@ -1279,3 +1282,9 @@ async def test_run_channel_builds_rate_limited_client_for_opencli( assert result.items == [{"title": "news"}] mock_rlc.assert_called_once() + from backend.pipeline.http_client import TokenBucket, parse_rate + + bucket = mock_rlc.call_args.args[1] + assert isinstance(bucket, TokenBucket) + assert bucket.rate == parse_rate(OpenCLIChannel().capabilities.default_rate) + assert bucket.rate == parse_rate("60/min")