fix(pipeline): notify without write lock + real success counts [修复组①]

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,14 +125,23 @@ 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,
         prompt_template=resolved_config.get("prompt_template", ""),
         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

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}

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,
     )

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