From 4d69050a1640ab249e2ee47b6a7a975feb3e9648 Mon Sep 17 00:00:00 2001 From: Curry Date: Sun, 19 Jul 2026 02:06:51 +0800 Subject: [PATCH] perf(pipeline): LLM timeout+concurrency, bulk AI persist, off-loop feed parse, batched events [C8,C21,C22,C24,C25] - C8/C25: openai/claude/local processors get an explicit per-request LLM timeout (new Settings.llm_request_timeout_seconds, default 120s, config["timeout"]-overridable) and bound per-record concurrency via asyncio.gather + Semaphore (new Settings.llm_max_concurrency, default 4) instead of a sequential await-in-a-for-loop. Order is preserved (gather), and each record's LLM call keeps its own try/except so one failure can't abort the batch. - C21: pipeline.py's AI-enrichment persist step replaces one session.get() per record with a single bulk SELECT ... WHERE id IN (...) + in-memory id->row map before one commit. Same field writes (ai_enrichment, status="ai_processed"). - C22: rss_channel's feedparser.parse() (collect() and fetch()) and web_scraper_channel's BeautifulSoup parse now run via asyncio.to_thread instead of blocking the event loop inline. - C24: new events.emit_many(run_id, events) writes a whole step trace in one session + bulk insert + one commit; skill_channel's per-step loop (the only tight-loop emit() caller) now uses it. All other emit() call sites are untouched. Tests: LLM timeout/concurrency/order/failure-isolation per processor, AI-persist query-count spy, feedparser/BeautifulSoup off-thread checks, emit_many one-commit behavior. Fakes/monkeypatch only, no real network/DB. --- backend/channels/rss_channel.py | 12 +- backend/channels/skill_channel.py | 72 +++--- backend/channels/web_scraper_channel.py | 7 +- backend/config.py | 12 + backend/pipeline/events.py | 34 +++ backend/pipeline/pipeline.py | 26 ++- backend/processors/claude_processor.py | 58 +++-- backend/processors/local_processor.py | 93 +++++--- backend/processors/openai_processor.py | 60 +++-- tests/unit/channels/test_rss_channel.py | 38 ++++ tests/unit/channels/test_rss_fetch.py | 26 +++ .../test_skill_channel_emit_batching.py | 78 +++++++ .../unit/channels/test_web_scraper_channel.py | 35 +++ tests/unit/pipeline/test_events.py | 80 +++++++ tests/unit/pipeline/test_pipeline.py | 95 ++++++++ tests/unit/test_claude_processor.py | 196 ++++++++++++++++ tests/unit/test_local_processor.py | 210 ++++++++++++++++++ tests/unit/test_openai_processor.py | 205 +++++++++++++++++ 18 files changed, 1234 insertions(+), 103 deletions(-) create mode 100644 tests/unit/channels/test_skill_channel_emit_batching.py create mode 100644 tests/unit/pipeline/test_events.py create mode 100644 tests/unit/test_claude_processor.py create mode 100644 tests/unit/test_local_processor.py create mode 100644 tests/unit/test_openai_processor.py diff --git a/backend/channels/rss_channel.py b/backend/channels/rss_channel.py index 20aa4a0..f317324 100644 --- a/backend/channels/rss_channel.py +++ b/backend/channels/rss_channel.py @@ -1,5 +1,6 @@ """RSS channel using feedparser.""" +import asyncio from typing import Any import feedparser @@ -75,7 +76,12 @@ async def collect( f"Failed to fetch RSS feed: {exc}", error_type=type(exc).__name__ ) - parsed = feedparser.parse(content) + # AUDIT C22: feedparser.parse() is a synchronous, potentially + # multi-second call for a large feed — running it inline would freeze + # the whole event loop (every other request/task on this process) + # for the duration. asyncio.to_thread runs it in the default executor + # instead; the parsed result is used identically either way. + parsed = await asyncio.to_thread(feedparser.parse, content) if parsed.bozo and not parsed.entries: return ChannelResult.fail( f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}" @@ -153,7 +159,9 @@ async def fetch(self, ctx: FetchContext) -> FetchResult: return FetchResult(items=[], next_cursor=(cursor or None), has_more=False) response.raise_for_status() - parsed = feedparser.parse(response.text) + # AUDIT C22: see collect()'s twin comment — off-load the synchronous + # parse instead of blocking the event loop. + parsed = await asyncio.to_thread(feedparser.parse, response.text) if parsed.bozo and not parsed.entries: raise ChannelFetchError( f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}" diff --git a/backend/channels/skill_channel.py b/backend/channels/skill_channel.py index e438ac3..48643b0 100644 --- a/backend/channels/skill_channel.py +++ b/backend/channels/skill_channel.py @@ -259,57 +259,62 @@ async def model_call( return model_call -def _emit_loop_events(run_id: str, result: LoopResult) -> list[Any]: - """Build the per-step ``events.emit`` coroutines for a finished loop. +def _emit_loop_events(result: LoopResult) -> list[dict[str, Any]]: + """Build the per-step ``TaskRunEvent`` payloads for a finished loop. The loop is *pure of the spine* (it only self-emits ``awaiting_confirm`` on a gate block); spine event emission is this channel's job. We walk the ordered - ``result.steps`` and emit one event each — ``skill_extract`` for ``extract`` - verbs, ``skill_step`` for everything else — bracketed by a leading - ``skill_perceive`` and a trailing ``skill_done`` carrying the outcome. Every - ``emit`` is best-effort and never raises (see ``events.emit``). + ``result.steps`` and build one event payload each — ``skill_extract`` for + ``extract`` verbs, ``skill_step`` for everything else — bracketed by a + leading ``skill_perceive`` and a trailing ``skill_done`` carrying the + outcome. + + AUDIT C24: this used to return a list of already-built ``events.emit(...)`` + coroutines, awaited one at a time by the caller (one session + INSERT + + commit/fsync per step). It now returns plain dicts so the caller can hand + the whole trace to ``events.emit_many`` — one session, one bulk insert, one + commit for the entire run's step trace. """ - coros: list[Any] = [] - coros.append( - events.emit( - run_id, STEP_PERCEIVE, - f"开始执行技能 | 步数={len(result.steps)}", - detail={"step_count": len(result.steps)}, - ) - ) + payloads: list[dict[str, Any]] = [ + { + "step": STEP_PERCEIVE, + "message": f"开始执行技能 | 步数={len(result.steps)}", + "detail": {"step_count": len(result.steps)}, + } + ] for step in result.steps: verb = step.verb or "?" is_extract = verb == "extract" - coros.append( - events.emit( - run_id, - STEP_EXTRACT if is_extract else STEP_STEP, - f"步骤 {step.index} | {verb}" + (f" | 错误: {step.error}" if step.error else ""), - level="warning" if step.error else "info", - detail={ + payloads.append( + { + "step": STEP_EXTRACT if is_extract else STEP_STEP, + "message": f"步骤 {step.index} | {verb}" + + (f" | 错误: {step.error}" if step.error else ""), + "level": "warning" if step.error else "info", + "detail": { "index": step.index, "verb": step.verb, "target": step.target, "error": step.error, "result": step.result, }, - elapsed_ms=step.elapsed_ms, - ) + "elapsed_ms": step.elapsed_ms, + } ) - coros.append( - events.emit( - run_id, STEP_DONE, - f"技能执行结束 | 结果={result.outcome} 提取={len(result.extracts)}", - level="warning" if result.outcome in ("error", "done_failed") else "info", - detail={ + payloads.append( + { + "step": STEP_DONE, + "message": f"技能执行结束 | 结果={result.outcome} 提取={len(result.extracts)}", + "level": "warning" if result.outcome in ("error", "done_failed") else "info", + "detail": { "outcome": result.outcome, "extract_count": len(result.extracts), "awaiting_confirm": result.awaiting_confirm, "summary": result.summary, }, - ) + } ) - return coros + return payloads def _extracts_to_items(result: LoopResult) -> list[dict[str, Any]]: @@ -616,9 +621,10 @@ async def collect( await skill_page.aclose() # Emit per-step events (best-effort; no-op when no run_id). + # AUDIT C24: one session + bulk insert + one commit for the + # whole step trace, instead of one commit (fsync) per step. if run_id: - for coro in _emit_loop_events(run_id, result): - await coro + await events.emit_many(run_id, _emit_loop_events(result)) items = _extracts_to_items(result) diff --git a/backend/channels/web_scraper_channel.py b/backend/channels/web_scraper_channel.py index cabd51f..1d54ac2 100644 --- a/backend/channels/web_scraper_channel.py +++ b/backend/channels/web_scraper_channel.py @@ -1,5 +1,6 @@ """Web scraper channel using httpx + BeautifulSoup.""" +import asyncio import logging from typing import Any from urllib.parse import urlparse @@ -117,7 +118,11 @@ async def fetch(self, ctx: FetchContext) -> FetchResult: async with client as opened_client: response = await self._get(opened_client, url, timeout) - soup = BeautifulSoup(response.text, "lxml") + # AUDIT C22: BeautifulSoup's lxml parse is synchronous and can take + # seconds on a large page — run it off the event loop so it can't + # freeze every other request/task on this process. Same object, + # same parser, just executed in the default thread-pool executor. + soup = await asyncio.to_thread(BeautifulSoup, response.text, "lxml") if list_selector: containers = soup.select(list_selector) diff --git a/backend/config.py b/backend/config.py index 40bef82..20fef3a 100644 --- a/backend/config.py +++ b/backend/config.py @@ -125,6 +125,18 @@ def cdp_endpoints(self) -> list[str]: # WS dispatch timeout when center sends a task over a reverse WS channel agent_ws_timeout: int = 130 + # AI enrichment processors (processors/openai_processor.py, claude_processor.py, + # local_processor.py): explicit per-request timeout on the LLM API call itself + # (AUDIT C8) — the SDKs' own default is a 600s x 2-retry black hole that can + # otherwise pin a whole batch in ai_processing for hours behind a dead/slow + # gateway. A source's ai_config can still override this per call via + # config["timeout"]; this is only the fallback default. + llm_request_timeout_seconds: int = 120 + # Bound how many per-record LLM calls run concurrently within one enrichment + # batch (AUDIT C25) — replaces a plain await-in-a-for-loop, where wall-clock + # cost was record_count x per-call latency. + llm_max_concurrency: int = 4 + # Webhooks webhook_secret: str = "change-me-webhook-secret" diff --git a/backend/pipeline/events.py b/backend/pipeline/events.py index bcf3bb3..cabccf0 100644 --- a/backend/pipeline/events.py +++ b/backend/pipeline/events.py @@ -30,3 +30,37 @@ async def emit( await session.commit() except Exception as exc: logger.warning("emit event failed: %s", exc) + + +async def emit_many(run_id: str, events: list[dict[str, Any]]) -> None: + """Write multiple TaskRunEvent rows for one run_id in a single session + + bulk insert + one commit (AUDIT C24) — the batched counterpart to + :func:`emit` for a caller that already has a whole step trace in hand + (e.g. the skill channel's per-step spine events) instead of one commit + (fsync) per event in a tight loop. + + Each item in ``events`` accepts the same keys as ``emit``'s kwargs: + ``step`` and ``message`` (required), ``level``/``detail``/``elapsed_ms`` + (optional, same defaults as ``emit``). Best-effort: never raises, mirrors + ``emit``. A no-op for an empty list (no session is even opened). + """ + if not events: + return + try: + from backend.database import AsyncSessionLocal + from backend.models.task import TaskRunEvent + async with AsyncSessionLocal() as session: + session.add_all([ + TaskRunEvent( + run_id=run_id, + level=event.get("level", "info"), + step=event["step"], + message=event["message"], + detail=event.get("detail"), + elapsed_ms=event.get("elapsed_ms"), + ) + for event in events + ]) + await session.commit() + except Exception as exc: + logger.warning("emit_many event failed: %s", exc) diff --git a/backend/pipeline/pipeline.py b/backend/pipeline/pipeline.py index 6170364..567f3cf 100644 --- a/backend/pipeline/pipeline.py +++ b/backend/pipeline/pipeline.py @@ -6,6 +6,8 @@ from email.utils import parsedate_to_datetime from typing import Any +from sqlalchemy import select + from backend.channels.base import ChannelFetchError from backend.control.error_kinds import map_error_type, map_exception from backend.control.recorder import FreshnessInfo, record_run_measurement @@ -400,16 +402,28 @@ async def run_pipeline( # in backend.pipeline.runner phase 2, so it's used as-is. resolve_provider=agent_config is None, ) - # Persist enrichments — new_records are detached after step3 session closed + # Persist enrichments — new_records are detached after step3 session + # closed. AUDIT C21: one bulk SELECT ... WHERE id IN (...) + an + # in-memory id->row map, instead of one `session.get` per record + # (N+1) — same field writes (ai_enrichment, status="ai_processed"). from backend.models.record import CollectedRecord - async with AsyncSessionLocal() as session: - for rec in new_records: - if rec.ai_enrichment is not None: - db_rec = await session.get(CollectedRecord, rec.id) + enriched_ids = [rec.id for rec in new_records if rec.ai_enrichment is not None] + if enriched_ids: + async with AsyncSessionLocal() as session: + db_recs = ( + await session.execute( + select(CollectedRecord).where(CollectedRecord.id.in_(enriched_ids)) + ) + ).scalars().all() + db_recs_by_id = {db_rec.id: db_rec for db_rec in db_recs} + for rec in new_records: + if rec.ai_enrichment is None: + continue + db_rec = db_recs_by_id.get(rec.id) if db_rec: db_rec.ai_enrichment = rec.ai_enrichment db_rec.status = "ai_processed" - await session.commit() + await session.commit() logger.info("[task:%s] step4/ai done | processed=%d", task_id, ai_count) if run_id: # AUDIT C3: ai_count is now the real enrichment count (0 when diff --git a/backend/processors/claude_processor.py b/backend/processors/claude_processor.py index a3d16b0..d824e9c 100644 --- a/backend/processors/claude_processor.py +++ b/backend/processors/claude_processor.py @@ -1,5 +1,6 @@ """Claude (Anthropic) AI processor.""" +import asyncio import json import logging import os @@ -41,12 +42,26 @@ async def process( success=False, error="anthropic package not installed" ) + from backend.config import get_settings + + settings = get_settings() + api_key = config.get("api_key") or os.environ.get("ANTHROPIC_API_KEY", "") model = config.get("model", "claude-haiku-4-5-20251001") max_tokens = config.get("max_tokens", 1024) - - logger.info("claude processor | model=%s max_tokens=%d records=%d", - model, max_tokens, len(records)) + # AUDIT C8: explicit per-request timeout (see openai_processor's twin + # comment) — a source's ai_config can still override this per call + # via config["timeout"]. + request_timeout = config.get("timeout", settings.llm_request_timeout_seconds) + # AUDIT C25: bound how many records are in flight at once instead of + # a plain await-in-a-for-loop (wall clock == record_count x latency). + max_concurrency = max(1, settings.llm_max_concurrency) + + logger.info( + "claude processor | model=%s max_tokens=%d records=%d " + "timeout=%s max_concurrency=%d", + model, max_tokens, len(records), request_timeout, max_concurrency, + ) # GOAL-6 PR-E: client construction consolidated through # backend.llm.anthropic.AnthropicAdapter (via @@ -56,18 +71,24 @@ async def process( # no behavior change. adapter = build_anthropic_adapter(api_key=api_key) client = await adapter.get_client() - enrichments: list[dict[str, Any]] = [] - try: - for i, record in enumerate(records): - prompt = _render(prompt_template, record.normalized_data) - logger.debug("claude req [%d/%d] | prompt_preview=%s", - i + 1, len(records), prompt[:200]) + semaphore = asyncio.Semaphore(max_concurrency) + + async def _process_one(i: int, record: "CollectedRecord") -> dict[str, Any]: + # AUDIT C25: the semaphore (not the for-loop) is what bounds + # concurrency now — every record's coroutine is created up front + # and handed to gather, but only `max_concurrency` run their LLM + # call at once. + async with semaphore: try: + prompt = _render(prompt_template, record.normalized_data) + logger.debug("claude req [%d/%d] | prompt_preview=%s", + i + 1, len(records), prompt[:200]) response = await client.messages.create( model=model, max_tokens=max_tokens, messages=[{"role": "user", "content": prompt}], + timeout=request_timeout, ) text = response.content[0].text usage = response.usage @@ -76,13 +97,24 @@ async def process( usage.input_tokens, usage.output_tokens, text[:200]) try: - enrichment = json.loads(text) + return json.loads(text) except json.JSONDecodeError: - enrichment = {"analysis": text} - enrichments.append(enrichment) + return {"analysis": text} except Exception as exc: + # A single record's failure must not abort the batch — + # it becomes an {"error": ...} enrichment, exactly like + # the old sequential loop's inner except did. logger.error("claude error [%d/%d] | %s", i + 1, len(records), exc) - enrichments.append({"error": str(exc)}) + return {"error": str(exc)} + + try: + # asyncio.gather returns results in the same order as the input + # awaitables (not completion order), so enrichments[i] still + # lines up with records[i] — process_with_ai's zip(records, + # enrichments) contract (and the C3 enriched-count fix) holds. + enrichments: list[dict[str, Any]] = list(await asyncio.gather( + *(_process_one(i, record) for i, record in enumerate(records)) + )) finally: await adapter.aclose() diff --git a/backend/processors/local_processor.py b/backend/processors/local_processor.py index 189e1a2..dec7480 100644 --- a/backend/processors/local_processor.py +++ b/backend/processors/local_processor.py @@ -11,11 +11,12 @@ OpenAI ``/v1/chat/completions`` shape. There is no adapter call that reaches ``/api/generate`` without changing what gets sent over the wire. * even the ``api_style="openai"`` branch has a per-call configurable - ``timeout`` (``config.get("timeout", 120)``) threaded straight into the - raw ``httpx.AsyncClient`` — ``OpenAICompatAdapter`` (frozen behavior, - PR-B, 1599-test baseline) has no parameter to accept a caller-supplied - timeout, so swapping in the adapter here would silently drop that - config knob for anyone using it. + ``timeout`` (``config.get("timeout", ...)``, defaulting to the shared + ``Settings.llm_request_timeout_seconds`` — AUDIT C8) threaded straight + into the raw ``httpx.AsyncClient`` — ``OpenAICompatAdapter`` (frozen + behavior, PR-B, 1599-test baseline) has no parameter to accept a + caller-supplied timeout, so swapping in the adapter here would silently + drop that config knob for anyone using it. Also has no SSRF guard today for either branch (raw ``httpx.AsyncClient``, no ``url_guard`` call) — unlike ``openai_processor``/``skill_channel``. This @@ -24,6 +25,7 @@ regression" scope. """ +import asyncio import json import re from typing import TYPE_CHECKING, Any @@ -55,43 +57,64 @@ async def process( prompt_template: str, config: dict[str, Any], ) -> ProcessingResult: + from backend.config import get_settings + + settings = get_settings() + base_url = config.get("base_url", "http://localhost:11434") model = config.get("model", "llama3") - timeout = config.get("timeout", 120) + # AUDIT C8: fallback now comes from the shared llm_request_timeout_seconds + # setting instead of a hardcoded 120 — an explicit config["timeout"] + # still wins, unchanged. + timeout = config.get("timeout", settings.llm_request_timeout_seconds) # Support both Ollama (/api/generate) and OpenAI-compatible (/v1/chat/completions) api_style = config.get("api_style", "ollama") + # AUDIT C25: bound how many records are in flight at once instead of + # a plain await-in-a-for-loop (wall clock == record_count x latency). + max_concurrency = max(1, settings.llm_max_concurrency) - enrichments: list[dict[str, Any]] = [] + semaphore = asyncio.Semaphore(max_concurrency) async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client: - for record in records: - prompt = _render(prompt_template, record.normalized_data) - try: - if api_style == "ollama": - resp = await client.post( - "/api/generate", - json={"model": model, "prompt": prompt, "stream": False}, - ) - resp.raise_for_status() - text = resp.json().get("response", "") - else: - # OpenAI-compatible - resp = await client.post( - "/v1/chat/completions", - json={ - "model": model, - "messages": [{"role": "user", "content": prompt}], - }, - ) - resp.raise_for_status() - text = resp.json()["choices"][0]["message"]["content"] - + # httpx.AsyncClient is safe to share across concurrent requests + # (connection-pool backed) — the semaphore just caps how many of + # `records` are in flight through it at once. + async def _process_one(record: "CollectedRecord") -> dict[str, Any]: + async with semaphore: + prompt = _render(prompt_template, record.normalized_data) try: - enrichment = json.loads(text) - except json.JSONDecodeError: - enrichment = {"analysis": text} - enrichments.append(enrichment) - except Exception as exc: - enrichments.append({"error": str(exc)}) + if api_style == "ollama": + resp = await client.post( + "/api/generate", + json={"model": model, "prompt": prompt, "stream": False}, + ) + resp.raise_for_status() + text = resp.json().get("response", "") + else: + # OpenAI-compatible + resp = await client.post( + "/v1/chat/completions", + json={ + "model": model, + "messages": [{"role": "user", "content": prompt}], + }, + ) + resp.raise_for_status() + text = resp.json()["choices"][0]["message"]["content"] + + try: + return json.loads(text) + except json.JSONDecodeError: + return {"analysis": text} + except Exception as exc: + # A single record's failure must not abort the batch. + return {"error": str(exc)} + + # asyncio.gather returns results in the same order as the input + # awaitables (not completion order), so enrichments[i] still + # lines up with records[i]. + enrichments: list[dict[str, Any]] = list(await asyncio.gather( + *(_process_one(record) for record in records) + )) return ProcessingResult(success=True, enrichments=enrichments) diff --git a/backend/processors/openai_processor.py b/backend/processors/openai_processor.py index 67ec4fa..9a9c136 100644 --- a/backend/processors/openai_processor.py +++ b/backend/processors/openai_processor.py @@ -1,5 +1,6 @@ """OpenAI AI processor.""" +import asyncio import json import logging import os @@ -40,14 +41,30 @@ async def process( except ImportError: return ProcessingResult(success=False, error="openai package not installed") + from backend.config import get_settings + + settings = get_settings() + api_key = config.get("api_key") or os.environ.get("OPENAI_API_KEY", "") base_url: str | None = config.get("base_url") or None model = config.get("model", "gpt-4o-mini") max_tokens = config.get("max_tokens", 1024) use_json_mode = config.get("json_mode", base_url is None) - - logger.info("openai processor | model=%s base_url=%s max_tokens=%d records=%d", - model, base_url or "(default)", max_tokens, len(records)) + # AUDIT C8: explicit per-request timeout — the SDK default (600s x 2 + # retries) can otherwise pin a whole batch in ai_processing for hours + # behind a dead/slow gateway. A source's ai_config can still override + # this per call via config["timeout"]. + request_timeout = config.get("timeout", settings.llm_request_timeout_seconds) + # AUDIT C25: bound how many records are in flight at once instead of + # a plain await-in-a-for-loop (wall clock == record_count x latency). + max_concurrency = max(1, settings.llm_max_concurrency) + + logger.info( + "openai processor | model=%s base_url=%s max_tokens=%d records=%d " + "timeout=%s max_concurrency=%d", + model, base_url or "(default)", max_tokens, len(records), + request_timeout, max_concurrency, + ) # GOAL-6 PR-E: client construction (SSRF guard + DNS-rebind pinning) # is consolidated through backend.llm.openai_compat.OpenAICompatAdapter @@ -62,18 +79,24 @@ async def process( client = await adapter.get_client() except LlmAdapterError as exc: return ProcessingResult(success=False, error=f"openai processor: {exc}") - enrichments: list[dict[str, Any]] = [] - try: - for i, record in enumerate(records): - prompt = _render(prompt_template, record.normalized_data) - logger.debug("openai req [%d/%d] | prompt_preview=%s", - i + 1, len(records), prompt[:200]) + semaphore = asyncio.Semaphore(max_concurrency) + + async def _process_one(i: int, record: "CollectedRecord") -> dict[str, Any]: + # AUDIT C25: the semaphore (not the for-loop) is what bounds + # concurrency now — every record's coroutine is created up front + # and handed to gather, but only `max_concurrency` run their LLM + # call at once. + async with semaphore: try: + prompt = _render(prompt_template, record.normalized_data) + logger.debug("openai req [%d/%d] | prompt_preview=%s", + i + 1, len(records), prompt[:200]) kwargs: dict[str, Any] = dict( model=model, max_tokens=max_tokens, messages=[{"role": "user", "content": prompt}], + timeout=request_timeout, ) if use_json_mode: kwargs["response_format"] = {"type": "json_object"} @@ -86,13 +109,24 @@ async def process( usage.completion_tokens if usage else -1, text[:200]) try: - enrichment = json.loads(text) + return json.loads(text) except json.JSONDecodeError: - enrichment = {"analysis": text} - enrichments.append(enrichment) + return {"analysis": text} except Exception as exc: + # A single record's failure must not abort the batch — + # it becomes an {"error": ...} enrichment, exactly like + # the old sequential loop's inner except did. logger.error("openai error [%d/%d] | %s", i + 1, len(records), exc) - enrichments.append({"error": str(exc)}) + return {"error": str(exc)} + + try: + # asyncio.gather returns results in the same order as the input + # awaitables (not completion order), so enrichments[i] still + # lines up with records[i] — process_with_ai's zip(records, + # enrichments) contract (and the C3 enriched-count fix) holds. + enrichments: list[dict[str, Any]] = list(await asyncio.gather( + *(_process_one(i, record) for i, record in enumerate(records)) + )) finally: # AsyncOpenAI does not close an externally-supplied http_client # (it doesn't own it) — close ours ourselves, same as the diff --git a/tests/unit/channels/test_rss_channel.py b/tests/unit/channels/test_rss_channel.py index a52e511..6f8fd19 100644 --- a/tests/unit/channels/test_rss_channel.py +++ b/tests/unit/channels/test_rss_channel.py @@ -1,5 +1,6 @@ """Unit tests for the RSS channel.""" +import threading from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -118,6 +119,43 @@ async def test_collect_metadata_includes_feed_title(channel): assert result.metadata.get("feed_title") == "Test Feed" +# ── AUDIT C22: feedparser.parse() off-loaded via asyncio.to_thread ───────────── + +@pytest.mark.asyncio +async def test_collect_parses_feed_off_event_loop_thread(channel): + """feedparser.parse() must run via asyncio.to_thread, not inline on the + event loop — a multi-MB feed would otherwise freeze every other + request/task on this process for the duration of the parse.""" + mock_response = MagicMock() + mock_response.text = VALID_RSS_XML + mock_response.raise_for_status = MagicMock() + + 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) + + import feedparser + + seen_threads: list[int] = [] + real_parse = feedparser.parse + + def spy_parse(content): + seen_threads.append(threading.get_ident()) + return real_parse(content) + + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + with patch("feedparser.parse", side_effect=spy_parse): + result = await channel.collect( + {"feed_url": "https://example.com/rss"}, {} + ) + + assert result.success is True + assert len(seen_threads) == 1 + assert seen_threads[0] != threading.get_ident() + + # ── collect: error cases ─────────────────────────────────────────────────────── @pytest.mark.asyncio diff --git a/tests/unit/channels/test_rss_fetch.py b/tests/unit/channels/test_rss_fetch.py index 9cb5f6c..580f8da 100644 --- a/tests/unit/channels/test_rss_fetch.py +++ b/tests/unit/channels/test_rss_fetch.py @@ -1,6 +1,7 @@ """RSS thick-contract fetch(): conditional GET (etag/304), identity(), and the runner driving it end to end (the thin-channel / thick-runner vertical slice).""" +import threading from types import SimpleNamespace from unittest.mock import patch @@ -73,6 +74,31 @@ async def test_fetch_200_returns_items_and_advances_cursor(): assert result.has_more is False +@pytest.mark.asyncio +async def test_fetch_parses_feed_off_event_loop_thread(): + """AUDIT C22: feedparser.parse() must run via asyncio.to_thread on the + fetch() path too — a multi-MB feed would otherwise freeze every other + request/task on this process for the duration of the parse.""" + import feedparser + + http = _Http(_Resp(200, text=_RSS, headers={})) + ctx = FetchContext(config={"feed_url": "https://x/feed"}, params={}, cursor=None, http=http) + + seen_threads: list[int] = [] + real_parse = feedparser.parse + + def spy_parse(content): + seen_threads.append(threading.get_ident()) + return real_parse(content) + + with patch("feedparser.parse", side_effect=spy_parse): + result = await RSSChannel().fetch(ctx) + + assert [i["id"] for i in result.items] == ["id-a", "id-b"] + assert len(seen_threads) == 1 + assert seen_threads[0] != threading.get_ident() + + @pytest.mark.asyncio async def test_fetch_304_no_new_items_keeps_cursor_and_sends_conditional(): http = _Http(_Resp(304)) diff --git a/tests/unit/channels/test_skill_channel_emit_batching.py b/tests/unit/channels/test_skill_channel_emit_batching.py new file mode 100644 index 0000000..4919d12 --- /dev/null +++ b/tests/unit/channels/test_skill_channel_emit_batching.py @@ -0,0 +1,78 @@ +"""Unit tests for skill_channel's AUDIT C24 event-batching refactor: +_emit_loop_events must build plain dict payloads (not pre-built +events.emit(...) coroutines) so the caller can hand the whole per-run step +trace to events.emit_many in one session/commit instead of awaiting one +events.emit(...) per step. + +End-to-end coverage of the real spine (events actually landing in the DB +under the right run_id/step names) already lives in +tests/skills/test_skill_channel.py — this file is the focused unit-level +check on the refactored builder function itself. +""" + + +def test_emit_loop_events_returns_plain_dicts_not_coroutines(): + from backend.channels.skill_channel import ( + STEP_DONE, + STEP_EXTRACT, + STEP_PERCEIVE, + STEP_STEP, + _emit_loop_events, + ) + from backend.skills.loop import LoopResult, StepRecord + + result = LoopResult( + steps=[ + StepRecord(index=0, verb="navigate", args={}, result={"ok": True}), + StepRecord(index=1, verb="extract", args={}, result={"data": {"title": "x"}}), + ], + extracts=[{"title": "x"}], + outcome="done_success", + summary={"note": "ok"}, + ) + + payloads = _emit_loop_events(result) + + # plain dicts — nothing here is an unawaited coroutine (which would leak + # a "coroutine was never awaited" warning and never reach the DB). + assert all(isinstance(p, dict) for p in payloads) + + # bracketed by a leading skill_perceive and trailing skill_done. + assert payloads[0]["step"] == STEP_PERCEIVE + assert payloads[-1]["step"] == STEP_DONE + # one payload per StepRecord in order, extract verb keyed distinctly. + assert payloads[1]["step"] == STEP_STEP + assert payloads[2]["step"] == STEP_EXTRACT + + # every payload carries at least the keys events.emit_many/TaskRunEvent need. + for p in payloads: + assert "message" in p + + +def test_emit_loop_events_step_error_is_warning_level(): + from backend.channels.skill_channel import STEP_STEP, _emit_loop_events + from backend.skills.loop import LoopResult, StepRecord + + result = LoopResult( + steps=[StepRecord(index=0, verb="click", args={}, error="blocked")], + outcome="error", + ) + + payloads = _emit_loop_events(result) + step_payload = next(p for p in payloads if p["step"] == STEP_STEP) + + assert step_payload["level"] == "warning" + assert "blocked" in step_payload["message"] + + +def test_emit_loop_events_done_outcome_reflects_error_level(): + from backend.channels.skill_channel import STEP_DONE, _emit_loop_events + from backend.skills.loop import LoopResult + + ok_payloads = _emit_loop_events(LoopResult(outcome="done_success")) + done_ok = next(p for p in ok_payloads if p["step"] == STEP_DONE) + assert done_ok["level"] == "info" + + failed_payloads = _emit_loop_events(LoopResult(outcome="done_failed")) + done_failed = next(p for p in failed_payloads if p["step"] == STEP_DONE) + assert done_failed["level"] == "warning" diff --git a/tests/unit/channels/test_web_scraper_channel.py b/tests/unit/channels/test_web_scraper_channel.py index 85b6d26..393d377 100644 --- a/tests/unit/channels/test_web_scraper_channel.py +++ b/tests/unit/channels/test_web_scraper_channel.py @@ -1,5 +1,6 @@ """Unit tests for the web scraper channel.""" +import threading from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -167,6 +168,40 @@ async def test_collect_metadata_includes_url_and_status(channel): assert result.metadata.get("status_code") == 200 +# ── AUDIT C22: BeautifulSoup parse off-loaded via asyncio.to_thread ──────────── + +@pytest.mark.asyncio +async def test_collect_parses_html_off_event_loop_thread(channel): + """BeautifulSoup(...) must run via asyncio.to_thread, not inline on the + event loop — a large scraped page would otherwise freeze every other + request/task on this process for the duration of the parse.""" + response = _make_mock_response() + mock_client_ctx = _make_mock_client(response) + + seen_threads: list[int] = [] + real_soup = BeautifulSoup + + def spy_soup(markup, parser): + seen_threads.append(threading.get_ident()) + return real_soup(markup, parser) + + with patch("httpx.AsyncClient", return_value=mock_client_ctx), patch( + "backend.channels.web_scraper_channel.BeautifulSoup", side_effect=spy_soup + ): + result = await channel.collect( + { + "url": "https://example.com", + "selectors": {"page_title": "h1.page-title"}, + }, + {}, + ) + + assert result.success is True + assert result.items[0]["page_title"] == "Products" + assert len(seen_threads) == 1 + assert seen_threads[0] != threading.get_ident() + + # ── collect: error cases ─────────────────────────────────────────────────────── @pytest.mark.asyncio diff --git a/tests/unit/pipeline/test_events.py b/tests/unit/pipeline/test_events.py new file mode 100644 index 0000000..9000a81 --- /dev/null +++ b/tests/unit/pipeline/test_events.py @@ -0,0 +1,80 @@ +"""Unit tests for backend.pipeline.events (AUDIT C24: emit_many batches a +whole step trace into one session + bulk insert + one commit, instead of one +session+INSERT+commit per event).""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from backend.pipeline import events + + +@pytest.mark.asyncio +async def test_emit_many_one_session_one_commit_for_n_events(): + """N events must cost exactly one session + one commit, not N.""" + mock_session = AsyncMock() + mock_session.add_all = MagicMock() + mock_session.commit = AsyncMock() + session_cm = AsyncMock() + session_cm.__aenter__ = AsyncMock(return_value=mock_session) + session_cm.__aexit__ = AsyncMock(return_value=False) + + session_ctor_calls: list[int] = [] + + def fake_session_local(): + session_ctor_calls.append(1) + return session_cm + + payloads = [ + {"step": "skill_perceive", "message": "start"}, + {"step": "skill_step", "message": "step 1", "detail": {"i": 1}}, + { + "step": "skill_step", + "message": "step 2", + "level": "warning", + "detail": {"i": 2}, + "elapsed_ms": 10, + }, + {"step": "skill_done", "message": "done"}, + ] + + with patch("backend.database.AsyncSessionLocal", side_effect=fake_session_local): + await events.emit_many("run-1", payloads) + + # one session opened regardless of how many events were in the batch. + assert len(session_ctor_calls) == 1 + # one bulk insert call, not one add() per event. + mock_session.add_all.assert_called_once() + inserted = mock_session.add_all.call_args.args[0] + assert len(inserted) == 4 + # one commit, not one per event. + assert mock_session.commit.await_count == 1 + + # field mapping: shared run_id, defaults applied, explicit values kept. + assert all(row.run_id == "run-1" for row in inserted) + assert inserted[0].level == "info" # default when not specified + assert inserted[2].level == "warning" + assert inserted[2].elapsed_ms == 10 + assert inserted[1].detail == {"i": 1} + assert inserted[3].step == "skill_done" + + +@pytest.mark.asyncio +async def test_emit_many_empty_list_is_noop_no_session_opened(): + session_ctor_calls: list[int] = [] + + def fake_session_local(): + session_ctor_calls.append(1) + raise AssertionError("should not open a session for an empty batch") + + with patch("backend.database.AsyncSessionLocal", side_effect=fake_session_local): + await events.emit_many("run-1", []) + + assert session_ctor_calls == [] + + +@pytest.mark.asyncio +async def test_emit_many_never_raises_on_db_failure(): + """Best-effort, mirrors emit(): a DB failure must not propagate.""" + with patch("backend.database.AsyncSessionLocal", side_effect=RuntimeError("db down")): + await events.emit_many("run-1", [{"step": "x", "message": "y"}]) # must not raise diff --git a/tests/unit/pipeline/test_pipeline.py b/tests/unit/pipeline/test_pipeline.py index 69695be..81fcd38 100644 --- a/tests/unit/pipeline/test_pipeline.py +++ b/tests/unit/pipeline/test_pipeline.py @@ -1,6 +1,7 @@ """Unit tests for the pipeline orchestrator.""" import pytest +from typing import Any from unittest.mock import AsyncMock, MagicMock, patch from backend.channels.base import ChannelResult @@ -103,6 +104,22 @@ async def test_run_pipeline_collector_exception(db_session): assert "collect crash" in result.error +class _FakeScalarsResult: + """Minimal stand-in for what `session.execute(select(...))` returns — + supports the sync `.scalars().all()` chain the AUDIT C21 bulk-fetch uses + (unlike a bare AsyncMock, whose auto-created child attributes are also + async and break that chain — see the N+1 test below for the same fake).""" + + def __init__(self, rows): + self._rows = rows + + def scalars(self): + return self + + def all(self): + return self._rows + + @pytest.mark.asyncio async def test_run_pipeline_with_ai(db_session): """Pipeline with ai_config calls process_with_ai.""" @@ -121,9 +138,16 @@ async def test_run_pipeline_with_ai(db_session): "prompt_template": "Summarize: {{content}}", } + # AUDIT C21: the persist step now does one bulk `session.execute(select(... + # ).where(id.in_(...)))` instead of one `session.get` per record — the DB + # row it should find and update. + db_row = MagicMock() + db_row.id = "rec-ai-1" + # Mock the inner AsyncSessionLocal calls used for AI status update and enrichment save mock_inner_session = AsyncMock() mock_inner_session.get = AsyncMock(return_value=MagicMock()) + mock_inner_session.execute = AsyncMock(return_value=_FakeScalarsResult([db_row])) mock_inner_session.commit = AsyncMock() inner_cm = AsyncMock() inner_cm.__aenter__ = AsyncMock(return_value=mock_inner_session) @@ -145,6 +169,77 @@ async def test_run_pipeline_with_ai(db_session): assert result.success is True assert result.ai_processed == 1 + # same field writes as before the N+1 fix: enrichment + status landed on + # the row the bulk SELECT found. + assert db_row.ai_enrichment == {"summary": "AI summary"} + assert db_row.status == "ai_processed" + + +@pytest.mark.asyncio +async def test_run_pipeline_ai_persist_is_bulk_not_n_plus_one(db_session): + """AUDIT C21: persisting N enriched records must cost exactly one + `session.execute` (bulk SELECT ... WHERE id IN (...)) + one commit, not N + `session.get` calls — spies on how many times execute() is invoked.""" + source, task = await _setup_source_task(db_session) + + n = 5 + mock_items = [{"title": f"Item {i}", "url": f"https://ex.com/{i}"} for i in range(n)] + channel_result = ChannelResult.ok(mock_items) + + mock_records = [] + for i in range(n): + rec = MagicMock() + rec.id = f"rec-{i}" + rec.ai_enrichment = {"summary": f"s{i}"} + mock_records.append(rec) + + db_rows = [] + for rec in mock_records: + row = MagicMock() + row.id = rec.id + db_rows.append(row) + + agent_config = {"processor_type": "claude", "model": "claude-3-haiku-20240307"} + + execute_calls: list[Any] = [] + + def _spy_execute(stmt): + execute_calls.append(stmt) + return _FakeScalarsResult(db_rows) + + mock_inner_session = AsyncMock() + mock_inner_session.get = AsyncMock(return_value=MagicMock()) # ai_processing status row + mock_inner_session.execute = AsyncMock(side_effect=_spy_execute) + mock_inner_session.commit = AsyncMock() + inner_cm = AsyncMock() + inner_cm.__aenter__ = AsyncMock(return_value=mock_inner_session) + inner_cm.__aexit__ = AsyncMock(return_value=False) + + with ( + patch("backend.pipeline.collector.collect", return_value=channel_result), + patch("backend.pipeline.storer.store_records", new=AsyncMock(return_value=(mock_records, 0))), + patch("backend.pipeline.ai_processor.process_with_ai", new=AsyncMock(return_value=n)), + patch("backend.database.AsyncSessionLocal", return_value=inner_cm), + ): + result = await run_pipeline( + task.id, + source, + agent_config=agent_config, + enable_ai=True, + enable_notifications=False, + ) + + assert result.success is True + assert result.ai_processed == n + + # exactly one execute() call persists all n enrichments — not n (N+1). + assert len(execute_calls) == 1 + + # same field writes as before: every row got its enrichment + status set. + for i, row in enumerate(db_rows): + assert row.ai_enrichment == {"summary": f"s{i}"} + assert row.status == "ai_processed" + assert mock_inner_session.commit.await_count >= 1 @pytest.mark.asyncio diff --git a/tests/unit/test_claude_processor.py b/tests/unit/test_claude_processor.py new file mode 100644 index 0000000..0806767 --- /dev/null +++ b/tests/unit/test_claude_processor.py @@ -0,0 +1,196 @@ +"""Unit tests for the Claude (Anthropic) AI processor (AUDIT C8/C25: explicit +per-request timeout + bounded concurrency instead of a sequential +await-in-a-for-loop). + +The SDK client is mocked at the ``anthropic.AsyncAnthropic`` class level, +same convention as tests/unit/llm/test_adapters.py — no real network call. +""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from backend.processors.claude_processor import ClaudeProcessor + + +def _record(rec_id: str, content: str = "hello") -> MagicMock: + record = MagicMock() + record.id = rec_id + record.normalized_data = {"content": content} + return record + + +def _msg_response(text: str) -> SimpleNamespace: + return SimpleNamespace( + content=[SimpleNamespace(text=text)], + usage=SimpleNamespace(input_tokens=1, output_tokens=1), + ) + + +# ── C8: explicit per-request timeout ──────────────────────────────────────── + +@pytest.mark.asyncio +async def test_process_sets_explicit_timeout_from_settings_default(): + """Every call carries an explicit timeout instead of relying on the SDK's + own (600s x 2 retries) default.""" + mock_client = MagicMock() + mock_client.messages.create = AsyncMock(return_value=_msg_response('{"a": 1}')) + + with patch("anthropic.AsyncAnthropic", return_value=mock_client): + result = await ClaudeProcessor().process( + records=[_record("r1")], prompt_template="{{content}}", config={} + ) + + assert result.success is True + _, kwargs = mock_client.messages.create.call_args + assert kwargs["timeout"] == 120 # Settings.llm_request_timeout_seconds default + + +@pytest.mark.asyncio +async def test_process_timeout_overridable_via_config(): + """An explicit ai_config["timeout"] wins over the settings default.""" + mock_client = MagicMock() + mock_client.messages.create = AsyncMock(return_value=_msg_response('{"a": 1}')) + + with patch("anthropic.AsyncAnthropic", return_value=mock_client): + await ClaudeProcessor().process( + records=[_record("r1")], + prompt_template="{{content}}", + config={"timeout": 45}, + ) + + _, kwargs = mock_client.messages.create.call_args + assert kwargs["timeout"] == 45 + + +@pytest.mark.asyncio +async def test_process_timeout_configurable_via_settings_env(monkeypatch): + """LLM_REQUEST_TIMEOUT_SECONDS env var changes the fallback default.""" + from backend.config import get_settings + + monkeypatch.setenv("LLM_REQUEST_TIMEOUT_SECONDS", "45") + get_settings.cache_clear() + try: + mock_client = MagicMock() + mock_client.messages.create = AsyncMock(return_value=_msg_response('{"a": 1}')) + with patch("anthropic.AsyncAnthropic", return_value=mock_client): + await ClaudeProcessor().process( + records=[_record("r1")], prompt_template="{{content}}", config={} + ) + _, kwargs = mock_client.messages.create.call_args + assert kwargs["timeout"] == 45 + finally: + get_settings.cache_clear() + + +# ── C25: bounded concurrency, order, failure isolation ────────────────────── + +@pytest.mark.asyncio +async def test_process_bounds_concurrency_to_semaphore_limit(monkeypatch): + """No more than LLM_MAX_CONCURRENCY records' LLM calls run at once, even + when many records are enriched in one batch.""" + from backend.config import get_settings + + monkeypatch.setenv("LLM_MAX_CONCURRENCY", "2") + get_settings.cache_clear() + + state = {"cur": 0, "peak": 0} + + async def fake_create(**kwargs): + state["cur"] += 1 + state["peak"] = max(state["peak"], state["cur"]) + await asyncio.sleep(0.02) + state["cur"] -= 1 + return _msg_response('{"ok": true}') + + mock_client = MagicMock() + mock_client.messages.create = fake_create + + try: + with patch("anthropic.AsyncAnthropic", return_value=mock_client): + result = await ClaudeProcessor().process( + records=[_record(f"r{i}") for i in range(6)], + prompt_template="{{content}}", + config={}, + ) + finally: + get_settings.cache_clear() + + assert result.success is True + assert len(result.enrichments) == 6 + assert state["peak"] <= 2 + + +@pytest.mark.asyncio +async def test_process_preserves_record_order_despite_uneven_latency(): + """gather preserves input order regardless of which call finishes first — + enrichments[i] must still correspond to records[i].""" + delays = {"r0": 0.03, "r1": 0.0, "r2": 0.0} + + async def fake_create(**kwargs): + rec_id = kwargs["messages"][0]["content"] + await asyncio.sleep(delays[rec_id]) + return _msg_response(f'{{"seen": "{rec_id}"}}') + + mock_client = MagicMock() + mock_client.messages.create = fake_create + + records = [ + _record("r0", content="r0"), + _record("r1", content="r1"), + _record("r2", content="r2"), + ] + + with patch("anthropic.AsyncAnthropic", return_value=mock_client): + result = await ClaudeProcessor().process( + records=records, prompt_template="{{content}}", config={} + ) + + assert [e["seen"] for e in result.enrichments] == ["r0", "r1", "r2"] + + +@pytest.mark.asyncio +async def test_process_one_record_failure_does_not_abort_batch(): + """One record's LLM call raising must not prevent the other records in + the same batch from being enriched.""" + + async def fake_create(**kwargs): + prompt = kwargs["messages"][0]["content"] + if prompt == "boom": + raise RuntimeError("gateway exploded") + return _msg_response('{"ok": true}') + + mock_client = MagicMock() + mock_client.messages.create = fake_create + + records = [ + _record("r0", content="ok"), + _record("r1", content="boom"), + _record("r2", content="ok"), + ] + + with patch("anthropic.AsyncAnthropic", return_value=mock_client): + result = await ClaudeProcessor().process( + records=records, prompt_template="{{content}}", config={} + ) + + assert result.success is True + assert len(result.enrichments) == 3 + assert result.enrichments[0] == {"ok": True} + assert "error" in result.enrichments[1] + assert "gateway exploded" in result.enrichments[1]["error"] + assert result.enrichments[2] == {"ok": True} + + +@pytest.mark.asyncio +async def test_process_no_anthropic_package_returns_failed_result(): + """The pre-existing import-availability guard must still short-circuit + cleanly (not raise) when the anthropic package isn't installed.""" + with patch.dict("sys.modules", {"anthropic": None}): + result = await ClaudeProcessor().process( + records=[_record("r1")], prompt_template="{{content}}", config={} + ) + assert result.success is False + assert "not installed" in result.error diff --git a/tests/unit/test_local_processor.py b/tests/unit/test_local_processor.py new file mode 100644 index 0000000..e98f51b --- /dev/null +++ b/tests/unit/test_local_processor.py @@ -0,0 +1,210 @@ +"""Unit tests for the local (Ollama/vLLM) AI processor (AUDIT C8/C25: +settings-driven timeout default + bounded concurrency instead of a +sequential await-in-a-for-loop). + +httpx.AsyncClient is mocked at the class level, same convention as +tests/unit/channels/test_rss_channel.py / test_web_scraper_channel.py — no +real network call. +""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from backend.processors.local_processor import LocalProcessor + + +def _record(rec_id: str, content: str = "hello") -> MagicMock: + record = MagicMock() + record.id = rec_id + record.normalized_data = {"content": content} + return record + + +def _make_client_ctx(post_impl): + mock_client = AsyncMock() + mock_client.post = post_impl + mock_client_ctx = AsyncMock() + mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_ctx.__aexit__ = AsyncMock(return_value=False) + return mock_client_ctx + + +def _ok_response(payload): + resp = MagicMock() + resp.raise_for_status = MagicMock() + resp.json = MagicMock(return_value=payload) + return resp + + +# ── C8: client timeout default sourced from settings ──────────────────────── + +@pytest.mark.asyncio +async def test_process_client_timeout_defaults_from_settings(): + captured = {} + + def fake_ctor(**kwargs): + captured.update(kwargs) + return _make_client_ctx(AsyncMock(return_value=_ok_response({"response": "{}"}))) + + with patch("httpx.AsyncClient", side_effect=fake_ctor): + result = await LocalProcessor().process( + records=[_record("r1")], prompt_template="{{content}}", config={} + ) + + assert result.success is True + assert captured["timeout"] == 120 # Settings.llm_request_timeout_seconds default + + +@pytest.mark.asyncio +async def test_process_client_timeout_overridable_via_config(): + captured = {} + + def fake_ctor(**kwargs): + captured.update(kwargs) + return _make_client_ctx(AsyncMock(return_value=_ok_response({"response": "{}"}))) + + with patch("httpx.AsyncClient", side_effect=fake_ctor): + await LocalProcessor().process( + records=[_record("r1")], + prompt_template="{{content}}", + config={"timeout": 30}, + ) + + assert captured["timeout"] == 30 + + +@pytest.mark.asyncio +async def test_process_client_timeout_configurable_via_settings_env(monkeypatch): + from backend.config import get_settings + + monkeypatch.setenv("LLM_REQUEST_TIMEOUT_SECONDS", "45") + get_settings.cache_clear() + captured = {} + + def fake_ctor(**kwargs): + captured.update(kwargs) + return _make_client_ctx(AsyncMock(return_value=_ok_response({"response": "{}"}))) + + try: + with patch("httpx.AsyncClient", side_effect=fake_ctor): + await LocalProcessor().process( + records=[_record("r1")], prompt_template="{{content}}", config={} + ) + assert captured["timeout"] == 45 + finally: + get_settings.cache_clear() + + +# ── C25: bounded concurrency, order, failure isolation ────────────────────── + +@pytest.mark.asyncio +async def test_process_bounds_concurrency_to_semaphore_limit(monkeypatch): + from backend.config import get_settings + + monkeypatch.setenv("LLM_MAX_CONCURRENCY", "2") + get_settings.cache_clear() + + state = {"cur": 0, "peak": 0} + + async def fake_post(url, json): + state["cur"] += 1 + state["peak"] = max(state["peak"], state["cur"]) + await asyncio.sleep(0.02) + state["cur"] -= 1 + return _ok_response({"response": '{"ok": true}'}) + + mock_client_ctx = _make_client_ctx(fake_post) + + try: + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + result = await LocalProcessor().process( + records=[_record(f"r{i}") for i in range(6)], + prompt_template="{{content}}", + config={}, + ) + finally: + get_settings.cache_clear() + + assert result.success is True + assert len(result.enrichments) == 6 + assert state["peak"] <= 2 + + +@pytest.mark.asyncio +async def test_process_preserves_record_order_despite_uneven_latency(): + delays = {"r0": 0.03, "r1": 0.0, "r2": 0.0} + + async def fake_post(url, json): + rec_id = json["prompt"] + await asyncio.sleep(delays[rec_id]) + return _ok_response({"response": f'{{"seen": "{rec_id}"}}'}) + + mock_client_ctx = _make_client_ctx(fake_post) + + records = [ + _record("r0", content="r0"), + _record("r1", content="r1"), + _record("r2", content="r2"), + ] + + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + result = await LocalProcessor().process( + records=records, prompt_template="{{content}}", config={} + ) + + assert [e["seen"] for e in result.enrichments] == ["r0", "r1", "r2"] + + +@pytest.mark.asyncio +async def test_process_one_record_failure_does_not_abort_batch(): + async def fake_post(url, json): + if json["prompt"] == "boom": + raise RuntimeError("connection reset") + return _ok_response({"response": '{"ok": true}'}) + + mock_client_ctx = _make_client_ctx(fake_post) + + records = [ + _record("r0", content="ok"), + _record("r1", content="boom"), + _record("r2", content="ok"), + ] + + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + result = await LocalProcessor().process( + records=records, prompt_template="{{content}}", config={} + ) + + assert result.success is True + assert len(result.enrichments) == 3 + assert result.enrichments[0] == {"ok": True} + assert "error" in result.enrichments[1] + assert "connection reset" in result.enrichments[1]["error"] + assert result.enrichments[2] == {"ok": True} + + +@pytest.mark.asyncio +async def test_process_openai_compatible_api_style_posts_chat_completions(): + """api_style="openai" hits /v1/chat/completions with the OpenAI shape — + unaffected by the C8/C25 changes.""" + captured = {} + + async def fake_post(url, json): + captured["url"] = url + captured["json"] = json + return _ok_response({"choices": [{"message": {"content": '{"ok": true}'}}]}) + + mock_client_ctx = _make_client_ctx(fake_post) + + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + result = await LocalProcessor().process( + records=[_record("r1")], + prompt_template="{{content}}", + config={"api_style": "openai"}, + ) + + assert result.success is True + assert captured["url"] == "/v1/chat/completions" + assert result.enrichments == [{"ok": True}] diff --git a/tests/unit/test_openai_processor.py b/tests/unit/test_openai_processor.py new file mode 100644 index 0000000..95d2cc9 --- /dev/null +++ b/tests/unit/test_openai_processor.py @@ -0,0 +1,205 @@ +"""Unit tests for the OpenAI AI processor (AUDIT C8/C25: explicit per-request +timeout + bounded concurrency instead of a sequential await-in-a-for-loop). + +The SDK client is mocked at the ``openai.AsyncOpenAI`` class level, same +convention as tests/unit/llm/test_adapters.py — no real network call. +""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from backend.processors.openai_processor import OpenAIProcessor + + +def _record(rec_id: str, content: str = "hello") -> MagicMock: + record = MagicMock() + record.id = rec_id + record.normalized_data = {"content": content} + return record + + +def _chat_response(text: str) -> SimpleNamespace: + return SimpleNamespace( + choices=[SimpleNamespace(message=SimpleNamespace(content=text))], + usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1), + ) + + +# ── C8: explicit per-request timeout ──────────────────────────────────────── + +@pytest.mark.asyncio +async def test_process_sets_explicit_timeout_from_settings_default(): + """Every call carries an explicit timeout instead of relying on the SDK's + own (600s x 2 retries) default.""" + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=_chat_response('{"a": 1}')) + + with patch("openai.AsyncOpenAI", return_value=mock_client): + result = await OpenAIProcessor().process( + records=[_record("r1")], + prompt_template="{{content}}", + config={"base_url": None}, + ) + + assert result.success is True + _, kwargs = mock_client.chat.completions.create.call_args + assert kwargs["timeout"] == 120 # Settings.llm_request_timeout_seconds default + + +@pytest.mark.asyncio +async def test_process_timeout_overridable_via_config(): + """An explicit ai_config["timeout"] wins over the settings default.""" + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=_chat_response('{"a": 1}')) + + with patch("openai.AsyncOpenAI", return_value=mock_client): + await OpenAIProcessor().process( + records=[_record("r1")], + prompt_template="{{content}}", + config={"base_url": None, "timeout": 45}, + ) + + _, kwargs = mock_client.chat.completions.create.call_args + assert kwargs["timeout"] == 45 + + +@pytest.mark.asyncio +async def test_process_timeout_configurable_via_settings_env(monkeypatch): + """LLM_REQUEST_TIMEOUT_SECONDS env var changes the fallback default.""" + from backend.config import get_settings + + monkeypatch.setenv("LLM_REQUEST_TIMEOUT_SECONDS", "45") + get_settings.cache_clear() + try: + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=_chat_response('{"a": 1}')) + with patch("openai.AsyncOpenAI", return_value=mock_client): + await OpenAIProcessor().process( + records=[_record("r1")], + prompt_template="{{content}}", + config={"base_url": None}, + ) + _, kwargs = mock_client.chat.completions.create.call_args + assert kwargs["timeout"] == 45 + finally: + get_settings.cache_clear() + + +# ── C25: bounded concurrency, order, failure isolation ────────────────────── + +@pytest.mark.asyncio +async def test_process_bounds_concurrency_to_semaphore_limit(monkeypatch): + """No more than LLM_MAX_CONCURRENCY records' LLM calls run at once, even + when many records are enriched in one batch.""" + from backend.config import get_settings + + monkeypatch.setenv("LLM_MAX_CONCURRENCY", "2") + get_settings.cache_clear() + + state = {"cur": 0, "peak": 0} + + async def fake_create(**kwargs): + state["cur"] += 1 + state["peak"] = max(state["peak"], state["cur"]) + await asyncio.sleep(0.02) + state["cur"] -= 1 + return _chat_response('{"ok": true}') + + mock_client = MagicMock() + mock_client.chat.completions.create = fake_create + + try: + with patch("openai.AsyncOpenAI", return_value=mock_client): + result = await OpenAIProcessor().process( + records=[_record(f"r{i}") for i in range(6)], + prompt_template="{{content}}", + config={"base_url": None}, + ) + finally: + get_settings.cache_clear() + + assert result.success is True + assert len(result.enrichments) == 6 + assert state["peak"] <= 2 + + +@pytest.mark.asyncio +async def test_process_preserves_record_order_despite_uneven_latency(): + """gather preserves input order regardless of which call finishes first — + enrichments[i] must still correspond to records[i].""" + # record 0's call takes longer than 1/2's, so completion order is 1, 2, 0 + # — but the result list must stay in submission order. + delays = {"r0": 0.03, "r1": 0.0, "r2": 0.0} + + async def fake_create(**kwargs): + rec_id = kwargs["messages"][0]["content"] # prompt renders straight to id + await asyncio.sleep(delays[rec_id]) + return _chat_response(f'{{"seen": "{rec_id}"}}') + + mock_client = MagicMock() + mock_client.chat.completions.create = fake_create + + records = [ + _record("r0", content="r0"), + _record("r1", content="r1"), + _record("r2", content="r2"), + ] + + with patch("openai.AsyncOpenAI", return_value=mock_client): + result = await OpenAIProcessor().process( + records=records, + prompt_template="{{content}}", + config={"base_url": None}, + ) + + assert [e["seen"] for e in result.enrichments] == ["r0", "r1", "r2"] + + +@pytest.mark.asyncio +async def test_process_one_record_failure_does_not_abort_batch(): + """One record's LLM call raising must not prevent the other records in + the same batch from being enriched.""" + + async def fake_create(**kwargs): + prompt = kwargs["messages"][0]["content"] + if prompt == "boom": + raise RuntimeError("gateway exploded") + return _chat_response('{"ok": true}') + + mock_client = MagicMock() + mock_client.chat.completions.create = fake_create + + records = [ + _record("r0", content="ok"), + _record("r1", content="boom"), + _record("r2", content="ok"), + ] + + with patch("openai.AsyncOpenAI", return_value=mock_client): + result = await OpenAIProcessor().process( + records=records, + prompt_template="{{content}}", + config={"base_url": None}, + ) + + assert result.success is True + assert len(result.enrichments) == 3 + assert result.enrichments[0] == {"ok": True} + assert "error" in result.enrichments[1] + assert "gateway exploded" in result.enrichments[1]["error"] + assert result.enrichments[2] == {"ok": True} + + +@pytest.mark.asyncio +async def test_process_no_openai_package_returns_failed_result(): + """The pre-existing import-availability guard must still short-circuit + cleanly (not raise) when the openai package isn't installed.""" + with patch.dict("sys.modules", {"openai": None}): + result = await OpenAIProcessor().process( + records=[_record("r1")], prompt_template="{{content}}", config={} + ) + assert result.success is False + assert "not installed" in result.error