fix(gateway): recognize Weixin's rate-limit wording in the delivery ledger's flood classifier

gateway/delivery_ledger.py

@@ -74,10 +74,18 @@
 # merely suggests retrying is never mistaken for a flood.
 _RAW_FLOOD_RE = re.compile(r"flood control exceeded.*?retry in\s+(\d+(?:\.\d+)?)", re.IGNORECASE)
 
+# Some adapters describe their own throttling without either the ``flood_control:`` prefix or PTB's
+# wording — e.g. Weixin's circuit breaker fails a send closed with "...cooldown active for 12.3s"
+# (``gateway/platforms/weixin.py``). The cooldown length is still embedded in the text, so it is worth
+# extracting precisely rather than falling back to the generic default every time.
+_COOLDOWN_WAIT_RE = re.compile(r"cooldown active for\s+(\d+(?:\.\d+)?)s", re.IGNORECASE)
+
 
 def _raw_flood_wait(text: str) -> Optional[float]:
-    """Seconds asked for by a flood error still carrying the platform's own wording, else ``None``."""
-    match = _RAW_FLOOD_RE.search(text or "")
+    """Seconds asked for by a flood error still carrying the platform's own wording (PTB's "flood
+    control exceeded... retry in N seconds", or an adapter's own "cooldown active for Ns"), else
+    ``None``."""
+    match = _RAW_FLOOD_RE.search(text or "") or _COOLDOWN_WAIT_RE.search(text or "")
     if not match:
         return None
     try:
@@ -86,11 +94,34 @@ def _raw_flood_wait(text: str) -> Optional[float]:
         return None
 
 
+def _classified_rate_limited(text: str) -> bool:
+    """Whether ``text`` matches the platform-neutral "rate_limited" send-error classification
+    (``gateway.platforms.base.classify_send_error``), so any adapter's rate-limit wording is
+    recognised here too and not just Telegram's. Imported locally (this module is loaded well before
+    ``gateway.platforms.base`` in some paths, e.g. CLI commands that only touch the ledger) and
+    best-effort like everything else in this module."""
+    if not text:
+        return False
+    try:
+        from gateway.platforms.base import classify_send_error
+    except Exception:
+        return False
+    try:
+        return classify_send_error(None, text) == "rate_limited"
+    except Exception:
+        return False
+
+
 def is_flood_error(error: Any) -> bool:
-    """True for a flood refusal: the adapters' fail-closed ``flood_control:<seconds>`` result, or a
-    row still carrying the platform's own flood wording (see ``_RAW_FLOOD_RE``)."""
+    """True for a flood refusal: the adapters' fail-closed ``flood_control:<seconds>`` result, a row
+    still carrying the platform's own flood/cooldown wording (see ``_RAW_FLOOD_RE``), or any other
+    error text the platform-neutral classifier recognises as a rate limit (see
+    ``_classified_rate_limited``) — so an adapter that surfaces a rate limit without either of the
+    two specific shapes above (e.g. Weixin's bare ``RuntimeError``) still gets a timed redelivery
+    instead of being treated as an ordinary failure and retried immediately inside its own cooldown."""
     text = str(error or "").strip().lower()
-    return text.startswith(FLOOD_ERROR_PREFIX) or _raw_flood_wait(text) is not None
+    return (text.startswith(FLOOD_ERROR_PREFIX) or _raw_flood_wait(text) is not None
+            or _classified_rate_limited(text))
 
 
 def flood_wait_seconds(error: Any, default: float = FLOOD_RETRY_DEFAULT_SECONDS) -> float:

tests/gateway/test_delivery_flood_invariants.py

@@ -5,11 +5,17 @@
 
 from gateway import delivery_ledger as dl
 
+# Weixin's circuit breaker fails a send closed with this shape (gateway/platforms/weixin.py) — a bare
+# error string with neither the ``flood_control:<seconds>`` prefix adapters normally use nor PTB's raw
+# "Flood control exceeded... retry in N seconds" wording, but the same platform-neutral "rate_limited"
+# classification (gateway.platforms.base.classify_send_error) that gates the in-attempt send backoff.
+WEIXIN_RATE_LIMIT_ERROR = 'iLink sendmessage rate limited; cooldown active for 12.3s'
 
-def record(oid, *, profile=None):
-    dl.record_obligation(obligation_id=oid, session_key='session-' + oid, platform='telegram',
+
+def record(oid, *, profile=None, platform='telegram', error='flood_control:185'):
+    dl.record_obligation(obligation_id=oid, session_key='session-' + oid, platform=platform,
                          chat_id='123', thread_id='77', content='.' * 3000, adapter_profile=profile)
-    dl.mark_failed(oid, 'flood_control:185')
+    dl.mark_failed(oid, error)
 
 
 def read(oid):
@@ -59,3 +65,66 @@ def test_boot_adopts_waiting_rows_without_spending_or_losing_deadline():
     claimed = dl.sweep_failed_for_runtime('telegram', now=original['updated_at'] + 186)
     assert len(claimed) == 1 and claimed[0]['needs_marker']
     assert read('boot')['state'] == 'attempting' and read('boot')['last_error'] is None
+
+
+def test_weixin_rate_limit_text_classifies_as_flood_and_extracts_its_own_wait():
+    """``is_flood_error``/``flood_wait_seconds`` must recognise Weixin's own rate-limit wording (no
+    ``flood_control:`` prefix, no PTB text) via the platform-neutral classifier, and read the cooldown
+    seconds Weixin itself embeds rather than falling back to the generic default."""
+    assert dl.is_flood_error(WEIXIN_RATE_LIMIT_ERROR)
+    assert dl.flood_wait_seconds(WEIXIN_RATE_LIMIT_ERROR) == 12.3
+    # An ordinary Weixin failure (not rate-limited) must still read as an ordinary failure.
+    assert not dl.is_flood_error('iLink sendmessage error: ret=1 errcode=2 errmsg=unknown error')
+
+
+def test_weixin_rate_limit_arms_timed_redelivery_not_immediate_retry():
+    """The ledger must treat a Weixin rate-limit failure exactly like a Telegram flood failure: adopted
+    without spending an attempt while still inside the wait, then claimable once it has passed — never
+    claimed as an ordinary failure that spends an attempt inside the still-active cooldown."""
+    record('weixin-due', platform='weixin', error=WEIXIN_RATE_LIMIT_ERROR)
+    stamp = read('weixin-due')['updated_at']
+    # Still inside Weixin's own 12.3s cooldown: the runtime sweep must not claim it yet.
+    assert dl.sweep_failed_for_runtime('weixin', now=stamp + 11) == []
+    assert read('weixin-due')['attempts'] == 0
+    # Past the cooldown: claimable, carries a marker (rate-limit wording), and clears the stale error.
+    claimed = dl.sweep_failed_for_runtime('weixin', now=stamp + 13)
+    assert [r['obligation_id'] for r in claimed] == ['weixin-due']
+    assert claimed[0]['needs_marker'] and 'rate limit' in claimed[0]['marker']
+    assert read('weixin-due')['last_error'] is None
+
+
+def test_weixin_rate_limit_boot_adoption_does_not_spend_attempt():
+    """A dead-owner boot sweep must adopt a still-cooling-down Weixin rate-limit row (no attempt spent,
+    deadline preserved) exactly as it does for Telegram's ``flood_control:`` rows — not treat it as an
+    ordinary failed row and immediately spend a redelivery attempt inside the cooldown."""
+    record('weixin-boot', platform='weixin', error=WEIXIN_RATE_LIMIT_ERROR)
+    original = read('weixin-boot')
+    with sqlite3.connect(dl._db_path()) as conn:
+        conn.execute("UPDATE delivery_obligations SET owner_pid=NULL, owner_started_at=NULL, adapter_profile=NULL")
+    claimed = dl.sweep_recoverable(now=original['updated_at'] + 5,
+                                    deliverable_targets={('weixin', None)})
+    assert len(claimed) == 1 and claimed[0].get('adopted')
+    adopted = read('weixin-boot')
+    assert adopted['owner_pid'] == os.getpid()
+    assert adopted['adapter_profile'] == 'default'
+    assert adopted['attempts'] == 0 and adopted['updated_at'] == original['updated_at']
+    assert adopted['last_error'] == WEIXIN_RATE_LIMIT_ERROR
+    # A second boot-sweep pass finds this process still the live owner: nothing more to adopt.
+    assert dl.sweep_recoverable(now=original['updated_at'] + 6) == []
+    # Still inside the cooldown: the runtime sweep (this process claiming its own adopted row) must wait.
+    assert dl.sweep_failed_for_runtime('weixin', now=original['updated_at'] + 11) == []
+    # Past the deadline: claimed, marker set, error cleared.
+    claimed = dl.sweep_failed_for_runtime('weixin', now=original['updated_at'] + 13)
+    assert len(claimed) == 1 and claimed[0]['needs_marker']
+    assert read('weixin-boot')['state'] == 'attempting' and read('weixin-boot')['last_error'] is None
+
+
+def test_telegram_flood_detection_unaffected_by_broader_classifier():
+    """Widening ``is_flood_error`` to also consult the platform-neutral classifier must not change
+    Telegram's own, already-correct behavior for its two established shapes."""
+    assert dl.is_flood_error('flood_control:185')
+    assert dl.is_flood_error('Flood control exceeded. Retry in 185 seconds')
+    assert dl.flood_wait_seconds('flood_control:185') == 185.0
+    assert dl.flood_wait_seconds('Flood control exceeded. Retry in 185 seconds') == 185.0
+    # A permanent Telegram rejection must never be mistaken for a flood.
+    assert not dl.is_flood_error('Forbidden: bot was blocked by the user')