fix: wire schema drift errors and constrain notification triggers

backend/channels/cli_channel.py

@@ -5,8 +5,6 @@
 import logging
 import os
 import re
-import shlex
-import shutil
 from typing import Any
 
 from backend.channels.base import AbstractChannel, ChannelResult
@@ -80,7 +78,7 @@ async def collect(
                 **_process_group_kwargs(),
             )
             stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
-        except asyncio.TimeoutError as exc:
+        except TimeoutError as exc:
             # Don't orphan the child: wait_for only cancels communicate();
             # the subprocess itself keeps running until explicitly killed. A
             # bare proc.kill() only reaches the direct child — shell-wrapped
@@ -116,7 +114,12 @@ async def collect(
                 data = json.loads(output)
                 items = data if isinstance(data, list) else [data]
             except json.JSONDecodeError as exc:
-                return ChannelResult.fail(f"Failed to parse CLI JSON output: {exc}")
+                # WIRING_GAP_LEDGER W1: error_type must be set so the SCHEMA_DRIFT
+                # chain (error_kinds -> control.recorder) actually fires instead of
+                # being dropped by recorder's `elif error_type is not None` guard.
+                return ChannelResult.fail(
+                    f"Failed to parse CLI JSON output: {exc}", error_type=type(exc).__name__
+                )
         else:
             # Plain text: each line is a record
             items = [{"line": line} for line in output.splitlines() if line.strip()]

backend/channels/rss_channel.py

@@ -22,6 +22,26 @@
 )
 
 
+def _bozo_error_type(parsed: Any) -> str:
+    """error_type for a bozo (unparseable) feed — WIRING_GAP_LEDGER W1.
+
+    feedparser sets ``bozo_exception`` to the underlying parse failure (e.g.
+    ``xml.sax._exceptions.SAXParseException`` for malformed markup, truncated
+    declarations, or an encoding mismatch — verified empirically across all
+    of those shapes; see backend/control/error_kinds.py's SCHEMA_DRIFT set).
+    Its class name is a real structured ``error_type`` like every other
+    channel already produces, so it flows through ChannelResult.fail() /
+    ChannelFetchError into control.recorder instead of being dropped by the
+    ``elif error_type is not None`` guard there. Falls back to a generic
+    "ParseError" (also mapped to SCHEMA_DRIFT) on the defensive case where
+    feedparser didn't attach one.
+    """
+    bozo_exception = getattr(parsed, "bozo_exception", None)
+    if bozo_exception is not None:
+        return type(bozo_exception).__name__
+    return "ParseError"
+
+
 @register_channel
 class RSSChannel(AbstractChannel):
     """Collect entries from RSS/Atom feeds."""
@@ -84,7 +104,8 @@ async def collect(
         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')}"
+                f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}",
+                error_type=_bozo_error_type(parsed),
             )
 
         entries = parsed.entries[:max_entries]
@@ -179,7 +200,8 @@ async def fetch(self, ctx: FetchContext) -> FetchResult:
         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')}"
+                f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}",
+                error_type=_bozo_error_type(parsed),
             )
         items = [self._entry_to_dict(entry) for entry in parsed.entries[:max_entries]]
 

backend/control/error_kinds.py

@@ -72,6 +72,11 @@ class ErrorKind(str, Enum):
     "JSONDecodeError": ErrorKind.SCHEMA_DRIFT,
     "SchemaDriftError": ErrorKind.SCHEMA_DRIFT,
     "ParseError": ErrorKind.SCHEMA_DRIFT,
+    # feedparser's bozo_exception class for malformed/truncated XML or an
+    # encoding mismatch (rss_channel.py's bozo branch) — verified empirically
+    # across markup corruption, truncated declarations, and encoding
+    # mismatches; feedparser's strict parser wraps all of them the same way.
+    "SAXParseException": ErrorKind.SCHEMA_DRIFT,
     # Validation / permanent bad input (including SSRF rejections — a
     # config/validation problem with the source, not a transient network fault)
     "ValueError": ErrorKind.VALIDATION,

backend/schemas/notification.py

@@ -9,7 +9,14 @@
 class NotificationRuleCreate(BaseModel):
     name: str = Field(..., min_length=1, max_length=255)
     source_id: str | None = None
-    trigger_event: str
+    # WIRING_GAP_LEDGER W2: dispatch_notifications() only ever queries/fires
+    # rules with trigger_event == "on_new_record" (backend/pipeline/
+    # notifier_dispatch.py) — there is no producer for any other value.
+    # Before this Literal, a rule saved with any other string (the free-text
+    # str field previously accepted anything) became permanently, silently
+    # inert. Constrained here at the schema layer so the API rejects an
+    # unsupported value loudly instead of persisting a dead rule.
+    trigger_event: Literal["on_new_record"]
     notifier_type: str
     notifier_config: dict[str, Any] = Field(default_factory=dict)
     filter_conditions: dict[str, Any] | None = None
@@ -18,7 +25,7 @@ class NotificationRuleCreate(BaseModel):
 
 class NotificationRuleUpdate(BaseModel):
     name: str | None = None
-    trigger_event: str | None = None
+    trigger_event: Literal["on_new_record"] | None = None
     notifier_type: str | None = None
     notifier_config: dict[str, Any] | None = None
     filter_conditions: dict[str, Any] | None = None