feat: acquisition hardening — opencli fetch() migration, catalog-driven demand matching, dedup proof - #42
Conversation
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).
…r 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.
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.
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
📝 WalkthroughWalkthroughOpenCLIChannel now implements the fetch contract with rate configuration and typed errors. Demand assembly uses OpenCLI adapter catalog matching before legacy fallback. Schema documentation and storer tests expand identity and batch deduplication coverage. ChangesOpenCLI fetch migration
Catalog-aware demand assembly
Ingestion deduplication documentation and tests
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
✅ Health: 7.1 📋 At a glance Files & modules (2)
📌 Before you merge
🗺️ Change map flowchart LR
subgraph PR ["Changed in this PR (2 with dependents)"]
f_backend_channels_opencli_channel_py["backend/channels/opencli_channel.py 🔥"]:::changed
f_tests_unit_pipeline_test_storer_py[".../pipeline/test_storer.py"]:::changed
end
f_backend_agent_server_py["backend/agent_server.py"]
f_backend_channels_opencli_channel_py --> f_backend_agent_server_py
f_backend_api_v1___init___py[".../v1/__init__.py"]
f_backend_channels_opencli_channel_py --> f_backend_api_v1___init___py
f_backend_api_v1_browsers_py[".../v1/browsers.py"]
f_backend_channels_opencli_channel_py --> f_backend_api_v1_browsers_py
f_backend_api_v1_nodes_py[".../v1/nodes.py"]
f_backend_channels_opencli_channel_py --> f_backend_api_v1_nodes_py
more(["+2 more dependents"])
PR --> more
w_backend_agent_server_py(["⚠️ backend/agent_server.py changed together 10×, not in PR"]):::warn
f_backend_channels_opencli_channel_py -.- w_backend_agent_server_py
t_tests_integration_test_opencli_channel_api_py(["✅ tests/integration/test_opencli_channel_api.py"]):::guard
t_tests_integration_test_opencli_channel_api_py -.-> f_backend_channels_opencli_channel_py
t_tests_conftest_py(["✅ tests/conftest.py"]):::guard
t_tests_conftest_py -.-> f_tests_unit_pipeline_test_storer_py
classDef changed fill:#dbeafe,stroke:#1d4ed8,color:#1e3a5f
classDef warn fill:#fef3c7,stroke:#b45309,color:#78350f
classDef guard fill:#dcfce7,stroke:#15803d,color:#14532d
Solid arrows: code that imports the changed files (6 direct dependents, from the last indexed snapshot). Dashed: history/tests. 🚨 Change risk: 9.4/10 (high)
🔎 More signals (2)🔥 Hotspots touched (2)
🔗 Hidden coupling (1 file)
👀 Suggested reviewers @xujinghua 📊 Full report · ⭐ Star Repowise · 📥 Install bot · Last updated 2026-07-25 09:56 UTC |
|
@coderabbitai review |
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
backend/channels/opencli_channel.py (1)
812-849: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDelegate instead of duplicating
AbstractChannel.fetch()'s body.This override exists purely to flip
type(chan).fetch is not AbstractChannel.fetchforchannel_runner's migration check — the docstring says so explicitly. But the body (Lines 842-848) is a verbatim copy ofAbstractChannel.fetch()'s default implementation (backend/channels/base.py). If that default ever changes (e.g. different metadata merging, different error wrapping), this copy silently drifts out of sync with no test or type system enforcement to catch it.Delegating to the base implementation achieves the same identity-check goal with zero duplication risk:
♻️ Proposed refactor
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)(
type(chan).fetchstill resolves toOpenCLIChannel.fetch, a distinct function object fromAbstractChannel.fetch, so the migration check still passes — this just removes the duplicated logic.)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/channels/opencli_channel.py` around lines 812 - 849, Update OpenCLIChannel.fetch to delegate directly to AbstractChannel.fetch instead of duplicating its collect, error-wrapping, and FetchResult construction logic. Preserve the override itself so type(chan).fetch remains distinct from AbstractChannel.fetch for the migration check, and pass through the existing context unchanged.tests/unit/channels/test_opencli_channel.py (2)
1242-1281: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAssertion doesn't verify the claimed rate value.
The docstring says this proves the client is "built from its declared default_rate", but
mock_rlc.assert_called_once()only checks thatRateLimitedClientwas constructed once — it doesn't check the constructor was actually given a token bucket configured for"60/min". A regression that hardcoded a different rate elsewhere wouldn't be caught here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/channels/test_opencli_channel.py` around lines 1242 - 1281, Update test_run_channel_builds_rate_limited_client_for_opencli to inspect the arguments passed to RateLimitedClient and assert its token-bucket configuration uses the OpenCLI channel’s declared default_rate of “60/min”, while preserving the existing construction-count assertion and result checks.
1135-1157: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider also asserting metadata passthrough.
The test only checks
result.items;FetchResult.metadataforwarding (called out inbackend/channels/base.py'sFetchResultdocstring as important for opencli'snode_url/chrome_mode) isn't exercised here. Not a functional gap, just an easy strengthening of this specific test's coverage.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/channels/test_opencli_channel.py` around lines 1135 - 1157, Strengthen test_fetch_returns_items_via_mocked_subprocess by asserting that result.metadata preserves the metadata returned through the mocked opencli fetch flow, including opencli-relevant fields such as node_url and chrome_mode. Keep the existing items assertion and use the established FetchResult metadata shape.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/workflow/demand_assembler.py`:
- Around line 542-554: Update _catalog_match_tier so tier 0 is awarded only when
the normalized text contains a valid site token; remove the command-based
condition from that tier. Leave alias matching and description-token matching
unchanged.
---
Nitpick comments:
In `@backend/channels/opencli_channel.py`:
- Around line 812-849: Update OpenCLIChannel.fetch to delegate directly to
AbstractChannel.fetch instead of duplicating its collect, error-wrapping, and
FetchResult construction logic. Preserve the override itself so type(chan).fetch
remains distinct from AbstractChannel.fetch for the migration check, and pass
through the existing context unchanged.
In `@tests/unit/channels/test_opencli_channel.py`:
- Around line 1242-1281: Update
test_run_channel_builds_rate_limited_client_for_opencli to inspect the arguments
passed to RateLimitedClient and assert its token-bucket configuration uses the
OpenCLI channel’s declared default_rate of “60/min”, while preserving the
existing construction-count assertion and result checks.
- Around line 1135-1157: Strengthen
test_fetch_returns_items_via_mocked_subprocess by asserting that result.metadata
preserves the metadata returned through the mocked opencli fetch flow, including
opencli-relevant fields such as node_url and chrome_mode. Keep the existing
items assertion and use the established FetchResult metadata shape.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 41c7267b-eddf-4f4b-9ddb-e9cb3e9c74bb
📒 Files selected for processing (6)
backend/channels/opencli_channel.pybackend/workflow/demand_assembler.pydocs/schema.mdtests/unit/channels/test_opencli_channel.pytests/unit/pipeline/test_storer.pytests/unit/test_demand_assembler.py
- 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)
1 participant
What changed
fetch()contract: runner-level error taxonomy + rate-limited client wiring;collect()behavior untouched (0a190bc)list_opencli_adapter_nodes(include_write=False)): exact site/command > Chinese alias > description keyword, deterministic order, capped at 3 slots; catalog failure fails soft to the legacy Xiaohongshu/Bilibili keyword floor; no-match still fail-closed torequest_missing_capability(4cb6c94)uq_source_content+ unconditional content_hash + in-batch dedup + SAVEPOINT retry + duplicates measurement chain) — added proof tests + schema docs instead of a redundant second mechanism (c0d9915)Why
Acquisition-layer hardening pass: retry/limit machinery coverage for the biggest source surface, needs-to-source strategy upgraded from a 2-entry keyword table to catalog lookup, and the dedup guarantee made explicit and regression-locked.
Validation
tests/unit: 1431 passed, 1 skipped (2 pre-existing Windows-only GBK env failures excluded, tracked separately)Notes
Stacked on #41 (
codex/dataflow-native-cleaning, base of this PR). Retarget tomainafter #41 merges.