From 553a294912dfb047fb9300807a195080c3d89817 Mon Sep 17 00:00:00 2001 From: Curry Date: Sat, 18 Jul 2026 21:31:09 +0800 Subject: [PATCH] fix(pipeline): notify without write lock + real success counts [C1,C3,C12,C18,C19,C23] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restructure notifier_dispatch.dispatch_notifications into three phases so notification sends never hold the SQLite write lock, and callers get real outcome counts instead of unconditional success: - C1/C23: dispatch_notifications now runs Phase A (caller session: query rules, create pending NotificationLog rows, flush+commit), Phase B (no session open: sequential sends, one resolved notifier instance per rule), Phase C (fresh short session: bulk-persist outcomes, single commit). Returns {"sent": n, "failed": m}. - C12: pipeline.py step 5 consumes that aggregate, assigns PipelineResult.notifications_sent (previously dead), and logs/emits real sent/failed numbers - warning level when every attempted send failed. - C3: ai_processor.process_with_ai now returns the number of records actually enriched (0 on empty input or unregistered processor_type, with a warning log) instead of None. pipeline.py step 4 uses that real count for ai_processed and only emits "AI 处理完成" when something was actually enriched; a config error now emits a warning-level "失败/跳过" event. - C18: unknown notifier_type on a rule now logs a warning with the rule id and notifier_type instead of skipping silently. - C19: runner.py's terminal-failure handler now prefixes the persisted error_message with the same retry-classification pipeline.py already computes (effective_error_type), so ops can distinguish retryable from permanent failures from the stored message alone. Tests extended in tests/unit/pipeline/{test_notifier_dispatch, test_notifier_dispatch_errors,test_ai_processor,test_pipeline}.py and tests/unit/test_runner.py: phase-commit-before-send ordering, partial/total send failure aggregates, unknown notifier_type/processor_type warnings, and the exact error_message prefix format. --- backend/pipeline/ai_processor.py | 24 ++- backend/pipeline/notifier_dispatch.py | 163 ++++++++++++++---- backend/pipeline/pipeline.py | 55 ++++-- backend/pipeline/runner.py | 14 +- tests/unit/pipeline/test_ai_processor.py | 22 ++- tests/unit/pipeline/test_notifier_dispatch.py | 27 ++- .../pipeline/test_notifier_dispatch_errors.py | 151 +++++++++++++++- tests/unit/pipeline/test_pipeline.py | 147 +++++++++++++++- tests/unit/test_runner.py | 6 + 9 files changed, 545 insertions(+), 64 deletions(-) diff --git a/backend/pipeline/ai_processor.py b/backend/pipeline/ai_processor.py index a83ac12..c5d6638 100644 --- a/backend/pipeline/ai_processor.py +++ b/backend/pipeline/ai_processor.py @@ -78,8 +78,9 @@ async def process_with_ai( *, source_id: Any = None, resolve_provider: bool = True, -) -> None: - """Enrich records with AI processing in-place. +) -> int: + """Enrich records with AI processing in-place. Returns the number of + records actually enriched. ai_config keys: processor_type: claude | openai | local @@ -106,9 +107,15 @@ async def process_with_ai( misfire on every agent-driven run, not just legacy inline ``DataSource.ai_config`` — so agent-sourced configs skip resolution entirely and are used exactly as before. + + AUDIT C3: an unknown/misconfigured ``processor_type`` used to no-op with + no signal at all, while the caller (``pipeline.py``) unconditionally + reported ``len(new_records)`` as "processed" regardless of what actually + happened here. This now returns 0 and logs a warning so the caller can + tell the difference between "enriched" and "silently skipped". """ if not ai_config or not records: - return + return 0 resolved_config = ( await _resolve_llm_config(ai_config, source_id) if resolve_provider else ai_config @@ -118,7 +125,12 @@ async def process_with_ai( try: processor = get_processor(processor_type) except ValueError: - return + logger.warning( + "DataSource %s ai_config.processor_type=%r is not a registered " + "processor; AI enrichment skipped for %d record(s)", + source_id, processor_type, len(records), + ) + return 0 result = await processor.process( records=records, @@ -126,6 +138,10 @@ async def process_with_ai( config=resolved_config, ) + enriched = 0 for record, enrichment in zip(records, result.enrichments): record.ai_enrichment = enrichment record.status = "ai_processed" + enriched += 1 + + return enriched diff --git a/backend/pipeline/notifier_dispatch.py b/backend/pipeline/notifier_dispatch.py index 12357f6..748380f 100644 --- a/backend/pipeline/notifier_dispatch.py +++ b/backend/pipeline/notifier_dispatch.py @@ -1,13 +1,34 @@ -"""Pipeline Step 5: Dispatch notifications based on rules.""" +"""Pipeline Step 5: Dispatch notifications based on rules. + +Restructured (AUDIT C1/C23/C12/C18) into three phases so a slow notifier +send never holds the SQLite write lock, and the caller gets a real +sent/failed aggregate instead of an unconditional "done": + +* Phase A (``session``, caller-provided, same contract as before): query + matching rules, create every NotificationLog row as "pending", flush + + COMMIT. No network I/O happens here. +* Phase B (no session open at all): perform the actual sends, sequentially, + reusing one resolved notifier instance per rule (not re-looked-up per + record). +* Phase C (a brand-new short-lived session opened internally): bulk-apply + the phase B outcomes onto the phase A NotificationLog rows, single commit. +""" + +import logging +import uuid +from dataclasses import dataclass +from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from backend.models.notification import NotificationLog, NotificationRule from backend.models.record import CollectedRecord -from backend.notifiers.base import NotificationPayload, NotificationSendResult +from backend.notifiers.base import AbstractNotifier, NotificationPayload, NotificationSendResult from backend.notifiers.registry import get_notifier +logger = logging.getLogger(__name__) + def _ack_secret(config: dict) -> str: return str(config.get("ack_secret") or "") @@ -19,16 +40,67 @@ def _normalize_send_result(result: bool | NotificationSendResult) -> tuple[bool, return bool(result), None +@dataclass +class _PendingSend: + """One (rule, record) send queued in phase A, executed in phase B. + + Plain data only (no ORM objects) — phase B runs with no session open, so + nothing here may depend on a live session/identity map. + """ + + log_id: str + notifier: AbstractNotifier + notifier_config: dict[str, Any] + payload: NotificationPayload + ack_required: bool + + +@dataclass +class _SendOutcome: + log_id: str + status: str + response_data: dict[str, Any] | None + error_message: str | None + ack_status: str + + +async def _send_one(task: _PendingSend) -> _SendOutcome: + """Perform a single notifier send, never raising — a bad send degrades to + a "failed" outcome so one broken rule can't abort the rest of phase B.""" + try: + success, response_data = _normalize_send_result( + await task.notifier.send(task.notifier_config, task.payload) + ) + status = "sent" if success else "failed" + error_msg = None + except Exception as exc: + status = "failed" + response_data = None + error_msg = str(exc) + + ack_status = "pending" if status == "sent" and task.ack_required else "not_required" + return _SendOutcome( + log_id=task.log_id, + status=status, + response_data=response_data, + error_message=error_msg, + ack_status=ack_status, + ) + + async def dispatch_notifications( session: AsyncSession, source_id: str, records: list[CollectedRecord], trigger_event: str = "on_new_record", -) -> None: - """Find matching notification rules and dispatch.""" +) -> dict[str, int]: + """Find matching notification rules and dispatch. Returns ``{"sent": n, + "failed": m}`` — the real aggregate outcome (AUDIT C12: this used to be + silent per-row-only bookkeeping the caller never inspected).""" if not records: - return + return {"sent": 0, "failed": 0} + # ── Phase A: query rules + create pending logs (caller's session) ─────── result = await session.execute( select(NotificationRule).where( NotificationRule.enabled.is_(True), @@ -39,43 +111,68 @@ async def dispatch_notifications( ) rules = result.scalars().all() + send_plan: list[_PendingSend] = [] for rule in rules: try: notifier = get_notifier(rule.notifier_type) except ValueError: + # AUDIT C18: a misconfigured rule used to die silently forever — + # at least surface which rule/type so it's discoverable. + logger.warning( + "notification rule %s references unknown notifier_type=%r; skipping", + rule.id, rule.notifier_type, + ) continue + ack_required = bool(_ack_secret(rule.notifier_config)) for record in records: - log = NotificationLog( + log_id = str(uuid.uuid4()) + session.add(NotificationLog( + id=log_id, rule_id=rule.id, record_id=record.id, status="pending", ack_status="not_required", - ) - session.add(log) - await session.flush() + )) + send_plan.append(_PendingSend( + log_id=log_id, + notifier=notifier, + notifier_config=rule.notifier_config, + payload=NotificationPayload( + event=trigger_event, + source_id=source_id, + delivery_id=log_id, + record_id=record.id, + data=record.normalized_data, + ai_enrichment=record.ai_enrichment, + ), + ack_required=ack_required, + )) - payload = NotificationPayload( - event=trigger_event, - source_id=source_id, - delivery_id=log.id, - record_id=record.id, - data=record.normalized_data, - ai_enrichment=record.ai_enrichment, - ) - try: - success, response_data = _normalize_send_result( - await notifier.send(rule.notifier_config, payload) - ) - status = "sent" if success else "failed" - error_msg = None - except Exception as exc: - status = "failed" - response_data = None - error_msg = str(exc) - - log.status = status - log.response_data = response_data - log.error_message = error_msg - if status == "sent" and _ack_secret(rule.notifier_config): - log.ack_status = "pending" + if not send_plan: + return {"sent": 0, "failed": 0} + + await session.flush() + await session.commit() + + # ── Phase B: sends, sequential, no session open ────────────────────────── + outcomes = [await _send_one(task) for task in send_plan] + + sent = sum(1 for outcome in outcomes if outcome.status == "sent") + failed = len(outcomes) - sent + + # ── Phase C: persist outcomes in a fresh short-lived session ───────────── + from backend.database import AsyncSessionLocal + + async with AsyncSessionLocal() as write_session: + for outcome in outcomes: + log = await write_session.get(NotificationLog, outcome.log_id) + if log is None: + continue + log.status = outcome.status + log.response_data = outcome.response_data + log.error_message = outcome.error_message + log.ack_status = outcome.ack_status + await write_session.commit() + + return {"sent": sent, "failed": failed} diff --git a/backend/pipeline/pipeline.py b/backend/pipeline/pipeline.py index a44e5b4..6170364 100644 --- a/backend/pipeline/pipeline.py +++ b/backend/pipeline/pipeline.py @@ -390,7 +390,7 @@ async def run_pipeline( task_row.status = "ai_processing" await session.commit() try: - await ai_processor.process_with_ai( + ai_count = await ai_processor.process_with_ai( new_records, effective_ai_config, source_id=source.id, @@ -410,14 +410,24 @@ async def run_pipeline( db_rec.ai_enrichment = rec.ai_enrichment db_rec.status = "ai_processed" await session.commit() - ai_count = len(new_records) logger.info("[task:%s] step4/ai done | processed=%d", task_id, ai_count) if run_id: - await events.emit( - run_id, "ai_process", - f"AI 处理完成 | {ai_count} 条", - detail={"processed": ai_count}, - ) + # AUDIT C3: ai_count is now the real enrichment count (0 when + # processor_type is unknown/misconfigured) — a config error + # must read as "failed/skipped", never "完成". + if ai_count > 0: + await events.emit( + run_id, "ai_process", + f"AI 处理完成 | {ai_count} 条", + detail={"processed": ai_count}, + ) + else: + await events.emit( + run_id, "ai_process", + f"AI 处理失败/跳过 | processor_type={processor_type!r} 未产生任何富化记录", + level="warning", + detail={"processed": 0, "processor_type": processor_type}, + ) except Exception as exc: logger.warning("[task:%s] step4/ai failed | %s", task_id, exc) if run_id: @@ -432,15 +442,37 @@ async def run_pipeline( await events.emit(run_id, "ai_process", "跳过 AI 处理（未配置）") # Step 5: Notify + notifications_sent = 0 if enable_notifications and new_records: logger.info("[task:%s] step5/notify start | records=%d", task_id, len(new_records)) try: + # dispatch_notifications manages its own write-lock-free phasing + # internally (AUDIT C1/C23): it commits phase A's pending rows on + # this session, performs the sends with no session open, then + # persists outcomes via its own short-lived session — so no + # explicit commit is needed here. async with AsyncSessionLocal() as session: - await notifier_dispatch.dispatch_notifications(session, source.id, new_records) - await session.commit() - logger.info("[task:%s] step5/notify done", task_id) + notify_summary = await notifier_dispatch.dispatch_notifications( + session, source.id, new_records + ) + notifications_sent = notify_summary.get("sent", 0) + notifications_failed = notify_summary.get("failed", 0) + # AUDIT C12: report the real aggregate, not an unconditional + # "done" — and if every attempted send failed, that's a warning, + # not routine info. + all_failed = notifications_failed > 0 and notifications_sent == 0 + log_fn = logger.warning if all_failed else logger.info + log_fn( + "[task:%s] step5/notify done | sent=%d failed=%d", + task_id, notifications_sent, notifications_failed, + ) if run_id: - await events.emit(run_id, "notify", "通知发送完成") + await events.emit( + run_id, "notify", + f"通知发送完成 | 成功 {notifications_sent} 失败 {notifications_failed}", + level="warning" if all_failed else "info", + detail={"sent": notifications_sent, "failed": notifications_failed}, + ) except Exception as exc: logger.warning("[task:%s] step5/notify failed | %s", task_id, exc) if run_id: @@ -501,6 +533,7 @@ async def run_pipeline( stored=len(new_records), skipped=skipped, ai_processed=ai_count, + notifications_sent=notifications_sent, duration_ms=duration_ms, metadata=channel_result.metadata, ) diff --git a/backend/pipeline/runner.py b/backend/pipeline/runner.py index 18f3640..6619e59 100644 --- a/backend/pipeline/runner.py +++ b/backend/pipeline/runner.py @@ -8,6 +8,7 @@ from backend.database import AsyncSessionLocal from backend.models.task import CollectionTask, TaskRun from backend.pipeline import events +from backend.pipeline.error_taxonomy import effective_error_type from backend.pipeline.pipeline import run_pipeline logger = logging.getLogger(__name__) @@ -170,15 +171,24 @@ async def run_collection_pipeline( # off and retries. Each retry is a fresh run_collection_pipeline # call, so it gets its own new TaskRun row (see TaskRun docstring: # "a single execution attempt"). + # + # AUDIT C19: persist the same retry-classification pipeline.py + # already computed (effective_error_type is a pure function of + # the exception, so recomputing it here reproduces that exact + # classification) instead of discarding it into a bare str(exc) — + # ops otherwise has no way to tell retryable from permanent from + # the stored error_message alone. + error_type = effective_error_type(exc) + error_message = f"[{error_type}] {exc}" async with AsyncSessionLocal() as session: err_task = await session.get(CollectionTask, task_id) err_run = await session.get(TaskRun, run_id) if err_task: err_task.status = "failed" - err_task.error_message = str(exc) + err_task.error_message = error_message if err_run: err_run.status = "failed" - err_run.error_message = str(exc) + err_run.error_message = error_message err_run.finished_at = datetime.now(UTC) await session.commit() raise diff --git a/tests/unit/pipeline/test_ai_processor.py b/tests/unit/pipeline/test_ai_processor.py index c8fc36f..5879797 100644 --- a/tests/unit/pipeline/test_ai_processor.py +++ b/tests/unit/pipeline/test_ai_processor.py @@ -12,21 +12,28 @@ @pytest.mark.asyncio async def test_process_with_ai_no_config(): records = [MagicMock()] - await process_with_ai(records, None) + result = await process_with_ai(records, None) # Should do nothing + assert result == 0 @pytest.mark.asyncio async def test_process_with_ai_no_records(): - await process_with_ai([], {"processor_type": "claude"}) + result = await process_with_ai([], {"processor_type": "claude"}) # Should do nothing + assert result == 0 @pytest.mark.asyncio -async def test_process_with_ai_unknown_processor(): +async def test_process_with_ai_unknown_processor(caplog): records = [MagicMock()] - # Should silently skip unknown processor - await process_with_ai(records, {"processor_type": "unknown_processor_xyz"}) + # Should silently skip unknown processor (AUDIT C3: "silently" only in + # the sense of not raising — it must now warn and report 0 enriched). + with caplog.at_level(logging.WARNING): + result = await process_with_ai(records, {"processor_type": "unknown_processor_xyz"}) + + assert result == 0 + assert _logged(caplog, "unknown_processor_xyz") @pytest.mark.asyncio @@ -42,11 +49,14 @@ async def test_process_with_ai_enriches_records(): mock_processor.process = AsyncMock(return_value=mock_result) with patch("backend.pipeline.ai_processor.get_processor", return_value=mock_processor): - await process_with_ai(records, {"processor_type": "claude", "prompt_template": "Summarize: {{content}}"}) + result = await process_with_ai( + records, {"processor_type": "claude", "prompt_template": "Summarize: {{content}}"} + ) assert records[0].ai_enrichment == {"summary": "Summary 1"} assert records[1].ai_enrichment == {"summary": "Summary 2"} assert records[0].status == "ai_processed" + assert result == 2 # ─── GOAL-6 PR-F (decision #9): DataSource.ai_config <-> ModelProvider ───── diff --git a/tests/unit/pipeline/test_notifier_dispatch.py b/tests/unit/pipeline/test_notifier_dispatch.py index ef29a7e..efad1ea 100644 --- a/tests/unit/pipeline/test_notifier_dispatch.py +++ b/tests/unit/pipeline/test_notifier_dispatch.py @@ -1,11 +1,24 @@ """Unit tests for notifier_dispatch.""" import pytest +from sqlalchemy import select from unittest.mock import AsyncMock, MagicMock, patch from backend.pipeline.notifier_dispatch import dispatch_notifications +def _session_cm(session): + """Wrap an already-open (test-fixture) AsyncSession in the async context + manager shape ``backend.database.AsyncSessionLocal()`` normally returns, + so phase C (a fresh internally-opened session) transparently reuses the + real ``db_session`` fixture instead of hitting the module-level + production engine. Same pattern as ``tests/unit/pipeline/test_ai_processor.py``.""" + cm = AsyncMock() + cm.__aenter__ = AsyncMock(return_value=session) + cm.__aexit__ = AsyncMock(return_value=False) + return cm + + @pytest.mark.asyncio async def test_dispatch_empty_records(db_session): await dispatch_notifications(db_session, "src-1", [], "on_new_record") @@ -54,7 +67,17 @@ async def test_dispatch_with_matching_rule(db_session): mock_notifier = AsyncMock() mock_notifier.send = AsyncMock(return_value=True) - with patch("backend.pipeline.notifier_dispatch.get_notifier", return_value=mock_notifier): - await dispatch_notifications(db_session, source.id, [record], "on_new_record") + with ( + patch("backend.pipeline.notifier_dispatch.get_notifier", return_value=mock_notifier), + patch("backend.database.AsyncSessionLocal", return_value=_session_cm(db_session)), + ): + outcome = await dispatch_notifications(db_session, source.id, [record], "on_new_record") mock_notifier.send.assert_awaited_once() + assert outcome == {"sent": 1, "failed": 0} + + from backend.models.notification import NotificationLog + + logs = (await db_session.execute(select(NotificationLog))).scalars().all() + assert len(logs) == 1 + assert logs[0].status == "sent" diff --git a/tests/unit/pipeline/test_notifier_dispatch_errors.py b/tests/unit/pipeline/test_notifier_dispatch_errors.py index 758ee13..fe5cabe 100644 --- a/tests/unit/pipeline/test_notifier_dispatch_errors.py +++ b/tests/unit/pipeline/test_notifier_dispatch_errors.py @@ -1,13 +1,27 @@ """Tests for error paths in notifier_dispatch.""" +import logging + import pytest from unittest.mock import AsyncMock, patch from backend.pipeline.notifier_dispatch import dispatch_notifications +def _session_cm(session): + """Wrap an already-open (test-fixture) AsyncSession in the async context + manager shape ``backend.database.AsyncSessionLocal()`` normally returns, + so phase C (a fresh internally-opened session) transparently reuses the + real ``db_session`` fixture instead of hitting the module-level + production engine. Same pattern as ``tests/unit/pipeline/test_ai_processor.py``.""" + cm = AsyncMock() + cm.__aenter__ = AsyncMock(return_value=session) + cm.__aexit__ = AsyncMock(return_value=False) + return cm + + @pytest.mark.asyncio -async def test_dispatch_unknown_notifier_type_skipped(db_session): +async def test_dispatch_unknown_notifier_type_skipped(db_session, caplog): from backend.models.notification import NotificationRule from backend.models.source import DataSource @@ -34,8 +48,14 @@ async def test_dispatch_unknown_notifier_type_skipped(db_session): record.normalized_data = {"title": "Test"} record.ai_enrichment = None - # Should silently skip unknown notifier (ValueError from get_notifier) - await dispatch_notifications(db_session, source.id, [record], "on_new_record") + # Should silently skip unknown notifier (ValueError from get_notifier), + # but (AUDIT C18) it must no longer be silent in the logs. + with caplog.at_level(logging.WARNING): + outcome = await dispatch_notifications(db_session, source.id, [record], "on_new_record") + + assert outcome == {"sent": 0, "failed": 0} + warnings = [r.getMessage() for r in caplog.records if r.levelno >= logging.WARNING] + assert any(rule.id in msg and "nonexistent_notifier" in msg for msg in warnings) @pytest.mark.asyncio @@ -69,6 +89,129 @@ async def test_dispatch_notifier_send_exception_logged(db_session): mock_notifier = AsyncMock() mock_notifier.send = AsyncMock(side_effect=Exception("connection refused")) - with patch("backend.pipeline.notifier_dispatch.get_notifier", return_value=mock_notifier): + with ( + patch("backend.pipeline.notifier_dispatch.get_notifier", return_value=mock_notifier), + patch("backend.database.AsyncSessionLocal", return_value=_session_cm(db_session)), + ): # Should catch exception and log as failed, not raise + outcome = await dispatch_notifications(db_session, source.id, [record], "on_new_record") + + assert outcome == {"sent": 0, "failed": 1} + + +@pytest.mark.asyncio +async def test_dispatch_aggregate_counts_partial_failure(db_session): + """One rule, two records: one send succeeds, one fails — the aggregate + must reflect both, not just report unconditional success (AUDIT C12).""" + from backend.models.notification import NotificationRule + from backend.models.source import DataSource + + source = DataSource( + name="Partial Fail Src", + channel_type="rss", + channel_config={"feed_url": "https://ex.com/feed"}, + ) + db_session.add(source) + await db_session.flush() + + rule = NotificationRule( + name="Partial Rule", + trigger_event="on_new_record", + notifier_type="webhook", + notifier_config={"url": "https://hooks.ex.com"}, + enabled=True, + ) + db_session.add(rule) + await db_session.flush() + + record_ok = AsyncMock() + record_ok.id = "rec-ok" + record_ok.normalized_data = {} + record_ok.ai_enrichment = None + + record_bad = AsyncMock() + record_bad.id = "rec-bad" + record_bad.normalized_data = {} + record_bad.ai_enrichment = None + + mock_notifier = AsyncMock() + + async def _send(config, payload): + if payload.record_id == "rec-bad": + raise Exception("timeout") + return True + + mock_notifier.send = AsyncMock(side_effect=_send) + + with ( + patch("backend.pipeline.notifier_dispatch.get_notifier", return_value=mock_notifier), + patch("backend.database.AsyncSessionLocal", return_value=_session_cm(db_session)), + ): + outcome = await dispatch_notifications( + db_session, source.id, [record_ok, record_bad], "on_new_record" + ) + + assert outcome == {"sent": 1, "failed": 1} + + +@pytest.mark.asyncio +async def test_dispatch_pending_rows_committed_before_sends_start(db_session): + """Structural regression test for AUDIT C1/C23: the NotificationLog rows + must already be committed as "pending" (phase A) BEFORE any network send + (phase B) begins — i.e. no write transaction/lock is held across the + send. Asserted via call order: session.commit() must fire before + notifier.send().""" + from backend.models.notification import NotificationRule + from backend.models.source import DataSource + + source = DataSource( + name="Order Src", + channel_type="rss", + channel_config={"feed_url": "https://ex.com/feed"}, + ) + db_session.add(source) + await db_session.flush() + + rule = NotificationRule( + name="Order Rule", + trigger_event="on_new_record", + notifier_type="webhook", + notifier_config={"url": "https://hooks.ex.com"}, + enabled=True, + ) + db_session.add(rule) + await db_session.flush() + + record = AsyncMock() + record.id = "rec-order" + record.normalized_data = {} + record.ai_enrichment = None + + call_order: list[str] = [] + original_commit = db_session.commit + + async def _tracked_commit(*args, **kwargs): + call_order.append("commit") + return await original_commit(*args, **kwargs) + + mock_notifier = AsyncMock() + + async def _send(config, payload): + call_order.append("send") + return True + + mock_notifier.send = AsyncMock(side_effect=_send) + + with ( + patch("backend.pipeline.notifier_dispatch.get_notifier", return_value=mock_notifier), + patch("backend.database.AsyncSessionLocal", return_value=_session_cm(db_session)), + patch.object(db_session, "commit", side_effect=_tracked_commit), + ): await dispatch_notifications(db_session, source.id, [record], "on_new_record") + + # Phase A's commit (the caller-session pending-rows commit) must precede + # the send — proving the row was durably "pending" with no open write + # transaction before the network call started. + assert "commit" in call_order + assert "send" in call_order + assert call_order.index("commit") < call_order.index("send") diff --git a/tests/unit/pipeline/test_pipeline.py b/tests/unit/pipeline/test_pipeline.py index fa6a4b9..69695be 100644 --- a/tests/unit/pipeline/test_pipeline.py +++ b/tests/unit/pipeline/test_pipeline.py @@ -132,7 +132,7 @@ async def test_run_pipeline_with_ai(db_session): with ( patch("backend.pipeline.collector.collect", return_value=channel_result), patch("backend.pipeline.storer.store_records", new=AsyncMock(return_value=([mock_record], 0))), - patch("backend.pipeline.ai_processor.process_with_ai", new=AsyncMock()), + patch("backend.pipeline.ai_processor.process_with_ai", new=AsyncMock(return_value=1)), patch("backend.database.AsyncSessionLocal", return_value=inner_cm), ): result = await run_pipeline( @@ -147,6 +147,67 @@ async def test_run_pipeline_with_ai(db_session): assert result.ai_processed == 1 +@pytest.mark.asyncio +async def test_run_pipeline_ai_zero_enriched_warns_not_done(db_session): + """AUDIT C3: process_with_ai returning 0 (unknown/misconfigured + processor_type, no exception raised) must surface as a failed/skipped + warning event, never the "AI 处理完成" success message — that used to be + emitted unconditionally regardless of what process_with_ai actually did.""" + source, task = await _setup_source_task(db_session) + + mock_items = [{"title": "AI Item", "url": "https://ex.com/ai"}] + channel_result = ChannelResult.ok(mock_items) + mock_record = MagicMock() + mock_record.ai_enrichment = None + + agent_config = {"processor_type": "bogus_processor", "model": "n/a"} + + mock_inner_session = AsyncMock() + mock_inner_session.get = AsyncMock(return_value=MagicMock()) + mock_inner_session.commit = AsyncMock() + # .add() is sync on a real Session (only .commit/.flush/.get etc. are + # awaited) — pin it to a plain MagicMock so record_run_measurement's + # (triggered below by run_id being set) internal session.add(row) doesn't + # return an unawaited coroutine. + mock_inner_session.add = MagicMock() + inner_cm = AsyncMock() + inner_cm.__aenter__ = AsyncMock(return_value=mock_inner_session) + inner_cm.__aexit__ = AsyncMock(return_value=False) + + emitted: list[dict] = [] + + async def fake_emit(run_id, step, message, level="info", detail=None, elapsed_ms=None): + emitted.append({"step": step, "level": level, "message": message, "detail": detail}) + + with ( + patch("backend.pipeline.collector.collect", return_value=channel_result), + patch( + "backend.pipeline.storer.store_records", + new=AsyncMock(return_value=([mock_record], 0)), + ), + patch("backend.pipeline.ai_processor.process_with_ai", new=AsyncMock(return_value=0)), + patch("backend.database.AsyncSessionLocal", return_value=inner_cm), + patch("backend.pipeline.events.emit", new=fake_emit), + ): + result = await run_pipeline( + task.id, + source, + agent_config=agent_config, + enable_ai=True, + enable_notifications=False, + run_id="run-ai-zero", + ) + + assert result.success is True + assert result.ai_processed == 0 + + ai_events = [e for e in emitted if e["step"] == "ai_process"] + assert len(ai_events) == 1 + assert ai_events[0]["level"] == "warning" + assert "完成" not in ai_events[0]["message"] + assert "失败" in ai_events[0]["message"] or "跳过" in ai_events[0]["message"] + + @pytest.mark.asyncio async def test_run_pipeline_with_notifications(db_session): """Pipeline with new records dispatches notifications.""" @@ -159,7 +220,42 @@ async def test_run_pipeline_with_notifications(db_session): with ( patch("backend.pipeline.collector.collect", return_value=channel_result), patch("backend.pipeline.storer.store_records", new=AsyncMock(return_value=([mock_record], 0))), - patch("backend.pipeline.notifier_dispatch.dispatch_notifications", new=AsyncMock()), + patch( + "backend.pipeline.notifier_dispatch.dispatch_notifications", + new=AsyncMock(return_value={"sent": 1, "failed": 0}), + ), + ): + result = await run_pipeline( + task.id, + source, + enable_ai=False, + enable_notifications=True, + ) + + assert result.success is True + assert result.notifications_sent == 1 + + +@pytest.mark.asyncio +async def test_run_pipeline_notifications_partial_failure_reported(db_session): + """AUDIT C12: notifications_sent/failed must reflect the real aggregate, + including the partial-failure case, not an unconditional success.""" + source, task = await _setup_source_task(db_session) + + mock_items = [{"title": "Notify Item", "url": "https://ex.com/n"}] + channel_result = ChannelResult.ok(mock_items) + mock_record = MagicMock() + + with ( + patch("backend.pipeline.collector.collect", return_value=channel_result), + patch( + "backend.pipeline.storer.store_records", + new=AsyncMock(return_value=([mock_record], 0)), + ), + patch( + "backend.pipeline.notifier_dispatch.dispatch_notifications", + new=AsyncMock(return_value={"sent": 1, "failed": 1}), + ), ): result = await run_pipeline( task.id, @@ -169,6 +265,53 @@ async def test_run_pipeline_with_notifications(db_session): ) assert result.success is True + assert result.notifications_sent == 1 + + +@pytest.mark.asyncio +async def test_run_pipeline_notifications_all_failed_warns(db_session): + """AUDIT C12: when every attempted send failed (sent=0, failed>0), the + emitted event and PipelineResult must say so at warning level, not the + unconditional "通知发送完成" info-level message this replaces.""" + source, task = await _setup_source_task(db_session) + + mock_items = [{"title": "Notify Item", "url": "https://ex.com/n"}] + channel_result = ChannelResult.ok(mock_items) + mock_record = MagicMock() + + emitted: list[dict] = [] + + async def fake_emit(run_id, step, message, level="info", detail=None, elapsed_ms=None): + emitted.append({"step": step, "level": level, "message": message, "detail": detail}) + + with ( + patch("backend.pipeline.collector.collect", return_value=channel_result), + patch( + "backend.pipeline.storer.store_records", + new=AsyncMock(return_value=([mock_record], 0)), + ), + patch( + "backend.pipeline.notifier_dispatch.dispatch_notifications", + new=AsyncMock(return_value={"sent": 0, "failed": 2}), + ), + patch("backend.pipeline.events.emit", new=fake_emit), + ): + result = await run_pipeline( + task.id, + source, + enable_ai=False, + enable_notifications=True, + run_id="run-notify-all-failed", + ) + + assert result.success is True + assert result.notifications_sent == 0 + + notify_events = [e for e in emitted if e["step"] == "notify"] + assert len(notify_events) == 1 + assert notify_events[0]["level"] == "warning" + assert "成功 0" in notify_events[0]["message"] + assert "失败 2" in notify_events[0]["message"] @pytest.mark.asyncio diff --git a/tests/unit/test_runner.py b/tests/unit/test_runner.py index e3c4c11..5259a8b 100644 --- a/tests/unit/test_runner.py +++ b/tests/unit/test_runner.py @@ -590,3 +590,9 @@ def capture_add(obj): assert "upstream reset" in task.error_message assert run.status == "failed" assert "upstream reset" in run.error_message + + # AUDIT C19: the persisted message is prefixed with the taxonomy + # classification (same one backend.pipeline.error_taxonomy computes for + # the pipeline layer's own retry decision), not a bare str(exc). + assert task.error_message == "[ConnectionError] upstream reset" + assert run.error_message == "[ConnectionError] upstream reset"