diff --git a/backend/channels/opencli_channel.py b/backend/channels/opencli_channel.py
index c654734..03c1b4f 100644
--- a/backend/channels/opencli_channel.py
+++ b/backend/channels/opencli_channel.py
@@ -14,7 +14,13 @@
 
 import yaml
 
-from backend.channels.base import AbstractChannel, Capabilities, ChannelResult
+from backend.channels.base import (
+    AbstractChannel,
+    Capabilities,
+    ChannelResult,
+    FetchContext,
+    FetchResult,
+)
 from backend.channels.registry import register_channel
 from backend.opencli_runtime import configured_opencli_bin, resolve_opencli_bin
 
@@ -610,12 +616,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 +808,38 @@ 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 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).
+
+        ``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.
+        """
+        return await AbstractChannel.fetch(self, ctx)
+
     async def validate_config(self, config: dict[str, Any]) -> list[str]:
         errors: list[str] = []
         if not config.get("site"):
diff --git a/backend/workflow/demand_assembler.py b/backend/workflow/demand_assembler.py
index 429063e..fe13092 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,98 @@ 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:
+    # Command names (node.command) repeat across unrelated sites, so a
+    # command-only hit must never claim the exact-site tier.
+    site = node.site.lower()
+    if len(site) >= 2 and site 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/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/channels/test_opencli_channel.py b/tests/unit/channels/test_opencli_channel.py
index e9576aa..0670e3c 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,177 @@ 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"}]
+    # 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
+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 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")
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",
         ),
     ]
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"},
+        },
+    ]