From 650f35b74af10c7832bb6203ce4f665e81af0ba6 Mon Sep 17 00:00:00 2001 From: Curry Date: Sun, 19 Jul 2026 02:37:00 +0800 Subject: [PATCH] fix(collection): wire identity dedup + chunk IN() + kill process tree + help cache TTL + safe CDP cleanup [C7,C15,C16,C17,C20] --- .claude/CLAUDE.md | 123 +++++++++++ .claude/launch.json | 17 ++ .../hud-cache.json | 1 + .../hud-cache.json | 1 + .../hud-cache.json | 1 + backend/channels/cli_channel.py | 20 +- backend/channels/opencli_channel.py | 98 +++++++-- ...5_add_identity_key_to_collected_records.py | 42 ++++ backend/models/record.py | 18 +- backend/pipeline/pipeline.py | 12 +- backend/pipeline/sinks/legacy_db_sink.py | 26 +++ backend/pipeline/storer.py | 147 +++++++++++-- tests/unit/channels/test_cli_channel.py | 44 +++- tests/unit/channels/test_opencli_channel.py | 176 +++++++++++++++ tests/unit/pipeline/test_legacy_db_sink.py | 61 ++++++ tests/unit/pipeline/test_pipeline.py | 106 +++++++++ tests/unit/pipeline/test_storer.py | 202 ++++++++++++++++++ 17 files changed, 1050 insertions(+), 45 deletions(-) create mode 100644 .claude/CLAUDE.md create mode 100644 .claude/launch.json create mode 100644 .omc/state/sessions/b4f7d02c-1da1-4b4d-9835-33e44401355e/hud-cache.json create mode 100644 .omc/state/sessions/dc752497-c4fd-4221-ae35-e78f40e96767/hud-cache.json create mode 100644 .omc/state/sessions/de0b7a77-e9c7-46ac-961a-f0e90ffa1f07/hud-cache.json create mode 100644 backend/migrations/versions/u0a1b2c3d4e5_add_identity_key_to_collected_records.py diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md new file mode 100644 index 0000000..b9789df --- /dev/null +++ b/.claude/CLAUDE.md @@ -0,0 +1,123 @@ +# CLAUDE.md + + + + + +## IMPORTANT: Codebase Intelligence Instructions for opencli-admin + +> This repository is indexed by [Repowise](https://repowise.dev). +> Use the MCP tools below for orientation, discovery, and enriched context +> (documentation, ownership, history, decisions). **Always verify against +> actual source files before making changes** — the index may be stale. + +Last indexed: 2026-07-12 (commit c326a2a) +### Entry Points +- `backend/main.py` +- `odp-rs/crates/odp-ingest/src/main.rs` +- `odp-rs/crates/odp-store/src/main.rs` +- `iii/workers/collector-discord/src/main.py` +- `iii/workers/collector-opencli/src/main.py` +- `iii/workers/odp-ingest-bridge/src/main.py` +- `iii/workers/schedule-bootstrap/src/main.py` +- `odp-rs/crates/odp-contracts/src/lib.rs` +### Tech Stack +**Languages:** Python, TypeScript +**Frameworks:** FastAPI, HTTPX, Pydantic +**Databases:** Alembic, SQLAlchemy +**Infra:** Celery, Docker, Docker Compose### Architectural Layers +| Layer | Files | Purpose | +|-------|-------|---------| +| Application | 431 | | +| API | 48 | | +| UI | 84 | | +| Config | 4 | | +| Types | 22 | | +| Data | 25 | | +| Service | 83 | | +| Docs & Tooling | 197 | | +| Test | 170 | | + +### Guided Tour (12 steps) +1. `README_HANDOVER.md` +2. `main.py` +3. `main.rs` +4. `main.rs` +5. `frontend.mjs` +6. `schema.ts` +... and 6 more steps +### Hotspots (High Churn) +| File | Churn | 90d Commits | Owner | +|------|-------|-------------|-------| +| `backend/channels/opencli_channel.py` | 100.0th %ile | 8 | xujinghua | +| `tests/integration/test_workflow_compile_api.py` | 99.9th %ile | 15 | lunnt | +| `backend/workflow/opencli_hda_tracer.py` | 99.8th %ile | 6 | lunnt | +| `tests/integration/test_workflow_opencli_hda_trace_api.py` | 99.6th %ile | 9 | lunnt | +| `frontend/lib/flow/store.ts` | 99.5th %ile | 7 | lunnt | + +## Code health +Three signals: **defect risk** (the overall score), **maintainability** (smells that hurt readability/change-cost without predicting bugs), and **performance** (static performance RISK: I/O-in-loop / N+1 shapes that waste work, high-precision/low-recall). Maintainability and performance are co-equal views, never blended into the defect headline. See `docs/CODE_HEALTH.md`. + +Defect risk, Hotspot health: 5.42/10 (stable) · +Average: 8.0/10 · +Worst: 1.0/10 (`frontend/lib/flow/store.ts`) +Maintainability, Average: 8.88/10 +Performance risk, Average: 9.93/10 + +### Critical biomarkers +- `frontend/components/flow/workflow-editor.tsx` — churn risk — impact −2.3 +- `tests/integration/test_workflow_opencli_hda_trace_api.py` — change entropy — impact −2.3 +- `backend/schemas/control.py` — change entropy — impact −2.1 +- `backend/config.py` — untested hotspot — impact −2.0 +- `frontend/components/ui/card.tsx` — untested hotspot — impact −2.0 + +### Repowise MCP Tools + +This repo has the Repowise MCP server configured. The tools below answer questions `grep`/`Read` cannot. Every response carries an `_meta` envelope with `index_age_days`, `indexed_commit`, and a `stale_warning` only when the index has actually diverged from HEAD — silence means the index is current. + +**When to call which tool:** + +| Tool | What only this tool answers | +|------|------------------------------| +| `get_answer(question)` | Synthesised answer with citations and a content-grounded `confidence`. First call for "how does X work" / "where is Y" / "why is Z" — and the one-shot for "explain function/method/class X": when the question names an indexed symbol, get_answer anchors its defining file (even when fuzzy retrieval would miss it) and returns the symbol's full live body in `symbol_bodies` — read that instead of a `get_symbol` follow-up. Value questions may return `grounding: "extracted"` (verbatim source line). On low confidence returns `best_guesses` with one-line justifications, plus `code_rationale` (rationale comments mined live from the candidate source) when the "why" lives in a code comment the wiki never captured. | +| `get_context(targets=[...])` | Triage card for files/modules/symbols — title, summary, signatures, `hotspot` bit, `decision_records` titles, `symbol_id`s. File targets auto-upgrade to a `verified` skeleton (every signature, ~37% of a full Read). `include=["callers"]` works on file targets too (import + call rollup). | +| `get_symbol(...)` | Source bytes with live-verified bounds. Three forms: `"path.py::Name"` (indexed symbol), `"path.py:140-180"` (live range read, ≤200 lines), `"repowise#"` (omission ref). Index misses return `fallback_lines` from a live grep instead of a dead end. | +| `search_codebase(query, mode?, kind?, symbol_kind?)` | Hybrid code search. `mode="auto"` (default) routes by query shape: an identifier → indexed symbol hits (`symbol_id`/`file`/line bounds — pipe into `get_symbol`), a path → file pages (pipe into `get_context`), prose → wiki-semantic search, mixed → hybrid (symbols first). Force a branch with `mode=symbol\|path\|concept\|hybrid`. Concept hits carry `search_method` (`embedding` vs `bm25` fallback); decision records rank below file pages unless the query is why-shaped. | +| `get_why(query, targets?)` | Architectural decision archaeology — *why* the code is shaped this way. Call before refactors or pattern divergences. Falls back to git archaeology when no ADRs exist for a file, and to `code_rationale` (rationale comments mined from the source) when neither decisions nor git history explain it. | +| `get_risk(targets, changed_files?)` | What history says about touching these files: churn, owners, blast radius. Pass `changed_files` for PR mode → returns a `directive` (`will_break`, `missing_cochanges`, `missing_tests`). | +| `get_health(targets?, include?)` | Code-health scores + biomarker findings (defect / maintainability / performance pillars). Self-check before a PR — read the same signals the merge-gate judges your change on. Default is lean; opt in with `include`: `["accuracy"]` (does the score rank the buggy files first — precision@K + `lift`), `["signals"]` (per-file prior-defects / churn / owners / degree, targeted mode), `["churn_complexity"]` (volatile-and-complex danger-zone files), `["biomarkers"]` (all findings), and a dimension name `["performance"]` / `["defect"]` / `["maintainability"]` to filter findings to one pillar. | +| `get_dead_code(...)` | Tiered unreachable / unused-export / zombie-package findings. Run before a cleanup sprint, not before a targeted fix. | +| `get_overview(repo?)` | Architecture map + `tool_guide` recipes. One-time orientation; skip on subsequent calls in the same session. | + +**Trust protocol — when a response replaces reading the source:** +- `verified: true` on any response means the served content was checked against the live working tree. **Never follow a verified response with a Read of the same lines** — you would be paying twice for identical bytes. +- `get_answer` with `confidence: "high"` or `grounding: "extracted"` is content-grounded (asserted values were verified against retrieved source; ≥1 citation is source-backed). Cite it directly. `quotes` entries `{path, lines, quote}` are verbatim live source — quote them instead of re-reading. `symbol_bodies` entries `{path, name, lines, source}` are the full live body of a named definition — read that body, do not call `get_symbol` for the same symbol (a `continuation` token names the next range if it was truncated). +- `code_rationale` entries `{path, lines, comment}` (on a low-confidence `get_answer` or a fallen-back `get_why`) are rationale comments read live from the source. When the "why" you asked for is a code comment rather than an ADR, the comment is right there: cite it instead of opening the file to hunt for it. +- Reading code: `get_context` skeleton first (~37% of a full Read), then `get_symbol` for bodies, `"path.py:a-b"` range reads for anything between symbols. Raw `Read` is for files the index marks `mostly_full` or cannot serve. On a **large file** `get_symbol("path::Name")` is the win — it serves one verified body instead of pulling the whole file into context. +- The **only** re-read triggers: `bounds: "approximate"`, `_meta.stale_warning`, `search_method: "bm25"`, or `confidence: "low"`/`retrieval_quality: "weak"`. +- Disallowed rationalizations for extra reads: "just to be safe", "to double-check the tool", "to see the full context" (use the skeleton / a range read), "the file might have changed" (that is what `verified` already checked). +- When a plain `Grep` is the right tool, use it: an exhaustive literal-token sweep (rename every call site, find all occurrences) is one grep and unbeatable on cost. MCP's edge there is not fewer tokens — it is the `callers_total`/`callers_truncated` honesty signal that tells you whether a list is complete; reach for `get_context(include=["callers"])` when silent incompleteness would bite. + +**Composition tips:** +- `get_answer` → if `confidence` is `medium`/`low`, follow `best_guesses[0].file` or `fallback_targets[0]` into `get_context`, then `get_symbol` for bytes. +- `get_context` returns `decision_records` titles → `get_why(targets=[...])` for the rationale; `hotspot: true` → `get_risk` before editing. +- PR review → `get_risk(targets=[...], changed_files=[...])`; read the `directive` block first. +- A `tombstone` error means the file was deleted/renamed since indexing — follow `successor_paths`. + +### Output Distillation + +- Prefer `repowise distill ` for noisy commands — test runs, builds, `git status`/`log`/`diff`, searches, file listings. It runs the command unchanged (exit code preserved) and prints a compact, errors-first rendering; every error line survives. +- Output may contain a marker like `[repowise#a1b2c3d4e5f6: 230 lines omitted (~6.1k tokens); restore: repowise expand a1b2c3d4e5f6]`. The omitted content is fully preserved — run `repowise expand ` to retrieve it, or `repowise expand -q ` for just the matching lines. +- Never re-run a command to see omitted output; expand the marker instead. +- For structure-level questions about a large indexed file ("what's in here", "which function handles X"), `get_context(["path"], include=["skeleton"])` returns the file with bodies elided — every signature plus the bodies of the most central symbols — at a fraction of the cost of a full Read. + +### Codebase Conventions +**Commands:** +- Build: `npm run build` +- Test: `npm run test` +- Lint: `npm run lint` +- Dev: `npm run dev` +- Format: `npm run format` +- Typecheck: `npm run typecheck` + + diff --git a/.claude/launch.json b/.claude/launch.json new file mode 100644 index 0000000..24db2ba --- /dev/null +++ b/.claude/launch.json @@ -0,0 +1,17 @@ +{ + "version": "0.0.1", + "configurations": [ + { + "name": "frontend-dev", + "runtimeExecutable": "pnpm", + "runtimeArgs": ["-C", "frontend", "exec", "next", "dev", "-p", "8030"], + "port": 8030 + }, + { + "name": "backend-stack", + "runtimeExecutable": "docker", + "runtimeArgs": ["compose", "up", "api", "agent-1"], + "port": 8031 + } + ] +} diff --git a/.omc/state/sessions/b4f7d02c-1da1-4b4d-9835-33e44401355e/hud-cache.json b/.omc/state/sessions/b4f7d02c-1da1-4b4d-9835-33e44401355e/hud-cache.json new file mode 100644 index 0000000..05a59bf --- /dev/null +++ b/.omc/state/sessions/b4f7d02c-1da1-4b4d-9835-33e44401355e/hud-cache.json @@ -0,0 +1 @@ +{"session_id":"b4f7d02c-1da1-4b4d-9835-33e44401355e","context_samples":[{"ts_ms":1783878055117,"tokens":109500},{"ts_ms":1783878057926,"tokens":109939},{"ts_ms":1783878059350,"tokens":109939},{"ts_ms":1783878062169,"tokens":109939},{"ts_ms":1783878065137,"tokens":112312},{"ts_ms":1783878066527,"tokens":112312},{"ts_ms":1783878069643,"tokens":112437},{"ts_ms":1783878070036,"tokens":112437},{"ts_ms":1783878070541,"tokens":112437},{"ts_ms":1783878071081,"tokens":112437},{"ts_ms":1783878077649,"tokens":113096},{"ts_ms":1783878078111,"tokens":113096},{"ts_ms":1783878083683,"tokens":113513},{"ts_ms":1783878084794,"tokens":113513},{"ts_ms":1783878091897,"tokens":114044},{"ts_ms":1783878093626,"tokens":114044},{"ts_ms":1783878095116,"tokens":114044},{"ts_ms":1783878100551,"tokens":114749},{"ts_ms":1783878117149,"tokens":114749},{"ts_ms":1783878475480,"tokens":116909},{"ts_ms":1783878477603,"tokens":116909},{"ts_ms":1783878481720,"tokens":118189},{"ts_ms":1783878483660,"tokens":118189},{"ts_ms":1783878488025,"tokens":118636},{"ts_ms":1783878488449,"tokens":118636},{"ts_ms":1783878495026,"tokens":119031},{"ts_ms":1783878514349,"tokens":119031},{"ts_ms":1783878521236,"tokens":121158},{"ts_ms":1783878586833,"tokens":121464},{"ts_ms":1783878587475,"tokens":121464},{"ts_ms":1783878597857,"tokens":121836},{"ts_ms":1783878604356,"tokens":122080},{"ts_ms":1783878604705,"tokens":122080},{"ts_ms":1783878614717,"tokens":122341},{"ts_ms":1783878621949,"tokens":122341},{"ts_ms":1783878630484,"tokens":123658}],"last_updated_ms":1783878630484,"session_start_tokens":72839} \ No newline at end of file diff --git a/.omc/state/sessions/dc752497-c4fd-4221-ae35-e78f40e96767/hud-cache.json b/.omc/state/sessions/dc752497-c4fd-4221-ae35-e78f40e96767/hud-cache.json new file mode 100644 index 0000000..c23bf5d --- /dev/null +++ b/.omc/state/sessions/dc752497-c4fd-4221-ae35-e78f40e96767/hud-cache.json @@ -0,0 +1 @@ +{"session_id":"dc752497-c4fd-4221-ae35-e78f40e96767","context_samples":[{"ts_ms":1783877236367,"tokens":0},{"ts_ms":1784175633941,"tokens":0}],"last_updated_ms":1784175633941,"session_start_tokens":null} \ No newline at end of file diff --git a/.omc/state/sessions/de0b7a77-e9c7-46ac-961a-f0e90ffa1f07/hud-cache.json b/.omc/state/sessions/de0b7a77-e9c7-46ac-961a-f0e90ffa1f07/hud-cache.json new file mode 100644 index 0000000..7ff589f --- /dev/null +++ b/.omc/state/sessions/de0b7a77-e9c7-46ac-961a-f0e90ffa1f07/hud-cache.json @@ -0,0 +1 @@ +{"session_id":"de0b7a77-e9c7-46ac-961a-f0e90ffa1f07","context_samples":[{"ts_ms":1783867603857,"tokens":0},{"ts_ms":1783867659239,"tokens":0}],"last_updated_ms":1783867659239,"session_start_tokens":null} \ No newline at end of file diff --git a/backend/channels/cli_channel.py b/backend/channels/cli_channel.py index 35c4c1b..2b6e693 100644 --- a/backend/channels/cli_channel.py +++ b/backend/channels/cli_channel.py @@ -2,6 +2,7 @@ import asyncio import json +import logging import os import re import shlex @@ -9,8 +10,11 @@ from typing import Any from backend.channels.base import AbstractChannel, ChannelResult +from backend.channels.opencli_channel import _kill_subprocess, _process_group_kwargs from backend.channels.registry import register_channel +logger = logging.getLogger(__name__) + _TEMPLATE_RE = re.compile(r"\{\{(\w+)\}\}") @@ -73,16 +77,24 @@ async def collect( stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, + **_process_group_kwargs(), ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) except asyncio.TimeoutError as exc: # Don't orphan the child: wait_for only cancels communicate(); - # the subprocess itself keeps running until explicitly killed. - proc.kill() + # the subprocess itself keeps running until explicitly killed. A + # bare proc.kill() only reaches the direct child — shell-wrapped + # or forked grandchildren survive it — so kill the whole process + # tree instead (same helper opencli_channel uses: taskkill /T /F + # on Windows, killpg on POSIX; the process group is set up above). try: - await proc.wait() + await _kill_subprocess(proc) except Exception: - pass + logger.warning( + "failed to reap CLI subprocess (pid=%s) after timeout kill", + getattr(proc, "pid", "?"), + exc_info=True, + ) return ChannelResult.fail( f"CLI command timed out after {timeout}s", error_type=type(exc).__name__ ) diff --git a/backend/channels/opencli_channel.py b/backend/channels/opencli_channel.py index 872eafb..c654734 100644 --- a/backend/channels/opencli_channel.py +++ b/backend/channels/opencli_channel.py @@ -8,6 +8,7 @@ import os import re import subprocess +import time from typing import Any from urllib.parse import urlparse @@ -23,9 +24,37 @@ # Binary to invoke. Override with OPENCLI_BIN env var if needed. _OPENCLI_BIN = configured_opencli_bin() -# Cache: (bin, site, command) → frozenset of accepted --option names (excluding builtins) -_help_cache: dict[tuple[str, str, str], frozenset[str]] = {} -_browser_requirement_cache: dict[tuple[str, str, str], bool] = {} +# Cache: (bin, site, command) → (cached_at, value). TTL'd (C17): without +# expiry these live for the process lifetime, so an opencli binary upgrade +# that adds/renames options keeps getting routed with the old flag set until +# the admin process restarts. +_CACHE_TTL_SECONDS = 3600 # 1 hour + +_help_cache: dict[tuple[str, str, str], tuple[float, frozenset[str]]] = {} +_browser_requirement_cache: dict[tuple[str, str, str], tuple[float, bool]] = {} + + +def _cache_get( + cache: dict[tuple[str, str, str], tuple[float, Any]], + key: tuple[str, str, str], +) -> Any | None: + """Return the cached value for key if present and not past its TTL, else None.""" + entry = cache.get(key) + if entry is None: + return None + cached_at, value = entry + if time.monotonic() - cached_at > _CACHE_TTL_SECONDS: + del cache[key] + return None + return value + + +def _cache_set( + cache: dict[tuple[str, str, str], tuple[float, Any]], + key: tuple[str, str, str], + value: Any, +) -> None: + cache[key] = (time.monotonic(), value) def _split_routing_parameters( @@ -119,16 +148,31 @@ def _process_group_kwargs() -> dict: return {"start_new_session": True} +def _peek_named_options(bin_path: str, site: str, command: str) -> frozenset[str] | None: + """Cache-only lookup for the --option names accepted by a command. + + Returns None on a cache miss/expiry WITHOUT spawning a subprocess (C17). + For callers that only want a nicer display string when the answer is + already known for free — e.g. the collection-run event-log detail in + pipeline.py, which must not pay a `--help` subprocess on the hot path + just to format a log line. Real dispatch still calls + ``_get_named_options`` (below), which fetches-and-caches for real. + """ + return _cache_get(_help_cache, (bin_path, site, command)) + + async def _get_named_options(bin_path: str, site: str, command: str) -> frozenset[str]: """Return the set of --option names accepted by `opencli `. - Runs `--help` once per (bin, site, command) triple and caches the result. - Falls back to an empty set on any error so the caller can still try running. + Runs `--help` once per (bin, site, command) triple and caches the result + for _CACHE_TTL_SECONDS. Falls back to an empty set on any error so the + caller can still try running. """ import re key = (bin_path, site, command) - if key in _help_cache: - return _help_cache[key] + cached = _cache_get(_help_cache, key) + if cached is not None: + return cached proc = None try: proc = await asyncio.create_subprocess_exec( @@ -150,7 +194,7 @@ async def _get_named_options(bin_path: str, site: str, command: str) -> frozense await _kill_subprocess(proc) logger.debug("could not fetch --help for %s %s: %s", site, command, exc) names = frozenset() - _help_cache[key] = names + _cache_set(_help_cache, key, names) return names @@ -161,8 +205,9 @@ async def _command_requires_browser(bin_path: str, site: str, command: str) -> b browser=True so authenticated/dynamic sites keep the safer routed behavior. """ key = (bin_path, site, command) - if key in _browser_requirement_cache: - return _browser_requirement_cache[key] + cached = _cache_get(_browser_requirement_cache, key) + if cached is not None: + return cached proc = None try: proc = await asyncio.create_subprocess_exec( @@ -194,7 +239,7 @@ async def _command_requires_browser(bin_path: str, site: str, command: str) -> b await _kill_subprocess(proc) logger.debug("could not inspect browser requirement for %s %s: %s", site, command, exc) requires_browser = True - _browser_requirement_cache[key] = requires_browser + _cache_set(_browser_requirement_cache, key, requires_browser) return requires_browser @@ -413,15 +458,24 @@ async def _check_bridge_ready(daemon_host: str, daemon_port: int) -> str | None: return None -async def _snapshot_tab_ids(cdp_endpoint: str) -> set[str]: - """Return the set of tab IDs currently open in Chrome.""" +async def _snapshot_tab_ids(cdp_endpoint: str) -> set[str] | None: + """Return the set of tab IDs currently open in Chrome. + + Returns ``None`` (not an empty set) when the snapshot itself fails (C20): + an empty set means "genuinely no tabs are open right now", which is a + valid baseline to diff against; ``None`` means "we don't know what + existed before", and a caller that can't tell the two apart would treat + every currently-open tab as newly-opened and close tabs the user opened + themselves. + """ import httpx try: async with httpx.AsyncClient(timeout=5) as client: resp = await client.get(f"{cdp_endpoint}/json/list") return {t["id"] for t in resp.json() if "id" in t} - except Exception: - return set() + except Exception as exc: + logger.warning("cdp tab snapshot failed at %s: %s", cdp_endpoint, exc) + return None async def _cleanup_cdp_tabs(cdp_endpoint: str, pre_existing_ids: set[str]) -> None: @@ -704,7 +758,7 @@ async def collect( env["OPENCLI_CDP_ENDPOINT"] = cdp_endpoint logger.info("opencli cdp | cmd=%s cdp=%s", " ".join(cmd), cdp_endpoint) - pre_tab_ids: set[str] = set() + pre_tab_ids: set[str] | None = set() if mode == "cdp": pre_tab_ids = await _snapshot_tab_ids(cdp_endpoint) @@ -719,7 +773,17 @@ async def collect( ) if mode == "cdp": - await _cleanup_cdp_tabs(cdp_endpoint, pre_tab_ids) + if pre_tab_ids is None: + # C20: no trustworthy baseline, so we can't tell newly-opened + # tabs from ones the user already had open — skip cleanup + # this run rather than risk closing the user's own tabs. + logger.warning( + "cdp cleanup skipped for %s: pre-collection tab snapshot " + "failed, can't distinguish new tabs from pre-existing ones", + cdp_endpoint, + ) + else: + await _cleanup_cdp_tabs(cdp_endpoint, pre_tab_ids) return result diff --git a/backend/migrations/versions/u0a1b2c3d4e5_add_identity_key_to_collected_records.py b/backend/migrations/versions/u0a1b2c3d4e5_add_identity_key_to_collected_records.py new file mode 100644 index 0000000..8f81be1 --- /dev/null +++ b/backend/migrations/versions/u0a1b2c3d4e5_add_identity_key_to_collected_records.py @@ -0,0 +1,42 @@ +"""add identity_key to collected_records + +Revision ID: u0a1b2c3d4e5 +Revises: t9y0z1a2b3c4 +Create Date: 2026-07-19 + +Rebased 2026-07-19: originally authored as t9y0z1a2b3c4/down=s8x9y0z1a2b3 in a +parallel worktree; group ④ landed its own t9y0z1a2b3c4 (source_cursor version) +first, so this migration is re-chained after it to keep a single linear head. +""" + +import sqlalchemy as sa +from alembic import op + +revision = "u0a1b2c3d4e5" +down_revision = "t9y0z1a2b3c4" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "collected_records", + sa.Column("identity_key", sa.String(length=512), nullable=True), + ) + # Non-unique (C7): identity_key is a supplementary dedup key alongside + # content_hash, not a replacement — many rows sharing NULL is expected + # for channels that don't implement identity(). + op.create_index( + op.f("ix_collected_records_source_identity"), + "collected_records", + ["source_id", "identity_key"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index( + op.f("ix_collected_records_source_identity"), + table_name="collected_records", + ) + op.drop_column("collected_records", "identity_key") diff --git a/backend/models/record.py b/backend/models/record.py index 86cb7da..4f3bfcd 100644 --- a/backend/models/record.py +++ b/backend/models/record.py @@ -1,6 +1,6 @@ from typing import TYPE_CHECKING, Optional -from sqlalchemy import JSON, ForeignKey, String, Text, UniqueConstraint +from sqlalchemy import JSON, ForeignKey, Index, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column, relationship from backend.models.base import TimestampMixin @@ -13,7 +13,13 @@ class CollectedRecord(TimestampMixin): """A single data record collected from a source.""" __tablename__ = "collected_records" - __table_args__ = (UniqueConstraint("source_id", "content_hash", name="uq_source_content"),) + __table_args__ = ( + UniqueConstraint("source_id", "content_hash", name="uq_source_content"), + # Non-unique: identity_key is a supplementary dedup key (C7), not a + # replacement for content_hash. Several rows sharing NULL is normal + # for channels that don't implement identity(). + Index("ix_collected_records_source_identity", "source_id", "identity_key"), + ) task_id: Mapped[str] = mapped_column( String(36), ForeignKey("collection_tasks.id", ondelete="CASCADE"), nullable=False @@ -30,6 +36,14 @@ class CollectedRecord(TimestampMixin): # SHA-256 hash of normalized content for deduplication content_hash: Mapped[str] = mapped_column(String(64), nullable=False) + # Stable source-native id (RSS entry id, tweet id, ...) from the channel's + # identity() (C7). NULL for channels that don't implement it — those keep + # deduplicating on content_hash alone, unchanged. When present, it's a + # supplementary key: an item whose identity matches an existing row gets + # updated in place instead of inserted as a new row when its content + # changes (e.g. a feed fixing a typo in a title no longer duplicates). + identity_key: Mapped[Optional[str]] = mapped_column(String(512), nullable=True) + # Processing status # raw | normalized | ai_processed | notified | error status: Mapped[str] = mapped_column(String(50), nullable=False, default="raw") diff --git a/backend/pipeline/pipeline.py b/backend/pipeline/pipeline.py index 11c83eb..7d18d26 100644 --- a/backend/pipeline/pipeline.py +++ b/backend/pipeline/pipeline.py @@ -180,15 +180,21 @@ async def run_pipeline( "auto_confirm": bool(source.channel_config.get("auto_confirm", False)), } if source.channel_type == "opencli": - from backend.channels.opencli_channel import _get_named_options, _OPENCLI_BIN + from backend.channels.opencli_channel import _OPENCLI_BIN, _peek_named_options cfg = source.channel_config _site = cfg.get("site", "") _cmd = cfg.get("command", "") _raw_args = {**cfg.get("args", {}), **{k: v for k, v in params.items() if k != "chrome_endpoint"}} _pos = [str(v) for v in cfg.get("positional_args", [])] _fmt = cfg.get("format", "json") - # Apply same positional-resolution logic as the channel - _named_opts = await _get_named_options(_OPENCLI_BIN, _site, _cmd) + # Apply same positional-resolution logic as the channel, but this + # detail string is display-only (the channel's own collect() call + # re-derives named options for real dispatch) — so peek the cache + # instead of spawning an opencli --help subprocess on the + # collection hot path just to format a log line (C17). A cache + # miss here behaves exactly like a failed --help fetch always + # did: fall through and treat every raw arg as a named option. + _named_opts = _peek_named_options(_OPENCLI_BIN, _site, _cmd) or frozenset() _named_args, _extra_pos = {}, [] for k, v in _raw_args.items(): if _named_opts and k not in _named_opts: diff --git a/backend/pipeline/sinks/legacy_db_sink.py b/backend/pipeline/sinks/legacy_db_sink.py index 86d2104..733acd9 100644 --- a/backend/pipeline/sinks/legacy_db_sink.py +++ b/backend/pipeline/sinks/legacy_db_sink.py @@ -19,10 +19,13 @@ from __future__ import annotations +import logging from typing import Sequence from backend.pipeline.sinks.base import RunContext, SinkResult +logger = logging.getLogger(__name__) + class LegacyDbSink: """Persist collected items to the legacy ``collected_records`` table. @@ -48,11 +51,33 @@ async def write_batch(self, ctx: RunContext, items: Sequence[dict]) -> SinkResul # rebound per call so tests can patch ``backend.database.AsyncSessionLocal``, # and ``storer``/``normalizer`` are reached as module attributes so # ``patch("backend.pipeline.storer.store_records")`` takes effect. + from backend.channels.registry import get_channel from backend.database import AsyncSessionLocal from backend.pipeline import normalizer, storer triples = normalizer.normalize_items(list(items), ctx.source_id) + # C7: ask the channel for each item's stable native id (RSS entry id, + # etc.) so store_records can update an edited item in place instead + # of inserting a duplicate. A channel without identity() returns None + # for every item — identical to the pre-C7 behavior (falls through to + # unchanged content_hash-only dedup). Best-effort: any failure here + # (unregistered channel_type, a channel's identity() raising on odd + # input, ...) also degrades to that unchanged behavior rather than + # breaking storage over a dedup nicety. + try: + channel = get_channel(ctx.provider) + identities: list[str | None] | None = [ + channel.identity(raw) for raw, _, _ in triples + ] + except Exception as exc: + logger.debug( + "could not resolve identity() for provider=%s (falling back " + "to content_hash-only dedup this batch): %s", + ctx.provider, exc, + ) + identities = None + # The ODP shadow-forward still fires inside storer.store_records; the # forward_to_odp gate lets DualSink(LegacyDbSink + OdpSink) turn it off on # the legacy leg so ODP is not double-sent. @@ -60,6 +85,7 @@ async def write_batch(self, ctx: RunContext, items: Sequence[dict]) -> SinkResul new_records, skipped = await storer.store_records( session, ctx.task_id, ctx.source_id, triples, channel_type=ctx.provider, forward_to_odp=self.forward_to_odp, + identities=identities, ) await session.commit() diff --git a/backend/pipeline/storer.py b/backend/pipeline/storer.py index 118833a..c71d6f1 100644 --- a/backend/pipeline/storer.py +++ b/backend/pipeline/storer.py @@ -12,6 +12,56 @@ logger = logging.getLogger(__name__) +# SQLite's default SQLITE_MAX_VARIABLE_NUMBER is 999 on older builds (32766 on +# newer ones, but we can't assume the deployment target). A collection batch +# (MAX_PAGES worth of items) can exceed that in one content_hash IN(...), so +# chunk the lookup instead of binding one variable per hash unconditionally. +_HASH_CHUNK_SIZE = 500 + + +async def _existing_hashes(session: AsyncSession, source_id: str, hashes: list[str]) -> set[str]: + """content_hash values from ``hashes`` already stored for ``source_id``. + + Chunks the IN() so a large batch never binds more query variables than + SQLite allows (C15) — behavior for small batches is unchanged, just one + chunk. + """ + existing: set[str] = set() + for i in range(0, len(hashes), _HASH_CHUNK_SIZE): + chunk = hashes[i : i + _HASH_CHUNK_SIZE] + result = await session.execute( + select(CollectedRecord.content_hash).where( + CollectedRecord.source_id == source_id, + CollectedRecord.content_hash.in_(chunk), + ) + ) + existing.update(row[0] for row in result) + return existing + + +async def _existing_by_identity( + session: AsyncSession, source_id: str, identity_keys: list[str] +) -> dict[str, CollectedRecord]: + """CollectedRecord rows already stored for source_id, keyed by identity_key. + + Only queried when at least one channel-provided identity() value is + present in the batch (C7) — channels without identity() never touch + this path, and existing content_hash-only dedup behavior for them is + unchanged. Chunked for the same reason as _existing_hashes (C15). + """ + existing: dict[str, CollectedRecord] = {} + for i in range(0, len(identity_keys), _HASH_CHUNK_SIZE): + chunk = identity_keys[i : i + _HASH_CHUNK_SIZE] + result = await session.execute( + select(CollectedRecord).where( + CollectedRecord.source_id == source_id, + CollectedRecord.identity_key.in_(chunk), + ) + ) + for record in result.scalars(): + existing[record.identity_key] = record + return existing + async def store_records( session: AsyncSession, @@ -21,6 +71,7 @@ async def store_records( *, channel_type: str = "unknown", forward_to_odp: bool = False, + identities: list[str | None] | None = None, ) -> tuple[list[CollectedRecord], int]: """Insert new records; skip existing ones by content_hash. @@ -36,6 +87,25 @@ async def store_records( ``LegacyDbSink`` (the ``legacy`` strategy's sink) now passes this explicitly; nothing should rely on the old implicit-True default. + ``identities`` (C7), when given, is a list parallel to ``normalized_triples`` + (same length, index-aligned) of each item's channel-provided ``identity()`` + value, or None for items the channel can't identify. It is a SUPPLEMENTARY + key alongside content_hash, not a replacement: + - identity present AND matches an existing row for this source: that's + the same source-native item seen before. If its content_hash is + unchanged, it's a plain duplicate (skipped). If the content_hash + differs (e.g. the feed fixed a typo in the title), the existing row + is UPDATED in place instead of inserted as a new row — a title edit + no longer duplicates the item (the C7 bug). Updated rows are folded + into the returned new_records (their content actually changed, so + downstream AI enrichment/notification should see them as fresh). + - identity present but not matching any existing row: falls through + to the normal content_hash-based insert path, with identity_key set + on the new row so future edits can be matched. + - identity None (or ``identities`` not given at all): behavior is + completely unchanged — content_hash-only dedup, exactly as before + C7. This is the only path channels without identity() ever take. + Returns (new_records, skipped_count). """ if not normalized_triples: @@ -54,25 +124,60 @@ async def store_records( raise logger.warning("odp ingest forward failed (continuing sqlite path): %s", exc) - # Collect all hashes to check for duplicates in one query + # Collect all hashes to check for duplicates (chunked, see _existing_hashes) hashes = [h for _, _, h in normalized_triples] - result = await session.execute( - select(CollectedRecord.content_hash).where( - CollectedRecord.source_id == source_id, - CollectedRecord.content_hash.in_(hashes), - ) + existing_hashes = await _existing_hashes(session, source_id, hashes) + + # C7: resolve identity()-matched existing rows, if the caller supplied any. + # Channels without identity() (identities is None, or all-None entries) + # never populate identity_keys, so existing_by_identity stays empty and + # every item below falls straight through to the unchanged content_hash path. + identity_keys = [i for i in (identities or []) if i is not None] + existing_by_identity = ( + await _existing_by_identity(session, source_id, identity_keys) + if identity_keys + else {} ) - existing_hashes = {row[0] for row in result} new_records: list[CollectedRecord] = [] + updated_records: list[CollectedRecord] = [] skipped = 0 # Dedup within this batch too: two triples can share a content_hash (e.g. two # CLI sub-commands that normalize to identical content). Without this, both # pass the existing_hashes check, both get added, and flush() fails the whole # batch atomically on the UNIQUE(source_id, content_hash) constraint. seen_in_batch: set[str] = set() + seen_identities_in_batch: set[str] = set() + + for idx, (raw, normalized, content_hash) in enumerate(normalized_triples): + identity = identities[idx] if identities else None + + if identity is not None: + existing = existing_by_identity.get(identity) + if existing is not None: + if existing.content_hash == content_hash: + # Same source-native item, content unchanged: a genuine + # duplicate, same as the content_hash-only path always did. + skipped += 1 + else: + # Same source-native item, content changed (e.g. a feed + # fixed a title typo) — update in place instead of + # inserting a duplicate row (C7's fix). + existing.raw_data = raw + existing.normalized_data = normalized + existing.content_hash = content_hash + existing.status = "normalized" + existing.ai_enrichment = None + updated_records.append(existing) + continue + if identity in seen_identities_in_batch: + # Two triples in this same batch share an identity (e.g. a + # feed listed the same entry twice) — keep the first, skip + # the rest rather than fight over which one "wins". + skipped += 1 + continue + seen_identities_in_batch.add(identity) - for raw, normalized, content_hash in normalized_triples: if content_hash in existing_hashes or content_hash in seen_in_batch: skipped += 1 continue @@ -85,6 +190,7 @@ async def store_records( normalized_data=normalized, content_hash=content_hash, status="normalized", + identity_key=identity, ) session.add(record) new_records.append(record) @@ -99,13 +205,26 @@ async def store_records( # (retries becoming real in PR-B makes this reachable in practice, not # just theoretical). await session.rollback() - recheck = await session.execute( - select(CollectedRecord.content_hash).where( - CollectedRecord.source_id == source_id, - CollectedRecord.content_hash.in_([r.content_hash for r in new_records]), + # C7: session.rollback() undoes the WHOLE transaction, not just the + # failed insert — so any identity-matched update mutations flushed + # in the same attempt are reverted too, and (expire_on_commit aside) + # are not worth replaying here: we don't retry them in this call. + # That's safe, not silent data loss — the row's stored content_hash + # in the DB is still the old value, so the next natural collection + # run will see the same "identity matches, content differs" mismatch + # and retry the update then. Only log if this actually cost us something. + if updated_records: + logger.warning( + "identity-matched update(s) reverted for source=%s alongside " + "an insert collision (%d update(s) deferred to next " + "collection run)", + source_id, + len(updated_records), ) + updated_records = [] + already_there = await _existing_hashes( + session, source_id, [r.content_hash for r in new_records] ) - already_there = {row[0] for row in recheck} survivors: list[CollectedRecord] = [] for record in new_records: if record.content_hash in already_there: @@ -124,4 +243,4 @@ async def store_records( skipped += 1 new_records = survivors - return new_records, skipped + return new_records + updated_records, skipped diff --git a/tests/unit/channels/test_cli_channel.py b/tests/unit/channels/test_cli_channel.py index 45d71b8..6fcf93e 100644 --- a/tests/unit/channels/test_cli_channel.py +++ b/tests/unit/channels/test_cli_channel.py @@ -6,7 +6,7 @@ """ import sys -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import AsyncMock, patch import pytest @@ -199,16 +199,19 @@ async def test_collect_text_output(channel): @pytest.mark.asyncio async def test_collect_timeout(channel): - """asyncio.TimeoutError returns failed ChannelResult and kills the child - so a timed-out subprocess is never orphaned (issue 05).""" + """asyncio.TimeoutError returns failed ChannelResult and kills the whole + process tree so a timed-out subprocess (and any shell-wrapped or forked + grandchild) is never orphaned (issue 05, C16).""" import asyncio mock_proc = AsyncMock() - mock_proc.kill = Mock() with ( _allow(sys.executable), patch("asyncio.create_subprocess_exec", return_value=mock_proc), patch("asyncio.wait_for", side_effect=asyncio.TimeoutError()), + patch( + "backend.channels.cli_channel._kill_subprocess", new=AsyncMock() + ) as kill_tree, ): result = await channel.collect( { @@ -221,7 +224,38 @@ async def test_collect_timeout(channel): assert result.success is False assert "timed out" in result.error.lower() - mock_proc.kill.assert_called_once() + kill_tree.assert_awaited_once_with(mock_proc) + + +@pytest.mark.asyncio +async def test_collect_timeout_logs_reap_failure(channel): + """If killing/reaping the process tree itself raises, that failure is + logged (not silently swallowed) and the timeout is still reported (C16).""" + import asyncio + + mock_proc = AsyncMock() + with ( + _allow(sys.executable), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + patch("asyncio.wait_for", side_effect=asyncio.TimeoutError()), + patch( + "backend.channels.cli_channel._kill_subprocess", + new=AsyncMock(side_effect=RuntimeError("reap failed")), + ), + patch("backend.channels.cli_channel.logger") as mock_logger, + ): + result = await channel.collect( + { + "binary": sys.executable, + "command": ["-c", "import time; time.sleep(10)"], + "timeout": 1, + }, + {}, + ) + + assert result.success is False + assert "timed out" in result.error.lower() + mock_logger.warning.assert_called_once() @pytest.mark.asyncio diff --git a/tests/unit/channels/test_opencli_channel.py b/tests/unit/channels/test_opencli_channel.py index f77c768..e9576aa 100644 --- a/tests/unit/channels/test_opencli_channel.py +++ b/tests/unit/channels/test_opencli_channel.py @@ -88,6 +88,182 @@ async def test_named_options_cancellation_kills_the_help_process(): kill_tree.assert_awaited_once_with(process) +# ── C17: --help/browser-requirement cache TTL + cache-only peek ───────────── + + +@pytest.mark.asyncio +async def test_named_options_cache_expires_after_ttl(monkeypatch): + """A cached --help result older than the TTL is refetched (C17): without + expiry, an opencli binary upgrade that adds/renames flags keeps being + routed with the old flag set until the admin process restarts.""" + import backend.channels.opencli_channel as oc + + process = AsyncMock() + process.communicate.return_value = (b"--alpha ", b"") + with patch("asyncio.create_subprocess_exec", return_value=process) as spawn: + first = await _get_named_options("ttl-opencli", "site", "cmd") + assert first == frozenset({"alpha"}) + assert spawn.call_count == 1 + + # Age the cache entry past the TTL without a real wait. + key = ("ttl-opencli", "site", "cmd") + cached_at, value = oc._help_cache[key] + oc._help_cache[key] = (cached_at - oc._CACHE_TTL_SECONDS - 1, value) + + process2 = AsyncMock() + process2.communicate.return_value = (b"--beta ", b"") + with patch("asyncio.create_subprocess_exec", return_value=process2) as spawn2: + second = await _get_named_options("ttl-opencli", "site", "cmd") + assert second == frozenset({"beta"}) + assert spawn2.call_count == 1 # refetched, not served the stale entry + + +@pytest.mark.asyncio +async def test_named_options_cache_hit_within_ttl_skips_subprocess(): + """A cache entry still within its TTL is served without refetching.""" + process = AsyncMock() + process.communicate.return_value = (b"--gamma ", b"") + with patch("asyncio.create_subprocess_exec", return_value=process): + await _get_named_options("ttl-fresh-opencli", "site", "cmd") + + with patch("asyncio.create_subprocess_exec") as spawn: + result = await _get_named_options("ttl-fresh-opencli", "site", "cmd") + + assert result == frozenset({"gamma"}) + spawn.assert_not_called() + + +def test_peek_named_options_cache_hit_no_subprocess(): + """Cache-only lookup (C17): a warm entry is returned synchronously with + no subprocess spawn — used by pipeline.py's event-log display string, + which must not pay a --help call on the collection hot path.""" + import time + + import backend.channels.opencli_channel as oc + from backend.channels.opencli_channel import _peek_named_options + + oc._help_cache[("peek-bin", "site", "cmd")] = (time.monotonic(), frozenset({"url"})) + assert _peek_named_options("peek-bin", "site", "cmd") == frozenset({"url"}) + + +def test_peek_named_options_cache_miss_returns_none_no_subprocess(): + """A cold (or expired) cache entry returns None without ever spawning a + subprocess (C17) — the caller falls back to treating all args as named, + exactly like a failed --help fetch always did.""" + from backend.channels.opencli_channel import _peek_named_options + + with patch("asyncio.create_subprocess_exec") as spawn: + result = _peek_named_options("never-cached-bin", "site", "cmd") + assert result is None + spawn.assert_not_called() + + +# ── C20: CDP tab-snapshot failure must not drive tab cleanup ──────────────── + + +@pytest.mark.asyncio +async def test_snapshot_tab_ids_returns_none_on_failure(): + """A failed pre-collection tab snapshot returns None, not an empty set + (C20): the two are not interchangeable. None means 'no trustworthy + baseline' and must skip cleanup entirely; an empty set would be treated + as 'genuinely zero tabs existed before', making every currently-open tab + look new and get closed.""" + from backend.channels.opencli_channel import _snapshot_tab_ids + + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=OSError("connection refused")) + mock_client_ctx = AsyncMock() + mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_ctx.__aexit__ = AsyncMock(return_value=False) + + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + result = await _snapshot_tab_ids("http://bad-endpoint:1") + + assert result is None + + +@pytest.mark.asyncio +async def test_snapshot_tab_ids_success_returns_set(): + from backend.channels.opencli_channel import _snapshot_tab_ids + + mock_response = MagicMock() + mock_response.json = MagicMock( + return_value=[{"id": "tab-1"}, {"id": "tab-2"}, {"no_id": True}] + ) + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client_ctx = AsyncMock() + mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_ctx.__aexit__ = AsyncMock(return_value=False) + + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + result = await _snapshot_tab_ids("http://good-endpoint:1") + + assert result == {"tab-1", "tab-2"} + + +@pytest.mark.asyncio +async def test_collect_cdp_skips_cleanup_when_snapshot_baseline_failed(channel): + """If the pre-collection tab snapshot fails, cleanup must not run at all + this pass (C20) — with no trustworthy baseline it can't tell tabs the + user already had open from ones opencli just opened, so closing + anything risks closing the user's own tabs.""" + 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( + "backend.channels.opencli_channel._snapshot_tab_ids", + new=AsyncMock(return_value=None), + ), + patch( + "backend.channels.opencli_channel._cleanup_cdp_tabs", new=AsyncMock() + ) as mock_cleanup, + ): + result = await channel.collect( + {"site": "example.com", "command": "list", "format": "json"}, {} + ) + + assert result.success is True + mock_cleanup.assert_not_called() + + +@pytest.mark.asyncio +async def test_collect_cdp_runs_cleanup_when_snapshot_baseline_ok(channel): + """Sanity check for the above: when the snapshot succeeds, cleanup still + runs exactly as before — C20 only changes the failure path.""" + 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( + "backend.channels.opencli_channel._snapshot_tab_ids", + new=AsyncMock(return_value={"pre-existing-tab"}), + ), + patch( + "backend.channels.opencli_channel._cleanup_cdp_tabs", new=AsyncMock() + ) as mock_cleanup, + ): + result = await channel.collect( + {"site": "example.com", "command": "list", "format": "json"}, {} + ) + + assert result.success is True + mock_cleanup.assert_awaited_once_with("http://chrome:9222", {"pre-existing-tab"}) + + def test_managed_profile_requirement_is_not_forwarded_as_a_cli_argument(): from backend.channels.opencli_channel import _split_routing_parameters diff --git a/tests/unit/pipeline/test_legacy_db_sink.py b/tests/unit/pipeline/test_legacy_db_sink.py index 3bfd390..df78b45 100644 --- a/tests/unit/pipeline/test_legacy_db_sink.py +++ b/tests/unit/pipeline/test_legacy_db_sink.py @@ -55,6 +55,67 @@ async def test_legacy_sink_normalizes_then_stores(): assert triples[0][1]["title"] == "A" +@pytest.mark.asyncio +async def test_legacy_sink_resolves_channel_identity_for_c7(): + """C7: the sink asks the channel for each item's identity() and passes + it through to store_records — this is what lets an RSS entry with a + stable id be matched across re-fetches even after its title changes.""" + items = [ + {"title": "A", "url": "https://x/a", "id": "guid-a"}, + {"title": "B", "url": "https://x/b"}, # no "id" key: identity() is None for this one + ] + store_mock = AsyncMock(return_value=([], 0)) + + with ( + patch("backend.pipeline.storer.store_records", new=store_mock), + patch("backend.database.AsyncSessionLocal", return_value=_session_cm()), + ): + await LegacyDbSink().write_batch(_ctx(), items) + + _, kwargs = store_mock.call_args + # RSS's identity() reads item["id"] (real feed fetches populate it via + # _entry_to_dict's fallback-to-link — out of scope here, this test is + # at the sink layer with hand-built raw dicts): present for item 1, + # absent for item 2. + assert kwargs["identities"] == ["guid-a", None] + + +@pytest.mark.asyncio +async def test_legacy_sink_falls_back_when_channel_has_no_identity(): + """A channel_type with no identity() override (or unresolvable) passes + identities=None through — store_records' documented content_hash-only + fallback, unchanged from before C7.""" + items = [{"title": "A", "url": "https://x/a"}] + store_mock = AsyncMock(return_value=([], 0)) + + with ( + patch("backend.pipeline.storer.store_records", new=store_mock), + patch("backend.database.AsyncSessionLocal", return_value=_session_cm()), + ): + await LegacyDbSink().write_batch(_ctx(provider="cli"), items) + + _, kwargs = store_mock.call_args + assert kwargs["identities"] == [None] + + +@pytest.mark.asyncio +async def test_legacy_sink_unknown_provider_degrades_to_no_identities(): + """An unregistered channel_type (get_channel raises) must not break + storage — it degrades to identities=None, the unchanged pre-C7 path.""" + items = [{"title": "A", "url": "https://x/a"}] + store_mock = AsyncMock(return_value=([], 0)) + + with ( + patch("backend.pipeline.storer.store_records", new=store_mock), + patch("backend.database.AsyncSessionLocal", return_value=_session_cm()), + ): + result = await LegacyDbSink().write_batch(_ctx(provider="no-such-channel"), items) + + assert result.accepted == 0 # ran to completion, no exception raised + _, kwargs = store_mock.call_args + assert kwargs["identities"] is None + + @pytest.mark.asyncio async def test_legacy_sink_empty_items(): store_mock = AsyncMock(return_value=([], 0)) diff --git a/tests/unit/pipeline/test_pipeline.py b/tests/unit/pipeline/test_pipeline.py index 81fcd38..7d0d6b2 100644 --- a/tests/unit/pipeline/test_pipeline.py +++ b/tests/unit/pipeline/test_pipeline.py @@ -576,6 +576,112 @@ async def test_run_pipeline_opencli_auto_binding(db_session): ) +# ── C17: opencli display-string must not spawn --help on the hot path ────── + +@pytest.mark.asyncio +async def test_run_pipeline_opencli_display_string_cold_cache_no_subprocess(db_session): + """The event-log 'command' display string must not spawn an opencli + --help subprocess on the collection hot path (C17). A cold cache falls + back to exactly the same 'treat every arg as named' behavior a failed + --help fetch always produced — just without ever spawning anything.""" + from backend.models.source import DataSource + from backend.models.task import CollectionTask + + source = DataSource( + name="OpenCLI Display Source", + channel_type="opencli", + channel_config={ + "site": "", "command": "cold-cache-observe-cmd", + "args": {"query": "x"}, + }, + ) + 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() + + channel_result = ChannelResult.ok([{"title": "t"}]) + captured: list[dict] = [] + + async def fake_emit(run_id, step, message, level="info", detail=None, elapsed_ms=None): + captured.append(detail or {}) + + with ( + patch("backend.pipeline.collector.collect", return_value=channel_result), + patch("backend.pipeline.storer.store_records", new=AsyncMock(return_value=([], 0))), + patch("backend.pipeline.events.emit", new=fake_emit), + patch("asyncio.create_subprocess_exec") as spawn, + ): + result = await run_pipeline( + task.id, source, parameters={}, enable_ai=False, + enable_notifications=False, run_id="run-cold", + ) + + assert result.success is True + spawn.assert_not_called() + commands = [d["command"] for d in captured if "command" in d] + assert len(commands) == 1 + assert "--query x" in commands[0] + + +@pytest.mark.asyncio +async def test_run_pipeline_opencli_display_string_uses_warm_cache(db_session): + """A warm --help cache (as populated by a real prior collection run + through the channel itself) still produces the nicer named/positional + split in the display string — C17 only removes the redundant spawn on + a cache miss, it doesn't regress the warm-cache case.""" + import time + + import backend.channels.opencli_channel as oc + from backend.channels.opencli_channel import _OPENCLI_BIN + from backend.models.source import DataSource + from backend.models.task import CollectionTask + + site, command = "", "warm-cache-observe-cmd" + oc._help_cache[(_OPENCLI_BIN, site, command)] = ( + time.monotonic(), frozenset({"query"}) + ) + + source = DataSource( + name="OpenCLI Display Source Warm", + channel_type="opencli", + channel_config={ + "site": site, "command": command, + "args": {"query": "x"}, "positional_args": ["extra"], + }, + ) + 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() + + channel_result = ChannelResult.ok([{"title": "t"}]) + captured: list[dict] = [] + + async def fake_emit(run_id, step, message, level="info", detail=None, elapsed_ms=None): + captured.append(detail or {}) + + with ( + patch("backend.pipeline.collector.collect", return_value=channel_result), + patch("backend.pipeline.storer.store_records", new=AsyncMock(return_value=([], 0))), + patch("backend.pipeline.events.emit", new=fake_emit), + patch("asyncio.create_subprocess_exec") as spawn, + ): + result = await run_pipeline( + task.id, source, parameters={}, enable_ai=False, + enable_notifications=False, run_id="run-warm", + ) + + assert result.success is True + spawn.assert_not_called() # peek-only, never spawns regardless of cache state + commands = [d["command"] for d in captured if "command" in d] + assert len(commands) == 1 + assert "--query x" in commands[0] + assert "extra" in commands[0] + + # ── P1-7: DualSink shadow errors must surface, not vanish ────────────────── @pytest.mark.asyncio diff --git a/tests/unit/pipeline/test_storer.py b/tests/unit/pipeline/test_storer.py index e8f2d85..5ebb519 100644 --- a/tests/unit/pipeline/test_storer.py +++ b/tests/unit/pipeline/test_storer.py @@ -149,3 +149,205 @@ async def blind_first_call(*args, **kwargs): assert skipped == 1 assert len(new_records) == 1 assert new_records[0].content_hash == "clean_hash" + + +async def _setup_source_task(db_session, channel_type="rss"): + from backend.models.source import DataSource + from backend.models.task import CollectionTask + + source = DataSource( + name="Identity Source", channel_type=channel_type, + 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() + return source, task + + +def _triple(source_id, title, content_hash): + return ( + {"title": title}, + { + "title": title, "url": "", "content": "", "author": "", + "published_at": "", "source_id": source_id, + }, + content_hash, + ) + + +# ── C7: identity()-based dedup/update-in-place ────────────────────────────── + +@pytest.mark.asyncio +async def test_store_identity_match_updates_in_place(db_session): + """An item whose identity() matches an existing row, but whose content + changed (e.g. a feed fixed a typo in the title), updates that row in + place instead of inserting a duplicate (C7's fix).""" + source, task = await _setup_source_task(db_session) + + first = _triple(source.id, "Original Title", "hash_v1") + records1, skipped1 = await store_records( + db_session, task.id, source.id, [first], identities=["entry-42"], + ) + assert len(records1) == 1 + assert skipped1 == 0 + assert records1[0].identity_key == "entry-42" + + edited = _triple(source.id, "Original Title (fixed)", "hash_v2") + records2, skipped2 = await store_records( + db_session, task.id, source.id, [edited], identities=["entry-42"], + ) + + # Updated in place, not inserted as a new row. + assert skipped2 == 0 + assert len(records2) == 1 + assert records2[0].content_hash == "hash_v2" + assert records2[0].id == records1[0].id + + from sqlalchemy import select as sa_select + + from backend.models.record import CollectedRecord + + result = await db_session.execute( + sa_select(CollectedRecord).where(CollectedRecord.source_id == source.id) + ) + rows = result.scalars().all() + assert len(rows) == 1 # still just one row for this source-native item + assert rows[0].content_hash == "hash_v2" + + +@pytest.mark.asyncio +async def test_store_identity_match_same_hash_is_plain_duplicate(db_session): + """Same identity AND same content_hash: a genuine duplicate, skipped — + exactly like the content_hash-only path always did for an unedited + re-fetch of the same item.""" + source, task = await _setup_source_task(db_session) + + triple = _triple(source.id, "Same", "same_hash") + await store_records(db_session, task.id, source.id, [triple], identities=["entry-1"]) + records2, skipped2 = await store_records( + db_session, task.id, source.id, [triple], identities=["entry-1"] + ) + + assert len(records2) == 0 + assert skipped2 == 1 + + +@pytest.mark.asyncio +async def test_store_without_identities_is_unchanged(db_session): + """Channels without identity() (identities=None, the default) keep + deduplicating on content_hash alone — completely unaffected by C7.""" + source, task = await _setup_source_task(db_session) + + triple = _triple(source.id, "No Identity", "no_identity_hash") + records1, skipped1 = await store_records(db_session, task.id, source.id, [triple]) + assert len(records1) == 1 + assert records1[0].identity_key is None + + # An "edit" with no identity info at all is content_hash-only: a + # different hash is a brand new row, not an update — pre-C7 behavior. + edited = _triple(source.id, "No Identity Edited", "no_identity_hash_v2") + records2, skipped2 = await store_records(db_session, task.id, source.id, [edited]) + assert len(records2) == 1 + assert records2[0].id != records1[0].id # inserted as a new row, not updated + + +@pytest.mark.asyncio +async def test_store_mixed_batch_some_items_without_identity(db_session): + """identities can mix real values and None per item — items with None + fall back to content_hash dedup individually, others use identity.""" + source, task = await _setup_source_task(db_session) + + triples = [ + _triple(source.id, "Has Identity", "hash_a"), + _triple(source.id, "No Identity", "hash_b"), + ] + records, skipped = await store_records( + db_session, task.id, source.id, triples, identities=["entry-x", None], + ) + assert skipped == 0 + by_hash = {r.content_hash: r for r in records} + assert by_hash["hash_a"].identity_key == "entry-x" + assert by_hash["hash_b"].identity_key is None + + +@pytest.mark.asyncio +async def test_store_identity_duplicated_within_same_batch(db_session): + """Two triples in the same batch sharing an identity (e.g. a feed listed + the same entry twice): keep the first, skip the rest.""" + source, task = await _setup_source_task(db_session) + + triples = [ + _triple(source.id, "First", "hash_first"), + _triple(source.id, "Duplicate Within Batch", "hash_dup"), + ] + records, skipped = await store_records( + db_session, task.id, source.id, triples, + identities=["entry-same", "entry-same"], + ) + assert len(records) == 1 + assert skipped == 1 + assert records[0].content_hash == "hash_first" + + +# ── C15: dedup lookup chunking across the SQLite variable limit ──────────── + +@pytest.mark.asyncio +async def test_store_dedup_chunks_across_batches(db_session, monkeypatch): + """A dedup lookup spanning more hashes than one chunk still correctly + finds every pre-existing hash, regardless of which chunk it falls into + (C15) — proves the union-across-chunks logic, not just 'doesn't crash'.""" + import backend.pipeline.storer as storer_module + + monkeypatch.setattr(storer_module, "_HASH_CHUNK_SIZE", 2) + + source, task = await _setup_source_task(db_session) + + # Seed 5 existing rows — spans more than one chunk of size 2. + seed_triples = [_triple(source.id, f"Seed {i}", f"seed_hash_{i}") for i in range(5)] + await store_records(db_session, task.id, source.id, seed_triples) + + # A new batch mixing all 5 pre-existing hashes with 2 genuinely new ones. + batch = seed_triples + [ + _triple(source.id, "New A", "new_hash_a"), + _triple(source.id, "New B", "new_hash_b"), + ] + new_records, skipped = await store_records(db_session, task.id, source.id, batch) + + assert skipped == 5 # every seeded hash correctly detected, across chunks + assert len(new_records) == 2 + assert {r.content_hash for r in new_records} == {"new_hash_a", "new_hash_b"} + + +@pytest.mark.asyncio +async def test_store_identity_lookup_chunks_across_batches(db_session, monkeypatch): + """The identity-based existence lookup is chunked the same way (C7+C15 + share the same chunk size) — a batch with more identities than one + chunk still matches every existing one correctly.""" + import backend.pipeline.storer as storer_module + + monkeypatch.setattr(storer_module, "_HASH_CHUNK_SIZE", 2) + + source, task = await _setup_source_task(db_session) + + seed_triples = [_triple(source.id, f"Seed {i}", f"seed_hash_{i}") for i in range(5)] + seed_identities = [f"entry-{i}" for i in range(5)] + await store_records( + db_session, task.id, source.id, seed_triples, identities=seed_identities, + ) + + # Re-submit all 5 with edited content (same identities, new hashes) — + # every one should be matched and updated in place, none inserted. + edited_triples = [ + _triple(source.id, f"Seed {i} edited", f"seed_hash_{i}_v2") for i in range(5) + ] + records, skipped = await store_records( + db_session, task.id, source.id, edited_triples, identities=seed_identities, + ) + + assert skipped == 0 + assert len(records) == 5 + assert {r.content_hash for r in records} == {f"seed_hash_{i}_v2" for i in range(5)}