"""PROD-P5-01: suppress duplicate or stale delegate completion WeChat pushes.""" from __future__ import annotations import logging import re import threading import time from types import TracebackType from datetime import datetime, timezone from butler.core.best_effort import safe_best_effort from butler.defaults.env_defaults import ( GATEWAY_DELEGATE_PUSH_DEDUP_DEFAULT, GATEWAY_DEFER_DELEGATE_PUSH_DURING_INBOUND_DEFAULT, GATEWAY_DELEGATE_PUSH_MAX_AGE_SECONDS, ) from butler.gateway.outbound_bridge import get_gateway_bridge_optional from butler.runtime.notify import push_runtime_message from butler.runtime.task_store import get_task from butler.utilities.env_parse import env_truthy, float_env logger = logging.getLogger(__name__) _TASK_ID_RE = re.compile(r"\btask_[a-f0-9]{8,}\b", re.I) _LOCK = threading.Lock() _PUSHED: dict[str, float] = {} _ACTIVE_INBOUND: dict[str, int] = {} _DEFERRED: dict[str, list[tuple[str, str]]] = {} def delegate_push_dedup_enabled() -> bool: return bool(env_truthy("BUTLER_GATEWAY_DELEGATE_PUSH_DEDUP", default=GATEWAY_DELEGATE_PUSH_DEDUP_DEFAULT)) def defer_delegate_push_during_inbound_enabled() -> bool: return bool(env_truthy("BUTLER_GATEWAY_DEFER_DELEGATE_PUSH_DURING_INBOUND", default=GATEWAY_DEFER_DELEGATE_PUSH_DURING_INBOUND_DEFAULT)) def delegate_push_max_age_seconds() -> float: return float(max(60.0, float_env("BUTLER_GATEWAY_DELEGATE_PUSH_MAX_AGE_SECONDS", GATEWAY_DELEGATE_PUSH_MAX_AGE_SECONDS))) def extract_task_id_from_text(text: str) -> str: match = _TASK_ID_RE.search(str(text or "")) return match.group(0) if match else "" def _dedup_key(chat_id: str, task_id: str) -> str: return f"{str(chat_id or '').strip()}::{str(task_id or '').strip()}" def _task_completed_epoch(task_id: str) -> float | None: def _run() -> float: tid = str(task_id or "").strip() if not tid: raise ValueError("task id missing") row = get_task(tid) if not isinstance(row, dict): raise ValueError("task row missing") if str(row.get("status") or "") not in ("completed", "failed"): raise ValueError("task not terminal") raw = str(row.get("updated_at") or row.get("created_at") or "").strip() if not raw: raise ValueError("task timestamp missing") dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return float(dt.timestamp()) result = safe_best_effort( _run, label="delegate_push_dedup.task_epoch", default=None, ) return float(result) if isinstance(result, (int, float)) else None def should_deliver_delegate_push( chat_id: str, task_id: str, *, body: str = "", ) -> tuple[bool, str]: """Return (ok, reason). reason empty when ok.""" cid = str(chat_id or "").strip() tid = str(task_id or "").strip() or extract_task_id_from_text(body) if not delegate_push_dedup_enabled(): return True, "" if not tid: return True, "" key = _dedup_key(cid, tid) with _LOCK: if key in _PUSHED: return False, "dedup:task_already_pushed" completed_at = _task_completed_epoch(tid) if completed_at is not None: age = time.time() - completed_at if age > delegate_push_max_age_seconds(): return False, f"stale:age_{int(age)}s" return True, "" def mark_delegate_push_delivered(chat_id: str, task_id: str, *, body: str = "") -> None: cid = str(chat_id or "").strip() tid = str(task_id or "").strip() or extract_task_id_from_text(body) if not tid: return key = _dedup_key(cid, tid) with _LOCK: _PUSHED[key] = time.time() if len(_PUSHED) > 500: cutoff = time.time() - 86400 * 2 for k, ts in list(_PUSHED.items()): if ts < cutoff: _PUSHED.pop(k, None) def is_inbound_active(chat_id: str) -> bool: cid = str(chat_id or "").strip() if not cid: return False with _LOCK: return _ACTIVE_INBOUND.get(cid, 0) > 0 def defer_delegate_push(chat_id: str, body: str, *, kind: str = "delegate") -> None: cid = str(chat_id or "").strip() if not cid or not (body or "").strip(): return with _LOCK: _DEFERRED.setdefault(cid, []).append((str(body), str(kind or "delegate"))) def flush_deferred_delegate_pushes(chat_id: str) -> list[tuple[str, str]]: cid = str(chat_id or "").strip() if not cid: return [] with _LOCK: items = _DEFERRED.pop(cid, []) return items class gateway_inbound_guard: """Mark chat inbound active; flush deferred delegate pushes on exit.""" def __init__(self, chat_id: str) -> None: self.chat_id = str(chat_id or "").strip() def __enter__(self) -> gateway_inbound_guard: if not self.chat_id: return self with _LOCK: _ACTIVE_INBOUND[self.chat_id] = _ACTIVE_INBOUND.get(self.chat_id, 0) + 1 return self def __exit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None, ) -> None: if not self.chat_id: return flush = False with _LOCK: depth = _ACTIVE_INBOUND.get(self.chat_id, 0) - 1 if depth <= 0: _ACTIVE_INBOUND.pop(self.chat_id, None) flush = True else: _ACTIVE_INBOUND[self.chat_id] = depth if flush and defer_delegate_push_during_inbound_enabled(): self._flush_deferred() def _flush_deferred(self) -> None: items = flush_deferred_delegate_pushes(self.chat_id) if not items: return def _run() -> None: bridge = get_gateway_bridge_optional() if bridge is None or str(getattr(bridge, "chat_id", "") or "") != self.chat_id: for body, kind in items: _schedule_deferred_push_standalone(self.chat_id, body, kind=kind) return for body, kind in items: if kind == "delegate": ok, reason = should_deliver_delegate_push(self.chat_id, "", body=body) if not ok: logger.info( "deferred delegate push suppressed chat=%s reason=%s", self.chat_id[:12], reason, ) continue bridge.schedule_completion_push(body, kind=kind) safe_best_effort(_run, label="delegate_push_dedup.flush", default=None) def _schedule_deferred_push_standalone(chat_id: str, body: str, *, kind: str) -> None: """Best-effort when no live bridge (reuse runtime push path).""" def _run() -> None: ok, reason = should_deliver_delegate_push(chat_id, "", body=body) if not ok: logger.info( "deferred delegate push suppressed chat=%s reason=%s", chat_id[:12], reason, ) return if push_runtime_message("[Butler] 委派完成", body): mark_delegate_push_delivered(chat_id, "", body=body) safe_best_effort(_run, label="delegate_push_dedup.standalone_push", default=None) def maybe_defer_delegate_push(chat_id: str, body: str, *, kind: str) -> bool: """If inbound active, queue push for after turn. Returns True if deferred.""" if kind != "delegate" or not defer_delegate_push_during_inbound_enabled(): return False if not is_inbound_active(chat_id): return False defer_delegate_push(chat_id, body, kind=kind) logger.info("delegate push deferred during inbound chat=%s", str(chat_id or "")[:12]) return True __all__ = [ "defer_delegate_push_during_inbound_enabled", "delegate_push_dedup_enabled", "delegate_push_max_age_seconds", "extract_task_id_from_text", "flush_deferred_delegate_pushes", "gateway_inbound_guard", "is_inbound_active", "mark_delegate_push_delivered", "maybe_defer_delegate_push", "should_deliver_delegate_push", ]