fix(OMN-19520): the stall alert fires on handler errors that never form a run

docs/evidence/OMN-15423-relation-inventory.json

@@ -6365,6 +6365,7 @@
       "partitioning": [],
       "producer": "node_projection_work_events",
       "readers": [
+        "node_consumer_flow_stall_alert_effect",
         "service:omnimarket_projection_migration_runner"
       ],
       "target_schema": "omninode_internal",

src/omnimarket/nodes/node_consumer_flow_stall_alert_effect/contract.yaml

@@ -151,6 +151,17 @@ alert_policy:
   # the consumer is not taking messages at all, which needs no second witness.
   require_failure_evidence_for:
     - STALLED
+  # OMN-19520. HANDLER ERRORS fire on their own, not only inside a run.
+  # A window whose handler errors counter is > 0 took messages, committed their
+  # offsets and wrote nothing. The run rule above damps flow-rate symptoms and
+  # is right to; applied to loss it hid a sparse topic completely. Measured on
+  # the .201 dev lane 2026-09-17..24: the work_events session-started group
+  # failed 261 windows, one event at a time between IDLE windows, and the run
+  # rule posted 18 of them, titled STALLED. Counted across the whole
+  # history_windows read, so it must stay <= history_windows. The renotify
+  # bucket (keyed on the newest failing window) keeps it to one post per
+  # consumer, topic and bucket through node_slack_publish_effect's ledger.
+  handler_error_windows: 1
   # UNKNOWN is NOT in alerting_states and must never be: a missed heartbeat is
   # the absence of an observation, not evidence of a stall (OMN-16777 AC5).
   # It surfaces as its own degraded outcome instead.

src/omnimarket/nodes/node_consumer_flow_stall_alert_effect/handlers/build_slack_command.py

@@ -39,6 +39,22 @@ def render_alert_text(payload: ModelStallAlertPayload) -> str:
     consumer, which topic, what went in, what came out, what was dead-lettered,
     and how long it has been that way.
     """
+    if payload.handler_error_windows is not None:
+        # OMN-19520: a handler error is lost data, so the headline says so
+        # instead of borrowing the flow state of the failing window.
+        return (
+            f":x: *HANDLER ERRORS* — `{payload.consumer_group}` on "
+            f"`{payload.topic}`: {payload.handler_error_windows} of the last "
+            f"{_counter(payload.observed_windows)} windows failed messages "
+            "in the handler\n"
+            f"handler_errors={_counter(payload.handler_errors)} "
+            f"dlq={_counter(payload.messages_dlq)} "
+            f"in={_counter(payload.messages_in)} "
+            f"out={_counter(payload.messages_out)}\n"
+            f"failing windows {payload.window_start.isoformat()} → "
+            f"{payload.window_end.isoformat()}\n"
+            f"correlation_id={payload.correlation_id}"
+        )
     return (
         f":rotating_light: *{payload.flow_state.value}* — "
         f"`{payload.consumer_group}` on `{payload.topic}` for "

src/omnimarket/nodes/node_consumer_flow_stall_alert_effect/handlers/decide_stall_alert.py

@@ -154,6 +154,89 @@ def _last_alerting_run_was_confirmed(
     return run >= policy.confirm_windows
 
 
+def _sum_observed(values: list[int | None]) -> int | None:
+    """Total of the observed counters, or ``None`` when none was observed."""
+    observed = [value for value in values if value is not None]
+    return sum(observed) if observed else None
+
+
+def _handler_error_decision(
+    request: ModelConsumerFlowStallAlertRequest,
+    alerting_run: int,
+    unknown_run: int,
+) -> ModelConsumerFlowStallAlertDecision | None:
+    """OMN-19520: fire on handler errors anywhere in the read, run or no run.
+
+    The run rule exists to damp flap on flow-rate symptoms. A handler error is
+    not a symptom: the message was taken, its offset committed, and nothing was
+    written. On a sparse topic -- one session-started event every few minutes
+    against 30 s windows -- the failing windows are separated by IDLE ones, so
+    no run ever forms and the run rule graded every one of them
+    PENDING_CONFIRMATION or NO_ALERT. Measured on the .201 dev lane from
+    2026-09-17 to 2026-09-24: 261 failing windows on the work_events
+    session-started group, 18 posts, each titled STALLED.
+
+    A failing run that already reached ``confirm_windows`` posted as a confirmed
+    stall; it is not announced a second time here, and its recovery keeps its
+    own RECOVERING grading.
+
+    The idempotency key is bucketed on the NEWEST failing window, so each of
+    the ``history_windows`` evaluations that still see one failure hands the
+    publish node the same key, and its ledger collapses them into one post.
+    """
+    policy = request.policy
+    failing = tuple(
+        window
+        for window in request.windows
+        if policy.has_handler_error(window.handler_errors)
+    )
+    if len(failing) < policy.handler_error_windows:
+        return None
+    if _last_alerting_run_was_confirmed(request.windows, policy):
+        return None
+
+    newest_failing = failing[-1]
+    payload = ModelStallAlertPayload(
+        consumer_group=request.consumer_group,
+        topic=request.topic,
+        flow_state=newest_failing.flow_state,
+        consecutive_windows=alerting_run,
+        messages_in=_sum_observed([window.messages_in for window in failing]),
+        messages_out=_sum_observed([window.messages_out for window in failing]),
+        messages_dlq=_sum_observed([window.messages_dlq for window in failing]),
+        handler_errors=_sum_observed([window.handler_errors for window in failing]),
+        window_start=failing[0].window_start,
+        window_end=newest_failing.window_end,
+        node_id=request.node_id,
+        correlation_id=request.correlation_id,
+        handler_error_windows=len(failing),
+        observed_windows=len(request.windows),
+    )
+    bucket = _renotify_bucket(
+        int(newest_failing.window_start.timestamp()),
+        policy.renotify_after_seconds,
+    )
+    return ModelConsumerFlowStallAlertDecision(
+        consumer_group=request.consumer_group,
+        topic=request.topic,
+        outcome=EnumStallAlertOutcome.FAIL_HANDLER_ERRORS,
+        severity=EnumStallAlertSeverity.FAIL,
+        consecutive_alerting_windows=alerting_run,
+        consecutive_unknown_windows=unknown_run,
+        should_publish=True,
+        reason=(
+            f"{request.consumer_group} failed messages in its handler on "
+            f"{request.topic} in {len(failing)} of the last "
+            f"{len(request.windows)} windows (threshold "
+            f"{policy.handler_error_windows})"
+        ),
+        idempotency_key=(
+            f"{bucket}|{request.consumer_group}|{request.topic}|HANDLER_ERRORS"
+        ),
+        alert=payload,
+    )
+
+
 def _renotify_bucket(window_start_epoch: int, renotify_after_seconds: int) -> int:
     """Bucket index used to collapse repeats of a standing condition.
 
@@ -227,6 +310,10 @@ def decide_stall_alert(
             alert=payload,
         )
 
+    handler_errors = _handler_error_decision(request, alerting_run, unknown_run)
+    if handler_errors is not None:
+        return handler_errors
+
     if alerting_run > 0:
         return ModelConsumerFlowStallAlertDecision(
             consumer_group=request.consumer_group,

src/omnimarket/nodes/node_consumer_flow_stall_alert_effect/models/enum_stall_alert_outcome.py

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: MIT
 """What one evaluation of a consumer's window history concluded (OMN-16778).
 
-Five outcomes, because the four the platform could previously express were not
+Six outcomes (five before OMN-19520), because the four the platform could previously express were not
 enough to keep an alert channel readable.  ``FAIL`` and ``WARN`` are different
 facts and the ticket forbids conflating them: a channel that receives both at
 the same volume gets muted, and a muted channel is worth exactly as much as no
@@ -21,6 +21,15 @@ class EnumStallAlertOutcome(StrEnum):
     #: is the only outcome that publishes a Slack alert by default.
     FAIL_CONFIRMED_STALL = "FAIL_CONFIRMED_STALL"
 
+    #: OMN-19520. At least ``handler_error_windows`` windows of the trailing
+    #: history carry a handler error, and no confirmed stall already covers
+    #: them. Fires on its own, without a consecutive run: a handler error is
+    #: lost data, not a flow-rate blip, and on a sparse topic the failures
+    #: never touch each other (the work_events session-started projection
+    #: failed 261 windows in a week on the .201 dev lane and the run rule
+    #: posted 18 of them).
+    FAIL_HANDLER_ERRORS = "FAIL_HANDLER_ERRORS"
+
     #: An alerting run is under way but has not yet reached confirm_windows.
     #: Deliberately silent: firing here is the flap that made the .201 host
     #: reporter's own alerts unreadable (OMN-16789).

src/omnimarket/nodes/node_consumer_flow_stall_alert_effect/models/model_stall_alert_decision.py

@@ -33,13 +33,36 @@ class ModelStallAlertPayload(BaseModel):
     consumer_group: str = Field(..., min_length=1)
     topic: str = Field(..., min_length=1)
     flow_state: EnumConsumerFlowState
-    consecutive_windows: int = Field(..., ge=1)
+    consecutive_windows: int = Field(
+        ...,
+        ge=0,
+        description=(
+            "Length of the trailing alerting run. At least 1 for "
+            "FAIL_CONFIRMED_STALL; may be 0 for FAIL_HANDLER_ERRORS, whose "
+            "failing windows need not be consecutive or newest (OMN-19520)."
+        ),
+    )
     messages_in: int | None = None
     messages_out: int | None = None
     messages_dlq: int | None = None
     handler_errors: int | None = None
     window_start: datetime
     window_end: datetime
+    handler_error_windows: int | None = Field(
+        default=None,
+        ge=1,
+        description=(
+            "OMN-19520. Set exactly on a FAIL_HANDLER_ERRORS payload: how many "
+            "windows of the read history carried a handler error. On that "
+            "payload the counters are totals over those windows and "
+            "window_start/window_end span them, oldest to newest."
+        ),
+    )
+    observed_windows: int | None = Field(
+        default=None,
+        ge=1,
+        description="OMN-19520. How many windows the history read held.",
+    )
     node_id: UUID | None = None
     correlation_id: UUID
 

src/omnimarket/nodes/node_consumer_flow_stall_alert_effect/models/model_stall_alert_policy.py

@@ -77,6 +77,16 @@ class ModelStallAlertPolicy(BaseModel):
             "Consecutive UNKNOWN windows before the missed-heartbeat WARN is surfaced."
         ),
     )
+    handler_error_windows: int = Field(
+        ...,
+        ge=1,
+        description=(
+            "OMN-19520. Windows in the trailing history that must carry a "
+            "handler error before FAIL_HANDLER_ERRORS fires. Counted across "
+            "the whole read, not as a consecutive run: on a sparse topic the "
+            "failing windows are separated by IDLE ones and a run never forms."
+        ),
+    )
     deliver_warnings: bool = Field(
         ...,
         description=(
@@ -90,6 +100,14 @@ def is_alerting(self, state: EnumConsumerFlowState) -> bool:
         """Whether ``state`` counts as a stall under this policy."""
         return state in self.alerting_states
 
+    @staticmethod
+    def has_handler_error(handler_errors: int | None) -> bool:
+        """Whether a window's handler-error counter records a failure.
+
+        ``None`` is an unobserved counter, not a failure (OMN-16777 AC5).
+        """
+        return bool(handler_errors)
+
     def needs_failure_evidence(self, state: EnumConsumerFlowState) -> bool:
         """Whether ``state`` may only alert alongside a DLQ or handler error.